Function readXInfoStreamConsumers

command.go:3951–4037

(rd *proto.Reader)

}

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++ {
		nn, err := rd.ReadMapLen()
		if err != nil {
			return nil, err
		}

		c := XInfoStreamConsumer{}

		for f := 0; f < nn; 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.UnixMilli(seen)
			case "active-time":
				active, err := rd.ReadInt()
				if err != nil {
					return nil, err
				}
				c.ActiveTime = time.UnixMilli(active)
			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
					}