sync master to v9 (#1760)

* Added missing idle args in XPendingExtArgs (#1750)

Added missing idle args in XPendingExtArgs

* fix #1754 (#1756)

* Replace go-pg with bun

* fix #1755

Signed-off-by: monkey <golang@88.com>

* fix read data

Signed-off-by: monkey <golang@88.com>

* fix #1758 (#1759)

fix #1758

Co-authored-by: Parvez <syedparvez72@gmail.com>
Co-authored-by: Vladimir Mihailenco <vladimir.webdev@gmail.com>

.github/workflows/build.yml

@@ -2,7 +2,7 @@ name: Go
 
 on:
   push:
-    branches: [master]
+    branches: [master, v9]
   pull_request:
     branches: [master, v9]
 

.github/workflows/golangci-lint.yml

@@ -7,6 +7,7 @@ on:
     branches:
       - master
       - main
+      - v9
   pull_request:
 
 jobs:

command.go

@@ -1520,7 +1520,7 @@ type XInfoConsumer struct {
 	Idle    int64
 }
 
-var _ Cmder = (*XInfoGroupsCmd)(nil)
+var _ Cmder = (*XInfoConsumersCmd)(nil)
 
 func NewXInfoConsumersCmd(ctx context.Context, stream string, group string) *XInfoConsumersCmd {
 	return &XInfoConsumersCmd{
@@ -1722,8 +1722,14 @@ func (cmd *XInfoStreamCmd) readReply(rd *proto.Reader) error {
 			cmd.val.LastGeneratedID, err = rd.ReadString()
 		case "first-entry":
 			cmd.val.FirstEntry, err = readXMessage(rd)
+			if err == Nil {
+				err = nil
+			}
 		case "last-entry":
 			cmd.val.LastEntry, err = readXMessage(rd)
+			if err == Nil {
+				err = nil
+			}
 		default:
 			return fmt.Errorf("redis: unexpected content %s "+
 				"in XINFO STREAM reply", key)
@@ -1737,6 +1743,277 @@ func (cmd *XInfoStreamCmd) readReply(rd *proto.Reader) error {
 
 //------------------------------------------------------------------------------
 
+type XInfoStreamFullCmd struct {
+	baseCmd
+	val *XInfoStreamFull
+}
+
+type XInfoStreamFull struct {
+	Length          int64
+	RadixTreeKeys   int64
+	RadixTreeNodes  int64
+	LastGeneratedID string
+	Entries         []XMessage
+	Groups          []XInfoStreamGroup
+}
+
+type XInfoStreamGroup struct {
+	Name            string
+	LastDeliveredID string
+	PelCount        int64
+	Pending         []XInfoStreamGroupPending
+	Consumers       []XInfoStreamConsumer
+}
+
+type XInfoStreamGroupPending struct {
+	ID            string
+	Consumer      string
+	DeliveryTime  time.Time
+	DeliveryCount int64
+}
+
+type XInfoStreamConsumer struct {
+	Name     string
+	SeenTime time.Time
+	PelCount int64
+	Pending  []XInfoStreamConsumerPending
+}
+
+type XInfoStreamConsumerPending struct {
+	ID            string
+	DeliveryTime  time.Time
+	DeliveryCount int64
+}
+
+var _ Cmder = (*XInfoStreamFullCmd)(nil)
+
+func NewXInfoStreamFullCmd(ctx context.Context, args ...interface{}) *XInfoStreamFullCmd {
+	return &XInfoStreamFullCmd{
+		baseCmd: baseCmd{
+			ctx:  ctx,
+			args: args,
+		},
+	}
+}
+
+func (cmd *XInfoStreamFullCmd) Val() *XInfoStreamFull {
+	return cmd.val
+}
+
+func (cmd *XInfoStreamFullCmd) Result() (*XInfoStreamFull, error) {
+	return cmd.val, cmd.err
+}
+
+func (cmd *XInfoStreamFullCmd) String() string {
+	return cmdString(cmd, cmd.val)
+}
+
+func (cmd *XInfoStreamFullCmd) readReply(rd *proto.Reader) error {
+	if err := rd.ReadFixedMapLen(6); err != nil {
+		return err
+	}
+
+	cmd.val = &XInfoStreamFull{}
+
+	for i := 0; i < 6; i++ {
+		key, err := rd.ReadString()
+		if err != nil {
+			return err
+		}
+
+		switch key {
+		case "length":
+			cmd.val.Length, err = rd.ReadInt()
+		case "radix-tree-keys":
+			cmd.val.RadixTreeKeys, err = rd.ReadInt()
+		case "radix-tree-nodes":
+			cmd.val.RadixTreeNodes, err = rd.ReadInt()
+		case "last-generated-id":
+			cmd.val.LastGeneratedID, err = rd.ReadString()
+		case "entries":
+			cmd.val.Entries, err = readXMessageSlice(rd)
+		case "groups":
+			cmd.val.Groups, err = readStreamGroups(rd)
+		default:
+			return fmt.Errorf("redis: unexpected content %s "+
+				"in XINFO STREAM FULL reply", key)
+		}
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func readStreamGroups(rd *proto.Reader) ([]XInfoStreamGroup, error) {
+	n, err := rd.ReadArrayLen()
+	if err != nil {
+		return nil, err
+	}
+	groups := make([]XInfoStreamGroup, 0, n)
+	for i := 0; i < n; i++ {
+		if err = rd.ReadFixedMapLen(5); err != nil {
+			return nil, err
+		}
+
+		group := XInfoStreamGroup{}
+
+		for f := 0; f < 5; f++ {
+			key, err := rd.ReadString()
+			if err != nil {
+				return nil, err
+			}
+
+			switch key {
+			case "name":
+				group.Name, err = rd.ReadString()
+			case "last-delivered-id":
+				group.LastDeliveredID, err = rd.ReadString()
+			case "pel-count":
+				group.PelCount, err = rd.ReadInt()
+			case "pending":
+				group.Pending, err = readXInfoStreamGroupPending(rd)
+			case "consumers":
+				group.Consumers, err = readXInfoStreamConsumers(rd)
+			default:
+				return nil, fmt.Errorf("redis: unexpected content %s "+
+					"in XINFO STREAM FULL reply", key)
+			}
+
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		groups = append(groups, group)
+	}
+
+	return groups, nil
+}
+
+func readXInfoStreamGroupPending(rd *proto.Reader) ([]XInfoStreamGroupPending, error) {
+	n, err := rd.ReadArrayLen()
+	if err != nil {
+		return nil, err
+	}
+
+	pending := make([]XInfoStreamGroupPending, 0, n)
+
+	for i := 0; i < n; i++ {
+		if err = rd.ReadFixedArrayLen(4); err != nil {
+			return nil, err
+		}
+
+		p := XInfoStreamGroupPending{}
+
+		p.ID, err = rd.ReadString()
+		if err != nil {
+			return nil, err
+		}
+
+		p.Consumer, err = rd.ReadString()
+		if err != nil {
+			return nil, err
+		}
+
+		delivery, err := rd.ReadInt()
+		if err != nil {
+			return nil, err
+		}
+		p.DeliveryTime = time.Unix(delivery/1000, delivery%1000*int64(time.Millisecond))
+
+		p.DeliveryCount, err = rd.ReadInt()
+		if err != nil {
+			return nil, err
+		}
+
+		pending = append(pending, p)
+	}
+
+	return pending, nil
+}
+
+func readXInfoStreamConsumers(rd *proto.Reader) ([]XInfoStreamConsumer, error) {
+	n, err := rd.ReadArrayLen()
+	if err != nil {
+		return nil, err
+	}
+
+	consumers := make([]XInfoStreamConsumer, 0, n)
+
+	for i := 0; i < n; i++ {
+		if err = rd.ReadFixedMapLen(4); err != nil {
+			return nil, err
+		}
+
+		c := XInfoStreamConsumer{}
+
+		for f := 0; f < 4; f++ {
+			cKey, err := rd.ReadString()
+			if err != nil {
+				return nil, err
+			}
+
+			switch cKey {
+			case "name":
+				c.Name, err = rd.ReadString()
+			case "seen-time":
+				seen, err := rd.ReadInt()
+				if err != nil {
+					return nil, err
+				}
+				c.SeenTime = time.Unix(seen/1000, seen%1000*int64(time.Millisecond))
+			case "pel-count":
+				c.PelCount, err = rd.ReadInt()
+			case "pending":
+				pendingNumber, err := rd.ReadArrayLen()
+				if err != nil {
+					return nil, err
+				}
+
+				c.Pending = make([]XInfoStreamConsumerPending, 0, pendingNumber)
+
+				for pn := 0; pn < pendingNumber; pn++ {
+					if err = rd.ReadFixedArrayLen(3); err != nil {
+						return nil, err
+					}
+
+					p := XInfoStreamConsumerPending{}
+
+					p.ID, err = rd.ReadString()
+					if err != nil {
+						return nil, err
+					}
+
+					delivery, err := rd.ReadInt()
+					if err != nil {
+						return nil, err
+					}
+					p.DeliveryTime = time.Unix(delivery/1000, delivery%1000*int64(time.Millisecond))
+
+					p.DeliveryCount, err = rd.ReadInt()
+					if err != nil {
+						return nil, err
+					}
+
+					c.Pending = append(c.Pending, p)
+				}
+			default:
+				return nil, fmt.Errorf("redis: unexpected content %s "+
+					"in XINFO STREAM FULL reply", cKey)
+			}
+			if err != nil {
+				return nil, err
+			}
+		}
+		consumers = append(consumers, c)
+	}
+
+	return consumers, nil
+}
+
+//------------------------------------------------------------------------------
+
 type ZSliceCmd struct {
 	baseCmd
 

commands.go

@@ -180,6 +180,7 @@ type Cmdable interface {
 	LInsertAfter(ctx context.Context, key string, pivot, value interface{}) *IntCmd
 	LLen(ctx context.Context, key string) *IntCmd
 	LPop(ctx context.Context, key string) *StringCmd
+	LPopCount(ctx context.Context, key string, count int) *StringSliceCmd
 	LPos(ctx context.Context, key string, value string, args LPosArgs) *IntCmd
 	LPosCount(ctx context.Context, key string, value string, count int64, args LPosArgs) *IntSliceCmd
 	LPush(ctx context.Context, key string, values ...interface{}) *IntCmd
@@ -1336,6 +1337,12 @@ func (c cmdable) LPop(ctx context.Context, key string) *StringCmd {
 	return cmd
 }
 
+func (c cmdable) LPopCount(ctx context.Context, key string, count int) *StringSliceCmd {
+	cmd := NewStringSliceCmd(ctx, "lpop", key, count)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
 type LPosArgs struct {
 	Rank, MaxLen int64
 }
@@ -1833,6 +1840,7 @@ func (c cmdable) XPending(ctx context.Context, stream, group string) *XPendingCm
 type XPendingExtArgs struct {
 	Stream   string
 	Group    string
+	Idle     time.Duration
 	Start    string
 	End      string
 	Count    int64
@@ -1840,8 +1848,12 @@ type XPendingExtArgs struct {
 }
 
 func (c cmdable) XPendingExt(ctx context.Context, a *XPendingExtArgs) *XPendingExtCmd {
-	args := make([]interface{}, 0, 7)
+	args := make([]interface{}, 0, 9)
-	args = append(args, "xpending", a.Stream, a.Group, a.Start, a.End, a.Count)
+	args = append(args, "xpending", a.Stream, a.Group)
+	if a.Idle != 0 {
+		args = append(args, "idle", formatMs(ctx, a.Idle))
+	}
+	args = append(args, a.Start, a.End, a.Count)
 	if a.Consumer != "" {
 		args = append(args, a.Consumer)
 	}
@@ -1916,6 +1928,19 @@ func (c cmdable) XInfoStream(ctx context.Context, key string) *XInfoStreamCmd {
 	return cmd
 }
 
+// XInfoStreamFull XINFO STREAM FULL [COUNT count]
+// redis-server >= 6.0.
+func (c cmdable) XInfoStreamFull(ctx context.Context, key string, count int) *XInfoStreamFullCmd {
+	args := make([]interface{}, 0, 6)
+	args = append(args, "xinfo", "stream", key, "full")
+	if count > 0 {
+		args = append(args, "count", count)
+	}
+	cmd := NewXInfoStreamFullCmd(ctx, args...)
+	_ = c(ctx, cmd)
+	return cmd
+}
+
 //------------------------------------------------------------------------------
 
 // Z represents sorted set member.

commands_test.go

@@ -2064,6 +2064,25 @@ var _ = Describe("Commands", func() {
 			Expect(lRange.Val()).To(Equal([]string{"two", "three"}))
 		})
 
+		It("should LPopCount", func() {
+			rPush := client.RPush(ctx, "list", "one")
+			Expect(rPush.Err()).NotTo(HaveOccurred())
+			rPush = client.RPush(ctx, "list", "two")
+			Expect(rPush.Err()).NotTo(HaveOccurred())
+			rPush = client.RPush(ctx, "list", "three")
+			Expect(rPush.Err()).NotTo(HaveOccurred())
+			rPush = client.RPush(ctx, "list", "four")
+			Expect(rPush.Err()).NotTo(HaveOccurred())
+
+			lPopCount := client.LPopCount(ctx, "list", 2)
+			Expect(lPopCount.Err()).NotTo(HaveOccurred())
+			Expect(lPopCount.Val()).To(Equal([]string{"one", "two"}))
+
+			lRange := client.LRange(ctx, "list", 0, -1)
+			Expect(lRange.Err()).NotTo(HaveOccurred())
+			Expect(lRange.Val()).To(Equal([]string{"three", "four"}))
+		})
+
 		It("should LPos", func() {
 			rPush := client.RPush(ctx, "list", "a")
 			Expect(rPush.Err()).NotTo(HaveOccurred())
@@ -4241,15 +4260,15 @@ var _ = Describe("Commands", func() {
 					Higher:    "3-0",
 					Consumers: map[string]int64{"consumer": 3},
 				}))
-
+				args := &redis.XPendingExtArgs{
-				infoExt, err := client.XPendingExt(ctx, &redis.XPendingExtArgs{
 					Stream:   "stream",
 					Group:    "group",
 					Start:    "-",
 					End:      "+",
 					Count:    10,
 					Consumer: "consumer",
-				}).Result()
+				}
+				infoExt, err := client.XPendingExt(ctx, args).Result()
 				Expect(err).NotTo(HaveOccurred())
 				for i := range infoExt {
 					infoExt[i].Idle = 0
@@ -4260,6 +4279,11 @@ var _ = Describe("Commands", func() {
 					{ID: "3-0", Consumer: "consumer", Idle: 0, RetryCount: 1},
 				}))
 
+				args.Idle = 72 * time.Hour
+				infoExt, err = client.XPendingExt(ctx, args).Result()
+				Expect(err).NotTo(HaveOccurred())
+				Expect(infoExt).To(HaveLen(0))
+
 				n, err := client.XGroupDelConsumer(ctx, "stream", "group", "consumer").Result()
 				Expect(err).NotTo(HaveOccurred())
 				Expect(n).To(Equal(int64(3)))
@@ -4382,6 +4406,153 @@ var _ = Describe("Commands", func() {
 					FirstEntry:      redis.XMessage{ID: "1-0", Values: map[string]interface{}{"uno": "un"}},
 					LastEntry:       redis.XMessage{ID: "3-0", Values: map[string]interface{}{"tres": "troix"}},
 				}))
+
+				// stream is empty
+				n, err := client.XDel(ctx, "stream", "1-0", "2-0", "3-0").Result()
+				Expect(err).NotTo(HaveOccurred())
+				Expect(n).To(Equal(int64(3)))
+
+				res, err = client.XInfoStream(ctx, "stream").Result()
+				Expect(err).NotTo(HaveOccurred())
+				res.RadixTreeKeys = 0
+				res.RadixTreeNodes = 0
+
+				Expect(res).To(Equal(&redis.XInfoStream{
+					Length:          0,
+					RadixTreeKeys:   0,
+					RadixTreeNodes:  0,
+					Groups:          2,
+					LastGeneratedID: "3-0",
+					FirstEntry:      redis.XMessage{},
+					LastEntry:       redis.XMessage{},
+				}))
+			})
+
+			It("should XINFO STREAM FULL", func() {
+				res, err := client.XInfoStreamFull(ctx, "stream", 2).Result()
+				Expect(err).NotTo(HaveOccurred())
+				res.RadixTreeKeys = 0
+				res.RadixTreeNodes = 0
+
+				// Verify DeliveryTime
+				now := time.Now()
+				maxElapsed := 10 * time.Minute
+				for k, g := range res.Groups {
+					for k2, p := range g.Pending {
+						Expect(now.Sub(p.DeliveryTime)).To(BeNumerically("<=", maxElapsed))
+						res.Groups[k].Pending[k2].DeliveryTime = time.Time{}
+					}
+					for k3, c := range g.Consumers {
+						Expect(now.Sub(c.SeenTime)).To(BeNumerically("<=", maxElapsed))
+						res.Groups[k].Consumers[k3].SeenTime = time.Time{}
+
+						for k4, p := range c.Pending {
+							Expect(now.Sub(p.DeliveryTime)).To(BeNumerically("<=", maxElapsed))
+							res.Groups[k].Consumers[k3].Pending[k4].DeliveryTime = time.Time{}
+						}
+					}
+				}
+
+				Expect(res).To(Equal(&redis.XInfoStreamFull{
+					Length:          3,
+					RadixTreeKeys:   0,
+					RadixTreeNodes:  0,
+					LastGeneratedID: "3-0",
+					Entries: []redis.XMessage{
+						{ID: "1-0", Values: map[string]interface{}{"uno": "un"}},
+						{ID: "2-0", Values: map[string]interface{}{"dos": "deux"}},
+					},
+					Groups: []redis.XInfoStreamGroup{
+						{
+							Name:            "group1",
+							LastDeliveredID: "3-0",
+							PelCount:        3,
+							Pending: []redis.XInfoStreamGroupPending{
+								{
+									ID:            "1-0",
+									Consumer:      "consumer1",
+									DeliveryTime:  time.Time{},
+									DeliveryCount: 1,
+								},
+								{
+									ID:            "2-0",
+									Consumer:      "consumer1",
+									DeliveryTime:  time.Time{},
+									DeliveryCount: 1,
+								},
+							},
+							Consumers: []redis.XInfoStreamConsumer{
+								{
+									Name:     "consumer1",
+									SeenTime: time.Time{},
+									PelCount: 2,
+									Pending: []redis.XInfoStreamConsumerPending{
+										{
+											ID:            "1-0",
+											DeliveryTime:  time.Time{},
+											DeliveryCount: 1,
+										},
+										{
+											ID:            "2-0",
+											DeliveryTime:  time.Time{},
+											DeliveryCount: 1,
+										},
+									},
+								},
+								{
+									Name:     "consumer2",
+									SeenTime: time.Time{},
+									PelCount: 1,
+									Pending: []redis.XInfoStreamConsumerPending{
+										{
+											ID:            "3-0",
+											DeliveryTime:  time.Time{},
+											DeliveryCount: 1,
+										},
+									},
+								},
+							},
+						},
+						{
+							Name:            "group2",
+							LastDeliveredID: "3-0",
+							PelCount:        2,
+							Pending: []redis.XInfoStreamGroupPending{
+								{
+									ID:            "2-0",
+									Consumer:      "consumer1",
+									DeliveryTime:  time.Time{},
+									DeliveryCount: 1,
+								},
+								{
+									ID:            "3-0",
+									Consumer:      "consumer1",
+									DeliveryTime:  time.Time{},
+									DeliveryCount: 1,
+								},
+							},
+							Consumers: []redis.XInfoStreamConsumer{
+								{
+									Name:     "consumer1",
+									SeenTime: time.Time{},
+									PelCount: 2,
+									Pending: []redis.XInfoStreamConsumerPending{
+										{
+											ID:            "2-0",
+											DeliveryTime:  time.Time{},
+											DeliveryCount: 1,
+										},
+										{
+											ID:            "3-0",
+											DeliveryTime:  time.Time{},
+											DeliveryCount: 1,
+										},
+									},
+								},
+							},
+						},
+					},
+				}))
 			})
 
 			It("should XINFO GROUPS", func() {