Method readFromVersion1

protocol/record_v1.go:55–150

(d *decoder)

}

func (rs *RecordSet) readFromVersion1(d *decoder) error {
	var records RecordReader

	b := newPageBuffer()
	defer b.unref()

	attributes, baseOffset, timestamp, key, value, err := readMessage(b, d)
	if err != nil {
		return err
	}

	if compression := Attributes(attributes).Compression(); compression == 0 {
		records = &message{
			Record: Record{
				Offset: baseOffset,
				Time:   makeTime(timestamp),
				Key:    key,
				Value:  value,
			},
		}
	} else {
		// Can we have a non-nil key when reading a compressed message?
		if key != nil {
			key.Close()
		}
		if value == nil {
			records = emptyRecordReader{}
		} else {
			defer value.Close()

			codec := compression.Codec()
			if codec == nil {
				return Errorf("unsupported compression codec: %d", compression)
			}
			decompressor := codec.NewReader(value)
			defer decompressor.Close()

			b := newPageBuffer()
			defer b.unref()

			d := &decoder{
				reader: decompressor,
				remain: math.MaxInt32,
			}

			r := &recordReader{
				records: make([]Record, 0, 32),
			}

			for !d.done() {
				_, offset, timestamp, key, value, err := readMessage(b, d)
				if err != nil {
					if errors.Is(err, io.ErrUnexpectedEOF) {
						break
					}
					for _, rec := range r.records {
						closeBytes(rec.Key)
						closeBytes(rec.Value)