Skip to content

Commit bcdb7d1

Browse files
htemelski-redisndyakov
htemelski-redis
and
ndyakov
authored
Added new stream commands (#3450)
* added new stream commands * updated docker image * fixed command return type * fixed tests * addressed PR comments --------- Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com>
1 parent 7bc12bb commit bcdb7d1

File tree

2 files changed

+129
-0
lines changed

2 files changed

+129
-0
lines changed

‎commands_test.go‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6169,6 +6169,34 @@ var _ = Describe("Commands", func() {
61696169
Expect(n).To(Equal(int64(3)))
61706170
})
61716171

6172+
It("should XTrimMaxLenMode", func() {
6173+
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
6174+
n, err := client.XTrimMaxLenMode(ctx, "stream", 0, "KEEPREF").Result()
6175+
Expect(err).NotTo(HaveOccurred())
6176+
Expect(n).To(BeNumerically(">=", 0))
6177+
})
6178+
6179+
It("should XTrimMaxLenApproxMode", func() {
6180+
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
6181+
n, err := client.XTrimMaxLenApproxMode(ctx, "stream", 0, 0, "KEEPREF").Result()
6182+
Expect(err).NotTo(HaveOccurred())
6183+
Expect(n).To(BeNumerically(">=", 0))
6184+
})
6185+
6186+
It("should XTrimMinIDMode", func() {
6187+
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
6188+
n, err := client.XTrimMinIDMode(ctx, "stream", "4-0", "KEEPREF").Result()
6189+
Expect(err).NotTo(HaveOccurred())
6190+
Expect(n).To(BeNumerically(">=", 0))
6191+
})
6192+
6193+
It("should XTrimMinIDApproxMode", func() {
6194+
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
6195+
n, err := client.XTrimMinIDApproxMode(ctx, "stream", "4-0", 0, "KEEPREF").Result()
6196+
Expect(err).NotTo(HaveOccurred())
6197+
Expect(n).To(BeNumerically(">=", 0))
6198+
})
6199+
61726200
It("should XAdd", func() {
61736201
id, err := client.XAdd(ctx, &redis.XAddArgs{
61746202
Stream: "stream",
@@ -6222,6 +6250,37 @@ var _ = Describe("Commands", func() {
62226250
Expect(n).To(Equal(int64(3)))
62236251
})
62246252

6253+
It("should XAckDel", func() {
6254+
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
6255+
// First, create a consumer group
6256+
err := client.XGroupCreate(ctx, "stream", "testgroup", "0").Err()
6257+
Expect(err).NotTo(HaveOccurred())
6258+
6259+
// Read messages to create pending entries
6260+
_, err = client.XReadGroup(ctx, &redis.XReadGroupArgs{
6261+
Group: "testgroup",
6262+
Consumer: "testconsumer",
6263+
Streams: []string{"stream", ">"},
6264+
}).Result()
6265+
Expect(err).NotTo(HaveOccurred())
6266+
6267+
// Test XAckDel with KEEPREF mode
6268+
n, err := client.XAckDel(ctx, "stream", "testgroup", "KEEPREF", "1-0", "2-0").Result()
6269+
Expect(err).NotTo(HaveOccurred())
6270+
Expect(n).To(HaveLen(2))
6271+
6272+
// Clean up
6273+
client.XGroupDestroy(ctx, "stream", "testgroup")
6274+
})
6275+
6276+
It("should XDelEx", func() {
6277+
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
6278+
// Test XDelEx with KEEPREF mode
6279+
n, err := client.XDelEx(ctx, "stream", "KEEPREF", "1-0", "2-0").Result()
6280+
Expect(err).NotTo(HaveOccurred())
6281+
Expect(n).To(HaveLen(2))
6282+
})
6283+
62256284
It("should XLen", func() {
62266285
n, err := client.XLen(ctx, "stream").Result()
62276286
Expect(err).NotTo(HaveOccurred())

‎stream_commands.go‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,9 @@ import (
77

88
type StreamCmdable interface {
99
XAdd(ctx context.Context, a *XAddArgs) *StringCmd
10+
XAckDel(ctx context.Context, stream string, group string, mode string, ids ...string) *SliceCmd
1011
XDel(ctx context.Context, stream string, ids ...string) *IntCmd
12+
XDelEx(ctx context.Context, stream string, mode string, ids ...string) *SliceCmd
1113
XLen(ctx context.Context, stream string) *IntCmd
1214
XRange(ctx context.Context, stream, start, stop string) *XMessageSliceCmd
1315
XRangeN(ctx context.Context, stream, start, stop string, count int64) *XMessageSliceCmd
@@ -31,8 +33,12 @@ type StreamCmdable interface {
3133
XAutoClaimJustID(ctx context.Context, a *XAutoClaimArgs) *XAutoClaimJustIDCmd
3234
XTrimMaxLen(ctx context.Context, key string, maxLen int64) *IntCmd
3335
XTrimMaxLenApprox(ctx context.Context, key string, maxLen, limit int64) *IntCmd
36+
XTrimMaxLenMode(ctx context.Context, key string, maxLen int64, mode string) *IntCmd
37+
XTrimMaxLenApproxMode(ctx context.Context, key string, maxLen, limit int64, mode string) *IntCmd
3438
XTrimMinID(ctx context.Context, key string, minID string) *IntCmd
3539
XTrimMinIDApprox(ctx context.Context, key string, minID string, limit int64) *IntCmd
40+
XTrimMinIDMode(ctx context.Context, key string, minID string, mode string) *IntCmd
41+
XTrimMinIDApproxMode(ctx context.Context, key string, minID string, limit int64, mode string) *IntCmd
3642
XInfoGroups(ctx context.Context, key string) *XInfoGroupsCmd
3743
XInfoStream(ctx context.Context, key string) *XInfoStreamCmd
3844
XInfoStreamFull(ctx context.Context, key string, count int) *XInfoStreamFullCmd
@@ -54,6 +60,7 @@ type XAddArgs struct {
5460
// Approx causes MaxLen and MinID to use "~" matcher (instead of "=").
5561
Approx bool
5662
Limit int64
63+
Mode string
5764
ID string
5865
Values interface{}
5966
}
@@ -81,6 +88,11 @@ func (c cmdable) XAdd(ctx context.Context, a *XAddArgs) *StringCmd {
8188
if a.Limit > 0 {
8289
args = append(args, "limit", a.Limit)
8390
}
91+
92+
if a.Mode != "" {
93+
args = append(args, a.Mode)
94+
}
95+
8496
if a.ID != "" {
8597
args = append(args, a.ID)
8698
} else {
@@ -93,6 +105,16 @@ func (c cmdable) XAdd(ctx context.Context, a *XAddArgs) *StringCmd {
93105
return cmd
94106
}
95107

108+
func (c cmdable) XAckDel(ctx context.Context, stream string, group string, mode string, ids ...string) *SliceCmd {
109+
args := []interface{}{"xackdel", stream, group, mode, "ids", len(ids)}
110+
for _, id := range ids {
111+
args = append(args, id)
112+
}
113+
cmd := NewSliceCmd(ctx, args...)
114+
_ = c(ctx, cmd)
115+
return cmd
116+
}
117+
96118
func (c cmdable) XDel(ctx context.Context, stream string, ids ...string) *IntCmd {
97119
args := []interface{}{"xdel", stream}
98120
for _, id := range ids {
@@ -103,6 +125,16 @@ func (c cmdable) XDel(ctx context.Context, stream string, ids ...string) *IntCmd
103125
return cmd
104126
}
105127

128+
func (c cmdable) XDelEx(ctx context.Context, stream string, mode string, ids ...string) *SliceCmd {
129+
args := []interface{}{"xdelex", stream, mode, "ids", len(ids)}
130+
for _, id := range ids {
131+
args = append(args, id)
132+
}
133+
cmd := NewSliceCmd(ctx, args...)
134+
_ = c(ctx, cmd)
135+
return cmd
136+
}
137+
106138
func (c cmdable) XLen(ctx context.Context, stream string) *IntCmd {
107139
cmd := NewIntCmd(ctx, "xlen", stream)
108140
_ = c(ctx, cmd)
@@ -375,6 +407,8 @@ func xClaimArgs(a *XClaimArgs) []interface{} {
375407
return args
376408
}
377409

410+
// TODO: refactor xTrim, xTrimMode and the wrappers over the functions
411+
378412
// xTrim If approx is true, add the "~" parameter, otherwise it is the default "=" (redis default).
379413
// example:
380414
//
@@ -418,6 +452,42 @@ func (c cmdable) XTrimMinIDApprox(ctx context.Context, key string, minID string,
418452
return c.xTrim(ctx, key, "minid", true, minID, limit)
419453
}
420454

455+
func (c cmdable) xTrimMode(
456+
ctx context.Context, key, strategy string,
457+
approx bool, threshold interface{}, limit int64,
458+
mode string,
459+
) *IntCmd {
460+
args := make([]interface{}, 0, 7)
461+
args = append(args, "xtrim", key, strategy)
462+
if approx {
463+
args = append(args, "~")
464+
}
465+
args = append(args, threshold)
466+
if limit > 0 {
467+
args = append(args, "limit", limit)
468+
}
469+
args = append(args, mode)
470+
cmd := NewIntCmd(ctx, args...)
471+
_ = c(ctx, cmd)
472+
return cmd
473+
}
474+
475+
func (c cmdable) XTrimMaxLenMode(ctx context.Context, key string, maxLen int64, mode string) *IntCmd {
476+
return c.xTrimMode(ctx, key, "maxlen", false, maxLen, 0, mode)
477+
}
478+
479+
func (c cmdable) XTrimMaxLenApproxMode(ctx context.Context, key string, maxLen, limit int64, mode string) *IntCmd {
480+
return c.xTrimMode(ctx, key, "maxlen", true, maxLen, limit, mode)
481+
}
482+
483+
func (c cmdable) XTrimMinIDMode(ctx context.Context, key string, minID string, mode string) *IntCmd {
484+
return c.xTrimMode(ctx, key, "minid", false, minID, 0, mode)
485+
}
486+
487+
func (c cmdable) XTrimMinIDApproxMode(ctx context.Context, key string, minID string, limit int64, mode string) *IntCmd {
488+
return c.xTrimMode(ctx, key, "minid", true, minID, limit, mode)
489+
}
490+
421491
func (c cmdable) XInfoConsumers(ctx context.Context, key string, group string) *XInfoConsumersCmd {
422492
cmd := NewXInfoConsumersCmd(ctx, key, group)
423493
_ = c(ctx, cmd)

0 commit comments

Comments
 (0)