Method readStreamEntries

core/stream.go:81–123

readStreamEntries read entries

()

// readStreamEntries read entries
func (dec *Decoder) readStreamEntries() ([]*model.StreamEntry, error) {
	length, _, err := dec.readLength()
	if err != nil {
		return nil, err
	}
	var result []*model.StreamEntry
	for i := uint64(0); i < length; i++ {
		header, err := dec.readString()
		if err != nil {
			return nil, err
		}
		cursor := 0
		msBin, err := readBytes(header, &cursor, 8)
		if err != nil {
			return nil, err
		}
		ms := binary.BigEndian.Uint64(msBin)
		seqBin, err := readBytes(header, &cursor, 8)
		if err != nil {
			return nil, err
		}
		seq := binary.BigEndian.Uint64(seqBin)
		firstId := &model.StreamId{
			Ms:       ms,
			Sequence: seq,
		}

		buf, err := dec.readString()
		if err != nil {
			return nil, err
		}
		cursor = 0
		// skip 4Byte total-bytes + 2Byte num-elements
		_, _ = readBytes(buf, &cursor, 6)
		entry, err := dec.readStreamEntryContent(buf, &cursor, firstId)
		if err != nil {
			return nil, err
		}
		entry.FirstMsgId = firstId
		result = append(result, entry)
	}
	return result, nil
}

// readStreamEntryContent read messages in a stream entry
func (dec *Decoder) readStreamEntryContent(buf []byte, cursor *int, firstId *model.StreamId) (*model.StreamEntry, error) {