added scaffolding for the req-resp manager

Author: htemelski-redis

command_policy_manager.go

@@ -0,0 +1,45 @@
+package redis
+
+import (
+	"sync"
+
+	"github.com/redis/go-redis/v9/internal/routing"
+)
+
+var defaultPolicies = map[string]*routing.CommandPolicy{}
+
+type commandPolicyManager struct {
+	rwmutex             *sync.RWMutex
+	clientPolicies      map[string]*routing.CommandPolicy
+	overwrittenPolicies map[string]*routing.CommandPolicy
+}
+
+func newCommandPolicyManager(overwrites interface{}) *commandPolicyManager {
+	return &commandPolicyManager{}
+}
+
+func (cpm *commandPolicyManager) updateClientPolicies(policies interface{}) {
+	cpm.rwmutex.Lock()
+	defer cpm.rwmutex.Unlock()
+}
+
+func (cpm *commandPolicyManager) getCmdPolicy(cmd Cmder) *routing.CommandPolicy {
+	cpm.rwmutex.RLock()
+	defer cpm.rwmutex.RUnlock()
+
+	cmdName := cmd.Name()
+
+	if policy, ok := cpm.overwrittenPolicies[cmdName]; ok {
+		return policy
+	}
+
+	if policy, ok := cpm.clientPolicies[cmdName]; ok {
+		return policy
+	}
+
+	if policy, ok := defaultPolicies[cmdName]; ok {
+		return policy
+	}
+
+	return nil
+}

osscluster.go

@@ -922,10 +922,11 @@ 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
+	opt              *ClusterOptions
+	nodes            *clusterNodes
+	state            *clusterStateHolder
+	cmdsInfoCache    *cmdsInfoCache
+	cmdPolicyManager *commandPolicyManager
 	cmdable
 	hooksMixin
 }
@@ -943,6 +944,7 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {
 	c.state = newClusterStateHolder(c.loadState)
 	c.cmdsInfoCache = newCmdsInfoCache(c.cmdsInfo)
 	c.cmdable = c.Process
+	c.cmdPolicyManager = newCommandPolicyManager(nil)
 	c.initHooks(hooks{
 		dial:       nil,
 		process:    c.process,
@@ -1334,7 +1336,7 @@ func (c *ClusterClient) mapCmdsByNode(ctx context.Context, cmdsMap *cmdsMap, cmd
 
 	if c.opt.ReadOnly && c.cmdsAreReadOnly(ctx, cmds) {
 		for _, cmd := range cmds {
-			policy := c.getCommandPolicy(ctx, cmd)
+			policy := c.cmdPolicyManager.getCmdPolicy(cmd)
 			if policy != nil && !policy.CanBeUsedInPipeline() {
 				return fmt.Errorf(
 					"redis: cannot pipeline command %q with request policy ReqAllNodes/ReqAllShards/ReqMultiShard; Note: This behavior is subject to change in the future", cmd.Name(),
@@ -1351,7 +1353,7 @@ func (c *ClusterClient) mapCmdsByNode(ctx context.Context, cmdsMap *cmdsMap, cmd
 	}
 
 	for _, cmd := range cmds {
-		policy := c.getCommandPolicy(ctx, cmd)
+		policy := c.cmdPolicyManager.getCmdPolicy(cmd)
 		if policy != nil && !policy.CanBeUsedInPipeline() {
 			return fmt.Errorf(
 				"redis: cannot pipeline command %q with request policy ReqAllNodes/ReqAllShards/ReqMultiShard; Note: This behavior is subject to change in the future", cmd.Name(),

osscluster_router.go

@@ -21,7 +21,7 @@ type slotResult struct {
 
 // routeAndRun routes a command to the appropriate cluster nodes and executes it
 func (c *ClusterClient) routeAndRun(ctx context.Context, cmd Cmder, node *clusterNode) error {
-	policy := c.getCommandPolicy(ctx, cmd)
+	policy := c.cmdPolicyManager.getCmdPolicy(cmd)
 
 	switch {
 	case policy != nil && policy.Request == routing.ReqAllNodes:
@@ -38,11 +38,13 @@ func (c *ClusterClient) routeAndRun(ctx context.Context, cmd Cmder, node *cluste
 }
 
 // getCommandPolicy retrieves the routing policy for a command
-func (c *ClusterClient) getCommandPolicy(ctx context.Context, cmd Cmder) *routing.CommandPolicy {
-	if cmdInfo := c.cmdInfo(ctx, cmd.Name()); cmdInfo != nil && cmdInfo.Tips != nil {
-		return cmdInfo.Tips
-	}
-	return nil
+func (c *ClusterClient) getCommandPolicy(cmd Cmder) *routing.CommandPolicy {
+
+	return c.cmdPolicyManager.getCmdPolicy(cmd)
+	// if cmdInfo := c.cmdInfo(ctx, cmd.Name()); cmdInfo != nil && cmdInfo.Tips != nil {
+	// 	return cmdInfo.Tips
+	// }
+	// return nil
 }
 
 // executeDefault handles standard command routing based on keys