Repository: go-redis/redis
Pull request #1780 (2021-06-02 20:14:58)

Add XAutoClaim command
Closes https://github.com/go-redis/redis/issues/1686

Redis docs: https://redis.io/commands/xautoclaim

I don't think `ID` is the right name for the cursor returned by Redis, but not sure what would be better. https://github.com/go-redis/redis/pull/1780/files#diff-4d7d1923693fc5ce892add2ea2907a744e77ea0b50c1939ccc5067cb48a466a3R1505

ericmillin: @monkey92t The assertion that failed passes locally. Are there any common race conditions you run into?

https://github.com/go-redis/redis/pull/1780/checks?check_run_id=2733660581#step:5:1586
monkey92t: @ericmillin Yes, there may be an unexpected data race, you can ignore it for now.
monkey92t: But the current workflow error is not accidental, it needs to be resolved
```go
      Expected
          <string>: 2-0
      to equal
          <string>: 3-0

      /home/runner/work/redis/redis/commands_test.go:4399
```
monkey92t: @ericmillin  it should be 2-0
```go
1) 1) "stream"
   2) 1) 1) "1-0"
         2) 1) "uno"
            2) "un"
      2) 1) "2-0"
         2) 1) "dos"
            2) "deux"
      3) 1) "3-0"
         2) 1) "tres"
            2) "troix"
127.0.0.1:6379> XAutoClaim stream group consumer 0 - count 2
1) "2-0"
2) 1) 1) "1-0"
      2) 1) "uno"
         2) "un"
   2) 1) "2-0"
      2) 1) "dos"
         2) "deux"
127.0.0.1:6379> 

```
ericmillin: @monkey92t  Do you have any opinions on this: https://github.com/go-redis/redis/pull/1780/files#diff-4d7d1923693fc5ce892add2ea2907a744e77ea0b50c1939ccc5067cb48a466a3R1556-R1559

Seems to match the Redis semantics, but looks a bit confusing. I'm fairly new to Redis, though, hence my hesitation.
monkey92t: I made a little modification. What do you think of the new return value method? Any better suggestions?
ericmillin: @monkey92t That looks great and had crossed my mind. If you're good with it, let's merge it.

Thanks for all your help :)

Patch:
diff --git a/command.go b/command.go
index 9a3018d..77d0bad 100644
--- a/command.go
+++ b/command.go
@@ -1501,6 +1501,122 @@ func (cmd *XPendingExtCmd) readReply(rd *proto.Reader) error {
 
 //------------------------------------------------------------------------------
 
+type XAutoClaimCmd struct {
+	baseCmd
+
+	start string
+	val   []XMessage
+}
+
+var _ Cmder = (*XAutoClaimCmd)(nil)
+
+func NewXAutoClaimCmd(ctx context.Context, args ...interface{}) *XAutoClaimCmd {
+	return &XAutoClaimCmd{
+		baseCmd: baseCmd{
+			ctx:  ctx,
+			args: args,
+		},
+	}
+}
+
+func (cmd *XAutoClaimCmd) Val() (messages []XMessage, start string) {
+	return cmd.val, cmd.start
+}
+
+func (cmd *XAutoClaimCmd) Result() (messages []XMessage, start string, err error) {
+	return cmd.val, cmd.start, cmd.err
+}
+
+func (cmd *XAutoClaimCmd) String() string {
+	return cmdString(cmd, cmd.val)
+}
+
+func (cmd *XAutoClaimCmd) readReply(rd *proto.Reader) error {
+	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
+		if n != 2 {
+			return nil, fmt.Errorf("got %d, wanted 2", n)
+		}
+		var err error
+
+		cmd.start, err = rd.ReadString()
+		if err != nil {
+			return nil, err
+		}
+
+		cmd.val, err = readXMessageSlice(rd)
+		if err != nil {
+			return nil, err
+		}
+
+		return nil, nil
+	})
+	return err
+}
+
+//------------------------------------------------------------------------------
+
+type XAutoClaimJustIDCmd struct {
+	baseCmd
+
+	start string
+	val   []string
+}
+
+var _ Cmder = (*XAutoClaimJustIDCmd)(nil)
+
+func NewXAutoClaimJustIDCmd(ctx context.Context, args ...interface{}) *XAutoClaimJustIDCmd {
+	return &XAutoClaimJustIDCmd{
+		baseCmd: baseCmd{
+			ctx:  ctx,
+			args: args,
+		},
+	}
+}
+
+func (cmd *XAutoClaimJustIDCmd) Val() (ids []string, start string) {
+	return cmd.val, cmd.start
+}
+
+func (cmd *XAutoClaimJustIDCmd) Result() (ids []string, start string, err error) {
+	return cmd.val, cmd.start, cmd.err
+}
+
+func (cmd *XAutoClaimJustIDCmd) String() string {
+	return cmdString(cmd, cmd.val)
+}
+
+func (cmd *XAutoClaimJustIDCmd) readReply(rd *proto.Reader) error {
+	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
+		if n != 2 {
+			return nil, fmt.Errorf("got %d, wanted 2", n)
+		}
+		var err error
+
+		cmd.start, err = rd.ReadString()
+		if err != nil {
+			return nil, err
+		}
+
+		nn, err := rd.ReadArrayLen()
+		if err != nil {
+			return nil, err
+		}
+
+		cmd.val = make([]string, nn)
+		for i := 0; i < nn; i++ {
+			cmd.val[i], err = rd.ReadString()
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		return nil, nil
+	})
+	return err
+}
+
+//------------------------------------------------------------------------------
+
 type XInfoConsumersCmd struct {
 	baseCmd
 	val []XInfoConsumer
diff --git a/commands.go b/commands.go
index 4ac42ec..b3b15c7 100644
--- a/commands.go
+++ b/commands.go
@@ -1845,6 +1845,39 @@ func (c cmdable) XPendingExt(ctx context.Context, a *XPendingExtArgs) *XPendingE
 	return cmd
 }
 
