Function testWriteProduceRequestV2

write_test.go:111–156

(t *testing.T)

}

func testWriteProduceRequestV2(t *testing.T) {
	key := []byte(nil)
	val := []byte("Hello World!")

	msg := messageSetItem{
		Offset: 10,
		Message: message{
			MagicByte:  1,
			Attributes: 0,
			Key:        key,
			Value:      val,
		},
	}
	msg.MessageSize = msg.Message.size()
	msg.Message.CRC = msg.Message.crc32(&crc32Writer{
		table: crc32.IEEETable,
	})

	const timeout = 100
	testWriteOptimization(t,
		requestHeader{
			ApiKey:        int16(produce),
			ApiVersion:    int16(v2),
			CorrelationID: testCorrelationID,
			ClientID:      testClientID,
		},
		produceRequestV2{
			RequiredAcks: -1,
			Timeout:      timeout,
			Topics: []produceRequestTopicV2{{
				TopicName: testTopic,
				Partitions: []produceRequestPartitionV2{{
					Partition:      testPartition,
					MessageSetSize: msg.size(), MessageSet: messageSet{msg},
				}},
			}},
		},
		func(w *writeBuffer) {
			w.writeProduceRequestV2(nil, testCorrelationID, testClientID, testTopic, testPartition, timeout*time.Millisecond, -1, Message{
				Offset: 10,
				Key:    key,
				Value:  val,
			})
		},
	)
}

func testWriteOptimization(t *testing.T, h requestHeader, r request, f func(*writeBuffer)) {
	b1 := &bytes.Buffer{}