improve code readability

- replace two similar functions `appendUniqueNode` and `appendIfNotExists` with a generic function.

- simplify the implementation of the `get` method in `clusterNodes`

- minimize line breaks in function definitions.

- keep the member name `_generation` of `clusterNodes` consistent with other types.

Signed-off-by: Xiaolong Chen <[email protected]>

Author: cxljs

geo_commands.go

@@ -61,9 +61,7 @@ func (c cmdable) GeoRadiusStore(
 }
 
 // GeoRadiusByMember is a read-only GEORADIUSBYMEMBER_RO command.
-func (c cmdable) GeoRadiusByMember(
-	ctx context.Context, key, member string, query *GeoRadiusQuery,
-) *GeoLocationCmd {
+func (c cmdable) GeoRadiusByMember(ctx context.Context, key, member string, query *GeoRadiusQuery) *GeoLocationCmd {
 	cmd := NewGeoLocationCmd(ctx, query, "georadiusbymember_ro", key, member)
 	if query.Store != "" || query.StoreDist != "" {
 		cmd.SetErr(errors.New("GeoRadiusByMember does not support Store or StoreDist"))
@@ -74,9 +72,7 @@ func (c cmdable) GeoRadiusByMember(
 }
 
 // GeoRadiusByMemberStore is a writing GEORADIUSBYMEMBER command.
-func (c cmdable) GeoRadiusByMemberStore(
-	ctx context.Context, key, member string, query *GeoRadiusQuery,
-) *IntCmd {
+func (c cmdable) GeoRadiusByMemberStore(ctx context.Context, key, member string, query *GeoRadiusQuery) *IntCmd {
 	args := geoLocationArgs(query, "georadiusbymember", key, member)
 	cmd := NewIntCmd(ctx, args...)
 	if query.Store == "" && query.StoreDist == "" {
@@ -96,9 +92,7 @@ func (c cmdable) GeoSearch(ctx context.Context, key string, q *GeoSearchQuery) *
 	return cmd
 }
 
-func (c cmdable) GeoSearchLocation(
-	ctx context.Context, key string, q *GeoSearchLocationQuery,
-) *GeoSearchLocationCmd {
+func (c cmdable) GeoSearchLocation(ctx context.Context, key string, q *GeoSearchLocationQuery) *GeoSearchLocationCmd {
 	args := make([]interface{}, 0, 16)
 	args = append(args, "geosearch", key)
 	args = geoSearchLocationArgs(q, args)
@@ -119,9 +113,7 @@ func (c cmdable) GeoSearchStore(ctx context.Context, key, store string, q *GeoSe
 	return cmd
 }
 
-func (c cmdable) GeoDist(
-	ctx context.Context, key string, member1, member2, unit string,
-) *FloatCmd {
+func (c cmdable) GeoDist(ctx context.Context, key string, member1, member2, unit string) *FloatCmd {
 	if unit == "" {
 		unit = "km"
 	}

internal/pool/conn.go

@@ -69,9 +69,7 @@ func (cn *Conn) RemoteAddr() net.Addr {
 	return nil
 }
 
-func (cn *Conn) WithReader(
-	ctx context.Context, timeout time.Duration, fn func(rd *proto.Reader) error,
-) error {
+func (cn *Conn) WithReader(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
@@ -80,9 +78,7 @@ func (cn *Conn) WithReader(
 	return fn(cn.rd)
 }
 
-func (cn *Conn) WithWriter(
-	ctx context.Context, timeout time.Duration, fn func(wr *proto.Writer) error,
-) error {
+func (cn *Conn) WithWriter(ctx context.Context, timeout time.Duration, fn func(wr *proto.Writer) error) error {
 	if timeout >= 0 {
 		if err := cn.netConn.SetWriteDeadline(cn.deadline(ctx, timeout)); err != nil {
 			return err

osscluster.go

@@ -342,8 +342,7 @@ type clusterNode struct {
 	failing    uint32 // atomic
 	loaded     uint32 // atomic
 
-	// last time the latency measurement was performed for the node, stored in nanoseconds
-	// from epoch
+	// last time the latency measurement was performed for the node, stored in nanoseconds from epoch
 	lastLatencyMeasurement int64 // atomic
 }
 
@@ -480,13 +479,12 @@ type clusterNodes struct {
 	closed      bool
 	onNewNode   []func(rdb *Client)
 
-	_generation uint32 // atomic
+	generation uint32 // atomic
 }
 
 func newClusterNodes(opt *ClusterOptions) *clusterNodes {
 	return &clusterNodes{
-		opt: opt,
-
+		opt:   opt,
 		addrs: opt.Addrs,
 		nodes: make(map[string]*clusterNode),
 	}
@@ -546,12 +544,11 @@ func (c *clusterNodes) Addrs() ([]string, error) {
 }
 
 func (c *clusterNodes) NextGeneration() uint32 {
-	return atomic.AddUint32(&c._generation, 1)
+	return atomic.AddUint32(&c.generation, 1)
 }
 
 // GC removes unused nodes.
 func (c *clusterNodes) GC(generation uint32) {
-	//nolint:prealloc
 	var collected []*clusterNode
 
 	c.mu.Lock()
@@ -604,23 +601,20 @@ func (c *clusterNodes) GetOrCreate(addr string) (*clusterNode, error) {
 		fn(node.Client)
 	}
 
-	c.addrs = appendIfNotExists(c.addrs, addr)
+	c.addrs = appendIfNotExist(c.addrs, addr)
 	c.nodes[addr] = node
 
 	return node, nil
 }
 
 func (c *clusterNodes) get(addr string) (*clusterNode, error) {
-	var node *clusterNode
-	var err error
 	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	if c.closed {
-		err = pool.ErrClosed
-	} else {
-		node = c.nodes[addr]
+		return nil, pool.ErrClosed
 	}
-	c.mu.RUnlock()
-	return node, err
+	return c.nodes[addr], nil
 }
 
 func (c *clusterNodes) All() ([]*clusterNode, error) {
@@ -651,8 +645,9 @@ func (c *clusterNodes) Random() (*clusterNode, error) {
 //------------------------------------------------------------------------------
 
 type clusterSlot struct {
-	start, end int
-	nodes      []*clusterNode
+	start int
+	end   int
+	nodes []*clusterNode
 }
 
 type clusterSlotSlice []*clusterSlot
@@ -680,9 +675,7 @@ type clusterState struct {
 	createdAt  time.Time
 }
 
-func newClusterState(
-	nodes *clusterNodes, slots []ClusterSlot, origin string,
-) (*clusterState, error) {
+func newClusterState(nodes *clusterNodes, slots []ClusterSlot, origin string) (*clusterState, error) {
 	c := clusterState{
 		nodes: nodes,
 
@@ -712,9 +705,9 @@ func newClusterState(
 			nodes = append(nodes, node)
 
 			if i == 0 {
-				c.Masters = appendUniqueNode(c.Masters, node)
+				c.Masters = appendIfNotExist(c.Masters, node)
 			} else {
-				c.Slaves = appendUniqueNode(c.Slaves, node)
+				c.Slaves = appendIfNotExist(c.Slaves, node)
 			}
 		}
 
@@ -1273,7 +1266,7 @@ func (c *ClusterClient) loadState(ctx context.Context) (*clusterState, error) {
 			continue
 		}
 
-		return newClusterState(c.nodes, slots, node.Client.opt.Addr)
+		return newClusterState(c.nodes, slots, addr)
 	}
 
 	/*
@@ -1932,11 +1925,7 @@ func cmdSlot(cmd Cmder, pos int, preferredRandomSlot int) int {
 	return hashtag.Slot(firstKey)
 }
 
-func (c *ClusterClient) cmdNode(
-	ctx context.Context,
-	cmdName string,
-	slot int,
-) (*clusterNode, error) {
+func (c *ClusterClient) cmdNode(ctx context.Context, cmdName string, slot int) (*clusterNode, error) {
 	state, err := c.state.Get(ctx)
 	if err != nil {
 		return nil, err
@@ -2005,26 +1994,13 @@ func (c *ClusterClient) context(ctx context.Context) context.Context {
 	return context.Background()
 }
 
-func appendUniqueNode(nodes []*clusterNode, node *clusterNode) []*clusterNode {
-	for _, n := range nodes {
-		if n == node {
-			return nodes
-		}
-	}
-	return append(nodes, node)
-}
-
-func appendIfNotExists(ss []string, es ...string) []string {
-loop:
-	for _, e := range es {
-		for _, s := range ss {
-			if s == e {
-				continue loop
-			}
+func appendIfNotExist[T comparable](vals []T, newVal T) []T {
+	for _, v := range vals {
+		if v == newVal {
+			return vals
 		}
-		ss = append(ss, e)
 	}
-	return ss
+	return append(vals, newVal)
 }
 
 //------------------------------------------------------------------------------

redis.go

@@ -383,7 +383,7 @@ func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error {
 
 	// for redis-server versions that do not support the HELLO command,
 	// RESP2 will continue to be used.
-  if err = conn.Hello(ctx, c.opt.Protocol, username, password, c.opt.ClientName).Err(); err == nil {
+	if err = conn.Hello(ctx, c.opt.Protocol, username, password, c.opt.ClientName).Err(); err == nil {
 		// Authentication successful with HELLO command
 	} else if !isRedisError(err) {
 		// When the server responds with the RESP protocol and the result is not a normal
@@ -460,9 +460,7 @@ func (c *baseClient) releaseConn(ctx context.Context, cn *pool.Conn, err error)
 	}
 }
 
-func (c *baseClient) withConn(
-	ctx context.Context, fn func(context.Context, *pool.Conn) error,
-) error {
+func (c *baseClient) withConn(ctx context.Context, fn func(context.Context, *pool.Conn) error) error {
 	cn, err := c.getConn(ctx)
 	if err != nil {
 		return err
@@ -610,9 +608,7 @@ func (c *baseClient) processTxPipeline(ctx context.Context, cmds []Cmder) error
 
 type pipelineProcessor func(context.Context, *pool.Conn, []Cmder) (bool, error)
 
-func (c *baseClient) generalProcessPipeline(
-	ctx context.Context, cmds []Cmder, p pipelineProcessor,
-) error {
+func (c *baseClient) generalProcessPipeline(ctx context.Context, cmds []Cmder, p pipelineProcessor) error {
 	var lastErr error
 	for attempt := 0; attempt <= c.opt.MaxRetries; attempt++ {
 		if attempt > 0 {
@@ -636,9 +632,7 @@ func (c *baseClient) generalProcessPipeline(
 	return lastErr
 }
 
-func (c *baseClient) pipelineProcessCmds(
-	ctx context.Context, cn *pool.Conn, cmds []Cmder,
-) (bool, error) {
+func (c *baseClient) pipelineProcessCmds(ctx context.Context, cn *pool.Conn, cmds []Cmder) (bool, error) {
 	if err := cn.WithWriter(c.context(ctx), c.opt.WriteTimeout, func(wr *proto.Writer) error {
 		return writeCmds(wr, cmds)
 	}); err != nil {
@@ -668,9 +662,7 @@ func pipelineReadCmds(rd *proto.Reader, cmds []Cmder) error {
 	return cmds[0].Err()
 }
 
-func (c *baseClient) txPipelineProcessCmds(
-	ctx context.Context, cn *pool.Conn, cmds []Cmder,
-) (bool, error) {
+func (c *baseClient) txPipelineProcessCmds(ctx context.Context, cn *pool.Conn, cmds []Cmder) (bool, error) {
 	if err := cn.WithWriter(c.context(ctx), c.opt.WriteTimeout, func(wr *proto.Writer) error {
 		return writeCmds(wr, cmds)
 	}); err != nil {

ring.go

@@ -667,10 +667,7 @@ func (c *Ring) OnNewNode(fn func(rdb *Client)) {
 
 // ForEachShard concurrently calls the fn on each live shard in the ring.
 // It returns the first error if any.
-func (c *Ring) ForEachShard(
-	ctx context.Context,
-	fn func(ctx context.Context, client *Client) error,
-) error {
+func (c *Ring) ForEachShard(ctx context.Context, fn func(ctx context.Context, client *Client) error) error {
 	// note: `c.List()` return a shadow copy of `[]*ringShard`.
 	shards := c.sharding.List()
 	var wg sync.WaitGroup
@@ -779,9 +776,7 @@ func (c *Ring) TxPipeline() Pipeliner {
 	return &pipe
 }
 
-func (c *Ring) generalProcessPipeline(
-	ctx context.Context, cmds []Cmder, tx bool,
-) error {
+func (c *Ring) generalProcessPipeline(ctx context.Context, cmds []Cmder, tx bool) error {
 	if tx {
 		// Trim multi .. exec.
 		cmds = cmds[1 : len(cmds)-1]

sentinel.go

@@ -447,9 +447,7 @@ func NewFailoverClient(failoverOpt *FailoverOptions) *Client {
 	return rdb
 }
 
-func masterReplicaDialer(
-	failover *sentinelFailover,
-) func(ctx context.Context, network, addr string) (net.Conn, error) {
+func masterReplicaDialer(failover *sentinelFailover) func(ctx context.Context, network, addr string) (net.Conn, error) {
 	return func(ctx context.Context, network, _ string) (net.Conn, error) {
 		var addr string
 		var err error

stream_commands.go

@@ -382,10 +382,7 @@ func xClaimArgs(a *XClaimArgs) []interface{} {
 //	XTRIM key MAXLEN/MINID ~ threshold LIMIT limit.
 //
 // The redis-server version is lower than 6.2, please set limit to 0.
-func (c cmdable) xTrim(
-	ctx context.Context, key, strategy string,
-	approx bool, threshold interface{}, limit int64,
-) *IntCmd {
+func (c cmdable) xTrim(ctx context.Context, key, strategy string, approx bool, threshold interface{}, limit int64) *IntCmd {
 	args := make([]interface{}, 0, 7)
 	args = append(args, "xtrim", key, strategy)
 	if approx {