+type XAutoClaimArgs struct {
+	Stream   string
+	Group    string
+	MinIdle  time.Duration
+	Start    string
+	Count    int64
+	Consumer string
+}
+
+func (c cmdable) XAutoClaim(ctx context.Context, a *XAutoClaimArgs) *XAutoClaimCmd {
+	args := xAutoClaimArgs(ctx, a)
+	cmd := NewXAutoClaimCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+func (c cmdable) XAutoClaimJustID(ctx context.Context, a *XAutoClaimArgs) *XAutoClaimJustIDCmd {
+	args := xAutoClaimArgs(ctx, a)
+	args = append(args, "justid")
+	cmd := NewXAutoClaimJustIDCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
+func xAutoClaimArgs(ctx context.Context, a *XAutoClaimArgs) []interface{} {
+	args := make([]interface{}, 0, 9)
+	args = append(args, "xautoclaim", a.Stream, a.Group, a.Consumer, formatMs(ctx, a.MinIdle), a.Start)
+	if a.Count > 0 {
+		args = append(args, "count", a.Count)
+	}
+	return args
+}
+
 type XClaimArgs struct {
 	Stream   string
 	Group    string


Test patch:
diff --git a/commands_test.go b/commands_test.go
index 427fec7..11b8592 100644
--- a/commands_test.go
+++ b/commands_test.go
@@ -4386,6 +4386,43 @@ var _ = Describe("Commands", func() {
 				Expect(n).To(Equal(int64(3)))
 			})
 
+			It("should XAutoClaim", func() {
+				xca := &redis.XAutoClaimArgs{
+					Stream:   "stream",
+					Group:    "group",
+					Consumer: "consumer",
+					Start:    "-",
+					Count:    2,
+				}
+				msgs, start, err := client.XAutoClaim(ctx, xca).Result()
+				Expect(err).NotTo(HaveOccurred())
+				Expect(start).To(Equal("2-0"))
+				Expect(msgs).To(Equal([]redis.XMessage{{
+					ID:     "1-0",
+					Values: map[string]interface{}{"uno": "un"},
+				}, {
+					ID:     "2-0",
+					Values: map[string]interface{}{"dos": "deux"},
+				}}))
+
+				xca.Start = start
+				msgs, start, err = client.XAutoClaim(ctx, xca).Result()
+				Expect(err).NotTo(HaveOccurred())
+				Expect(start).To(Equal("3-0"))
+				Expect(msgs).To(Equal([]redis.XMessage{{
+					ID:     "2-0",
+					Values: map[string]interface{}{"dos": "deux"},
+				}, {
+					ID:     "3-0",
+					Values: map[string]interface{}{"tres": "troix"},
+				}}))
+
+				ids, start, err := client.XAutoClaimJustID(ctx, xca).Result()
+				Expect(err).NotTo(HaveOccurred())
+				Expect(start).To(Equal("3-0"))
+				Expect(ids).To(Equal([]string{"2-0", "3-0"}))
+			})
+
 			It("should XClaim", func() {
 				msgs, err := client.XClaim(ctx, &redis.XClaimArgs{
 					Stream:   "stream",