Skip to content

Commit

Permalink
Add support for XREAD last entry (#3005)
Browse files Browse the repository at this point in the history
* add support for XREAD last entry

* handle reading from multiple streams

* add test to ensure we block for empty stream

* small tweak

* add an option to XReadArgs instead

* modify test comment

* small preallocation optimization

* Changed argument to generic ID, skip tests on Enterprise

* Fix test case

* Updated expiration command

---------

Co-authored-by: Vladyslav Vildanov <117659936+vladvildanov@users.noreply.github.com>
Co-authored-by: vladvildanov <divinez122@outlook.com>
  • Loading branch information
3 people authored and ofekshenawa committed Jun 20, 2024
1 parent 0b95fd7 commit da8623e
Show file tree
Hide file tree
Showing 2 changed files with 82 additions and 3 deletions.
77 changes: 75 additions & 2 deletions commands_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2588,13 +2588,14 @@ var _ = Describe("Commands", func() {
Expect(sadd.Err()).NotTo(HaveOccurred())
}

res, err := client.HExpire(ctx, "myhash", 10, "key1", "key200").Result()
expireAt := time.Now().Add(10 * time.Second)
res, err := client.HPExpireAt(ctx, "myhash", expireAt, "key1", "key200").Result()
Expect(err).NotTo(HaveOccurred())
Expect(res).To(Equal([]int64{1, -2}))

res, err = client.HPExpireTime(ctx, "myhash", "key1", "key2", "key200").Result()
Expect(err).NotTo(HaveOccurred())
Expect(res).To(BeEquivalentTo([]int64{time.Now().Add(10 * time.Second).UnixMilli(), -1, -2}))
Expect(res).To(BeEquivalentTo([]int64{expireAt.UnixMilli(), -1, -2}))
})

It("should HTTL", Label("hash-expiration", "NonRedisEnterprise"), func() {
Expand Down Expand Up @@ -5888,6 +5889,78 @@ var _ = Describe("Commands", func() {
Expect(err).To(Equal(redis.Nil))
})

It("should XRead LastEntry", Label("NonRedisEnterprise"), func() {
res, err := client.XRead(ctx, &redis.XReadArgs{
Streams: []string{"stream"},
Count: 2, // we expect 1 message
ID: "+",
}).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"}},
},
},
}))
})

It("should XRead LastEntry from two streams", Label("NonRedisEnterprise"), func() {
res, err := client.XRead(ctx, &redis.XReadArgs{
Streams: []string{"stream", "stream"},
ID: "+",
}).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: "stream",
Messages: []redis.XMessage{
{ID: "3-0", Values: map[string]interface{}{"tres": "troix"}},
},
},
}))
})

It("should XRead LastEntry blocks", Label("NonRedisEnterprise"), func() {
start := time.Now()
go func() {
defer GinkgoRecover()

time.Sleep(100 * time.Millisecond)
id, err := client.XAdd(ctx, &redis.XAddArgs{
Stream: "empty",
ID: "4-0",
Values: map[string]interface{}{"quatro": "quatre"},
}).Result()
Expect(err).NotTo(HaveOccurred())
Expect(id).To(Equal("4-0"))
}()

res, err := client.XRead(ctx, &redis.XReadArgs{
Streams: []string{"empty"},
Block: 500 * time.Millisecond,
ID: "+",
}).Result()
Expect(err).NotTo(HaveOccurred())
// Ensure that the XRead call with LastEntry option blocked for at least 100ms.
Expect(time.Since(start)).To(BeNumerically(">=", 100*time.Millisecond))
Expect(res).To(Equal([]redis.XStream{
{
Stream: "empty",
Messages: []redis.XMessage{
{ID: "4-0", Values: map[string]interface{}{"quatro": "quatre"}},
},
},
}))
})

Describe("group", func() {
BeforeEach(func() {
err := client.XGroupCreate(ctx, "stream", "group", "0").Err()
Expand Down
8 changes: 7 additions & 1 deletion stream_commands.go
Original file line number Diff line number Diff line change
Expand Up @@ -137,10 +137,11 @@ type XReadArgs struct {
Streams []string // list of streams and ids, e.g. stream1 stream2 id1 id2
Count int64
Block time.Duration
ID string
}

func (c cmdable) XRead(ctx context.Context, a *XReadArgs) *XStreamSliceCmd {
args := make([]interface{}, 0, 6+len(a.Streams))
args := make([]interface{}, 0, 2*len(a.Streams)+6)
args = append(args, "xread")

keyPos := int8(1)
Expand All @@ -159,6 +160,11 @@ func (c cmdable) XRead(ctx context.Context, a *XReadArgs) *XStreamSliceCmd {
for _, s := range a.Streams {
args = append(args, s)
}
if a.ID != "" {
for range a.Streams {
args = append(args, a.ID)
}
}

cmd := NewXStreamSliceCmd(ctx, args...)
if a.Block >= 0 {
Expand Down

0 comments on commit da8623e

Please sign in to comment.