Add support for CLUSTER LINKS command (#2504)

* feat: Adding support for Cluster Links Command

Co-authored-by: Anuragkillswitch <70265851+Anuragkillswitch@users.noreply.github.com>
This commit is contained in:
Anurag Bandyopadhyay 2023-03-28 09:49:54 +05:30 committed by GitHub
parent f30d7c4a25
commit 42d730c143
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 120 additions and 0 deletions

View File

@ -677,6 +677,30 @@ var _ = Describe("ClusterClient", func() {
Expect(assertSlotsEqual(res, wanted)).NotTo(HaveOccurred())
})
It("should CLUSTER LINKS", func() {
res, err := client.ClusterLinks(ctx).Result()
Expect(err).NotTo(HaveOccurred())
Expect(res).NotTo(BeEmpty())
// Iterate over the ClusterLink results and validate the map keys.
for _, link := range res {
Expect(link.Direction).NotTo(BeEmpty())
Expect([]string{"from", "to"}).To(ContainElement(link.Direction))
Expect(link.Node).NotTo(BeEmpty())
Expect(link.CreateTime).To(BeNumerically(">", 0))
Expect(link.Events).NotTo(BeEmpty())
validEventChars := []rune{'r', 'w'}
for _, eventChar := range link.Events {
Expect(validEventChars).To(ContainElement(eventChar))
}
Expect(link.SendBufferAllocated).To(BeNumerically(">=", 0))
Expect(link.SendBufferUsed).To(BeNumerically(">=", 0))
}
})
It("should cluster client setname", func() {
err := client.ForEachShard(ctx, func(ctx context.Context, c *redis.Client) error {
return c.Ping(ctx).Err()

View File

@ -4246,3 +4246,92 @@ func (cmd *KeyFlagsCmd) readReply(rd *proto.Reader) error {
return nil
}
// ---------------------------------------------------------------------------------------------------
type ClusterLink struct {
Direction string
Node string
CreateTime int64
Events string
SendBufferAllocated int64
SendBufferUsed int64
}
type ClusterLinksCmd struct {
baseCmd
val []ClusterLink
}
var _ Cmder = (*ClusterLinksCmd)(nil)
func NewClusterLinksCmd(ctx context.Context, args ...interface{}) *ClusterLinksCmd {
return &ClusterLinksCmd{
baseCmd: baseCmd{
ctx: ctx,
args: args,
},
}
}
func (cmd *ClusterLinksCmd) SetVal(val []ClusterLink) {
cmd.val = val
}
func (cmd *ClusterLinksCmd) Val() []ClusterLink {
return cmd.val
}
func (cmd *ClusterLinksCmd) Result() ([]ClusterLink, error) {
return cmd.Val(), cmd.Err()
}
func (cmd *ClusterLinksCmd) String() string {
return cmdString(cmd, cmd.val)
}
func (cmd *ClusterLinksCmd) readReply(rd *proto.Reader) error {
n, err := rd.ReadArrayLen()
if err != nil {
return err
}
cmd.val = make([]ClusterLink, n)
for i := 0; i < len(cmd.val); i++ {
m, err := rd.ReadMapLen()
if err != nil {
return err
}
for j := 0; j < m; j++ {
key, err := rd.ReadString()
if err != nil {
return err
}
switch key {
case "direction":
cmd.val[i].Direction, err = rd.ReadString()
case "node":
cmd.val[i].Node, err = rd.ReadString()
case "create-time":
cmd.val[i].CreateTime, err = rd.ReadInt()
case "events":
cmd.val[i].Events, err = rd.ReadString()
case "send-buffer-allocated":
cmd.val[i].SendBufferAllocated, err = rd.ReadInt()
case "send-buffer-used":
cmd.val[i].SendBufferUsed, err = rd.ReadInt()
default:
return fmt.Errorf("redis: unexpected key %q in CLUSTER LINKS reply", key)
}
if err != nil {
return err
}
}
}
return nil
}

View File

@ -422,6 +422,7 @@ type Cmdable interface {
PubSubShardNumSub(ctx context.Context, channels ...string) *MapStringIntCmd
ClusterSlots(ctx context.Context) *ClusterSlotsCmd
ClusterLinks(ctx context.Context) *ClusterLinksCmd
ClusterNodes(ctx context.Context) *StringCmd
ClusterMeet(ctx context.Context, host, port string) *StatusCmd
ClusterForget(ctx context.Context, nodeID string) *StatusCmd
@ -3497,6 +3498,12 @@ func (c cmdable) ClusterSlots(ctx context.Context) *ClusterSlotsCmd {
return cmd
}
func (c cmdable) ClusterLinks(ctx context.Context) *ClusterLinksCmd {
cmd := NewClusterLinksCmd(ctx, "cluster", "links")
_ = c(ctx, cmd)
return cmd
}
func (c cmdable) ClusterNodes(ctx context.Context) *StringCmd {
cmd := NewStringCmd(ctx, "cluster", "nodes")
_ = c(ctx, cmd)