feat(streams): Add the possibility to have per stream ids

commands_test.go

@@ -7410,6 +7410,81 @@ var _ = Describe("Commands", func() {
 			}))
 		})
 
+		It("should XRead with per-stream IDs", func() {
+			id, err := client.XAdd(ctx, &redis.XAddArgs{
+				Stream: "stream2",
+				ID:     "1-0",
+				Values: map[string]interface{}{"alpha": "a"},
+			}).Result()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(id).To(Equal("1-0"))
+
+			id, err = client.XAdd(ctx, &redis.XAddArgs{
+				Stream: "stream2",
+				ID:     "2-0",
+				Values: map[string]interface{}{"beta": "b"},
+			}).Result()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(id).To(Equal("2-0"))
+			defer client.Del(ctx, "stream2")
+
+			// Use per-stream IDs: resume "stream" from 2-0 (expect only 3-0),
+			// and "stream2" from 0 (expect 1-0 and 2-0).
+			res, err := client.XRead(ctx, &redis.XReadArgs{
+				Streams: []string{"stream", "stream2"},
+				IDs:     []string{"2-0", "0"},
+			}).Result()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(res).To(Equal([]redis.XStream{
+				{
+					Stream: "stream",
+					Messages: []redis.XMessage{
+						{ID: "3-0", Values: map[string]interface{}{"tres": "troix"}},
+					},
+				},
+				{
+					Stream: "stream2",
+					Messages: []redis.XMessage{
+						{ID: "1-0", Values: map[string]interface{}{"alpha": "a"}},
+						{ID: "2-0", Values: map[string]interface{}{"beta": "b"}},
+					},
+				},
+			}))
+		})
+
+		It("should XRead IDs take precedence over ID", func() {
+			id, err := client.XAdd(ctx, &redis.XAddArgs{
+				Stream: "stream3",
+				ID:     "1-0",
+				Values: map[string]interface{}{"alpha": "a"},
+			}).Result()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(id).To(Equal("1-0"))
+			defer client.Del(ctx, "stream3")
+
+			// ID is set but should be ignored because IDs is non-empty.
+			res, err := client.XRead(ctx, &redis.XReadArgs{
+				Streams: []string{"stream", "stream3"},
+				ID:     "$",
+				IDs:    []string{"2-0", "0"},
+			}).Result()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(res).To(Equal([]redis.XStream{
+				{
+					Stream: "stream",
+					Messages: []redis.XMessage{
+						{ID: "3-0", Values: map[string]interface{}{"tres": "troix"}},
+					},
+				},
+				{
+					Stream: "stream3",
+					Messages: []redis.XMessage{
+						{ID: "1-0", Values: map[string]interface{}{"alpha": "a"}},
+					},
+				},
+			}))
+		})
+
 		It("should XRead LastEntry blocks", Label("NonRedisEnterprise"), func() {
 			SkipBeforeRedisVersion(7.4, "doesn't work with older redis stack images")
 			start := time.Now()

stream_commands.go

@@ -193,10 +193,16 @@ func (c cmdable) XRevRangeN(ctx context.Context, stream, start, stop string, cou
 }
 
 type XReadArgs struct {
-	Streams []string // list of streams and ids, e.g. stream1 stream2 id1 id2
+	Streams []string // list of streams, or streams and ids e.g. stream1 stream2 id1 id2
 	Count   int64
 	Block   time.Duration
-	ID      string
+	// ID is a single ID applied to every stream in Streams.
+	// When reading from multiple streams that require different IDs,
+	// use IDs instead so each stream can be resumed from its own last ID.
+	ID string
+	// IDs is the per-stream list of IDs (one per entry in Streams).
+	// When non-empty, IDs takes precedence over ID.
+	IDs []string
 }
 
 func (c cmdable) XRead(ctx context.Context, a *XReadArgs) *XStreamSliceCmd {
@@ -219,7 +225,12 @@ func (c cmdable) XRead(ctx context.Context, a *XReadArgs) *XStreamSliceCmd {
 	for _, s := range a.Streams {
 		args = append(args, s)
 	}
-	if a.ID != "" {
+	switch {
+	case len(a.IDs) > 0:
+		for _, id := range a.IDs {
+			args = append(args, id)
+		}
+	case a.ID != "":
 		for range a.Streams {
 			args = append(args, a.ID)
 		}