Skip to content

Commit

Permalink
sync master to v9 (#1760)
Browse files Browse the repository at this point in the history 
* 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>
  • Loading branch information
@monkey92t @parvez0 @vmihailenco
3 people committed May 19, 2021
1 parent d42071c commit fe9bc12
Show file tree
Hide file tree
Showing 5 changed files with 481 additions and 7 deletions.
2 changes: 1 addition & 1 deletion .github/workflows/build.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ name: Go

on:
push:
branches: [master]
branches: [master, v9]
pull_request:
branches: [master, v9]

Expand Down
1 change: 1 addition & 0 deletions .github/workflows/golangci-lint.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ on:
branches:
- master
- main
- v9
pull_request:

jobs:
Expand Down
279 changes: 278 additions & 1 deletion command.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand Down Expand Up @@ -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)
Expand All @@ -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

Expand Down
29 changes: 27 additions & 2 deletions commands.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -1833,15 +1840,20 @@ 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
Consumer string
}

func (c cmdable) XPendingExt(ctx context.Context, a *XPendingExtArgs) *XPendingExtCmd {
args := make([]interface{}, 0, 7)
args = append(args, "xpending", a.Stream, a.Group, a.Start, a.End, a.Count)
args := make([]interface{}, 0, 9)
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)
}
Expand Down Expand Up @@ -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.
Expand Down

0 comments on commit fe9bc12

Please sign in to comment.