redis/go-redis
1
0
Fork 0
mirror of https://github.com/redis/go-redis.git synced 2025-09-08 19:52:07 +03:00
Code Activity

added new stream commands

Browse Source
This commit is contained in:
Hristo Temelski
2025-07-28 18:08:43 +03:00
parent 23a87a2137
commit fbf892951b
2 changed files with 128 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
commands_test.go
View File

@@ -6169,6 +6169,34 @@ var _ = Describe("Commands", func() {
Expect(n).To(Equal(int64(3)))
})
It("should XTrimMaxLenMode", func() {
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
n, err := client.XTrimMaxLenMode(ctx, "stream", 0, "KEEPREF").Result()
Expect(err).NotTo(HaveOccurred())
Expect(n).To(BeNumerically(">=", 0))
})
It("should XTrimMaxLenApproxMode", func() {
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
n, err := client.XTrimMaxLenApproxMode(ctx, "stream", 0, 0, "KEEPREF").Result()
Expect(err).NotTo(HaveOccurred())
Expect(n).To(BeNumerically(">=", 0))
})
It("should XTrimMinIDMode", func() {
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
n, err := client.XTrimMinIDMode(ctx, "stream", "4-0", "KEEPREF").Result()
Expect(err).NotTo(HaveOccurred())
Expect(n).To(BeNumerically(">=", 0))
})
It("should XTrimMinIDApproxMode", func() {
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
n, err := client.XTrimMinIDApproxMode(ctx, "stream", "4-0", 0, "KEEPREF").Result()
Expect(err).NotTo(HaveOccurred())
Expect(n).To(BeNumerically(">=", 0))
})
It("should XAdd", func() {
id, err := client.XAdd(ctx, &redis.XAddArgs{
Stream: "stream",
@@ -6222,6 +6250,37 @@ var _ = Describe("Commands", func() {
Expect(n).To(Equal(int64(3)))
})
It("should XAckDel", func() {
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
// First, create a consumer group
err := client.XGroupCreate(ctx, "stream", "testgroup", "0").Err()
Expect(err).NotTo(HaveOccurred())
// Read messages to create pending entries
_, err = client.XReadGroup(ctx, &redis.XReadGroupArgs{
Group: "testgroup",
Consumer: "testconsumer",
Streams: []string{"stream", ">"},
}).Result()
Expect(err).NotTo(HaveOccurred())
// Test XAckDel with KEEPREF mode
n, err := client.XAckDel(ctx, "stream", "testgroup", "KEEPREF", "1-0", "2-0").Result()
Expect(err).NotTo(HaveOccurred())
Expect(n).To(BeNumerically(">=", 0))
// Clean up
client.XGroupDestroy(ctx, "stream", "testgroup")
})
It("should XDelEx", func() {
SkipBeforeRedisVersion(8.2, "doesn't work with older redis stack images")
// Test XDelEx with KEEPREF mode
n, err := client.XDelEx(ctx, "stream", "KEEPREF", "1-0", "2-0").Result()
Expect(err).NotTo(HaveOccurred())
Expect(n).To(BeNumerically(">=", 0))
})
It("should XLen", func() {
n, err := client.XLen(ctx, "stream").Result()
Expect(err).NotTo(HaveOccurred())