Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 24 additions & 38 deletions command.go
Original file line number Diff line number Diff line change
Expand Up @@ -3414,7 +3414,7 @@ type CommandInfo struct {
LastKeyPos int8
StepCount int8
ReadOnly bool
Tips map[string]string
Tips *routing.CommandPolicy
}

type CommandsInfoCmd struct {
Expand Down Expand Up @@ -3547,8 +3547,7 @@ func (cmd *CommandsInfoCmd) readReply(rd *proto.Reader) error {
return err
}

cmdInfo.Tips = make(map[string]string, tipsLen)

rawTips := make(map[string]string, tipsLen)
for f := 0; f < tipsLen; f++ {
tip, err := rd.ReadString()
if err != nil {
Expand All @@ -3557,7 +3556,7 @@ func (cmd *CommandsInfoCmd) readReply(rd *proto.Reader) error {

// Handle tips that don't have a colon (like "nondeterministic_output")
if !strings.Contains(tip, ":") {
cmdInfo.Tips[tip] = ""
rawTips[tip] = ""
continue
}

Expand All @@ -3566,8 +3565,9 @@ func (cmd *CommandsInfoCmd) readReply(rd *proto.Reader) error {
if !ok {
return fmt.Errorf("redis: unexpected tip %q in COMMAND reply", tip)
}
cmdInfo.Tips[k] = v
rawTips[k] = v
}
cmdInfo.Tips = parseCommandPolicies(rawTips)

if err := rd.DiscardNext(); err != nil {
return err
Expand Down Expand Up @@ -3620,47 +3620,33 @@ func (c *cmdsInfoCache) Get(ctx context.Context) (map[string]*CommandInfo, error
}

// ------------------------------------------------------------------------------
var BuiltinPolicies = map[string]routing.CommandPolicy{
"ft.create": {Request: routing.ReqSpecial, Response: routing.RespAllSucceeded},
"ft.alter": {Request: routing.ReqSpecial, Response: routing.RespAllSucceeded},
"ft.drop": {Request: routing.ReqSpecial, Response: routing.RespAllSucceeded},

"mset": {Request: routing.ReqMultiShard, Response: routing.RespAllSucceeded},
"mget": {Request: routing.ReqMultiShard, Response: routing.RespSpecial},
"del": {Request: routing.ReqMultiShard, Response: routing.RespAggSum},
}

func newCommandPolicies(commandInfo map[string]*CommandInfo) map[string]routing.CommandPolicy {

table := make(map[string]routing.CommandPolicy, len(commandInfo))
const requestPolicy = "request_policy"
const responsePolicy = "response_policy"

for name, info := range commandInfo {
req := routing.ReqDefault
resp := routing.RespAllSucceeded
func parseCommandPolicies(commandInfoTips map[string]string) *routing.CommandPolicy {
req := routing.ReqDefault
resp := routing.RespAllSucceeded

if tips := info.Tips; tips != nil {
if v, ok := tips["request_policy"]; ok {
if p, err := routing.ParseRequestPolicy(v); err == nil {
req = p
}
if commandInfoTips != nil {
if v, ok := commandInfoTips[requestPolicy]; ok {
if p, err := routing.ParseRequestPolicy(v); err == nil {
req = p
}
if v, ok := tips["response_policy"]; ok {
if p, err := routing.ParseResponsePolicy(v); err == nil {
resp = p
}
}
if v, ok := commandInfoTips[responsePolicy]; ok {
if p, err := routing.ParseResponsePolicy(v); err == nil {
resp = p
}
} else {
return BuiltinPolicies
}
table[name] = routing.CommandPolicy{Request: req, Response: resp}
}

if len(table) == 0 {
for k, v := range BuiltinPolicies {
table[k] = v
tips := make(map[string]string, len(commandInfoTips))
for k, v := range commandInfoTips {
if k == requestPolicy || k == responsePolicy {
continue
}
tips[k] = v
}
return table
return &routing.CommandPolicy{Request: req, Response: resp, Tips: tips}
}

//------------------------------------------------------------------------------
Expand Down
9 changes: 5 additions & 4 deletions commands_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (

"github.com/redis/go-redis/v9"
"github.com/redis/go-redis/v9/internal/proto"
"github.com/redis/go-redis/v9/internal/routing"
)

type TimeValue struct {
Expand Down Expand Up @@ -664,13 +665,13 @@ var _ = Describe("Commands", func() {

cmd := cmds["touch"]
Expect(cmd.Name).To(Equal("touch"))
Expect(cmd.Tips["request_policy"]).To(Equal("multi_shard"))
Expect(cmd.Tips["response_policy"]).To(Equal("agg_sum"))
Expect(cmd.Tips.Request).To(Equal(routing.ReqMultiShard))
Expect(cmd.Tips.Response).To(Equal(routing.RespAggSum))

cmd = cmds["flushall"]
Expect(cmd.Name).To(Equal("flushall"))
Expect(cmd.Tips["request_policy"]).To(Equal("all_shards"))
Expect(cmd.Tips["response_policy"]).To(Equal("all_succeeded"))
Expect(cmd.Tips.Request).To(Equal(routing.ReqAllShards))
Expect(cmd.Tips.Response).To(Equal(routing.RespAllSucceeded))
})

It("should return all command names", func() {
Expand Down
3 changes: 3 additions & 0 deletions internal/routing/policy.go
Original file line number Diff line number Diff line change
Expand Up @@ -115,4 +115,7 @@ func ParseResponsePolicy(raw string) (ResponsePolicy, error) {
type CommandPolicy struct {
Request RequestPolicy
Response ResponsePolicy
// Tips that are not request_policy or response_policy
// e.g nondeterministic_output, nondeterministic_output_order.
Tips map[string]string
}
11 changes: 4 additions & 7 deletions osscluster.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ import (
"github.com/redis/go-redis/v9/internal/pool"
"github.com/redis/go-redis/v9/internal/proto"
"github.com/redis/go-redis/v9/internal/rand"
"github.com/redis/go-redis/v9/internal/routing"
)

const (
Expand Down Expand Up @@ -913,11 +912,10 @@ func (c *clusterStateHolder) ReloadOrGet(ctx context.Context) (*clusterState, er
// or more underlying connections. It's safe for concurrent use by
// multiple goroutines.
type ClusterClient struct {
opt *ClusterOptions
nodes *clusterNodes
state *clusterStateHolder
cmdsInfoCache *cmdsInfoCache
commandPolicies map[string]routing.CommandPolicy
opt *ClusterOptions
nodes *clusterNodes
state *clusterStateHolder
cmdsInfoCache *cmdsInfoCache
cmdable
hooksMixin
}
Expand All @@ -934,7 +932,6 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {

c.state = newClusterStateHolder(c.loadState)
c.cmdsInfoCache = newCmdsInfoCache(c.cmdsInfo)
c.commandPolicies = newCommandPolicies(c.cmdsInfoCache.cmds)
c.cmdable = c.Process
c.initHooks(hooks{
dial: nil,
Expand Down