fix: send readonly command

cluster.go

@@ -18,7 +18,6 @@ import (
 
 // ErrNoSlot indicates that there is no redis node owns the key slot.
 var ErrNoSlot = errors.New("the slot has no redis node")
-var ErrReplicaOnlyConflict = errors.New("ReplicaOnly conflicts with SendToReplicas option")
 
 type retry struct {
 	cIndexes []int
@@ -93,17 +92,13 @@ type connrole struct {
 func newClusterClient(opt *ClientOption, connFn connFn) (client *clusterClient, err error) {
 	client = &clusterClient{
 		cmd:    cmds.NewBuilder(cmds.InitSlot),
-		opt:    opt,
 		connFn: connFn,
+		opt:    opt,
 		conns:  make(map[string]connrole),
 		retry:  !opt.DisableRetry,
 		aws:    len(opt.InitAddress) == 1 && strings.Contains(opt.InitAddress[0], "amazonaws.com"),
 	}
 
-	if opt.ReplicaOnly && opt.SendToReplicas != nil {
-		return nil, ErrReplicaOnlyConflict
-	}
-
 	client.connFn = func(dst string, opt *ClientOption) conn {
 		cc := connFn(dst, opt)
 		cc.SetOnCloseHook(func(err error) {
@@ -222,10 +217,42 @@ func (c *clusterClient) _refresh() (err error) {
 
 	groups := result.parse(c.opt.TLSConfig != nil)
 	conns := make(map[string]connrole, len(groups))
+	var wg sync.WaitGroup
 	for master, g := range groups {
-		conns[master] = connrole{conn: c.connFn(master, c.opt), replica: false}
+		conns[master] = connrole{
+			conn:    c.connFn(master, c.opt),
+			replica: false,
+		}
 		for _, addr := range g.nodes[1:] {
-			conns[addr] = connrole{conn: c.connFn(addr, c.opt), replica: true}
+			cc := c.connFn(addr, c.opt)
+
+			if !c.opt.ReplicaOnly {
+				conns[addr] = connrole{
+					conn:    cc,
+					replica: true,
+				}
+				continue
+			}
+
+			wg.Add(1)
+			go func(cc conn) {
+				defer wg.Done()
+
+				timeout := c.opt.Dialer.Timeout
+				if timeout <= 0 {
+					timeout = DefaultDialTimeout
+				}
+
+				ctx, cancel := context.WithTimeout(context.Background(), timeout)
+				defer cancel()
+
+				cc.Do(ctx, cmds.NewCompleted([]string{"READONLY"})) // ignore error
+			}(cc)
+
+			conns[addr] = connrole{
+				conn:    cc,
+				replica: true,
+			}
 		}
 	}
 	// make sure InitAddress always be present
@@ -236,12 +263,14 @@ func (c *clusterClient) _refresh() (err error) {
 			}
 		}
 	}
+	wg.Wait()
 
 	var removes []conn
 
 	c.mu.RLock()
 	for addr, cc := range c.conns {
-		if fresh, ok := conns[addr]; ok {
+		fresh, ok := conns[addr]
+		if ok && cc.replica == fresh.replica {
 			conns[addr] = connrole{
 				conn:    cc.conn,
 				replica: fresh.replica,
@@ -256,7 +285,7 @@ func (c *clusterClient) _refresh() (err error) {
 	var rslots []conn
 	for master, g := range groups {
 		switch {
-		case c.opt.ReplicaOnly && len(g.nodes) > 1:
+		case c.opt.SendToReplicas == nil && c.opt.ReplicaOnly && len(g.nodes) > 1:
 			nodesCount := len(g.nodes)
 			for _, slot := range g.slots {
 				for i := slot[0]; i <= slot[1]; i++ {

cluster_test.go

@@ -911,6 +911,74 @@ func TestClusterClientInit(t *testing.T) {
 			t.Fatalf("unexpected node assigned to rslot 16383")
 		}
 	})
+
+	t.Run("Refresh cluster which has multi nodes per shard with SendToReplica, ReplicaOnly options", func(t *testing.T) {
+		primaryNodeConn := &mockConn{
+			DoFn: func(cmd Completed) RedisResult {
+				if strings.Join(cmd.Commands(), " ") == "CLUSTER SLOTS" {
+					return slotsMultiResp
+				}
+				return RedisResult{
+					err: errors.New("unexpected call"),
+				}
+			},
+		}
+		replicaNodeConn := &mockConn{
+			DoFn: func(cmd Completed) RedisResult {
+				if strings.Join(cmd.Commands(), " ") == "READONLY" {
+					return newResult(RedisMessage{typ: '+', string: "READONLY"}, nil)
+				}
+				return RedisResult{
+					err: errors.New("unexpected call"),
+				}
+			},
+		}
+
+		client, err := newClusterClient(
+			&ClientOption{
+				InitAddress: []string{"127.0.0.1:0"},
+				SendToReplicas: func(cmd Completed) bool {
+					return true
+				},
+				ReplicaOnly: true,
+			},
+			func(dst string, opt *ClientOption) conn {
+				if dst == "127.0.0.1:0" || dst == "127.0.2.1:0" {
+					return primaryNodeConn
+				} else {
+					return replicaNodeConn
+				}
+			},
+		)
+		if err != nil {
+			t.Fatalf("unexpected err %v", err)
+		}
+
+		if client.pslots[0] != client.conns["127.0.0.1:0"].conn {
+			t.Fatalf("unexpected node assigned to pslot 0")
+		}
+		if client.pslots[8192] != client.conns["127.0.0.1:0"].conn {
+			t.Fatalf("unexpected node assigned to pslot 8192")
+		}
+		if client.pslots[8193] != client.conns["127.0.2.1:0"].conn {
+			t.Fatalf("unexpected node assigned to pslot 8193")
+		}
+		if client.pslots[16383] != client.conns["127.0.2.1:0"].conn {
+			t.Fatalf("unexpected node assigned to pslot 16383")
+		}
+		if client.rslots[0] != client.conns["127.0.1.1:1"].conn {
+			t.Fatalf("unexpected node assigned to rslot 0")
+		}
+		if client.rslots[8192] != client.conns["127.0.1.1:1"].conn {
+			t.Fatalf("unexpected node assigned to rslot 8192")
+		}
+		if client.rslots[8193] != client.conns["127.0.3.1:1"].conn {
+			t.Fatalf("unexpected node assigned to rslot 8193")
+		}
+		if client.rslots[16383] != client.conns["127.0.3.1:1"].conn {
+			t.Fatalf("unexpected node assigned to rslot 16383")
+		}
+	})
 }
 
 //gocyclo:ignore

rueidis.go

@@ -73,7 +73,6 @@ type ClientOption struct {
 
 	// SendToReplicas is a function that returns true if the command should be sent to replicas.
 	// currently only used for cluster client.
-	// NOTE: This function can't be used with ReplicaOnly option.
 	SendToReplicas func(cmd Completed) bool
 
 	// Sentinel options, including MasterSet and Auth options
@@ -161,6 +160,7 @@ type ClientOption struct {
 	ForceSingleClient bool
 
 	// ReplicaOnly indicates that this client will only try to connect to readonly replicas of redis setup.
+	// If SendToReplicas is set, client will use read-only operations through replicas connection.
 	ReplicaOnly bool
 
 	// ClientNoEvict sets the client eviction mode for the current connection.

rueidis_test.go

@@ -14,7 +14,6 @@ import (
 	"math/big"
 	"net"
 	"os"
-	"strings"
 	"testing"
 	"time"
 
@@ -132,30 +131,6 @@ func TestNewClusterClientError(t *testing.T) {
 		}
 		<-done
 	})
-
-	t.Run("replica only and send to replicas option conflict", func(t *testing.T) {
-		ln, err := net.Listen("tcp", "127.0.0.1:0")
-		if err != nil {
-			t.Fatal(err)
-		}
-		defer ln.Close()
-
-		_, port, _ := net.SplitHostPort(ln.Addr().String())
-		client, err := NewClient(ClientOption{
-			InitAddress: []string{"127.0.0.1:" + port},
-			ReplicaOnly: true,
-			SendToReplicas: func(cmd Completed) bool {
-				return true
-			},
-		})
-		if client != nil || err == nil {
-			t.Errorf("unexpected return %v %v", client, err)
-		}
-
-		if !strings.Contains(err.Error(), ErrReplicaOnlyConflict.Error()) {
-			t.Errorf("unexpected error %v", err)
-		}
-	})
 }
 
 func TestFallBackSingleClient(t *testing.T) {