Method bytes

builder_test.go:86–116

()

}

func (f v0MessageSetBuilder) bytes() []byte {
	bs := newWB().call(func(wb *kafkaWriteBuffer) {
		for _, msg := range f.msgs {
			bs := newWB().call(func(wb *kafkaWriteBuffer) {
				wb.writeInt64(msg.Offset) // offset
				wb.writeBytes(newWB().call(func(wb *kafkaWriteBuffer) {
					wb.writeInt32(-1) // crc, unused
					wb.writeInt8(0)   // magic
					wb.writeInt8(0)   // attributes -- zero, no compression for the inner message
					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[0].Offset) // offset
			wb.writeBytes(newWB().call(func(wb *kafkaWriteBuffer) {
				compressed := mustCompress(bs, f.codec)
				wb.writeInt32(-1)            // crc, unused
				wb.writeInt8(0)              // magic
				wb.writeInt8(f.codec.Code()) // attributes
				wb.writeBytes(nil)           // key is always nil for compressed
				wb.writeBytes(compressed)    // the value is the compressed message
			}))
		})
	}
	return bs
}

type v1MessageSetBuilder struct {
	msgs  []Message