Method writeToVersion1

protocol/record_v1.go:152–230

(buffer *pageBuffer, bufferOffset int64)

}

func (rs *RecordSet) writeToVersion1(buffer *pageBuffer, bufferOffset int64) error {
	attributes := rs.Attributes
	records := rs.Records

	if compression := attributes.Compression(); compression != 0 {
		if codec := compression.Codec(); codec != nil {
			// In the message format version 1, compression is achieved by
			// compressing the value of a message which recursively contains
			// the representation of the compressed message set.
			subset := *rs
			subset.Attributes &= ^7 // erase compression

			if err := subset.writeToVersion1(buffer, bufferOffset); err != nil {
				return err
			}

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

			compressor := codec.NewWriter(compressed)
			defer compressor.Close()

			var err error
			buffer.pages.scan(bufferOffset, buffer.Size(), func(b []byte) bool {
				_, err = compressor.Write(b)
				return err == nil
			})
			if err != nil {
				return err
			}
			if err := compressor.Close(); err != nil {
				return err
			}

			buffer.Truncate(int(bufferOffset))

			records = &message{
				Record: Record{
					Value: compressed,
				},
			}
		}
	}

	e := encoder{writer: buffer}
	currentTimestamp := timestamp(time.Now())

	return forEachRecord(records, func(i int, r *Record) error {
		t := timestamp(r.Time)
		if t == 0 {
			t = currentTimestamp
		}

		messageOffset := buffer.Size()
		e.writeInt64(int64(i))
		e.writeInt32(0) // message size placeholder
		e.writeInt32(0) // crc32 placeholder
		e.setCRC(crc32.IEEETable)