redis/go-redis
1
0
Fork 0
mirror of https://github.com/redis/go-redis.git synced 2026-10-22 19:49:00 +03:00
Code Activity

feat(command): Add MAXCOUNT and MAXSIZE params to XREAD and XREADGROUP commands (#3898)

Browse Source 
* Add MAXCOUNT and MAXSIZE params to XREAD and XREADGROUP commands

* Skip test before 8.10

---------

Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com>
This commit is contained in:
ofekshenawa
2026-07-20 18:49:31 +03:00
committed by GitHub
parent 26fde83feb
commit 24a75d0c51
3 changed files with 238 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
commands_test.go
View File

@@ -7999,6 +7999,77 @@ var _ = Describe("Commands", func() {
Expect(err).To(Equal(redis.Nil))
})
It("should XRead with MAXCOUNT capping cumulative entries across streams", func() {
SkipBeforeRedisVersion("8.10", "XREAD MAXCOUNT requires Redis 8.10+")
for _, id := range []string{"1-0", "2-0", "3-0"} {
_, err := client.XAdd(ctx, &redis.XAddArgs{Stream: "s1", ID: id, Values: map[string]interface{}{"f": "v"}}).Result()
Expect(err).NotTo(HaveOccurred())
_, err = client.XAdd(ctx, &redis.XAddArgs{Stream: "s2", ID: id, Values: map[string]interface{}{"f": "v"}}).Result()
Expect(err).NotTo(HaveOccurred())
}
// MAXCOUNT is cumulative across all streams. Streams are served in
// caller order, so s1 contributes its 3 entries and s2 contributes 1,
// reaching the total cap of 4.
res, err := client.XRead(ctx, &redis.XReadArgs{
Streams: []string{"s1", "s2", "0", "0"},
MaxCount: 4,
Block: -1,
}).Result()
Expect(err).NotTo(HaveOccurred())
total := 0
for _, s := range res {
total += len(s.Messages)
}
Expect(total).To(Equal(4))
})
It("should XRead with MAXSIZE returning at least the first available entry", func() {
SkipBeforeRedisVersion("8.10", "XREAD MAXSIZE requires Redis 8.10+")
// MAXSIZE is a soft cap; even a value smaller than a single entry must
// still return the first available entry (never suppress all output).
res, err := client.XRead(ctx, &redis.XReadArgs{
Streams: []string{"stream", "0"},
MaxSize: 1,
Block: -1,
}).Result()
Expect(err).NotTo(HaveOccurred())
Expect(res).To(HaveLen(1))
Expect(res[0].Stream).To(Equal("stream"))
Expect(len(res[0].Messages)).To(BeNumerically(">=", 1))
})
It("should XReadGroup with MAXCOUNT capping cumulative entries across streams", func() {
SkipBeforeRedisVersion("8.10", "XREADGROUP MAXCOUNT requires Redis 8.10+")
for _, id := range []string{"1-0", "2-0", "3-0"} {
_, err := client.XAdd(ctx, &redis.XAddArgs{Stream: "g1", ID: id, Values: map[string]interface{}{"f": "v"}}).Result()
Expect(err).NotTo(HaveOccurred())
_, err = client.XAdd(ctx, &redis.XAddArgs{Stream: "g2", ID: id, Values: map[string]interface{}{"f": "v"}}).Result()
Expect(err).NotTo(HaveOccurred())
}
Expect(client.XGroupCreate(ctx, "g1", "grp", "0").Err()).NotTo(HaveOccurred())
Expect(client.XGroupCreate(ctx, "g2", "grp", "0").Err()).NotTo(HaveOccurred())
res, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{
Group: "grp",
Consumer: "consumer",
Streams: []string{"g1", "g2", ">", ">"},
MaxCount: 4,
Block: -1,
}).Result()
Expect(err).NotTo(HaveOccurred())
total := 0
for _, s := range res {
total += len(s.Messages)
}
Expect(total).To(Equal(4))
})
It("should XRead LastEntry", Label("NonRedisEnterprise"), func() {
SkipBeforeRedisVersion("7.4", "doesn't work with older redis stack images")
res, err := client.XRead(ctx, &redis.XReadArgs{

32
stream_commands.go
View File

@@ -227,14 +227,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
Count int64
Block time.Duration
ID string
Streams []string // list of streams and ids, e.g. stream1 stream2 id1 id2
Count int64
MaxCount int64 // cumulative cap on total entries across all streams (Redis >= 8.10)
MaxSize int64 // soft cumulative cap on total reply size in bytes across all streams (Redis >= 8.10)
Block time.Duration
ID string
}
func (c cmdable) XRead(ctx context.Context, a *XReadArgs) *XStreamSliceCmd {
args := make([]interface{}, 0, 2*len(a.Streams)+6)
args := make([]interface{}, 0, 2*len(a.Streams)+10)
args = append(args, "xread")
keyPos := int8(1)
@@ -243,6 +245,14 @@ func (c cmdable) XRead(ctx context.Context, a *XReadArgs) *XStreamSliceCmd {
args = append(args, a.Count)
keyPos += 2
}
if a.MaxCount > 0 {
args = append(args, "maxcount", a.MaxCount)
keyPos += 2
}
if a.MaxSize > 0 {
args = append(args, "maxsize", a.MaxSize)
keyPos += 2
}
if a.Block >= 0 {
args = append(args, "block")
args = append(args, int64(a.Block/time.Millisecond))
@@ -322,13 +332,15 @@ type XReadGroupArgs struct {
Consumer string
Streams []string // list of streams and ids, e.g. stream1 stream2 id1 id2
Count int64
MaxCount int64 // cumulative cap on total entries across all streams (Redis >= 8.10)
MaxSize int64 // soft cumulative cap on total reply size in bytes across all streams (Redis >= 8.10)
Block time.Duration
NoAck bool
Claim time.Duration // Claim idle pending entries older than this duration
}
func (c cmdable) XReadGroup(ctx context.Context, a *XReadGroupArgs) *XStreamSliceCmd {
args := make([]interface{}, 0, 10+len(a.Streams))
args := make([]interface{}, 0, 14+len(a.Streams))
args = append(args, "xreadgroup", "group", a.Group, a.Consumer)
keyPos := int8(4)
@@ -336,6 +348,14 @@ func (c cmdable) XReadGroup(ctx context.Context, a *XReadGroupArgs) *XStreamSlic
args = append(args, "count", a.Count)
keyPos += 2
}
if a.MaxCount > 0 {
args = append(args, "maxcount", a.MaxCount)
keyPos += 2
}
if a.MaxSize > 0 {
args = append(args, "maxsize", a.MaxSize)
keyPos += 2
}
if a.Block >= 0 {
args = append(args, "block", int64(a.Block/time.Millisecond))
keyPos += 2

141
stream_commands_unit_test.go
View File

@@ -4,6 +4,7 @@ import (
"context"
"reflect"
"testing"
"time"
)
// These tests assert the argument lists built by the XTRIM/XADD trimming
@@ -194,3 +195,143 @@ func TestXAdd_TrimLimitArgs(t *testing.T) {
})
}
}
// These tests assert the argument lists built by XREAD for the MAXCOUNT and
// MAXSIZE options (Redis >= 8.10) without dispatching to a server:
// - both default to unset and emit no token when zero;
// - when set they are emitted in canonical order, after COUNT and before
// BLOCK, and before the STREAMS keys;
// - the first-key position (used for cluster routing) accounts for the new
// tokens, since they appear before STREAMS.
func TestXRead_MaxArgs(t *testing.T) {
ctx := context.Background()
tests := []struct {
name string
args *XReadArgs
want []interface{}
wantPos int8
}{
{
name: "no_max_options",
args: &XReadArgs{Streams: []string{"s1", "0"}, Block: -1},
want: []interface{}{"xread", "streams", "s1", "0"},
// keyPos: 1 (xread) + 1 (streams) = 2
wantPos: 2,
},
{
name: "maxcount_only",
args: &XReadArgs{Streams: []string{"s1", "s2", "0", "0"}, MaxCount: 80, Block: -1},
want: []interface{}{"xread", "maxcount", int64(80), "streams", "s1", "s2", "0", "0"},
// keyPos: 1 + 2 (maxcount) + 1 (streams) = 4
wantPos: 4,
},
{
name: "maxsize_only",
args: &XReadArgs{Streams: []string{"s1", "0"}, MaxSize: 65536, Block: -1},
want: []interface{}{"xread", "maxsize", int64(65536), "streams", "s1", "0"},
// keyPos: 1 + 2 (maxsize) + 1 (streams) = 4
wantPos: 4,
},
{
name: "count_maxcount_maxsize_block",
args: &XReadArgs{
Streams: []string{"s1", "s2", "0", "0"},
Count: 50,
MaxCount: 80,
MaxSize: 65536,
Block: 5 * time.Second,
},
want: []interface{}{
"xread", "count", int64(50), "maxcount", int64(80), "maxsize", int64(65536),
"block", int64(5000), "streams", "s1", "s2", "0", "0",
},
// keyPos: 1 + 2 (count) + 2 (maxcount) + 2 (maxsize) + 2 (block) + 1 (streams) = 10
wantPos: 10,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
var captured Cmder
c := captureCmdable(&captured)
cmd := c.XRead(ctx, tt.args)
if cmd == nil {
t.Fatalf("XRead returned nil")
}
if !reflect.DeepEqual(cmd.Args(), tt.want) {
t.Errorf("args mismatch\n got: %#v\nwant: %#v", cmd.Args(), tt.want)
}
if got := cmd.firstKeyPos(); got != tt.wantPos {
t.Errorf("firstKeyPos mismatch: got %d, want %d", got, tt.wantPos)
}
})
}
}
// These tests assert the argument lists built by XREADGROUP for the MAXCOUNT and
// MAXSIZE options (Redis >= 8.10), including their interaction with the existing
// NOACK and CLAIM options and the first-key position used for cluster routing.
func TestXReadGroup_MaxArgs(t *testing.T) {
ctx := context.Background()
tests := []struct {
name string
args *XReadGroupArgs
want []interface{}
wantPos int8
}{
{
name: "no_max_options",
args: &XReadGroupArgs{Group: "g", Consumer: "c", Streams: []string{"s1", ">"}, Block: -1},
want: []interface{}{"xreadgroup", "group", "g", "c", "streams", "s1", ">"},
// keyPos: 4 (group/consumer) + 1 (streams) = 5
wantPos: 5,
},
{
name: "maxcount_only",
args: &XReadGroupArgs{Group: "g", Consumer: "c", Streams: []string{"s1", ">"}, MaxCount: 80, Block: -1},
want: []interface{}{"xreadgroup", "group", "g", "c", "maxcount", int64(80), "streams", "s1", ">"},
// keyPos: 4 + 2 (maxcount) + 1 (streams) = 7
wantPos: 7,
},
{
name: "all_options",
args: &XReadGroupArgs{
Group: "g",
Consumer: "c",
Streams: []string{"s1", "s2", ">", ">"},
Count: 50,
MaxCount: 80,
MaxSize: 65536,
Block: 5 * time.Second,
NoAck: true,
Claim: 30 * time.Second,
},
want: []interface{}{
"xreadgroup", "group", "g", "c", "count", int64(50), "maxcount", int64(80),
"maxsize", int64(65536), "block", int64(5000), "noack", "claim", int64(30000),
"streams", "s1", "s2", ">", ">",
},
// keyPos: 4 + 2 (count) + 2 (maxcount) + 2 (maxsize) + 2 (block) + 1 (noack) + 2 (claim) + 1 (streams) = 16
wantPos: 16,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
var captured Cmder
c := captureCmdable(&captured)
cmd := c.XReadGroup(ctx, tt.args)
if cmd == nil {
t.Fatalf("XReadGroup returned nil")
}
if !reflect.DeepEqual(cmd.Args(), tt.want) {
t.Errorf("args mismatch\n got: %#v\nwant: %#v", cmd.Args(), tt.want)
}
if got := cmd.firstKeyPos(); got != tt.wantPos {
t.Errorf("firstKeyPos mismatch: got %d, want %d", got, tt.wantPos)
}
})
}
}