This is an automated email from the ASF dual-hosted git repository. twice pushed a commit to branch 2.9 in repository https://gitbox.apache.org/repos/asf/kvrocks.git
commit c8826e7024af19bac12677b630dba1a1027a997f Author: PragmaTwice <[email protected]> AuthorDate: Mon Jul 1 13:16:44 2024 +0900 chore: disable search and stream group commands --- src/commands/cmd_search.cc | 17 +- src/commands/cmd_stream.cc | 37 +- tests/gocase/unit/search/search_test.go | 2 + tests/gocase/unit/type/stream/stream_test.go | 2074 +++++++++++++------------- 4 files changed, 1067 insertions(+), 1063 deletions(-) diff --git a/src/commands/cmd_search.cc b/src/commands/cmd_search.cc index 1928672c..ea42f269 100644 --- a/src/commands/cmd_search.cc +++ b/src/commands/cmd_search.cc @@ -358,13 +358,14 @@ class CommandFTDrop : public Commander { }; }; -REDIS_REGISTER_COMMANDS(MakeCmdAttr<CommandFTCreate>("ft.create", -2, "write exclusive no-multi no-script", 0, 0, 0), - MakeCmdAttr<CommandFTSearchSQL>("ft.searchsql", 2, "read-only", 0, 0, 0), - MakeCmdAttr<CommandFTSearch>("ft.search", -3, "read-only", 0, 0, 0), - MakeCmdAttr<CommandFTExplainSQL>("ft.explainsql", -2, "read-only", 0, 0, 0), - MakeCmdAttr<CommandFTExplain>("ft.explain", -3, "read-only", 0, 0, 0), - MakeCmdAttr<CommandFTInfo>("ft.info", 2, "read-only", 0, 0, 0), - MakeCmdAttr<CommandFTList>("ft._list", 1, "read-only", 0, 0, 0), - MakeCmdAttr<CommandFTDrop>("ft.dropindex", 2, "write exclusive no-multi no-script", 0, 0, 0)); +// REDIS_REGISTER_COMMANDS(MakeCmdAttr<CommandFTCreate>("ft.create", -2, "write exclusive no-multi no-script", 0, 0, 0), +// MakeCmdAttr<CommandFTSearchSQL>("ft.searchsql", 2, "read-only", 0, 0, 0), +// MakeCmdAttr<CommandFTSearch>("ft.search", -3, "read-only", 0, 0, 0), +// MakeCmdAttr<CommandFTExplainSQL>("ft.explainsql", -2, "read-only", 0, 0, 0), +// MakeCmdAttr<CommandFTExplain>("ft.explain", -3, "read-only", 0, 0, 0), +// MakeCmdAttr<CommandFTInfo>("ft.info", 2, "read-only", 0, 0, 0), +// MakeCmdAttr<CommandFTList>("ft._list", 1, "read-only", 0, 0, 0), +// MakeCmdAttr<CommandFTDrop>("ft.dropindex", 2, "write exclusive no-multi no-script", 0, 0, +// 0)); } // namespace redis diff --git a/src/commands/cmd_stream.cc b/src/commands/cmd_stream.cc index 7faeb9c4..ee608be2 100644 --- a/src/commands/cmd_stream.cc +++ b/src/commands/cmd_stream.cc @@ -655,10 +655,10 @@ class CommandXInfo : public Commander { count_ = *parse_result; } - } else if (val == "groups" && args.size() == 3) { - subcommand_ = "groups"; - } else if (val == "consumers" && args.size() == 4) { - subcommand_ = "consumers"; + // } else if (val == "groups" && args.size() == 3) { + // subcommand_ = "groups"; + // } else if (val == "consumers" && args.size() == 4) { + // subcommand_ = "consumers"; } else { return {Status::RedisParseErr, errUnknownSubcommandOrWrongArguments}; } @@ -1736,19 +1736,20 @@ class CommandXSetId : public Commander { std::optional<uint64_t> entries_added_; }; -REDIS_REGISTER_COMMANDS(MakeCmdAttr<CommandXAck>("xack", -4, "write no-dbsize-check", 1, 1, 1), - MakeCmdAttr<CommandXAdd>("xadd", -5, "write", 1, 1, 1), - MakeCmdAttr<CommandXDel>("xdel", -3, "write no-dbsize-check", 1, 1, 1), - MakeCmdAttr<CommandXClaim>("xclaim", -6, "write", 1, 1, 1), - MakeCmdAttr<CommandAutoClaim>("xautoclaim", -6, "write", 1, 1, 1), - MakeCmdAttr<CommandXGroup>("xgroup", -4, "write", 2, 2, 1), - MakeCmdAttr<CommandXLen>("xlen", -2, "read-only", 1, 1, 1), - MakeCmdAttr<CommandXInfo>("xinfo", -2, "read-only", 0, 0, 0), - MakeCmdAttr<CommandXRange>("xrange", -4, "read-only", 1, 1, 1), - MakeCmdAttr<CommandXRevRange>("xrevrange", -2, "read-only", 1, 1, 1), - MakeCmdAttr<CommandXRead>("xread", -4, "read-only", 0, 0, 0), - MakeCmdAttr<CommandXReadGroup>("xreadgroup", -7, "write", 0, 0, 0), - MakeCmdAttr<CommandXTrim>("xtrim", -4, "write no-dbsize-check", 1, 1, 1), - MakeCmdAttr<CommandXSetId>("xsetid", -3, "write", 1, 1, 1)) +REDIS_REGISTER_COMMANDS( + // MakeCmdAttr<CommandXAck>("xack", -4, "write no-dbsize-check", 1, 1, 1), + MakeCmdAttr<CommandXAdd>("xadd", -5, "write", 1, 1, 1), + MakeCmdAttr<CommandXDel>("xdel", -3, "write no-dbsize-check", 1, 1, 1), + // MakeCmdAttr<CommandXClaim>("xclaim", -6, "write", 1, 1, 1), + // MakeCmdAttr<CommandAutoClaim>("xautoclaim", -6, "write", 1, 1, 1), + // MakeCmdAttr<CommandXGroup>("xgroup", -4, "write", 2, 2, 1), + MakeCmdAttr<CommandXLen>("xlen", -2, "read-only", 1, 1, 1), + MakeCmdAttr<CommandXInfo>("xinfo", -2, "read-only", 0, 0, 0), + MakeCmdAttr<CommandXRange>("xrange", -4, "read-only", 1, 1, 1), + MakeCmdAttr<CommandXRevRange>("xrevrange", -2, "read-only", 1, 1, 1), + MakeCmdAttr<CommandXRead>("xread", -4, "read-only", 0, 0, 0), + // MakeCmdAttr<CommandXReadGroup>("xreadgroup", -7, "write", 0, 0, 0), + MakeCmdAttr<CommandXTrim>("xtrim", -4, "write no-dbsize-check", 1, 1, 1), + MakeCmdAttr<CommandXSetId>("xsetid", -3, "write", 1, 1, 1)) } // namespace redis diff --git a/tests/gocase/unit/search/search_test.go b/tests/gocase/unit/search/search_test.go index bb87c19f..651323c1 100644 --- a/tests/gocase/unit/search/search_test.go +++ b/tests/gocase/unit/search/search_test.go @@ -29,6 +29,8 @@ import ( ) func TestSearch(t *testing.T) { + t.Skip("search commands is disabled") + srv := util.StartServer(t, map[string]string{}) defer srv.Close() diff --git a/tests/gocase/unit/type/stream/stream_test.go b/tests/gocase/unit/type/stream/stream_test.go index b8958e70..4ba461ff 100644 --- a/tests/gocase/unit/type/stream/stream_test.go +++ b/tests/gocase/unit/type/stream/stream_test.go @@ -867,1043 +867,1043 @@ func TestStreamOffset(t *testing.T) { require.EqualValues(t, providedSeqNum, seqNum) }) - t.Run("XGROUP CREATE with different kinds of commands and XGROUP DESTROY", func(t *testing.T) { - streamName := "test-stream-a" - groupName := "test-group-a" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - // No such stream (No such key) - require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$").Err()) - require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "ENTRIESREAD", "10").Err()) - require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "ENTRIESREAD").Err()) - require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "MKSTREAM", "ENTRIESREAD").Err()) - require.NoError(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "MKSTREAM").Err()) - require.NoError(t, rdb.XInfoStream(ctx, streamName).Err()) - require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$").Err()) - // Invalid syntax - groupName = "test-group-b" - require.Error(t, rdb.Do(ctx, "XGROUP", "CREAT", streamName, groupName, "$").Err()) - require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "ENTRIEREAD", "10").Err()) - require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "ENTRIESREAD", "-10").Err()) - require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, "1test-group-c", "$").Err()) - - require.NoError(t, rdb.Del(ctx, "myStream").Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{Stream: "myStream", Values: []string{"iTeM", "1", "vAluE", "a"}}).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, "myStream", "myGroup", "$").Err()) - result, err := rdb.XGroupDestroy(ctx, "myStream", "myGroup").Result() - require.NoError(t, err) - require.Equal(t, int64(1), result) - result, err = rdb.XGroupDestroy(ctx, "myStream", "myGroup").Result() - require.NoError(t, err) - require.Equal(t, int64(0), result) - }) - - t.Run("XGROUP CREATECONSUMER with different kinds of commands", func(t *testing.T) { - streamName := "test-stream" - groupName := "test-group" - consumerName := "test-consumer" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - //No such stream - require.Error(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: []string{"data", "a"}, - }).Err()) - //no such group - require.Error(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "$").Err()) - - r := rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Val() - require.Equal(t, int64(1), r) - r = rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Val() - require.Equal(t, int64(0), r) - }) - - t.Run("XGROUP DELCONSUMER with different kinds of commands", func(t *testing.T) { - streamName := "test-stream" - groupName := "test-group" - consumerName := "test-consumer" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - //No such stream - require.Error(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: []string{"data", "a"}, - }).Err()) - //no such group - require.Error(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "$").Err()) - require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) - - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"data1", "a1"}, - }).Err()) - require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 1, - NoAck: false, - }).Err()) - ri, erri := rdb.XInfoGroups(ctx, streamName).Result() - require.NoError(t, erri) - require.Equal(t, int64(1), ri[0].Consumers) - require.Equal(t, int64(1), ri[0].Pending) - - r, err := rdb.XGroupDelConsumer(ctx, streamName, groupName, consumerName).Result() - require.NoError(t, err) - require.Equal(t, int64(1), r) - ri, erri = rdb.XInfoGroups(ctx, streamName).Result() - require.NoError(t, erri) - require.Equal(t, int64(0), ri[0].Consumers) - require.Equal(t, int64(0), ri[0].Pending) - - require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"data2", "a2"}, - }).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"data3", "a3"}, - }).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"data4", "a4"}, - }).Err()) - require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 3, - NoAck: false, - }).Err()) - ri, erri = rdb.XInfoGroups(ctx, streamName).Result() - require.NoError(t, erri) - require.Equal(t, int64(1), ri[0].Consumers) - require.Equal(t, int64(3), ri[0].Pending) - r, err = rdb.XGroupDelConsumer(ctx, streamName, groupName, consumerName).Result() - require.NoError(t, err) - require.Equal(t, int64(3), r) - ri, erri = rdb.XInfoGroups(ctx, streamName).Result() - require.NoError(t, erri) - require.Equal(t, int64(0), ri[0].Consumers) - require.Equal(t, int64(0), ri[0].Pending) - }) - - t.Run("XGROUP SETID with different kinds of commands", func(t *testing.T) { - streamName := "test-stream" - groupName := "test-group" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - //No such stream - require.Error(t, rdb.XGroupSetID(ctx, streamName, groupName, "$").Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: []string{"data", "a"}, - }).Err()) - //No such group - require.Error(t, rdb.XGroupSetID(ctx, streamName, groupName, "$").Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "$").Err()) - - require.NoError(t, rdb.XGroupSetID(ctx, streamName, groupName, "0-0").Err()) - require.Error(t, rdb.Do(ctx, "xgroup", "setid", streamName, groupName, "$", "entries", "100").Err()) - require.Error(t, rdb.Do(ctx, "xgroup", "setid", streamName, groupName, "$", "entriesread", "-100").Err()) - require.NoError(t, rdb.Do(ctx, "xgroup", "setid", streamName, groupName, "$", "entriesread", "100").Err()) - }) - - t.Run("XINFO GROUPS and XINFO CONSUMERS", func(t *testing.T) { - streamName := "test-stream" - group1 := "t1" - group2 := "t2" - consumer1 := "c1" - consumer2 := "c2" - consumer3 := "c3" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: []string{"data", "a"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, group1, "$").Err()) - r := rdb.XInfoGroups(ctx, streamName).Val() - require.Equal(t, group1, r[0].Name) - require.Equal(t, int64(0), r[0].Consumers) - require.Equal(t, int64(0), r[0].Pending) - require.Equal(t, "1-0", r[0].LastDeliveredID) - require.Equal(t, int64(0), r[0].EntriesRead) - require.Equal(t, int64(0), r[0].Lag) - - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "2-0", - Values: []string{"data1", "b"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, group2, "$").Err()) - r = rdb.XInfoGroups(ctx, streamName).Val() - require.Equal(t, group2, r[1].Name) - require.Equal(t, "2-0", r[1].LastDeliveredID) - - require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, group1, consumer1).Err()) - require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, group1, consumer2).Err()) - require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, group2, consumer3).Err()) - r = rdb.XInfoGroups(ctx, streamName).Val() - require.Equal(t, int64(2), r[0].Consumers) - require.Equal(t, int64(1), r[1].Consumers) - - r1 := rdb.XInfoConsumers(ctx, streamName, group1).Val() - require.Equal(t, consumer1, r1[0].Name) - require.Equal(t, consumer2, r1[1].Name) - r1 = rdb.XInfoConsumers(ctx, streamName, group2).Val() - require.Equal(t, consumer3, r1[0].Name) - }) - - t.Run("XINFO after delete pending message and related consumer, for issue #2350", func(t *testing.T) { - streamName := "test-stream-2350" - groupName := "test-group-2350" - consumerName := "test-consumer-2350" - require.NoError(t, rdb.XGroupCreateMkStream(ctx, streamName, groupName, "$").Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"testing", "overflow"}, - }).Err()) - readRsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 1, - NoAck: false, - }) - require.NoError(t, readRsp.Err()) - require.Len(t, readRsp.Val(), 1) - streamRsp := readRsp.Val()[0] - require.Len(t, streamRsp.Messages, 1) - msgID := streamRsp.Messages[0] - require.NoError(t, rdb.XAck(ctx, streamName, groupName, msgID.ID).Err()) - require.NoError(t, rdb.XGroupDelConsumer(ctx, streamName, groupName, consumerName).Err()) - infoRsp := rdb.XInfoGroups(ctx, streamName) - require.NoError(t, infoRsp.Err()) - infoGroups := infoRsp.Val() - require.Len(t, infoGroups, 1) - infoGroup := infoGroups[0] - require.Equal(t, groupName, infoGroup.Name) - require.Equal(t, int64(0), infoGroup.Consumers) - require.Equal(t, int64(0), infoGroup.Pending) - require.Equal(t, msgID.ID, infoGroup.LastDeliveredID) - }) - - t.Run("XREAD After XGroupCreate and XGroupCreateConsumer, for issue #2109", func(t *testing.T) { - streamName := "test-stream" - group := "group" - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"data1", "b"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, group, "0").Err()) - require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, group, "consumer").Err()) - require.NoError(t, rdb.XRead(ctx, &redis.XReadArgs{ - Streams: []string{streamName, "0"}, - }).Err()) - }) - - t.Run("XREADGROUP with different kinds of commands", func(t *testing.T) { - streamName := "mystream" - groupName := "mygroup" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: []string{"field1", "data1"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) - consumerName := "myconsumer" - r, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 1, - NoAck: false, - }).Result() - require.NoError(t, err) - require.Equal(t, []redis.XStream{{ - Stream: streamName, - Messages: []redis.XMessage{{ID: "1-0", Values: map[string]interface{}{"field1": "data1"}}}, - }}, r) - - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "2-0", - Values: []string{"field2", "data2"}, - }).Err()) - r, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 1, - NoAck: false, - }).Result() - require.NoError(t, err) - require.Equal(t, []redis.XStream{{ - Stream: streamName, - Messages: []redis.XMessage{{ID: "2-0", Values: map[string]interface{}{"field2": "data2"}}}, - }}, r) - - r, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, "0"}, - Count: 2, - NoAck: false, - }).Result() - require.NoError(t, err) - require.Equal(t, []redis.XStream{{ - Stream: streamName, - Messages: []redis.XMessage{{ID: "1-0", Values: map[string]interface{}{"field1": "data1"}}, - {ID: "2-0", Values: map[string]interface{}{"field2": "data2"}}}, - }}, r) - - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "3-0", - Values: []string{"field3", "data3"}, - }).Err()) - r, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 1, - NoAck: true, - }).Result() - require.NoError(t, err) - require.Equal(t, []redis.XStream{{ - Stream: streamName, - Messages: []redis.XMessage{{ID: "3-0", Values: map[string]interface{}{"field3": "data3"}}}, - }}, r) - r, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, "0"}, - Count: 2, - NoAck: false, - }).Result() - require.NoError(t, err) - require.Equal(t, []redis.XStream{{ - Stream: streamName, - Messages: []redis.XMessage{{ID: "1-0", Values: map[string]interface{}{"field1": "data1"}}, - {ID: "2-0", Values: map[string]interface{}{"field2": "data2"}}}, - }}, r) - - c := srv.NewClient() - defer func() { require.NoError(t, c.Close()) }() - ch := make(chan []redis.XStream) - go func() { - ch <- c.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 2, - Block: 10 * time.Second, - NoAck: false, - }).Val() - }() - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "4-0", - Values: []string{"field4", "data4"}, - }).Err()) - r = <-ch - require.Equal(t, []redis.XStream{{ - Stream: streamName, - Messages: []redis.XMessage{{ID: "4-0", Values: map[string]interface{}{"field4": "data4"}}}, - }}, r) - - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "5-0", - Values: []string{"field5", "data5"}, - }).Err()) - require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 1, - NoAck: false, - }).Err()) - require.NoError(t, rdb.XDel(ctx, streamName, "5-0").Err()) - r, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, "5"}, - Count: 1, - NoAck: false, - }).Result() - require.NoError(t, err) - require.Equal(t, []redis.XStream{{ - Stream: streamName, - Messages: []redis.XMessage{{ID: "5-0", Values: map[string]interface{}(nil)}}, - }}, r) - }) - - t.Run("Check xreadgroup fetches the newest data after create consumer in the command", func(t *testing.T) { - streamName := "mystream" - groupName := "mygroup" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: []string{"field1", "data1"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) - consumerName := "myconsumer" - err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 1, - NoAck: false, - }).Err() - require.NoError(t, err) - ri, erri := rdb.XInfoGroups(ctx, streamName).Result() - require.NoError(t, erri) - require.Equal(t, int64(1), ri[0].Consumers) - }) - - t.Run("XACK with different kinds of commands", func(t *testing.T) { - streamName := "mystream" - groupName := "mygroup" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - r, err := rdb.XAck(ctx, streamName, groupName, "0-0").Result() - require.NoError(t, err) - require.Equal(t, int64(0), r) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: []string{"field1", "data1"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) - consumerName := "myconsumer" - err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 1, - NoAck: false, - }).Err() - require.NoError(t, err) - r, err = rdb.XAck(ctx, streamName, groupName, "1-0").Result() - require.NoError(t, err) - require.Equal(t, int64(1), r) - - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "2-0", - Values: []string{"field1", "data1"}, - }).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "3-0", - Values: []string{"field1", "data1"}, - }).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "4-0", - Values: []string{"field1", "data1"}, - }).Err()) - err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 3, - NoAck: false, - }).Err() - require.NoError(t, err) - r, err = rdb.XAck(ctx, streamName, groupName, "2-0", "3-0", "4-0").Result() - require.NoError(t, err) - require.Equal(t, int64(3), r) - }) - - t.Run("Simple XCLAIM command tests", func(t *testing.T) { - streamName := "mystream" - groupName := "mygroup" - consumerName := "myconsumer" - consumer1Name := "myconsumer1" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: []string{"field1", "data1"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) - r, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 1, - NoAck: false, - }).Result() - require.NoError(t, err) - require.Equal(t, []redis.XStream{{ - Stream: streamName, - Messages: []redis.XMessage{{ID: "1-0", Values: map[string]interface{}{"field1": "data1"}}}, - }}, r) - - claimedMessages, err := rdb.XClaim(ctx, &redis.XClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumer1Name, - MinIdle: 0, - Messages: []string{"1-0"}, - }).Result() - require.NoError(t, err) - require.Len(t, claimedMessages, 1, "Expected to claim 1 message") - require.Equal(t, "1-0", claimedMessages[0].ID, "Expected claimed message ID to match") - - time.Sleep(2000 * time.Millisecond) - minIdleTime := 1000 * time.Millisecond - claimedMessages, err = rdb.XClaim(ctx, &redis.XClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumerName, - MinIdle: minIdleTime, - Messages: []string{"1-0"}, - }).Result() - require.NoError(t, err) - require.Len(t, claimedMessages, 1, "Expected to claim 1 message if idle time is large enough") - require.Equal(t, "1-0", claimedMessages[0].ID, "Expected claimed message ID to match") - - minIdleTime = 60000 * time.Millisecond - claimedMessages, err = rdb.XClaim(ctx, &redis.XClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumer1Name, - MinIdle: minIdleTime, - Messages: []string{"1-0"}, - }).Result() - - require.NoError(t, err) - require.Empty(t, claimedMessages, "Expected no messages to be claimed due to insufficient idle time") - }) - - t.Run("XCLAIM with different timing situations and options", func(t *testing.T) { - streamName := "mystream" - groupName := "mygroup" - consumerName := "myconsumer" - consumer1Name := "myconsumer1" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: []string{"field1", "data1"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) - r, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumerName, - Streams: []string{streamName, ">"}, - Count: 1, - NoAck: false, - }).Result() - require.NoError(t, err) - require.Equal(t, []redis.XStream{{ - Stream: streamName, - Messages: []redis.XMessage{{ID: "1-0", Values: map[string]interface{}{"field1": "data1"}}}, - }}, r) - - rawClaimedMessages, err := rdb.Do(ctx, "XCLAIM", streamName, groupName, consumer1Name, "0", "1-0", "IDLE", "5000").Result() - require.NoError(t, err) - messages, ok := rawClaimedMessages.([]interface{}) - require.True(t, ok, "Expected the result to be a slice of interface{}") - firstMsg, ok := messages[0].([]interface{}) - require.True(t, ok, "Expected message details to be a slice of interface{}") - msgID, ok := firstMsg[0].(string) - require.True(t, ok, "Expected message ID to be a string") - require.Equal(t, "1-0", msgID, "Expected claimed message ID to match") - - claimedMessages, err := rdb.XClaim(ctx, &redis.XClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumerName, - MinIdle: 2000 * time.Millisecond, - Messages: []string{"1-0"}, - }).Result() - require.NoError(t, err) - require.Len(t, claimedMessages, 1, "Expected to claim 1 message if idle time is large enough") - require.Equal(t, "1-0", claimedMessages[0].ID, "Expected claimed message ID to match") - - tenSecondsAgo := time.Now().Add(-10 * time.Second).UnixMilli() - rawClaimedMessages, err = rdb.Do(ctx, "XCLAIM", streamName, groupName, consumer1Name, "0", "1-0", "TIME", tenSecondsAgo).Result() - require.NoError(t, err) - messages, ok = rawClaimedMessages.([]interface{}) - require.True(t, ok, "Expected the result to be a slice of interface{}") - firstMsg, ok = messages[0].([]interface{}) - require.True(t, ok, "Expected message details to be a slice of interface{}") - msgID, ok = firstMsg[0].(string) - require.True(t, ok, "Expected message ID to be a string") - require.Equal(t, "1-0", msgID, "Expected claimed message ID to match") - - claimedMessages, err = rdb.XClaim(ctx, &redis.XClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumerName, - MinIdle: 5000 * time.Millisecond, - Messages: []string{"1-0"}, - }).Result() - require.NoError(t, err) - require.Len(t, claimedMessages, 1, "Expected to claim 1 message if idle time is large enough") - require.Equal(t, "1-0", claimedMessages[0].ID, "Expected claimed message ID to match") - }) - - t.Run("XCLAIM command with different options", func(t *testing.T) { - streamName := "mystream" - groupName := "mygroup" - consumerName := "myconsumer" - consumer1Name := "myconsumer1" - - require.NoError(t, rdb.Del(ctx, streamName).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: []string{"field1", "data1"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) - - rawClaimedMessages, err := rdb.Do(ctx, "XCLAIM", streamName, groupName, consumerName, "0", "1-0", "FORCE").Result() - require.NoError(t, err) - messages, ok := rawClaimedMessages.([]interface{}) - require.True(t, ok, "Expected the result to be a slice of interface{}") - firstMsg, ok := messages[0].([]interface{}) - require.True(t, ok, "Expected message details to be a slice of interface{}") - msgID, ok := firstMsg[0].(string) - require.True(t, ok, "Expected message ID to be a string") - require.Equal(t, "1-0", msgID, "Expected claimed message ID to match") - - cmd := rdb.XClaimJustID(ctx, &redis.XClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumer1Name, - MinIdle: 0, - Messages: []string{"1-0"}, - }) - - claimedIDs, err := cmd.Result() - require.NoError(t, err) - require.Len(t, claimedIDs, 1, "Expected to claim exactly one message ID") - require.Equal(t, "1-0", claimedIDs[0], "Expected claimed message ID to match") - }) - - t.Run("XAUTOCLAIM can claim PEL items from another consume", func(t *testing.T) { - - streamName := "mystream" - groupName := "mygroup" - var id1 string - require.NoError(t, rdb.Del(ctx, streamName).Err()) - { - rsp := rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"a", "1"}, - }) - require.NoError(t, rsp.Err()) - id1 = rsp.Val() - } - var id2 string - { - rsp := rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"b", "2"}, - }) - require.NoError(t, rsp.Err()) - id2 = rsp.Val() - } - var id3 string - { - rsp := rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"c", "3"}, - }) - require.NoError(t, rsp.Err()) - id3 = rsp.Val() - } - var id4 string - { - rsp := rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"d", "4"}, - }) - require.NoError(t, rsp.Err()) - id4 = rsp.Val() - } - - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) - - consumer1 := "consumer1" - consumer2 := "consumer2" - { - rsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumer1, - Streams: []string{streamName, ">"}, - Count: 1, - }) - require.NoError(t, rsp.Err()) - require.Len(t, rsp.Val(), 1) - require.Len(t, rsp.Val()[0].Messages, 1) - require.Equal(t, id1, rsp.Val()[0].Messages[0].ID) - require.Len(t, rsp.Val()[0].Messages[0].Values, 1) - require.Equal(t, "1", rsp.Val()[0].Messages[0].Values["a"]) - } - - { - time.Sleep(200 * time.Millisecond) - rsp := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumer2, - MinIdle: 10 * time.Millisecond, - Count: 1, - Start: "-", - }) - require.NoError(t, rsp.Err()) - msgs, start := rsp.Val() - require.Equal(t, "0-0", start) - require.Len(t, msgs, 1) - require.Len(t, msgs[0].Values, 1) - require.Equal(t, "1", msgs[0].Values["a"]) - } - - { - rsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumer1, - Streams: []string{streamName, ">"}, - Count: 3, - }) - require.NoError(t, rsp.Err()) - - time.Sleep(time.Millisecond * 200) - require.NoError(t, rdb.XDel(ctx, streamName, id2).Err()) - } - - { - cmd := rdb.Do(ctx, "XAUTOCLAIM", streamName, groupName, consumer2, 10, "-", "COUNT", 3) - require.NoError(t, cmd.Err()) - require.Equal(t, []interface{}{ - id4, - []interface{}{ - []interface{}{ - id1, - []interface{}{"a", "1"}, - }, - []interface{}{ - id3, - []interface{}{"c", "3"}, - }, - }, - []interface{}{ - id2, - }, - }, cmd.Val()) - } - - { - time.Sleep(time.Millisecond * 200) - require.NoError(t, rdb.XDel(ctx, streamName, id4).Err()) - rsp := rdb.XAutoClaimJustID(ctx, &redis.XAutoClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumer2, - MinIdle: 10 * time.Millisecond, - Start: "-", - }) - require.NoError(t, rsp.Err()) - msgs, start := rsp.Val() - require.Equal(t, "0-0", start) - require.Len(t, msgs, 2) - require.Equal(t, id1, msgs[0]) - require.Equal(t, id3, msgs[1]) - } - }) - - t.Run("XAUTOCLAIM as an iterator", func(t *testing.T) { - streamName := "mystream" - groupName := "mygroup" - var id3, id5 string - require.NoError(t, rdb.Del(ctx, streamName).Err()) - { - rsp := rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"a", "1"}, - }) - require.NoError(t, rsp.Err()) - } - { - rsp := rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"b", "2"}, - }) - require.NoError(t, rsp.Err()) - } - { - rsp := rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"c", "3"}, - }) - require.NoError(t, rsp.Err()) - id3 = rsp.Val() - } - { - rsp := rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"d", "4"}, - }) - require.NoError(t, rsp.Err()) - } - { - rsp := rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "*", - Values: []string{"e", "5"}, - }) - require.NoError(t, rsp.Err()) - id5 = rsp.Val() - } - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) - - consumer1, consumer2 := "consumer1", "consumer2" - { - rsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: consumer1, - Streams: []string{streamName, ">"}, - Count: 90, - }) - require.NoError(t, rsp.Err()) - time.Sleep(200 * time.Millisecond) - } - { - rsp := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumer2, - MinIdle: 10 * time.Millisecond, - Count: 2, - Start: "-", - }) - require.NoError(t, rsp.Err()) - msgs, start := rsp.Val() - require.Equal(t, id3, start) - require.Len(t, msgs, 2) - require.Len(t, msgs[0].Values, 1) - require.Equal(t, "1", msgs[0].Values["a"]) - } - - { - rsp := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumer2, - MinIdle: 10 * time.Millisecond, - Start: id3, - Count: 2, - }) - require.NoError(t, rsp.Err()) - msgs, start := rsp.Val() - require.Equal(t, id5, start) - require.Len(t, msgs, 2) - require.Len(t, msgs[0].Values, 1) - require.Equal(t, "3", msgs[0].Values["c"]) - } - - { - rsp := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ - Stream: streamName, - Group: groupName, - Consumer: consumer2, - MinIdle: 10 * time.Millisecond, - Start: id5, - Count: 1, - }) - require.NoError(t, rsp.Err()) - msgs, start := rsp.Val() - require.Equal(t, "0-0", start) - require.Len(t, msgs, 1) - require.Len(t, msgs[0].Values, 1) - require.Equal(t, "5", msgs[0].Values["e"]) - } - }) - - t.Run("XAUTOCLAIM with XDEL", func(t *testing.T) { - streamName := "x" - groupName := "grp" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: map[string]interface{}{"f": "v"}, - }).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "2-0", - Values: map[string]interface{}{"f": "v"}, - }).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "3-0", - Values: map[string]interface{}{"f": "v"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) - { - rsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: "Alice", - Streams: []string{streamName, ">"}, - }) - require.NoError(t, rsp.Err()) - require.Len(t, rsp.Val(), 1) - require.Len(t, rsp.Val()[0].Messages, 3) - require.Equal(t, "1-0", rsp.Val()[0].Messages[0].ID) - require.Equal(t, "v", rsp.Val()[0].Messages[0].Values["f"]) - require.Equal(t, "2-0", rsp.Val()[0].Messages[1].ID) - require.Equal(t, "v", rsp.Val()[0].Messages[1].Values["f"]) - require.Equal(t, "3-0", rsp.Val()[0].Messages[2].ID) - require.Equal(t, "v", rsp.Val()[0].Messages[2].Values["f"]) - } - { - require.NoError(t, rdb.XDel(ctx, streamName, "2-0").Err()) - cmd := rdb.Do(ctx, "XAUTOCLAIM", streamName, groupName, "Bob", 0, "0-0") - require.NoError(t, cmd.Err()) - require.Equal(t, []interface{}{ - "0-0", - []interface{}{ - []interface{}{ - "1-0", - []interface{}{"f", "v"}, - }, - []interface{}{ - "3-0", - []interface{}{"f", "v"}, - }, - }, - []interface{}{ - "2-0", - }, - }, cmd.Val()) - } - }) - - t.Run("XAUTOCLAIM with XDEL and count", func(t *testing.T) { - streamName := "x" - groupName := "grp" - require.NoError(t, rdb.Del(ctx, streamName).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "1-0", - Values: map[string]interface{}{"f": "v"}, - }).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "2-0", - Values: map[string]interface{}{"f": "v"}, - }).Err()) - require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ - Stream: streamName, - ID: "3-0", - Values: map[string]interface{}{"f": "v"}, - }).Err()) - require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) - { - rsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ - Group: groupName, - Consumer: "Alice", - Streams: []string{streamName, ">"}, - }) - require.NoError(t, rsp.Err()) - require.Len(t, rsp.Val(), 1) - require.Len(t, rsp.Val()[0].Messages, 3) - require.Equal(t, "1-0", rsp.Val()[0].Messages[0].ID) - require.Equal(t, "v", rsp.Val()[0].Messages[0].Values["f"]) - require.Equal(t, "2-0", rsp.Val()[0].Messages[1].ID) - require.Equal(t, "v", rsp.Val()[0].Messages[1].Values["f"]) - require.Equal(t, "3-0", rsp.Val()[0].Messages[2].ID) - require.Equal(t, "v", rsp.Val()[0].Messages[2].Values["f"]) - } - { - require.NoError(t, rdb.XDel(ctx, streamName, "1-0").Err()) - require.NoError(t, rdb.XDel(ctx, streamName, "2-0").Err()) - cmd := rdb.Do(ctx, "XAUTOCLAIM", streamName, groupName, "Bob", 0, "0-0", "COUNT", 1) - require.NoError(t, cmd.Err()) - require.Equal(t, []interface{}{ - "2-0", - []interface{}{}, - []interface{}{ - "1-0", - }, - }, cmd.Val()) - } - { - cmd := rdb.Do(ctx, "XAUTOCLAIM", streamName, groupName, "Bob", 0, "2-0", "COUNT", 1) - require.NoError(t, cmd.Err()) - require.Equal(t, []interface{}{ - "3-0", - []interface{}{}, - []interface{}{ - "2-0", - }, - }, cmd.Val()) - } - { - cmd := rdb.Do(ctx, "XAUTOCLAIM", streamName, groupName, "Bob", 0, "3-0", "COUNT", 1) - require.NoError(t, cmd.Err()) - require.Equal(t, []interface{}{ - "0-0", - []interface{}{ - []interface{}{ - "3-0", - []interface{}{"f", "v"}, - }, - }, - []interface{}{}, - }, cmd.Val()) - } - // assert_equal [XPENDING x grp - + 10 Alice] {} - // add xpending to this test case when it is supported - }) - - t.Run("XAUTOCLAIM with out of range count", func(t *testing.T) { - err := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ - Stream: "x", - Group: "grp", - Consumer: "Bob", - MinIdle: 0, - Start: "3-0", - Count: 8070450532247928833, - }).Err() - require.Error(t, err) - require.True(t, strings.HasPrefix(err.Error(), "ERR COUNT")) - }) - - t.Run("XAUTOCLAIM COUNT must be > 0", func(t *testing.T) { - cmd := rdb.Do(ctx, "XAUTOCLAIM", "key", "group", "consumer", 1, 1, "COUNT", 0) - require.Error(t, cmd.Err()) - require.Equal(t, "ERR COUNT must be > 0", cmd.Err().Error()) - }) + // t.Run("XGROUP CREATE with different kinds of commands and XGROUP DESTROY", func(t *testing.T) { + // streamName := "test-stream-a" + // groupName := "test-group-a" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // // No such stream (No such key) + // require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$").Err()) + // require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "ENTRIESREAD", "10").Err()) + // require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "ENTRIESREAD").Err()) + // require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "MKSTREAM", "ENTRIESREAD").Err()) + // require.NoError(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "MKSTREAM").Err()) + // require.NoError(t, rdb.XInfoStream(ctx, streamName).Err()) + // require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$").Err()) + // // Invalid syntax + // groupName = "test-group-b" + // require.Error(t, rdb.Do(ctx, "XGROUP", "CREAT", streamName, groupName, "$").Err()) + // require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "ENTRIEREAD", "10").Err()) + // require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, groupName, "$", "ENTRIESREAD", "-10").Err()) + // require.Error(t, rdb.Do(ctx, "XGROUP", "CREATE", streamName, "1test-group-c", "$").Err()) + + // require.NoError(t, rdb.Del(ctx, "myStream").Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{Stream: "myStream", Values: []string{"iTeM", "1", "vAluE", "a"}}).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, "myStream", "myGroup", "$").Err()) + // result, err := rdb.XGroupDestroy(ctx, "myStream", "myGroup").Result() + // require.NoError(t, err) + // require.Equal(t, int64(1), result) + // result, err = rdb.XGroupDestroy(ctx, "myStream", "myGroup").Result() + // require.NoError(t, err) + // require.Equal(t, int64(0), result) + // }) + + // t.Run("XGROUP CREATECONSUMER with different kinds of commands", func(t *testing.T) { + // streamName := "test-stream" + // groupName := "test-group" + // consumerName := "test-consumer" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // //No such stream + // require.Error(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: []string{"data", "a"}, + // }).Err()) + // //no such group + // require.Error(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "$").Err()) + + // r := rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Val() + // require.Equal(t, int64(1), r) + // r = rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Val() + // require.Equal(t, int64(0), r) + // }) + + // t.Run("XGROUP DELCONSUMER with different kinds of commands", func(t *testing.T) { + // streamName := "test-stream" + // groupName := "test-group" + // consumerName := "test-consumer" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // //No such stream + // require.Error(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: []string{"data", "a"}, + // }).Err()) + // //no such group + // require.Error(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "$").Err()) + // require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) + + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"data1", "a1"}, + // }).Err()) + // require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 1, + // NoAck: false, + // }).Err()) + // ri, erri := rdb.XInfoGroups(ctx, streamName).Result() + // require.NoError(t, erri) + // require.Equal(t, int64(1), ri[0].Consumers) + // require.Equal(t, int64(1), ri[0].Pending) + + // r, err := rdb.XGroupDelConsumer(ctx, streamName, groupName, consumerName).Result() + // require.NoError(t, err) + // require.Equal(t, int64(1), r) + // ri, erri = rdb.XInfoGroups(ctx, streamName).Result() + // require.NoError(t, erri) + // require.Equal(t, int64(0), ri[0].Consumers) + // require.Equal(t, int64(0), ri[0].Pending) + + // require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, groupName, consumerName).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"data2", "a2"}, + // }).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"data3", "a3"}, + // }).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"data4", "a4"}, + // }).Err()) + // require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 3, + // NoAck: false, + // }).Err()) + // ri, erri = rdb.XInfoGroups(ctx, streamName).Result() + // require.NoError(t, erri) + // require.Equal(t, int64(1), ri[0].Consumers) + // require.Equal(t, int64(3), ri[0].Pending) + // r, err = rdb.XGroupDelConsumer(ctx, streamName, groupName, consumerName).Result() + // require.NoError(t, err) + // require.Equal(t, int64(3), r) + // ri, erri = rdb.XInfoGroups(ctx, streamName).Result() + // require.NoError(t, erri) + // require.Equal(t, int64(0), ri[0].Consumers) + // require.Equal(t, int64(0), ri[0].Pending) + // }) + + // t.Run("XGROUP SETID with different kinds of commands", func(t *testing.T) { + // streamName := "test-stream" + // groupName := "test-group" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // //No such stream + // require.Error(t, rdb.XGroupSetID(ctx, streamName, groupName, "$").Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: []string{"data", "a"}, + // }).Err()) + // //No such group + // require.Error(t, rdb.XGroupSetID(ctx, streamName, groupName, "$").Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "$").Err()) + + // require.NoError(t, rdb.XGroupSetID(ctx, streamName, groupName, "0-0").Err()) + // require.Error(t, rdb.Do(ctx, "xgroup", "setid", streamName, groupName, "$", "entries", "100").Err()) + // require.Error(t, rdb.Do(ctx, "xgroup", "setid", streamName, groupName, "$", "entriesread", "-100").Err()) + // require.NoError(t, rdb.Do(ctx, "xgroup", "setid", streamName, groupName, "$", "entriesread", "100").Err()) + // }) + + // t.Run("XINFO GROUPS and XINFO CONSUMERS", func(t *testing.T) { + // streamName := "test-stream" + // group1 := "t1" + // group2 := "t2" + // consumer1 := "c1" + // consumer2 := "c2" + // consumer3 := "c3" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: []string{"data", "a"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, group1, "$").Err()) + // r := rdb.XInfoGroups(ctx, streamName).Val() + // require.Equal(t, group1, r[0].Name) + // require.Equal(t, int64(0), r[0].Consumers) + // require.Equal(t, int64(0), r[0].Pending) + // require.Equal(t, "1-0", r[0].LastDeliveredID) + // require.Equal(t, int64(0), r[0].EntriesRead) + // require.Equal(t, int64(0), r[0].Lag) + + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "2-0", + // Values: []string{"data1", "b"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, group2, "$").Err()) + // r = rdb.XInfoGroups(ctx, streamName).Val() + // require.Equal(t, group2, r[1].Name) + // require.Equal(t, "2-0", r[1].LastDeliveredID) + + // require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, group1, consumer1).Err()) + // require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, group1, consumer2).Err()) + // require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, group2, consumer3).Err()) + // r = rdb.XInfoGroups(ctx, streamName).Val() + // require.Equal(t, int64(2), r[0].Consumers) + // require.Equal(t, int64(1), r[1].Consumers) + + // r1 := rdb.XInfoConsumers(ctx, streamName, group1).Val() + // require.Equal(t, consumer1, r1[0].Name) + // require.Equal(t, consumer2, r1[1].Name) + // r1 = rdb.XInfoConsumers(ctx, streamName, group2).Val() + // require.Equal(t, consumer3, r1[0].Name) + // }) + + // t.Run("XINFO after delete pending message and related consumer, for issue #2350", func(t *testing.T) { + // streamName := "test-stream-2350" + // groupName := "test-group-2350" + // consumerName := "test-consumer-2350" + // require.NoError(t, rdb.XGroupCreateMkStream(ctx, streamName, groupName, "$").Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"testing", "overflow"}, + // }).Err()) + // readRsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 1, + // NoAck: false, + // }) + // require.NoError(t, readRsp.Err()) + // require.Len(t, readRsp.Val(), 1) + // streamRsp := readRsp.Val()[0] + // require.Len(t, streamRsp.Messages, 1) + // msgID := streamRsp.Messages[0] + // require.NoError(t, rdb.XAck(ctx, streamName, groupName, msgID.ID).Err()) + // require.NoError(t, rdb.XGroupDelConsumer(ctx, streamName, groupName, consumerName).Err()) + // infoRsp := rdb.XInfoGroups(ctx, streamName) + // require.NoError(t, infoRsp.Err()) + // infoGroups := infoRsp.Val() + // require.Len(t, infoGroups, 1) + // infoGroup := infoGroups[0] + // require.Equal(t, groupName, infoGroup.Name) + // require.Equal(t, int64(0), infoGroup.Consumers) + // require.Equal(t, int64(0), infoGroup.Pending) + // require.Equal(t, msgID.ID, infoGroup.LastDeliveredID) + // }) + + // t.Run("XREAD After XGroupCreate and XGroupCreateConsumer, for issue #2109", func(t *testing.T) { + // streamName := "test-stream" + // group := "group" + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"data1", "b"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, group, "0").Err()) + // require.NoError(t, rdb.XGroupCreateConsumer(ctx, streamName, group, "consumer").Err()) + // require.NoError(t, rdb.XRead(ctx, &redis.XReadArgs{ + // Streams: []string{streamName, "0"}, + // }).Err()) + // }) + + // t.Run("XREADGROUP with different kinds of commands", func(t *testing.T) { + // streamName := "mystream" + // groupName := "mygroup" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: []string{"field1", "data1"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) + // consumerName := "myconsumer" + // r, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 1, + // NoAck: false, + // }).Result() + // require.NoError(t, err) + // require.Equal(t, []redis.XStream{{ + // Stream: streamName, + // Messages: []redis.XMessage{{ID: "1-0", Values: map[string]interface{}{"field1": "data1"}}}, + // }}, r) + + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "2-0", + // Values: []string{"field2", "data2"}, + // }).Err()) + // r, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 1, + // NoAck: false, + // }).Result() + // require.NoError(t, err) + // require.Equal(t, []redis.XStream{{ + // Stream: streamName, + // Messages: []redis.XMessage{{ID: "2-0", Values: map[string]interface{}{"field2": "data2"}}}, + // }}, r) + + // r, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, "0"}, + // Count: 2, + // NoAck: false, + // }).Result() + // require.NoError(t, err) + // require.Equal(t, []redis.XStream{{ + // Stream: streamName, + // Messages: []redis.XMessage{{ID: "1-0", Values: map[string]interface{}{"field1": "data1"}}, + // {ID: "2-0", Values: map[string]interface{}{"field2": "data2"}}}, + // }}, r) + + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "3-0", + // Values: []string{"field3", "data3"}, + // }).Err()) + // r, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 1, + // NoAck: true, + // }).Result() + // require.NoError(t, err) + // require.Equal(t, []redis.XStream{{ + // Stream: streamName, + // Messages: []redis.XMessage{{ID: "3-0", Values: map[string]interface{}{"field3": "data3"}}}, + // }}, r) + // r, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, "0"}, + // Count: 2, + // NoAck: false, + // }).Result() + // require.NoError(t, err) + // require.Equal(t, []redis.XStream{{ + // Stream: streamName, + // Messages: []redis.XMessage{{ID: "1-0", Values: map[string]interface{}{"field1": "data1"}}, + // {ID: "2-0", Values: map[string]interface{}{"field2": "data2"}}}, + // }}, r) + + // c := srv.NewClient() + // defer func() { require.NoError(t, c.Close()) }() + // ch := make(chan []redis.XStream) + // go func() { + // ch <- c.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 2, + // Block: 10 * time.Second, + // NoAck: false, + // }).Val() + // }() + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "4-0", + // Values: []string{"field4", "data4"}, + // }).Err()) + // r = <-ch + // require.Equal(t, []redis.XStream{{ + // Stream: streamName, + // Messages: []redis.XMessage{{ID: "4-0", Values: map[string]interface{}{"field4": "data4"}}}, + // }}, r) + + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "5-0", + // Values: []string{"field5", "data5"}, + // }).Err()) + // require.NoError(t, rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 1, + // NoAck: false, + // }).Err()) + // require.NoError(t, rdb.XDel(ctx, streamName, "5-0").Err()) + // r, err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, "5"}, + // Count: 1, + // NoAck: false, + // }).Result() + // require.NoError(t, err) + // require.Equal(t, []redis.XStream{{ + // Stream: streamName, + // Messages: []redis.XMessage{{ID: "5-0", Values: map[string]interface{}(nil)}}, + // }}, r) + // }) + + // t.Run("Check xreadgroup fetches the newest data after create consumer in the command", func(t *testing.T) { + // streamName := "mystream" + // groupName := "mygroup" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: []string{"field1", "data1"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) + // consumerName := "myconsumer" + // err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 1, + // NoAck: false, + // }).Err() + // require.NoError(t, err) + // ri, erri := rdb.XInfoGroups(ctx, streamName).Result() + // require.NoError(t, erri) + // require.Equal(t, int64(1), ri[0].Consumers) + // }) + + // t.Run("XACK with different kinds of commands", func(t *testing.T) { + // streamName := "mystream" + // groupName := "mygroup" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // r, err := rdb.XAck(ctx, streamName, groupName, "0-0").Result() + // require.NoError(t, err) + // require.Equal(t, int64(0), r) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: []string{"field1", "data1"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) + // consumerName := "myconsumer" + // err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 1, + // NoAck: false, + // }).Err() + // require.NoError(t, err) + // r, err = rdb.XAck(ctx, streamName, groupName, "1-0").Result() + // require.NoError(t, err) + // require.Equal(t, int64(1), r) + + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "2-0", + // Values: []string{"field1", "data1"}, + // }).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "3-0", + // Values: []string{"field1", "data1"}, + // }).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "4-0", + // Values: []string{"field1", "data1"}, + // }).Err()) + // err = rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 3, + // NoAck: false, + // }).Err() + // require.NoError(t, err) + // r, err = rdb.XAck(ctx, streamName, groupName, "2-0", "3-0", "4-0").Result() + // require.NoError(t, err) + // require.Equal(t, int64(3), r) + // }) + + // t.Run("Simple XCLAIM command tests", func(t *testing.T) { + // streamName := "mystream" + // groupName := "mygroup" + // consumerName := "myconsumer" + // consumer1Name := "myconsumer1" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: []string{"field1", "data1"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) + // r, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 1, + // NoAck: false, + // }).Result() + // require.NoError(t, err) + // require.Equal(t, []redis.XStream{{ + // Stream: streamName, + // Messages: []redis.XMessage{{ID: "1-0", Values: map[string]interface{}{"field1": "data1"}}}, + // }}, r) + + // claimedMessages, err := rdb.XClaim(ctx, &redis.XClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumer1Name, + // MinIdle: 0, + // Messages: []string{"1-0"}, + // }).Result() + // require.NoError(t, err) + // require.Len(t, claimedMessages, 1, "Expected to claim 1 message") + // require.Equal(t, "1-0", claimedMessages[0].ID, "Expected claimed message ID to match") + + // time.Sleep(2000 * time.Millisecond) + // minIdleTime := 1000 * time.Millisecond + // claimedMessages, err = rdb.XClaim(ctx, &redis.XClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumerName, + // MinIdle: minIdleTime, + // Messages: []string{"1-0"}, + // }).Result() + // require.NoError(t, err) + // require.Len(t, claimedMessages, 1, "Expected to claim 1 message if idle time is large enough") + // require.Equal(t, "1-0", claimedMessages[0].ID, "Expected claimed message ID to match") + + // minIdleTime = 60000 * time.Millisecond + // claimedMessages, err = rdb.XClaim(ctx, &redis.XClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumer1Name, + // MinIdle: minIdleTime, + // Messages: []string{"1-0"}, + // }).Result() + + // require.NoError(t, err) + // require.Empty(t, claimedMessages, "Expected no messages to be claimed due to insufficient idle time") + // }) + + // t.Run("XCLAIM with different timing situations and options", func(t *testing.T) { + // streamName := "mystream" + // groupName := "mygroup" + // consumerName := "myconsumer" + // consumer1Name := "myconsumer1" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: []string{"field1", "data1"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) + // r, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumerName, + // Streams: []string{streamName, ">"}, + // Count: 1, + // NoAck: false, + // }).Result() + // require.NoError(t, err) + // require.Equal(t, []redis.XStream{{ + // Stream: streamName, + // Messages: []redis.XMessage{{ID: "1-0", Values: map[string]interface{}{"field1": "data1"}}}, + // }}, r) + + // rawClaimedMessages, err := rdb.Do(ctx, "XCLAIM", streamName, groupName, consumer1Name, "0", "1-0", "IDLE", "5000").Result() + // require.NoError(t, err) + // messages, ok := rawClaimedMessages.([]interface{}) + // require.True(t, ok, "Expected the result to be a slice of interface{}") + // firstMsg, ok := messages[0].([]interface{}) + // require.True(t, ok, "Expected message details to be a slice of interface{}") + // msgID, ok := firstMsg[0].(string) + // require.True(t, ok, "Expected message ID to be a string") + // require.Equal(t, "1-0", msgID, "Expected claimed message ID to match") + + // claimedMessages, err := rdb.XClaim(ctx, &redis.XClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumerName, + // MinIdle: 2000 * time.Millisecond, + // Messages: []string{"1-0"}, + // }).Result() + // require.NoError(t, err) + // require.Len(t, claimedMessages, 1, "Expected to claim 1 message if idle time is large enough") + // require.Equal(t, "1-0", claimedMessages[0].ID, "Expected claimed message ID to match") + + // tenSecondsAgo := time.Now().Add(-10 * time.Second).UnixMilli() + // rawClaimedMessages, err = rdb.Do(ctx, "XCLAIM", streamName, groupName, consumer1Name, "0", "1-0", "TIME", tenSecondsAgo).Result() + // require.NoError(t, err) + // messages, ok = rawClaimedMessages.([]interface{}) + // require.True(t, ok, "Expected the result to be a slice of interface{}") + // firstMsg, ok = messages[0].([]interface{}) + // require.True(t, ok, "Expected message details to be a slice of interface{}") + // msgID, ok = firstMsg[0].(string) + // require.True(t, ok, "Expected message ID to be a string") + // require.Equal(t, "1-0", msgID, "Expected claimed message ID to match") + + // claimedMessages, err = rdb.XClaim(ctx, &redis.XClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumerName, + // MinIdle: 5000 * time.Millisecond, + // Messages: []string{"1-0"}, + // }).Result() + // require.NoError(t, err) + // require.Len(t, claimedMessages, 1, "Expected to claim 1 message if idle time is large enough") + // require.Equal(t, "1-0", claimedMessages[0].ID, "Expected claimed message ID to match") + // }) + + // t.Run("XCLAIM command with different options", func(t *testing.T) { + // streamName := "mystream" + // groupName := "mygroup" + // consumerName := "myconsumer" + // consumer1Name := "myconsumer1" + + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: []string{"field1", "data1"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) + + // rawClaimedMessages, err := rdb.Do(ctx, "XCLAIM", streamName, groupName, consumerName, "0", "1-0", "FORCE").Result() + // require.NoError(t, err) + // messages, ok := rawClaimedMessages.([]interface{}) + // require.True(t, ok, "Expected the result to be a slice of interface{}") + // firstMsg, ok := messages[0].([]interface{}) + // require.True(t, ok, "Expected message details to be a slice of interface{}") + // msgID, ok := firstMsg[0].(string) + // require.True(t, ok, "Expected message ID to be a string") + // require.Equal(t, "1-0", msgID, "Expected claimed message ID to match") + + // cmd := rdb.XClaimJustID(ctx, &redis.XClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumer1Name, + // MinIdle: 0, + // Messages: []string{"1-0"}, + // }) + + // claimedIDs, err := cmd.Result() + // require.NoError(t, err) + // require.Len(t, claimedIDs, 1, "Expected to claim exactly one message ID") + // require.Equal(t, "1-0", claimedIDs[0], "Expected claimed message ID to match") + // }) + + // t.Run("XAUTOCLAIM can claim PEL items from another consume", func(t *testing.T) { + + // streamName := "mystream" + // groupName := "mygroup" + // var id1 string + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // { + // rsp := rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"a", "1"}, + // }) + // require.NoError(t, rsp.Err()) + // id1 = rsp.Val() + // } + // var id2 string + // { + // rsp := rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"b", "2"}, + // }) + // require.NoError(t, rsp.Err()) + // id2 = rsp.Val() + // } + // var id3 string + // { + // rsp := rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"c", "3"}, + // }) + // require.NoError(t, rsp.Err()) + // id3 = rsp.Val() + // } + // var id4 string + // { + // rsp := rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"d", "4"}, + // }) + // require.NoError(t, rsp.Err()) + // id4 = rsp.Val() + // } + + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) + + // consumer1 := "consumer1" + // consumer2 := "consumer2" + // { + // rsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumer1, + // Streams: []string{streamName, ">"}, + // Count: 1, + // }) + // require.NoError(t, rsp.Err()) + // require.Len(t, rsp.Val(), 1) + // require.Len(t, rsp.Val()[0].Messages, 1) + // require.Equal(t, id1, rsp.Val()[0].Messages[0].ID) + // require.Len(t, rsp.Val()[0].Messages[0].Values, 1) + // require.Equal(t, "1", rsp.Val()[0].Messages[0].Values["a"]) + // } + + // { + // time.Sleep(200 * time.Millisecond) + // rsp := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumer2, + // MinIdle: 10 * time.Millisecond, + // Count: 1, + // Start: "-", + // }) + // require.NoError(t, rsp.Err()) + // msgs, start := rsp.Val() + // require.Equal(t, "0-0", start) + // require.Len(t, msgs, 1) + // require.Len(t, msgs[0].Values, 1) + // require.Equal(t, "1", msgs[0].Values["a"]) + // } + + // { + // rsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumer1, + // Streams: []string{streamName, ">"}, + // Count: 3, + // }) + // require.NoError(t, rsp.Err()) + + // time.Sleep(time.Millisecond * 200) + // require.NoError(t, rdb.XDel(ctx, streamName, id2).Err()) + // } + + // { + // cmd := rdb.Do(ctx, "XAUTOCLAIM", streamName, groupName, consumer2, 10, "-", "COUNT", 3) + // require.NoError(t, cmd.Err()) + // require.Equal(t, []interface{}{ + // id4, + // []interface{}{ + // []interface{}{ + // id1, + // []interface{}{"a", "1"}, + // }, + // []interface{}{ + // id3, + // []interface{}{"c", "3"}, + // }, + // }, + // []interface{}{ + // id2, + // }, + // }, cmd.Val()) + // } + + // { + // time.Sleep(time.Millisecond * 200) + // require.NoError(t, rdb.XDel(ctx, streamName, id4).Err()) + // rsp := rdb.XAutoClaimJustID(ctx, &redis.XAutoClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumer2, + // MinIdle: 10 * time.Millisecond, + // Start: "-", + // }) + // require.NoError(t, rsp.Err()) + // msgs, start := rsp.Val() + // require.Equal(t, "0-0", start) + // require.Len(t, msgs, 2) + // require.Equal(t, id1, msgs[0]) + // require.Equal(t, id3, msgs[1]) + // } + // }) + + // t.Run("XAUTOCLAIM as an iterator", func(t *testing.T) { + // streamName := "mystream" + // groupName := "mygroup" + // var id3, id5 string + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // { + // rsp := rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"a", "1"}, + // }) + // require.NoError(t, rsp.Err()) + // } + // { + // rsp := rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"b", "2"}, + // }) + // require.NoError(t, rsp.Err()) + // } + // { + // rsp := rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"c", "3"}, + // }) + // require.NoError(t, rsp.Err()) + // id3 = rsp.Val() + // } + // { + // rsp := rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"d", "4"}, + // }) + // require.NoError(t, rsp.Err()) + // } + // { + // rsp := rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "*", + // Values: []string{"e", "5"}, + // }) + // require.NoError(t, rsp.Err()) + // id5 = rsp.Val() + // } + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) + + // consumer1, consumer2 := "consumer1", "consumer2" + // { + // rsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: consumer1, + // Streams: []string{streamName, ">"}, + // Count: 90, + // }) + // require.NoError(t, rsp.Err()) + // time.Sleep(200 * time.Millisecond) + // } + // { + // rsp := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumer2, + // MinIdle: 10 * time.Millisecond, + // Count: 2, + // Start: "-", + // }) + // require.NoError(t, rsp.Err()) + // msgs, start := rsp.Val() + // require.Equal(t, id3, start) + // require.Len(t, msgs, 2) + // require.Len(t, msgs[0].Values, 1) + // require.Equal(t, "1", msgs[0].Values["a"]) + // } + + // { + // rsp := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumer2, + // MinIdle: 10 * time.Millisecond, + // Start: id3, + // Count: 2, + // }) + // require.NoError(t, rsp.Err()) + // msgs, start := rsp.Val() + // require.Equal(t, id5, start) + // require.Len(t, msgs, 2) + // require.Len(t, msgs[0].Values, 1) + // require.Equal(t, "3", msgs[0].Values["c"]) + // } + + // { + // rsp := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ + // Stream: streamName, + // Group: groupName, + // Consumer: consumer2, + // MinIdle: 10 * time.Millisecond, + // Start: id5, + // Count: 1, + // }) + // require.NoError(t, rsp.Err()) + // msgs, start := rsp.Val() + // require.Equal(t, "0-0", start) + // require.Len(t, msgs, 1) + // require.Len(t, msgs[0].Values, 1) + // require.Equal(t, "5", msgs[0].Values["e"]) + // } + // }) + + // t.Run("XAUTOCLAIM with XDEL", func(t *testing.T) { + // streamName := "x" + // groupName := "grp" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: map[string]interface{}{"f": "v"}, + // }).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "2-0", + // Values: map[string]interface{}{"f": "v"}, + // }).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "3-0", + // Values: map[string]interface{}{"f": "v"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) + // { + // rsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: "Alice", + // Streams: []string{streamName, ">"}, + // }) + // require.NoError(t, rsp.Err()) + // require.Len(t, rsp.Val(), 1) + // require.Len(t, rsp.Val()[0].Messages, 3) + // require.Equal(t, "1-0", rsp.Val()[0].Messages[0].ID) + // require.Equal(t, "v", rsp.Val()[0].Messages[0].Values["f"]) + // require.Equal(t, "2-0", rsp.Val()[0].Messages[1].ID) + // require.Equal(t, "v", rsp.Val()[0].Messages[1].Values["f"]) + // require.Equal(t, "3-0", rsp.Val()[0].Messages[2].ID) + // require.Equal(t, "v", rsp.Val()[0].Messages[2].Values["f"]) + // } + // { + // require.NoError(t, rdb.XDel(ctx, streamName, "2-0").Err()) + // cmd := rdb.Do(ctx, "XAUTOCLAIM", streamName, groupName, "Bob", 0, "0-0") + // require.NoError(t, cmd.Err()) + // require.Equal(t, []interface{}{ + // "0-0", + // []interface{}{ + // []interface{}{ + // "1-0", + // []interface{}{"f", "v"}, + // }, + // []interface{}{ + // "3-0", + // []interface{}{"f", "v"}, + // }, + // }, + // []interface{}{ + // "2-0", + // }, + // }, cmd.Val()) + // } + // }) + + // t.Run("XAUTOCLAIM with XDEL and count", func(t *testing.T) { + // streamName := "x" + // groupName := "grp" + // require.NoError(t, rdb.Del(ctx, streamName).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "1-0", + // Values: map[string]interface{}{"f": "v"}, + // }).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "2-0", + // Values: map[string]interface{}{"f": "v"}, + // }).Err()) + // require.NoError(t, rdb.XAdd(ctx, &redis.XAddArgs{ + // Stream: streamName, + // ID: "3-0", + // Values: map[string]interface{}{"f": "v"}, + // }).Err()) + // require.NoError(t, rdb.XGroupCreate(ctx, streamName, groupName, "0").Err()) + // { + // rsp := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{ + // Group: groupName, + // Consumer: "Alice", + // Streams: []string{streamName, ">"}, + // }) + // require.NoError(t, rsp.Err()) + // require.Len(t, rsp.Val(), 1) + // require.Len(t, rsp.Val()[0].Messages, 3) + // require.Equal(t, "1-0", rsp.Val()[0].Messages[0].ID) + // require.Equal(t, "v", rsp.Val()[0].Messages[0].Values["f"]) + // require.Equal(t, "2-0", rsp.Val()[0].Messages[1].ID) + // require.Equal(t, "v", rsp.Val()[0].Messages[1].Values["f"]) + // require.Equal(t, "3-0", rsp.Val()[0].Messages[2].ID) + // require.Equal(t, "v", rsp.Val()[0].Messages[2].Values["f"]) + // } + // { + // require.NoError(t, rdb.XDel(ctx, streamName, "1-0").Err()) + // require.NoError(t, rdb.XDel(ctx, streamName, "2-0").Err()) + // cmd := rdb.Do(ctx, "XAUTOCLAIM", streamName, groupName, "Bob", 0, "0-0", "COUNT", 1) + // require.NoError(t, cmd.Err()) + // require.Equal(t, []interface{}{ + // "2-0", + // []interface{}{}, + // []interface{}{ + // "1-0", + // }, + // }, cmd.Val()) + // } + // { + // cmd := rdb.Do(ctx, "XAUTOCLAIM", streamName, groupName, "Bob", 0, "2-0", "COUNT", 1) + // require.NoError(t, cmd.Err()) + // require.Equal(t, []interface{}{ + // "3-0", + // []interface{}{}, + // []interface{}{ + // "2-0", + // }, + // }, cmd.Val()) + // } + // { + // cmd := rdb.Do(ctx, "XAUTOCLAIM", streamName, groupName, "Bob", 0, "3-0", "COUNT", 1) + // require.NoError(t, cmd.Err()) + // require.Equal(t, []interface{}{ + // "0-0", + // []interface{}{ + // []interface{}{ + // "3-0", + // []interface{}{"f", "v"}, + // }, + // }, + // []interface{}{}, + // }, cmd.Val()) + // } + // // assert_equal [XPENDING x grp - + 10 Alice] {} + // // add xpending to this test case when it is supported + // }) + + // t.Run("XAUTOCLAIM with out of range count", func(t *testing.T) { + // err := rdb.XAutoClaim(ctx, &redis.XAutoClaimArgs{ + // Stream: "x", + // Group: "grp", + // Consumer: "Bob", + // MinIdle: 0, + // Start: "3-0", + // Count: 8070450532247928833, + // }).Err() + // require.Error(t, err) + // require.True(t, strings.HasPrefix(err.Error(), "ERR COUNT")) + // }) + + // t.Run("XAUTOCLAIM COUNT must be > 0", func(t *testing.T) { + // cmd := rdb.Do(ctx, "XAUTOCLAIM", "key", "group", "consumer", 1, 1, "COUNT", 0) + // require.Error(t, cmd.Err()) + // require.Equal(t, "ERR COUNT must be > 0", cmd.Err().Error()) + // }) } func parseStreamEntryID(id string) (ts int64, seqNum int64) {
