redis/go-redis
1
0
Fork 0
mirror of https://github.com/redis/go-redis.git synced 2025-07-20 22:42:59 +03:00
Code Activity

Create raw output

Browse Source
This commit is contained in:
ofekshenawa
2023-10-09 11:52:59 +03:00
parent c6fe509f4a
commit e9b6d43c11
7 changed files with 196 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

334
command.go
View File

@ -25,6 +25,7 @@ type Cmder interface {
readTimeout() *time.Duration
readReply(rd *proto.Reader) error
readRawReply(rd *proto.Reader) error
SetErr(error)
Err() error
@ -111,10 +112,12 @@ func cmdString(cmd Cmder, val interface{}) string {
//------------------------------------------------------------------------------
type baseCmd struct {
ctx context.Context
args []interface{}
err error
keyPos int8
ctx context.Context
args []interface{}
err error
keyPos int8
rawReplay []byte
rawReplayErr error
_readTimeout *time.Duration
}
@ -186,6 +189,11 @@ func (cmd *baseCmd) setReadTimeout(d time.Duration) {
cmd._readTimeout = &d
}
func (cmd *baseCmd) readRawReply(rd *proto.Reader) error {
cmd.rawReplay, cmd.rawReplayErr = rd.ReadLine()
return nil
}
//------------------------------------------------------------------------------
type Cmd struct {
@ -207,8 +215,12 @@ func (cmd *Cmd) String() string {
return cmdString(cmd, cmd.val)
}
func (cmd *Cmd) SetVal(val interface{}) {
cmd.val = val
// func (cmd *Cmd) SetVal(val interface{}) {
// cmd.val = val
// }
func (cmd *Cmd) RawResult() ([]byte, error) {
return cmd.rawReplay, cmd.rawReplayErr
}
func (cmd *Cmd) Val() interface{} {
@ -490,9 +502,9 @@ func NewSliceCmd(ctx context.Context, args ...interface{}) *SliceCmd {
}
}
func (cmd *SliceCmd) SetVal(val []interface{}) {
cmd.val = val
}
// func (cmd *SliceCmd) SetVal(val []interface{}) {
// cmd.val = val
// }
func (cmd *SliceCmd) Val() []interface{} {
return cmd.val
@ -550,9 +562,9 @@ func NewStatusCmd(ctx context.Context, args ...interface{}) *StatusCmd {
}
}
func (cmd *StatusCmd) SetVal(val string) {
cmd.val = val
}
// func (cmd *StatusCmd) SetVal(val string) {
// cmd.val = val
// }
func (cmd *StatusCmd) Val() string {
return cmd.val
@ -590,9 +602,9 @@ func NewIntCmd(ctx context.Context, args ...interface{}) *IntCmd {
}
}
func (cmd *IntCmd) SetVal(val int64) {
cmd.val = val
}
// func (cmd *IntCmd) SetVal(val int64) {
// cmd.val = val
// }
func (cmd *IntCmd) Val() int64 {
return cmd.val
@ -634,9 +646,9 @@ func NewIntSliceCmd(ctx context.Context, args ...interface{}) *IntSliceCmd {
}
}
func (cmd *IntSliceCmd) SetVal(val []int64) {
cmd.val = val
}
// func (cmd *IntSliceCmd) SetVal(val []int64) {
// cmd.val = val
// }
func (cmd *IntSliceCmd) Val() []int64 {
return cmd.val
@ -685,9 +697,9 @@ func NewDurationCmd(ctx context.Context, precision time.Duration, args ...interf
}
}
func (cmd *DurationCmd) SetVal(val time.Duration) {
cmd.val = val
}
// func (cmd *DurationCmd) SetVal(val time.Duration) {
// cmd.val = val
// }
func (cmd *DurationCmd) Val() time.Duration {
return cmd.val
@ -736,9 +748,9 @@ func NewTimeCmd(ctx context.Context, args ...interface{}) *TimeCmd {
}
}
func (cmd *TimeCmd) SetVal(val time.Time) {
cmd.val = val
}
// func (cmd *TimeCmd) SetVal(val time.Time) {
// cmd.val = val
// }
func (cmd *TimeCmd) Val() time.Time {
return cmd.val
@ -787,9 +799,9 @@ func NewBoolCmd(ctx context.Context, args ...interface{}) *BoolCmd {
}
}
func (cmd *BoolCmd) SetVal(val bool) {
cmd.val = val
}
// func (cmd *BoolCmd) SetVal(val bool) {
// cmd.val = val
// }
func (cmd *BoolCmd) Val() bool {
return cmd.val
@ -834,9 +846,9 @@ func NewStringCmd(ctx context.Context, args ...interface{}) *StringCmd {
}
}
func (cmd *StringCmd) SetVal(val string) {
cmd.val = val
}
// func (cmd *StringCmd) SetVal(val string) {
// cmd.val = val
// }
func (cmd *StringCmd) Val() string {
return cmd.val
@ -938,9 +950,9 @@ func NewFloatCmd(ctx context.Context, args ...interface{}) *FloatCmd {
}
}
func (cmd *FloatCmd) SetVal(val float64) {
cmd.val = val
}
// func (cmd *FloatCmd) SetVal(val float64) {
// cmd.val = val
// }
func (cmd *FloatCmd) Val() float64 {
return cmd.val
@ -978,9 +990,9 @@ func NewFloatSliceCmd(ctx context.Context, args ...interface{}) *FloatSliceCmd {
}
}
func (cmd *FloatSliceCmd) SetVal(val []float64) {
cmd.val = val
}
// func (cmd *FloatSliceCmd) SetVal(val []float64) {
// cmd.val = val
// }
func (cmd *FloatSliceCmd) Val() []float64 {
return cmd.val
@ -1033,9 +1045,9 @@ func NewStringSliceCmd(ctx context.Context, args ...interface{}) *StringSliceCmd
}
}
func (cmd *StringSliceCmd) SetVal(val []string) {
cmd.val = val
}
// func (cmd *StringSliceCmd) SetVal(val []string) {
// cmd.val = val
// }
func (cmd *StringSliceCmd) Val() []string {
return cmd.val
@ -1096,9 +1108,9 @@ func NewKeyValueSliceCmd(ctx context.Context, args ...interface{}) *KeyValueSlic
}
}
func (cmd *KeyValueSliceCmd) SetVal(val []KeyValue) {
cmd.val = val
}
// func (cmd *KeyValueSliceCmd) SetVal(val []KeyValue) {
// cmd.val = val
// }
func (cmd *KeyValueSliceCmd) Val() []KeyValue {
return cmd.val
@ -1185,9 +1197,9 @@ func NewBoolSliceCmd(ctx context.Context, args ...interface{}) *BoolSliceCmd {
}
}
func (cmd *BoolSliceCmd) SetVal(val []bool) {
cmd.val = val
}
// func (cmd *BoolSliceCmd) SetVal(val []bool) {
// cmd.val = val
// }
func (cmd *BoolSliceCmd) Val() []bool {
return cmd.val
@ -1238,9 +1250,9 @@ func (cmd *MapStringStringCmd) Val() map[string]string {
return cmd.val
}
func (cmd *MapStringStringCmd) SetVal(val map[string]string) {
cmd.val = val
}
// func (cmd *MapStringStringCmd) SetVal(val map[string]string) {
// cmd.val = val
// }
func (cmd *MapStringStringCmd) Result() (map[string]string, error) {
return cmd.val, cmd.err
@ -1313,9 +1325,9 @@ func NewMapStringIntCmd(ctx context.Context, args ...interface{}) *MapStringIntC
}
}
func (cmd *MapStringIntCmd) SetVal(val map[string]int64) {
cmd.val = val
}
// func (cmd *MapStringIntCmd) SetVal(val map[string]int64) {
// cmd.val = val
// }
func (cmd *MapStringIntCmd) Val() map[string]int64 {
return cmd.val
@ -1370,9 +1382,9 @@ func (cmd *MapStringSliceInterfaceCmd) String() string {
return cmdString(cmd, cmd.val)
}
func (cmd *MapStringSliceInterfaceCmd) SetVal(val map[string][]interface{}) {
cmd.val = val
}
// func (cmd *MapStringSliceInterfaceCmd) SetVal(val map[string][]interface{}) {
// cmd.val = val
// }
func (cmd *MapStringSliceInterfaceCmd) Result() (map[string][]interface{}, error) {
return cmd.val, cmd.err
@ -1429,9 +1441,9 @@ func NewStringStructMapCmd(ctx context.Context, args ...interface{}) *StringStru
}
}
func (cmd *StringStructMapCmd) SetVal(val map[string]struct{}) {
cmd.val = val
}
// func (cmd *StringStructMapCmd) SetVal(val map[string]struct{}) {
// cmd.val = val
// }
func (cmd *StringStructMapCmd) Val() map[string]struct{} {
return cmd.val
@ -1486,9 +1498,9 @@ func NewXMessageSliceCmd(ctx context.Context, args ...interface{}) *XMessageSlic
}
}
func (cmd *XMessageSliceCmd) SetVal(val []XMessage) {
cmd.val = val
}
// func (cmd *XMessageSliceCmd) SetVal(val []XMessage) {
// cmd.val = val
// }
func (cmd *XMessageSliceCmd) Val() []XMessage {
return cmd.val
@ -1592,9 +1604,9 @@ func NewXStreamSliceCmd(ctx context.Context, args ...interface{}) *XStreamSliceC
}
}
func (cmd *XStreamSliceCmd) SetVal(val []XStream) {
cmd.val = val
}
// func (cmd *XStreamSliceCmd) SetVal(val []XStream) {
// cmd.val = val
// }
func (cmd *XStreamSliceCmd) Val() []XStream {
return cmd.val
@ -1665,9 +1677,9 @@ func NewXPendingCmd(ctx context.Context, args ...interface{}) *XPendingCmd {
}
}
func (cmd *XPendingCmd) SetVal(val *XPending) {
cmd.val = val
}
// func (cmd *XPendingCmd) SetVal(val *XPending) {
// cmd.val = val
// }
func (cmd *XPendingCmd) Val() *XPending {
return cmd.val
@ -1748,9 +1760,9 @@ func NewXPendingExtCmd(ctx context.Context, args ...interface{}) *XPendingExtCmd
}
}
func (cmd *XPendingExtCmd) SetVal(val []XPendingExt) {
cmd.val = val
}
// func (cmd *XPendingExtCmd) SetVal(val []XPendingExt) {
// cmd.val = val
// }
func (cmd *XPendingExtCmd) Val() []XPendingExt {
return cmd.val
@ -1818,10 +1830,10 @@ func NewXAutoClaimCmd(ctx context.Context, args ...interface{}) *XAutoClaimCmd {
}
}
func (cmd *XAutoClaimCmd) SetVal(val []XMessage, start string) {
cmd.val = val
cmd.start = start
}
// func (cmd *XAutoClaimCmd) SetVal(val []XMessage, start string) {
// cmd.val = val
// cmd.start = start
// }
func (cmd *XAutoClaimCmd) Val() (messages []XMessage, start string) {
return cmd.val, cmd.start
@ -1888,10 +1900,10 @@ func NewXAutoClaimJustIDCmd(ctx context.Context, args ...interface{}) *XAutoClai
}
}
func (cmd *XAutoClaimJustIDCmd) SetVal(val []string, start string) {
cmd.val = val
cmd.start = start
}
// func (cmd *XAutoClaimJustIDCmd) SetVal(val []string, start string) {
// cmd.val = val
// cmd.start = start
// }
func (cmd *XAutoClaimJustIDCmd) Val() (ids []string, start string) {
return cmd.val, cmd.start
@ -1971,9 +1983,9 @@ func NewXInfoConsumersCmd(ctx context.Context, stream string, group string) *XIn
}
}
func (cmd *XInfoConsumersCmd) SetVal(val []XInfoConsumer) {
cmd.val = val
}
// func (cmd *XInfoConsumersCmd) SetVal(val []XInfoConsumer) {
// cmd.val = val
// }
func (cmd *XInfoConsumersCmd) Val() []XInfoConsumer {
return cmd.val
@ -2059,9 +2071,9 @@ func NewXInfoGroupsCmd(ctx context.Context, stream string) *XInfoGroupsCmd {
}
}
func (cmd *XInfoGroupsCmd) SetVal(val []XInfoGroup) {
cmd.val = val
}
// func (cmd *XInfoGroupsCmd) SetVal(val []XInfoGroup) {
// cmd.val = val
// }
func (cmd *XInfoGroupsCmd) Val() []XInfoGroup {
return cmd.val
@ -2171,9 +2183,9 @@ func NewXInfoStreamCmd(ctx context.Context, stream string) *XInfoStreamCmd {
}
}
func (cmd *XInfoStreamCmd) SetVal(val *XInfoStream) {
cmd.val = val
}
// func (cmd *XInfoStreamCmd) SetVal(val *XInfoStream) {
// cmd.val = val
// }
func (cmd *XInfoStreamCmd) Val() *XInfoStream {
return cmd.val
@ -2318,9 +2330,9 @@ func NewXInfoStreamFullCmd(ctx context.Context, args ...interface{}) *XInfoStrea
}
}
func (cmd *XInfoStreamFullCmd) SetVal(val *XInfoStreamFull) {
cmd.val = val
}
// func (cmd *XInfoStreamFullCmd) SetVal(val *XInfoStreamFull) {
// cmd.val = val
// }
func (cmd *XInfoStreamFullCmd) Val() *XInfoStreamFull {
return cmd.val
@ -2617,9 +2629,9 @@ func NewZSliceCmd(ctx context.Context, args ...interface{}) *ZSliceCmd {
}
}
func (cmd *ZSliceCmd) SetVal(val []Z) {
cmd.val = val
}
// func (cmd *ZSliceCmd) SetVal(val []Z) {
// cmd.val = val
// }
func (cmd *ZSliceCmd) Val() []Z {
return cmd.val
@ -2695,9 +2707,9 @@ func NewZWithKeyCmd(ctx context.Context, args ...interface{}) *ZWithKeyCmd {
}
}
func (cmd *ZWithKeyCmd) SetVal(val *ZWithKey) {
cmd.val = val
}
// func (cmd *ZWithKeyCmd) SetVal(val *ZWithKey) {
// cmd.val = val
// }
func (cmd *ZWithKeyCmd) Val() *ZWithKey {
return cmd.val
@ -2753,10 +2765,10 @@ func NewScanCmd(ctx context.Context, process cmdable, args ...interface{}) *Scan
}
}
func (cmd *ScanCmd) SetVal(page []string, cursor uint64) {
cmd.page = page
cmd.cursor = cursor
}
// func (cmd *ScanCmd) SetVal(page []string, cursor uint64) {
// cmd.page = page
// cmd.cursor = cursor
// }
func (cmd *ScanCmd) Val() (keys []string, cursor uint64) {
return cmd.page, cmd.cursor
@ -2833,9 +2845,9 @@ func NewClusterSlotsCmd(ctx context.Context, args ...interface{}) *ClusterSlotsC
}
}
func (cmd *ClusterSlotsCmd) SetVal(val []ClusterSlot) {
cmd.val = val
}
// func (cmd *ClusterSlotsCmd) SetVal(val []ClusterSlot) {
// cmd.val = val
// }
func (cmd *ClusterSlotsCmd) Val() []ClusterSlot {
return cmd.val
@ -3023,9 +3035,9 @@ func geoLocationArgs(q *GeoRadiusQuery, args ...interface{}) []interface{} {
return args
}
func (cmd *GeoLocationCmd) SetVal(locations []GeoLocation) {
cmd.locations = locations
}
// func (cmd *GeoLocationCmd) SetVal(locations []GeoLocation) {
// cmd.locations = locations
// }
func (cmd *GeoLocationCmd) Val() []GeoLocation {
return cmd.locations
@ -3203,9 +3215,9 @@ func NewGeoSearchLocationCmd(
}
}
func (cmd *GeoSearchLocationCmd) SetVal(val []GeoLocation) {
cmd.val = val
}
// func (cmd *GeoSearchLocationCmd) SetVal(val []GeoLocation) {
// cmd.val = val
// }
func (cmd *GeoSearchLocationCmd) Val() []GeoLocation {
return cmd.val
@ -3293,9 +3305,9 @@ func NewGeoPosCmd(ctx context.Context, args ...interface{}) *GeoPosCmd {
}
}
func (cmd *GeoPosCmd) SetVal(val []*GeoPos) {
cmd.val = val
}
// func (cmd *GeoPosCmd) SetVal(val []*GeoPos) {
// cmd.val = val
// }
func (cmd *GeoPosCmd) Val() []*GeoPos {
return cmd.val
@ -3374,9 +3386,9 @@ func NewCommandsInfoCmd(ctx context.Context, args ...interface{}) *CommandsInfoC
}
}
func (cmd *CommandsInfoCmd) SetVal(val map[string]*CommandInfo) {
cmd.val = val
}
// func (cmd *CommandsInfoCmd) SetVal(val map[string]*CommandInfo) {
// cmd.val = val
// }
func (cmd *CommandsInfoCmd) Val() map[string]*CommandInfo {
return cmd.val
@ -3564,9 +3576,9 @@ func NewSlowLogCmd(ctx context.Context, args ...interface{}) *SlowLogCmd {
}
}
func (cmd *SlowLogCmd) SetVal(val []SlowLog) {
cmd.val = val
}
// func (cmd *SlowLogCmd) SetVal(val []SlowLog) {
// cmd.val = val
// }
func (cmd *SlowLogCmd) Val() []SlowLog {
return cmd.val
@ -3663,9 +3675,9 @@ func NewMapStringInterfaceCmd(ctx context.Context, args ...interface{}) *MapStri
}
}
func (cmd *MapStringInterfaceCmd) SetVal(val map[string]interface{}) {
cmd.val = val
}
// func (cmd *MapStringInterfaceCmd) SetVal(val map[string]interface{}) {
// cmd.val = val
// }
func (cmd *MapStringInterfaceCmd) Val() map[string]interface{} {
return cmd.val
@ -3727,9 +3739,9 @@ func NewMapStringStringSliceCmd(ctx context.Context, args ...interface{}) *MapSt
}
}
func (cmd *MapStringStringSliceCmd) SetVal(val []map[string]string) {
cmd.val = val
}
// func (cmd *MapStringStringSliceCmd) SetVal(val []map[string]string) {
// cmd.val = val
// }
func (cmd *MapStringStringSliceCmd) Val() []map[string]string {
return cmd.val
@ -3791,9 +3803,9 @@ func NewMapStringInterfaceSliceCmd(ctx context.Context, args ...interface{}) *Ma
}
}
func (cmd *MapStringInterfaceSliceCmd) SetVal(val []map[string]interface{}) {
cmd.val = val
}
// func (cmd *MapStringInterfaceSliceCmd) SetVal(val []map[string]interface{}) {
// cmd.val = val
// }
func (cmd *MapStringInterfaceSliceCmd) Val() []map[string]interface{} {
return cmd.val
@ -3857,10 +3869,10 @@ func NewKeyValuesCmd(ctx context.Context, args ...interface{}) *KeyValuesCmd {
}
}
func (cmd *KeyValuesCmd) SetVal(key string, val []string) {
cmd.key = key
cmd.val = val
}
// func (cmd *KeyValuesCmd) SetVal(key string, val []string) {
// cmd.key = key
// cmd.val = val
// }
func (cmd *KeyValuesCmd) Val() (string, []string) {
return cmd.key, cmd.val
@ -3919,10 +3931,10 @@ func NewZSliceWithKeyCmd(ctx context.Context, args ...interface{}) *ZSliceWithKe
}
}
func (cmd *ZSliceWithKeyCmd) SetVal(key string, val []Z) {
cmd.key = key
cmd.val = val
}
// func (cmd *ZSliceWithKeyCmd) SetVal(key string, val []Z) {
// cmd.key = key
// cmd.val = val
// }
func (cmd *ZSliceWithKeyCmd) Val() (string, []Z) {
return cmd.key, cmd.val
@ -4012,9 +4024,9 @@ func NewFunctionListCmd(ctx context.Context, args ...interface{}) *FunctionListC
}
}
func (cmd *FunctionListCmd) SetVal(val []Library) {
cmd.val = val
}
// func (cmd *FunctionListCmd) SetVal(val []Library) {
// cmd.val = val
// }
func (cmd *FunctionListCmd) String() string {
return cmdString(cmd, cmd.val)
@ -4193,9 +4205,9 @@ func NewFunctionStatsCmd(ctx context.Context, args ...interface{}) *FunctionStat
}
}
func (cmd *FunctionStatsCmd) SetVal(val FunctionStats) {
cmd.val = val
}
// func (cmd *FunctionStatsCmd) SetVal(val FunctionStats) {
// cmd.val = val
// }
func (cmd *FunctionStatsCmd) String() string {
return cmdString(cmd, cmd.val)
@ -4429,9 +4441,9 @@ func NewLCSCmd(ctx context.Context, q *LCSQuery) *LCSCmd {
return cmd
}
func (cmd *LCSCmd) SetVal(val *LCSMatch) {
cmd.val = val
}
// func (cmd *LCSCmd) SetVal(val *LCSMatch) {
// cmd.val = val
// }
func (cmd *LCSCmd) String() string {
return cmdString(cmd, cmd.val)
@ -4559,9 +4571,9 @@ func NewKeyFlagsCmd(ctx context.Context, args ...interface{}) *KeyFlagsCmd {
}
}
func (cmd *KeyFlagsCmd) SetVal(val []KeyFlags) {
cmd.val = val
}
// func (cmd *KeyFlagsCmd) SetVal(val []KeyFlags) {
// cmd.val = val
// }
func (cmd *KeyFlagsCmd) Val() []KeyFlags {
return cmd.val
@ -4641,9 +4653,9 @@ func NewClusterLinksCmd(ctx context.Context, args ...interface{}) *ClusterLinksC
}
}
func (cmd *ClusterLinksCmd) SetVal(val []ClusterLink) {
cmd.val = val
}
// func (cmd *ClusterLinksCmd) SetVal(val []ClusterLink) {
// cmd.val = val
// }
func (cmd *ClusterLinksCmd) Val() []ClusterLink {
return cmd.val
@ -4743,9 +4755,9 @@ func NewClusterShardsCmd(ctx context.Context, args ...interface{}) *ClusterShard
}
}
func (cmd *ClusterShardsCmd) SetVal(val []ClusterShard) {
cmd.val = val
}
// func (cmd *ClusterShardsCmd) SetVal(val []ClusterShard) {
// cmd.val = val
// }
func (cmd *ClusterShardsCmd) Val() []ClusterShard {
return cmd.val
@ -4876,9 +4888,9 @@ func NewRankWithScoreCmd(ctx context.Context, args ...interface{}) *RankWithScor
}
}
func (cmd *RankWithScoreCmd) SetVal(val RankScore) {
cmd.val = val
}
// func (cmd *RankWithScoreCmd) SetVal(val RankScore) {
// cmd.val = val
// }
func (cmd *RankWithScoreCmd) Val() RankScore {
return cmd.val
@ -5022,9 +5034,9 @@ func NewClientInfoCmd(ctx context.Context, args ...interface{}) *ClientInfoCmd {
}
}
func (cmd *ClientInfoCmd) SetVal(val *ClientInfo) {
cmd.val = val
}
// func (cmd *ClientInfoCmd) SetVal(val *ClientInfo) {
// cmd.val = val
// }
func (cmd *ClientInfoCmd) String() string {
return cmdString(cmd, cmd.val)
@ -5216,9 +5228,9 @@ func NewACLLogCmd(ctx context.Context, args ...interface{}) *ACLLogCmd {
}
}
func (cmd *ACLLogCmd) SetVal(val []*ACLLogEntry) {
cmd.val = val
}
// func (cmd *ACLLogCmd) SetVal(val []*ACLLogEntry) {
// cmd.val = val
// }
func (cmd *ACLLogCmd) Val() []*ACLLogEntry {
return cmd.val

7
internal/pool/conn.go
View File

@ -64,14 +64,17 @@ func (cn *Conn) RemoteAddr() net.Addr {
}
func (cn *Conn) WithReader(
ctx context.Context, timeout time.Duration, fn func(rd *proto.Reader) error,
ctx context.Context, timeout time.Duration, fn []func(rd *proto.Reader) error,
) error {
if timeout >= 0 {
if err := cn.netConn.SetReadDeadline(cn.deadline(ctx, timeout)); err != nil {
return err
}
}
return fn(cn.rd)
for _, f := range fn {
f(cn.rd)
}
return nil
}
func (cn *Conn) WithWriter(

8
osscluster.go
View File

@ -1314,9 +1314,9 @@ func (c *ClusterClient) processPipelineNodeConn(
return err
}
return cn.WithReader(c.context(ctx), c.opt.ReadTimeout, func(rd *proto.Reader) error {
return cn.WithReader(c.context(ctx), c.opt.ReadTimeout, []func(rd *proto.Reader) error{func(rd *proto.Reader) error {
return c.pipelineReadCmds(ctx, node, rd, cmds, failedCmds)
})
}})
}
func (c *ClusterClient) pipelineReadCmds(
@ -1495,7 +1495,7 @@ func (c *ClusterClient) processTxPipelineNodeConn(
return err
}
return cn.WithReader(c.context(ctx), c.opt.ReadTimeout, func(rd *proto.Reader) error {
return cn.WithReader(c.context(ctx), c.opt.ReadTimeout, []func(rd *proto.Reader) error{func(rd *proto.Reader) error {
statusCmd := cmds[0].(*StatusCmd)
// Trim multi and exec.
trimmedCmds := cmds[1 : len(cmds)-1]
@ -1514,7 +1514,7 @@ func (c *ClusterClient) processTxPipelineNodeConn(
}
return pipelineReadCmds(rd, trimmedCmds)
})
}})
}
func (c *ClusterClient) txPipelineReadQueued(

4
pubsub.go
View File

@ -432,9 +432,9 @@ func (c *PubSub) ReceiveTimeout(ctx context.Context, timeout time.Duration) (int
return nil, err
}
err = cn.WithReader(context.Background(), timeout, func(rd *proto.Reader) error {
err = cn.WithReader(context.Background(), timeout, []func(rd *proto.Reader) error{func(rd *proto.Reader) error {
return c.cmd.readReply(rd)
})
}})
c.releaseConnWithLock(ctx, cn, err, timeout > 0)

10
redis.go
View File

@ -405,7 +405,7 @@ func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int) (bool
return err
}
if err := cn.WithReader(c.context(ctx), c.cmdTimeout(cmd), cmd.readReply); err != nil {
if err := cn.WithReader(c.context(ctx), c.cmdTimeout(cmd), []func(rd *proto.Reader) error{cmd.readReply, cmd.readRawReply}); err != nil {
if cmd.readTimeout() == nil {
atomic.StoreUint32(&retryTimeout, 1)
} else {
@ -511,9 +511,9 @@ func (c *baseClient) pipelineProcessCmds(
return true, err
}
if err := cn.WithReader(c.context(ctx), c.opt.ReadTimeout, func(rd *proto.Reader) error {
if err := cn.WithReader(c.context(ctx), c.opt.ReadTimeout, []func(rd *proto.Reader) error{func(rd *proto.Reader) error {
return pipelineReadCmds(rd, cmds)
}); err != nil {
}}); err != nil {
return true, err
}
@ -543,7 +543,7 @@ func (c *baseClient) txPipelineProcessCmds(
return true, err
}
if err := cn.WithReader(c.context(ctx), c.opt.ReadTimeout, func(rd *proto.Reader) error {
if err := cn.WithReader(c.context(ctx), c.opt.ReadTimeout, []func(rd *proto.Reader) error{func(rd *proto.Reader) error {
statusCmd := cmds[0].(*StatusCmd)
// Trim multi and exec.
trimmedCmds := cmds[1 : len(cmds)-1]
@ -554,7 +554,7 @@ func (c *baseClient) txPipelineProcessCmds(
}
return pipelineReadCmds(rd, trimmedCmds)
}); err != nil {
}}); err != nil {
return false, err
}

4
timeseries_commands.go
View File

@ -480,6 +480,10 @@ func (cmd *TSTimestampValueCmd) Val() TSTimestampValue {
return cmd.val
}
func (cmd *TSTimestampValueCmd) RawResult() ([]byte, error) {
return cmd.rawReplay, cmd.rawReplayErr
}
func (cmd *TSTimestampValueCmd) readReply(rd *proto.Reader) (err error) {
n, err := rd.ReadMapLen()
if err != nil {

4
timeseries_commands_test.go
View File

@ -284,10 +284,12 @@ var _ = Describe("RedisTimeseries commands", Label("timeseries"), func() {
resultGet, err := client.TSAddWithArgs(ctx, "foo", 2265985, 151, opt).Result()
Expect(err).NotTo(HaveOccurred())
Expect(resultGet).To(BeEquivalentTo(2265985))
result, err := client.TSGet(ctx, "foo").Result()
res := client.TSGet(ctx, "foo")
result, err := res.Result()
Expect(err).NotTo(HaveOccurred())
Expect(result.Timestamp).To(BeEquivalentTo(2265985))
Expect(result.Value).To(BeEquivalentTo(151))
// Expect(res.RawResult()).To(BeEquivalentTo(0))
})
It("should TSGet Latest", Label("timeseries", "tsgetlatest"), func() {