Method bytes

builder_test.go:174–223

()

}

func (f v2MessageSetBuilder) bytes() []byte {
	attributes := int16(0)
	if f.codec != nil {
		attributes = int16(f.codec.Code()) // set codec code on attributes
	}
	return newWB().call(func(wb *kafkaWriteBuffer) {
		wb.writeInt64(f.msgs[0].Offset)
		wb.writeBytes(newWB().call(func(wb *kafkaWriteBuffer) {
			wb.writeInt32(0)                            // leader epoch
			wb.writeInt8(2)                             // magic = 2
			wb.writeInt32(0)                            // crc, unused
			wb.writeInt16(attributes)                   // record set attributes
			wb.writeInt32(0)                            // record set last offset delta
			wb.writeInt64(1000 * f.msgs[0].Time.Unix()) // record set first timestamp
			wb.writeInt64(1000 * f.msgs[0].Time.Unix()) // record set last timestamp
			wb.writeInt64(0)                            // record set producer id
			wb.writeInt16(0)                            // record set producer epoch
			wb.writeInt32(0)                            // record set base sequence
			wb.writeInt32(int32(len(f.msgs)))           // record set count
			bs := newWB().call(func(wb *kafkaWriteBuffer) {
				for i, msg := range f.msgs {
					wb.Write(newWB().call(func(wb *kafkaWriteBuffer) {
						bs := newWB().call(func(wb *kafkaWriteBuffer) {
							wb.writeInt8(0)                                              // record attributes, not used here
							wb.writeVarInt(1000 * (time.Now().Unix() - msg.Time.Unix())) // timestamp
							wb.writeVarInt(int64(i))                                     // offset delta
							wb.writeVarInt(int64(len(msg.Key)))                          // key len
							wb.Write(msg.Key)                                            // key bytes
							wb.writeVarInt(int64(len(msg.Value)))                        // value len
							wb.Write(msg.Value)                                          // value bytes
							wb.writeVarInt(int64(len(msg.Headers)))                      // number of headers
							for _, header := range msg.Headers {
								wb.writeVarInt(int64(len(header.Key)))
								wb.Write([]byte(header.Key))
								wb.writeVarInt(int64(len(header.Value)))
								wb.Write(header.Value)
							}
						})
						wb.writeVarInt(int64(len(bs)))
						wb.Write(bs)
					}))
				}
			})
			if f.codec != nil {
				bs = mustCompress(bs, f.codec)
			}
			wb.Write(bs)
		}))
	})
}

// kafkaWriteBuffer is a write buffer that helps writing fetch responses.
type kafkaWriteBuffer struct {