Don't panic when cluster does not have valid nodes.

This commit is contained in:
Vladimir Mihailenco 2016-09-23 11:52:19 +00:00
parent 938235994e
commit c0b792a059
2 changed files with 24 additions and 8 deletions

View File

@ -160,17 +160,18 @@ func (c *ClusterClient) newNode(addr string) *clusterNode {
}
}
func (c *ClusterClient) slotNodes(slot int) []*clusterNode {
func (c *ClusterClient) slotNodes(slot int) (nodes []*clusterNode) {
c.mu.RLock()
nodes := c.slots[slot]
if slot < len(c.slots) {
nodes = c.slots[slot]
}
c.mu.RUnlock()
return nodes
}
// randomNode returns random live node.
func (c *ClusterClient) randomNode() (*clusterNode, error) {
var node *clusterNode
var err error
var nodeErr error
for i := 0; i < 10; i++ {
c.mu.RLock()
closed := c.closed
@ -182,17 +183,19 @@ func (c *ClusterClient) randomNode() (*clusterNode, error) {
}
n := rand.Intn(len(addrs))
node, err = c.nodeByAddr(addrs[n])
node, err := c.nodeByAddr(addrs[n])
if err != nil {
return nil, err
}
if node.Client.ClusterInfo().Err() == nil {
break
}
}
nodeErr = node.Client.ClusterInfo().Err()
if nodeErr == nil {
return node, nil
}
}
return nil, nodeErr
}
func (c *ClusterClient) slotMasterNode(slot int) (*clusterNode, error) {
nodes := c.slotNodes(slot)

View File

@ -515,6 +515,19 @@ var _ = Describe("ClusterClient", func() {
describeClusterClient()
})
Describe("ClusterClient without valid nodes", func() {
BeforeEach(func() {
client = redis.NewClusterClient(&redis.ClusterOptions{
Addrs: []string{redisAddr},
})
})
It("returns an error", func() {
err := client.Ping().Err()
Expect(err).To(MatchError("ERR This instance has cluster support disabled"))
})
})
})
//------------------------------------------------------------------------------