Method bytes

builder_test.go:127–163

()

}

func (f v1MessageSetBuilder) bytes() []byte {
	bs := newWB().call(func(wb *kafkaWriteBuffer) {
		for i, msg := range f.msgs {
			bs := newWB().call(func(wb *kafkaWriteBuffer) {
				if f.codec != nil {
					wb.writeInt64(int64(i)) // compressed inner message offsets are relative
				} else {
					wb.writeInt64(msg.Offset) // offset
				}
				wb.writeBytes(newWB().call(func(wb *kafkaWriteBuffer) {
					wb.writeInt32(-1)                     // crc, unused
					wb.writeInt8(1)                       // magic
					wb.writeInt8(0)                       // attributes -- zero, no compression for the inner message
					wb.writeInt64(1000 * msg.Time.Unix()) // timestamp
					wb.writeBytes(msg.Key)
					wb.writeBytes(msg.Value)
				}))
			})
			wb.Write(bs)
		}
	})
	if f.codec != nil {
		bs = newWB().call(func(wb *kafkaWriteBuffer) {
			wb.writeInt64(f.msgs[len(f.msgs)-1].Offset) // offset of the wrapper message is the last offset of the inner messages
			wb.writeBytes(newWB().call(func(wb *kafkaWriteBuffer) {
				bs := mustCompress(bs, f.codec)
				wb.writeInt32(-1)                           // crc, unused
				wb.writeInt8(1)                             // magic
				wb.writeInt8(f.codec.Code())                // attributes
				wb.writeInt64(1000 * f.msgs[0].Time.Unix()) // timestamp
				wb.writeBytes(nil)                          // key is always nil for compressed
				wb.writeBytes(bs)                           // the value is the compressed message
			}))
		})
	}
	return bs
}

type v2MessageSetBuilder struct {
	msgs  []Message