forked from mirror/redis
parent
3871963e2d
commit
76393b5b71
155
command.go
155
command.go
|
@ -1879,11 +1879,7 @@ func (cmd *XInfoStreamFullCmd) readReply(rd *proto.Reader) error {
|
||||||
case "entries":
|
case "entries":
|
||||||
cmd.val.Entries, err = readXMessageSlice(rd)
|
cmd.val.Entries, err = readXMessageSlice(rd)
|
||||||
case "groups":
|
case "groups":
|
||||||
groups, err := rd.ReadReply(readStreamGroups)
|
cmd.val.Groups, err = readStreamGroups(rd)
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
cmd.val.Groups = groups.([]XInfoStreamGroup)
|
|
||||||
default:
|
default:
|
||||||
return fmt.Errorf("redis: unexpected content %s "+
|
return fmt.Errorf("redis: unexpected content %s "+
|
||||||
"in XINFO STREAM reply", key)
|
"in XINFO STREAM reply", key)
|
||||||
|
@ -1895,9 +1891,13 @@ func (cmd *XInfoStreamFullCmd) readReply(rd *proto.Reader) error {
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func readStreamGroups(rd *proto.Reader, n int64) (interface{}, error) {
|
func readStreamGroups(rd *proto.Reader) ([]XInfoStreamGroup, error) {
|
||||||
|
n, err := rd.ReadArrayLen()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
groups := make([]XInfoStreamGroup, 0, n)
|
groups := make([]XInfoStreamGroup, 0, n)
|
||||||
for i := int64(0); i < n; i++ {
|
for i := 0; i < n; i++ {
|
||||||
nn, err := rd.ReadArrayLen()
|
nn, err := rd.ReadArrayLen()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
|
@ -1906,31 +1906,34 @@ func readStreamGroups(rd *proto.Reader, n int64) (interface{}, error) {
|
||||||
return nil, fmt.Errorf("redis: got %d elements in XINFO STREAM FULL reply,"+
|
return nil, fmt.Errorf("redis: got %d elements in XINFO STREAM FULL reply,"+
|
||||||
"wanted 10", nn)
|
"wanted 10", nn)
|
||||||
}
|
}
|
||||||
key, err := rd.ReadString()
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
group := XInfoStreamGroup{}
|
group := XInfoStreamGroup{}
|
||||||
|
|
||||||
switch key {
|
for f := 0; f < 5; f++ {
|
||||||
case "name":
|
key, err := rd.ReadString()
|
||||||
group.Name, err = rd.ReadString()
|
if err != nil {
|
||||||
case "last-delivered-id":
|
return nil, err
|
||||||
group.LastDeliveredID, err = rd.ReadString()
|
}
|
||||||
case "pel-count":
|
|
||||||
group.PelCount, err = rd.ReadIntReply()
|
|
||||||
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 reply", key)
|
|
||||||
}
|
|
||||||
|
|
||||||
if err != nil {
|
switch key {
|
||||||
return nil, err
|
case "name":
|
||||||
|
group.Name, err = rd.ReadString()
|
||||||
|
case "last-delivered-id":
|
||||||
|
group.LastDeliveredID, err = rd.ReadString()
|
||||||
|
case "pel-count":
|
||||||
|
group.PelCount, err = rd.ReadIntReply()
|
||||||
|
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 reply", key)
|
||||||
|
}
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
groups = append(groups, group)
|
groups = append(groups, group)
|
||||||
|
@ -2004,71 +2007,71 @@ func readXInfoStreamConsumers(rd *proto.Reader) ([]XInfoStreamConsumer, error) {
|
||||||
"wanted 8", nn)
|
"wanted 8", nn)
|
||||||
}
|
}
|
||||||
|
|
||||||
cKey, err := rd.ReadString()
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
c := XInfoStreamConsumer{}
|
c := XInfoStreamConsumer{}
|
||||||
|
|
||||||
switch cKey {
|
for f := 0; f < 4; f++ {
|
||||||
case "name":
|
cKey, err := rd.ReadString()
|
||||||
c.Name, err = rd.ReadString()
|
|
||||||
case "seen-time":
|
|
||||||
seen, err := rd.ReadIntReply()
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
c.SeenTime = time.Unix(seen/1000, seen%1000*int64(time.Millisecond))
|
|
||||||
case "pel-count":
|
|
||||||
c.PelCount, err = rd.ReadIntReply()
|
|
||||||
case "pending":
|
|
||||||
pendingNumber, err := rd.ReadArrayLen()
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
c.Pending = make([]XInfoStreamConsumerPending, 0, pendingNumber)
|
switch cKey {
|
||||||
|
case "name":
|
||||||
for f := 0; f < pendingNumber; f++ {
|
c.Name, err = rd.ReadString()
|
||||||
nn, err := rd.ReadArrayLen()
|
case "seen-time":
|
||||||
|
seen, err := rd.ReadIntReply()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if nn != 3 {
|
c.SeenTime = time.Unix(seen/1000, seen%1000*int64(time.Millisecond))
|
||||||
return nil, fmt.Errorf("redis: got %d elements in XINFO STREAM reply,"+
|
case "pel-count":
|
||||||
"wanted 3", nn)
|
c.PelCount, err = rd.ReadIntReply()
|
||||||
}
|
case "pending":
|
||||||
|
pendingNumber, err := rd.ReadArrayLen()
|
||||||
p := XInfoStreamConsumerPending{}
|
|
||||||
|
|
||||||
p.ID, err = rd.ReadString()
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
delivery, err := rd.ReadIntReply()
|
c.Pending = make([]XInfoStreamConsumerPending, 0, pendingNumber)
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
p.DeliveryTime = time.Unix(delivery/1000, delivery%1000*int64(time.Millisecond))
|
|
||||||
|
|
||||||
p.DeliveryCount, err = rd.ReadIntReply()
|
for f := 0; f < pendingNumber; f++ {
|
||||||
if err != nil {
|
nn, err := rd.ReadArrayLen()
|
||||||
return nil, err
|
if err != nil {
|
||||||
}
|
return nil, err
|
||||||
|
}
|
||||||
|
if nn != 3 {
|
||||||
|
return nil, fmt.Errorf("redis: got %d elements in XINFO STREAM reply,"+
|
||||||
|
"wanted 3", nn)
|
||||||
|
}
|
||||||
|
|
||||||
c.Pending = append(c.Pending, p)
|
p := XInfoStreamConsumerPending{}
|
||||||
|
|
||||||
|
p.ID, err = rd.ReadString()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
delivery, err := rd.ReadIntReply()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
p.DeliveryTime = time.Unix(delivery/1000, delivery%1000*int64(time.Millisecond))
|
||||||
|
|
||||||
|
p.DeliveryCount, err = rd.ReadIntReply()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
c.Pending = append(c.Pending, p)
|
||||||
|
}
|
||||||
|
default:
|
||||||
|
return nil, fmt.Errorf("redis: unexpected content %s "+
|
||||||
|
"in XINFO STREAM reply", cKey)
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
}
|
}
|
||||||
default:
|
|
||||||
return nil, fmt.Errorf("redis: unexpected content %s "+
|
|
||||||
"in XINFO STREAM reply", cKey)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
consumers = append(consumers, c)
|
consumers = append(consumers, c)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in New Issue