foreign/go/internal/command/message_test.go - iggy - Git at Google

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you under the Apache License, Version 2.0 (the
 // "License"); you may not use this file except in compliance
 // with the License.  You may obtain a copy of the License at
 //
 //   http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing,
 // software distributed under the License is distributed on an
 // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 // KIND, either express or implied.  See the License for the
 // specific language governing permissions and limitations
 // under the License.

 package command

 import (
 	"bytes"
 	"encoding/binary"
 	"encoding/hex"
 	"math"
 	"testing"

 	"github.com/apache/iggy/foreign/go/contracts"
 	"github.com/apache/iggy/foreign/go/internal/batch"
 	"github.com/google/uuid"
 	"github.com/klauspost/compress/s2"
 )

 func TestSerialize_TcpFetchMessagesRequest(t *testing.T) {
 	partitionId := uint32(123)
 	consumerId, _ := iggcon.NewIdentifier(uint32(42))
 	streamId, _ := iggcon.NewIdentifier("test_stream_id")
 	topicId, _ := iggcon.NewIdentifier("test_topic_id")
 	// Create a sample PollMessages
 	request := PollMessages{
 		Consumer:    iggcon.NewSingleConsumer(consumerId),
 		StreamId:    streamId,
 		TopicId:     topicId,
 		PartitionId: &partitionId,
 		Strategy:    iggcon.FirstPollingStrategy(),
 		Count:       100,
 		AutoCommit:  true,
 	}

 	// Serialize the request
 	serialized, err := request.MarshalBinary()
 	if err != nil {
 		t.Error(err)
 	}

 	// Expected serialized bytes based on the provided sample request
 	expected := []byte{
 		0x01,                 // Consumer Kind
 		0x01,                 // ConsumerId Kind (NumericId)
 		0x04,                 // ConsumerId Length (4)
 		0x2A, 0x00, 0x0, 0x0, // ConsumerId

 		0x02,                                                                               // StreamId Kind (StringId)
 		0x0E,                                                                               // StreamId Length (14)
 		0x74, 0x65, 0x73, 0x74, 0x5F, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6D, 0x5F, 0x69, 0x64, // StreamId

 		0x02,                                                                         // TopicId Kind (StringId)
 		0x0D,                                                                         // TopicId Length (13)
 		0x74, 0x65, 0x73, 0x74, 0x5F, 0x74, 0x6F, 0x70, 0x69, 0x63, 0x5F, 0x69, 0x64, // TopicId

 		0x01,                   // Partition present
 		0x7B, 0x00, 0x00, 0x00, // PartitionId (123)
 		0x03,                                           // PollingStrategy Kind
 		0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // PollingStrategy Value (0)
 		0x64, 0x00, 0x00, 0x00, // Count (100)
 		0x01, // AutoCommit
 	}

 	// Check if the serialized bytes match the expected bytes
 	if !areBytesEqual(serialized, expected) {
 		t.Errorf("Serialized bytes are incorrect. \nExpected:\t%v\nGot:\t\t%v", expected, serialized)
 	}
 }

 func areBytesEqual(a, b []byte) bool {
 	if len(a) != len(b) {
 		return false
 	}
 	for i := range a {
 		if a[i] != b[i] {
 			return false
 		}
 	}
 	return true
 }

 // Golden vectors generated by the Rust encoder for stream 1, topic 2,
 // balanced partitioning, and two messages: {id 7, origin timestamp 1000,
 // payload "first-payload"} and {id 8, origin timestamp 1050, payload
 // "second-payload", user headers "user-header-bytes"}.
 const (
 	goldenProduceBodyFull  = "12000000010401000000010402000000010002000000000000000000000000000000000000000000000000000000e8030000000000008c01000000000000a91f38c86307267c02000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000bfd2b9205a759675070000000000000000000000000000000000000000000000000000000d000000000000000000000066697273742d7061796c6f6164d66b7e1c758eb7c0080000000000000000000000000000000100000032000000110000000e00000000000000000000007365636f6e642d7061796c6f6164757365722d6865616465722d6279746573"
 	goldenProduceBatchOnly = "000000000000000000000000000000000000000000000000e8030000000000008c01000000000000a91f38c86307267c02000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000bfd2b9205a759675070000000000000000000000000000000000000000000000000000000d000000000000000000000066697273742d7061796c6f6164d66b7e1c758eb7c0080000000000000000000000000000000100000032000000110000000e00000000000000000000007365636f6e642d7061796c6f6164757365722d6865616465722d6279746573"
 )

 func TestSerialize_SendMessagesMatchesTheGoldenVector(t *testing.T) {
 	streamId, _ := iggcon.NewIdentifier(uint32(1))
 	topicId, _ := iggcon.NewIdentifier(uint32(2))
 	request := SendMessages{
 		StreamId:     streamId,
 		TopicId:      topicId,
 		Partitioning: iggcon.None(),
 		Messages: []iggcon.IggyMessage{
 			{
 				Header:  iggcon.MessageHeader{Id: iggcon.MessageID{7}, OriginTimestamp: 1000},
 				Payload: []byte("first-payload"),
 			},
 			{
 				Header:      iggcon.MessageHeader{Id: iggcon.MessageID{8}, OriginTimestamp: 1050},
 				Payload:     []byte("second-payload"),
 				UserHeaders: []byte("user-header-bytes"),
 			},
 		},
 		Compression: iggcon.MESSAGE_COMPRESSION_NONE,
 	}

 	serialized, err := request.MarshalBinary()
 	if err != nil {
 		t.Fatal(err)
 	}

 	if got := hex.EncodeToString(serialized); got != goldenProduceBodyFull {
 		t.Errorf("the full produce body diverges from the golden vector.\nExpected:\t%s\nGot:\t\t%s",
 			goldenProduceBodyFull, got)
 	}

 	metadataLength := binary.LittleEndian.Uint32(serialized[0:4])
 	batchOnly := serialized[4+metadataLength:]
 	if got := hex.EncodeToString(batchOnly); got != goldenProduceBatchOnly {
 		t.Errorf("the batch section diverges from the golden vector.\nExpected:\t%s\nGot:\t\t%s",
 			goldenProduceBatchOnly, got)
 	}
 }

 func TestSerialize_SendMessagesRequest(t *testing.T) {
 	message1 := generateTestMessage("data1")
 	streamId, _ := iggcon.NewIdentifier("test_stream_id")
 	topicId, _ := iggcon.NewIdentifier("test_topic_id")
 	request := SendMessages{
 		StreamId:     streamId,
 		TopicId:      topicId,
 		Partitioning: iggcon.PartitionId(1),
 		Messages: []iggcon.IggyMessage{
 			message1,
 		},
 		Compression: iggcon.MESSAGE_COMPRESSION_NONE,
 	}

 	// Serialize the request
 	serialized, err := request.MarshalBinary()
 	if err != nil {
 		t.Error(err)
 	}

 	expectedMetadata := []byte{
 		0x29, 0x0, 0x0, 0x0, // metadataLength
 		0x02,                                                                               // StreamId Kind (StringId)
 		0x0E,                                                                               // StreamId Length (14)
 		0x74, 0x65, 0x73, 0x74, 0x5F, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6D, 0x5F, 0x69, 0x64, // StreamId

 		0x02,                                                                         // TopicId Kind (StringId)
 		0x0D,                                                                         // TopicId Length (13)
 		0x74, 0x65, 0x73, 0x74, 0x5F, 0x74, 0x6F, 0x70, 0x69, 0x63, 0x5F, 0x69, 0x64, // TopicId
 		0x02,                   // PartitionIdKind
 		0x04,                   // Partitioning Length
 		0x01, 0x00, 0x00, 0x00, // PartitionId (1)
 		0x01, 0x0, 0x0, 0x0, // MessageCount
 	}
 	if !bytes.Equal(serialized[:len(expectedMetadata)], expectedMetadata) {
 		t.Fatalf("the metadata section is incorrect.\nExpected:\t%v\nGot:\t\t%v",
 			expectedMetadata, serialized[:len(expectedMetadata)])
 	}

 	record := serialized[len(expectedMetadata):]
 	header, err := batch.DecodeHeader(record)
 	if err != nil {
 		t.Fatal(err)
 	}
 	if header.PartitionId != 0 || header.BaseOffset != 0 || header.BaseTimestamp != 0 {
 		t.Errorf("the server-stamped fields must be zero, got %+v", header)
 	}
 	if header.OriginTimestamp != message1.Header.OriginTimestamp {
 		t.Errorf("batch origin timestamp = %d, want %d",
 			header.OriginTimestamp, message1.Header.OriginTimestamp)
 	}
 	if header.BatchLength != uint64(len(record)) {
 		t.Errorf("batch length = %d, the record holds %d bytes", header.BatchLength, len(record))
 	}
 	if header.MessageCount != 1 {
 		t.Errorf("message count = %d, want 1", header.MessageCount)
 	}
 	if got := header.Checksum(record[batch.HeaderSize : batch.HeaderSize+8]); got != header.BatchChecksum {
 		t.Errorf("batch checksum = %d, recomputed %d", header.BatchChecksum, got)
 	}

 	frame, err := batch.DecodeMessageHeader(record[batch.HeaderSize:])
 	if err != nil {
 		t.Fatal(err)
 	}
 	if frame.Id != [16]byte(message1.Header.Id) {
 		t.Errorf("frame id = %v, want %v", frame.Id, message1.Header.Id)
 	}
 	if frame.OffsetDelta != 0 || frame.TimestampDelta != 0 {
 		t.Errorf("a single-message batch has zero deltas, got %+v", frame)
 	}
 	if int(frame.PayloadLength) != len(message1.Payload) ||
 		int(frame.UserHeadersLength) != len(message1.UserHeaders) {
 		t.Errorf("frame lengths %d/%d do not match the message %d/%d",
 			frame.PayloadLength, frame.UserHeadersLength,
 			len(message1.Payload), len(message1.UserHeaders))
 	}

 	body := record[batch.HeaderSize+batch.MessageHeaderSize:]
 	expectedBody := append(append([]byte{}, message1.Payload...), message1.UserHeaders...)
 	if !bytes.Equal(body, expectedBody) {
 		t.Errorf("the frame body is incorrect.\nExpected:\t%v\nGot:\t\t%v", expectedBody, body)
 	}
 }

 func TestSerialize_SendMessagesMintsAZeroMessageId(t *testing.T) {
 	message, err := iggcon.NewIggyMessage([]byte("payload"))
 	if err != nil {
 		t.Fatal(err)
 	}
 	streamId, _ := iggcon.NewIdentifier(uint32(1))
 	topicId, _ := iggcon.NewIdentifier(uint32(1))
 	request := SendMessages{
 		StreamId:     streamId,
 		TopicId:      topicId,
 		Partitioning: iggcon.PartitionId(0),
 		Messages:     []iggcon.IggyMessage{message},
 		Compression:  iggcon.MESSAGE_COMPRESSION_NONE,
 	}

 	serialized, err := request.MarshalBinary()
 	if err != nil {
 		t.Fatal(err)
 	}

 	minted := request.Messages[0].Header.Id
 	if minted == (iggcon.MessageID{}) {
 		t.Fatal("a zero id must be minted before encoding")
 	}
 	metadataLength := binary.LittleEndian.Uint32(serialized[0:4])
 	frame, err := batch.DecodeMessageHeader(serialized[4+int(metadataLength)+batch.HeaderSize:])
 	if err != nil {
 		t.Fatal(err)
 	}
 	if frame.Id != [16]byte(minted) {
 		t.Errorf("the wire carries id %v, the message holds %v", frame.Id, minted)
 	}
 }

 func TestSerialize_SendMessagesRejectsATimestampDeltaPastU32(t *testing.T) {
 	streamId, _ := iggcon.NewIdentifier(uint32(1))
 	topicId, _ := iggcon.NewIdentifier(uint32(1))
 	request := SendMessages{
 		StreamId:     streamId,
 		TopicId:      topicId,
 		Partitioning: iggcon.PartitionId(0),
 		Messages: []iggcon.IggyMessage{
 			{
 				Header:  iggcon.MessageHeader{Id: iggcon.MessageID{1}, OriginTimestamp: 0},
 				Payload: []byte("early"),
 			},
 			{
 				Header: iggcon.MessageHeader{
 					Id:              iggcon.MessageID{2},
 					OriginTimestamp: uint64(math.MaxUint32) + 1,
 				},
 				Payload: []byte("late"),
 			},
 		},
 		Compression: iggcon.MESSAGE_COMPRESSION_NONE,
 	}

 	if _, err := request.MarshalBinary(); err == nil {
 		t.Fatal("a timestamp delta past u32 must fail the encode")
 	}
 }

 func TestSerialize_SendMessagesCompressesThePayloadCoherently(t *testing.T) {
 	// A compressible payload above the 32-byte floor.
 	payload := bytes.Repeat([]byte("abcdefgh"), 32)
 	message, err := iggcon.NewIggyMessage(payload)
 	if err != nil {
 		t.Fatal(err)
 	}
 	streamId, _ := iggcon.NewIdentifier(uint32(1))
 	topicId, _ := iggcon.NewIdentifier(uint32(1))
 	request := SendMessages{
 		StreamId:     streamId,
 		TopicId:      topicId,
 		Partitioning: iggcon.PartitionId(0),
 		Messages:     []iggcon.IggyMessage{message},
 		Compression:  iggcon.MESSAGE_COMPRESSION_S2,
 	}

 	serialized, err := request.MarshalBinary()
 	if err != nil {
 		t.Fatal(err)
 	}

 	compressed := request.Messages[0]
 	if got := int(compressed.Header.PayloadLength); got != len(compressed.Payload) {
 		t.Fatalf("header claims %d payload bytes, the slice holds %d: the wire would mis-frame",
 			got, len(compressed.Payload))
 	}
 	if len(compressed.Payload) >= len(payload) {
 		t.Fatalf("the payload did not compress: %d >= %d", len(compressed.Payload), len(payload))
 	}

 	// The frame must be sized by the compressed length: header, then exactly
 	// PayloadLength payload bytes, and nothing after.
 	frameStart := len(serialized) - int(compressed.Header.PayloadLength) - batch.MessageHeaderSize
 	frame, err := batch.DecodeMessageHeader(serialized[frameStart:])
 	if err != nil {
 		t.Fatal(err)
 	}
 	if frame.PayloadLength != compressed.Header.PayloadLength {
 		t.Fatalf("wire frame claims %d, in-memory header %d",
 			frame.PayloadLength, compressed.Header.PayloadLength)
 	}

 	decoded, err := s2.Decode(nil, serialized[frameStart+batch.MessageHeaderSize:])
 	if err != nil {
 		t.Fatal(err)
 	}
 	if !bytes.Equal(decoded, payload) {
 		t.Fatal("the compressed payload does not round-trip")
 	}
 }

 func TestSerialize_SendMessagesSkipsCompressionBelowTheFloor(t *testing.T) {
 	message, err := iggcon.NewIggyMessage([]byte("short"))
 	if err != nil {
 		t.Fatal(err)
 	}
 	streamId, _ := iggcon.NewIdentifier(uint32(1))
 	topicId, _ := iggcon.NewIdentifier(uint32(1))
 	request := SendMessages{
 		StreamId:     streamId,
 		TopicId:      topicId,
 		Partitioning: iggcon.PartitionId(0),
 		Messages:     []iggcon.IggyMessage{message},
 		Compression:  iggcon.MESSAGE_COMPRESSION_S2,
 	}

 	if _, err := request.MarshalBinary(); err != nil {
 		t.Fatal(err)
 	}
 	if !bytes.Equal(request.Messages[0].Payload, []byte("short")) {
 		t.Fatal("a payload under 32 bytes must pass through uncompressed")
 	}
 }

 func createDefaultMessageHeaders() []iggcon.HeaderEntry {
 	return []iggcon.HeaderEntry{
 		{Key: iggcon.HeaderKey{Kind: iggcon.String, Value: []byte("HeaderKey1")}, Value: iggcon.HeaderValue{Kind: iggcon.String, Value: []byte("Value 1")}},
 		{Key: iggcon.HeaderKey{Kind: iggcon.String, Value: []byte("HeaderKey2")}, Value: iggcon.HeaderValue{Kind: iggcon.Uint32, Value: []byte{0x01, 0x02, 0x03, 0x04}}},
 	}
 }

 func generateTestMessage(payload string) iggcon.IggyMessage {
 	msg, _ := iggcon.NewIggyMessage(
 		[]byte(payload),
 		iggcon.WithID(uuid.New()),
 		iggcon.WithUserHeaders(createDefaultMessageHeaders()))
 	return msg
 }
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you under the Apache License, Version 2.0 (the
	// "License"); you may not use this file except in compliance
	// with the License. You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing,
	// software distributed under the License is distributed on an
	// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	// KIND, either express or implied. See the License for the
	// specific language governing permissions and limitations
	// under the License.

	package command

	import (
	"bytes"
	"encoding/binary"
	"encoding/hex"
	"math"
	"testing"

	"github.com/apache/iggy/foreign/go/contracts"
	"github.com/apache/iggy/foreign/go/internal/batch"
	"github.com/google/uuid"
	"github.com/klauspost/compress/s2"
	)

	func TestSerialize_TcpFetchMessagesRequest(t *testing.T) {
	partitionId := uint32(123)
	consumerId, _ := iggcon.NewIdentifier(uint32(42))
	streamId, _ := iggcon.NewIdentifier("test_stream_id")
	topicId, _ := iggcon.NewIdentifier("test_topic_id")
	// Create a sample PollMessages
	request := PollMessages{
	Consumer: iggcon.NewSingleConsumer(consumerId),
	StreamId: streamId,
	TopicId: topicId,
	PartitionId: &partitionId,
	Strategy: iggcon.FirstPollingStrategy(),
	Count: 100,
	AutoCommit: true,
	}

	// Serialize the request
	serialized, err := request.MarshalBinary()
	if err != nil {
	t.Error(err)
	}

	// Expected serialized bytes based on the provided sample request
	expected := []byte{
	0x01, // Consumer Kind
	0x01, // ConsumerId Kind (NumericId)
	0x04, // ConsumerId Length (4)
	0x2A, 0x00, 0x0, 0x0, // ConsumerId

	0x02, // StreamId Kind (StringId)
	0x0E, // StreamId Length (14)
	0x74, 0x65, 0x73, 0x74, 0x5F, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6D, 0x5F, 0x69, 0x64, // StreamId

	0x02, // TopicId Kind (StringId)
	0x0D, // TopicId Length (13)
	0x74, 0x65, 0x73, 0x74, 0x5F, 0x74, 0x6F, 0x70, 0x69, 0x63, 0x5F, 0x69, 0x64, // TopicId

	0x01, // Partition present
	0x7B, 0x00, 0x00, 0x00, // PartitionId (123)
	0x03, // PollingStrategy Kind
	0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // PollingStrategy Value (0)
	0x64, 0x00, 0x00, 0x00, // Count (100)
	0x01, // AutoCommit
	}

	// Check if the serialized bytes match the expected bytes
	if !areBytesEqual(serialized, expected) {
	t.Errorf("Serialized bytes are incorrect. \nExpected:\t%v\nGot:\t\t%v", expected, serialized)
	}
	}

	func areBytesEqual(a, b []byte) bool {
	if len(a) != len(b) {
	return false
	}
	for i := range a {
	if a[i] != b[i] {
	return false
	}
	}
	return true
	}

	// Golden vectors generated by the Rust encoder for stream 1, topic 2,
	// balanced partitioning, and two messages: {id 7, origin timestamp 1000,
	// payload "first-payload"} and {id 8, origin timestamp 1050, payload
	// "second-payload", user headers "user-header-bytes"}.
	const (
	goldenProduceBodyFull = "12000000010401000000010402000000010002000000000000000000000000000000000000000000000000000000e8030000000000008c01000000000000a91f38c86307267c02000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000bfd2b9205a759675070000000000000000000000000000000000000000000000000000000d000000000000000000000066697273742d7061796c6f6164d66b7e1c758eb7c0080000000000000000000000000000000100000032000000110000000e00000000000000000000007365636f6e642d7061796c6f6164757365722d6865616465722d6279746573"
	goldenProduceBatchOnly = "000000000000000000000000000000000000000000000000e8030000000000008c01000000000000a91f38c86307267c02000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000bfd2b9205a759675070000000000000000000000000000000000000000000000000000000d000000000000000000000066697273742d7061796c6f6164d66b7e1c758eb7c0080000000000000000000000000000000100000032000000110000000e00000000000000000000007365636f6e642d7061796c6f6164757365722d6865616465722d6279746573"
	)

	func TestSerialize_SendMessagesMatchesTheGoldenVector(t *testing.T) {
	streamId, _ := iggcon.NewIdentifier(uint32(1))
	topicId, _ := iggcon.NewIdentifier(uint32(2))
	request := SendMessages{
	StreamId: streamId,
	TopicId: topicId,
	Partitioning: iggcon.None(),
	Messages: []iggcon.IggyMessage{
	{
	Header: iggcon.MessageHeader{Id: iggcon.MessageID{7}, OriginTimestamp: 1000},
	Payload: []byte("first-payload"),
	},
	{
	Header: iggcon.MessageHeader{Id: iggcon.MessageID{8}, OriginTimestamp: 1050},
	Payload: []byte("second-payload"),
	UserHeaders: []byte("user-header-bytes"),
	},
	},
	Compression: iggcon.MESSAGE_COMPRESSION_NONE,
	}

	serialized, err := request.MarshalBinary()
	if err != nil {
	t.Fatal(err)
	}

	if got := hex.EncodeToString(serialized); got != goldenProduceBodyFull {
	t.Errorf("the full produce body diverges from the golden vector.\nExpected:\t%s\nGot:\t\t%s",
	goldenProduceBodyFull, got)
	}

	metadataLength := binary.LittleEndian.Uint32(serialized[0:4])
	batchOnly := serialized[4+metadataLength:]
	if got := hex.EncodeToString(batchOnly); got != goldenProduceBatchOnly {
	t.Errorf("the batch section diverges from the golden vector.\nExpected:\t%s\nGot:\t\t%s",
	goldenProduceBatchOnly, got)
	}
	}

	func TestSerialize_SendMessagesRequest(t *testing.T) {
	message1 := generateTestMessage("data1")
	streamId, _ := iggcon.NewIdentifier("test_stream_id")
	topicId, _ := iggcon.NewIdentifier("test_topic_id")
	request := SendMessages{
	StreamId: streamId,
	TopicId: topicId,
	Partitioning: iggcon.PartitionId(1),
	Messages: []iggcon.IggyMessage{
	message1,
	},
	Compression: iggcon.MESSAGE_COMPRESSION_NONE,
	}

	// Serialize the request
	serialized, err := request.MarshalBinary()
	if err != nil {
	t.Error(err)
	}

	expectedMetadata := []byte{
	0x29, 0x0, 0x0, 0x0, // metadataLength
	0x02, // StreamId Kind (StringId)
	0x0E, // StreamId Length (14)
	0x74, 0x65, 0x73, 0x74, 0x5F, 0x73, 0x74, 0x72, 0x65, 0x61, 0x6D, 0x5F, 0x69, 0x64, // StreamId

	0x02, // TopicId Kind (StringId)
	0x0D, // TopicId Length (13)
	0x74, 0x65, 0x73, 0x74, 0x5F, 0x74, 0x6F, 0x70, 0x69, 0x63, 0x5F, 0x69, 0x64, // TopicId
	0x02, // PartitionIdKind
	0x04, // Partitioning Length
	0x01, 0x00, 0x00, 0x00, // PartitionId (1)
	0x01, 0x0, 0x0, 0x0, // MessageCount
	}
	if !bytes.Equal(serialized[:len(expectedMetadata)], expectedMetadata) {
	t.Fatalf("the metadata section is incorrect.\nExpected:\t%v\nGot:\t\t%v",
	expectedMetadata, serialized[:len(expectedMetadata)])
	}

	record := serialized[len(expectedMetadata):]
	header, err := batch.DecodeHeader(record)
	if err != nil {
	t.Fatal(err)
	}
	if header.PartitionId != 0 \|\| header.BaseOffset != 0 \|\| header.BaseTimestamp != 0 {
	t.Errorf("the server-stamped fields must be zero, got %+v", header)
	}
	if header.OriginTimestamp != message1.Header.OriginTimestamp {
	t.Errorf("batch origin timestamp = %d, want %d",
	header.OriginTimestamp, message1.Header.OriginTimestamp)
	}
	if header.BatchLength != uint64(len(record)) {
	t.Errorf("batch length = %d, the record holds %d bytes", header.BatchLength, len(record))
	}
	if header.MessageCount != 1 {
	t.Errorf("message count = %d, want 1", header.MessageCount)
	}
	if got := header.Checksum(record[batch.HeaderSize : batch.HeaderSize+8]); got != header.BatchChecksum {
	t.Errorf("batch checksum = %d, recomputed %d", header.BatchChecksum, got)
	}

	frame, err := batch.DecodeMessageHeader(record[batch.HeaderSize:])
	if err != nil {
	t.Fatal(err)
	}
	if frame.Id != [16]byte(message1.Header.Id) {
	t.Errorf("frame id = %v, want %v", frame.Id, message1.Header.Id)
	}
	if frame.OffsetDelta != 0 \|\| frame.TimestampDelta != 0 {
	t.Errorf("a single-message batch has zero deltas, got %+v", frame)
	}
	if int(frame.PayloadLength) != len(message1.Payload) \|\|
	int(frame.UserHeadersLength) != len(message1.UserHeaders) {
	t.Errorf("frame lengths %d/%d do not match the message %d/%d",
	frame.PayloadLength, frame.UserHeadersLength,
	len(message1.Payload), len(message1.UserHeaders))
	}

	body := record[batch.HeaderSize+batch.MessageHeaderSize:]
	expectedBody := append(append([]byte{}, message1.Payload...), message1.UserHeaders...)
	if !bytes.Equal(body, expectedBody) {
	t.Errorf("the frame body is incorrect.\nExpected:\t%v\nGot:\t\t%v", expectedBody, body)
	}
	}

	func TestSerialize_SendMessagesMintsAZeroMessageId(t *testing.T) {
	message, err := iggcon.NewIggyMessage([]byte("payload"))
	if err != nil {
	t.Fatal(err)
	}
	streamId, _ := iggcon.NewIdentifier(uint32(1))
	topicId, _ := iggcon.NewIdentifier(uint32(1))
	request := SendMessages{
	StreamId: streamId,
	TopicId: topicId,
	Partitioning: iggcon.PartitionId(0),
	Messages: []iggcon.IggyMessage{message},
	Compression: iggcon.MESSAGE_COMPRESSION_NONE,
	}

	serialized, err := request.MarshalBinary()
	if err != nil {
	t.Fatal(err)
	}

	minted := request.Messages[0].Header.Id
	if minted == (iggcon.MessageID{}) {
	t.Fatal("a zero id must be minted before encoding")
	}
	metadataLength := binary.LittleEndian.Uint32(serialized[0:4])
	frame, err := batch.DecodeMessageHeader(serialized[4+int(metadataLength)+batch.HeaderSize:])
	if err != nil {
	t.Fatal(err)
	}
	if frame.Id != [16]byte(minted) {
	t.Errorf("the wire carries id %v, the message holds %v", frame.Id, minted)
	}
	}

	func TestSerialize_SendMessagesRejectsATimestampDeltaPastU32(t *testing.T) {
	streamId, _ := iggcon.NewIdentifier(uint32(1))
	topicId, _ := iggcon.NewIdentifier(uint32(1))
	request := SendMessages{
	StreamId: streamId,
	TopicId: topicId,
	Partitioning: iggcon.PartitionId(0),
	Messages: []iggcon.IggyMessage{
	{
	Header: iggcon.MessageHeader{Id: iggcon.MessageID{1}, OriginTimestamp: 0},
	Payload: []byte("early"),
	},
	{
	Header: iggcon.MessageHeader{
	Id: iggcon.MessageID{2},
	OriginTimestamp: uint64(math.MaxUint32) + 1,
	},
	Payload: []byte("late"),
	},
	},
	Compression: iggcon.MESSAGE_COMPRESSION_NONE,
	}

	if _, err := request.MarshalBinary(); err == nil {
	t.Fatal("a timestamp delta past u32 must fail the encode")
	}
	}

	func TestSerialize_SendMessagesCompressesThePayloadCoherently(t *testing.T) {
	// A compressible payload above the 32-byte floor.
	payload := bytes.Repeat([]byte("abcdefgh"), 32)
	message, err := iggcon.NewIggyMessage(payload)
	if err != nil {
	t.Fatal(err)
	}
	streamId, _ := iggcon.NewIdentifier(uint32(1))
	topicId, _ := iggcon.NewIdentifier(uint32(1))
	request := SendMessages{
	StreamId: streamId,
	TopicId: topicId,
	Partitioning: iggcon.PartitionId(0),
	Messages: []iggcon.IggyMessage{message},
	Compression: iggcon.MESSAGE_COMPRESSION_S2,
	}

	serialized, err := request.MarshalBinary()
	if err != nil {
	t.Fatal(err)
	}

	compressed := request.Messages[0]
	if got := int(compressed.Header.PayloadLength); got != len(compressed.Payload) {
	t.Fatalf("header claims %d payload bytes, the slice holds %d: the wire would mis-frame",
	got, len(compressed.Payload))
	}
	if len(compressed.Payload) >= len(payload) {
	t.Fatalf("the payload did not compress: %d >= %d", len(compressed.Payload), len(payload))
	}

	// The frame must be sized by the compressed length: header, then exactly
	// PayloadLength payload bytes, and nothing after.
	frameStart := len(serialized) - int(compressed.Header.PayloadLength) - batch.MessageHeaderSize
	frame, err := batch.DecodeMessageHeader(serialized[frameStart:])
	if err != nil {
	t.Fatal(err)
	}
	if frame.PayloadLength != compressed.Header.PayloadLength {
	t.Fatalf("wire frame claims %d, in-memory header %d",
	frame.PayloadLength, compressed.Header.PayloadLength)
	}

	decoded, err := s2.Decode(nil, serialized[frameStart+batch.MessageHeaderSize:])
	if err != nil {
	t.Fatal(err)
	}
	if !bytes.Equal(decoded, payload) {
	t.Fatal("the compressed payload does not round-trip")
	}
	}

	func TestSerialize_SendMessagesSkipsCompressionBelowTheFloor(t *testing.T) {
	message, err := iggcon.NewIggyMessage([]byte("short"))
	if err != nil {
	t.Fatal(err)
	}
	streamId, _ := iggcon.NewIdentifier(uint32(1))
	topicId, _ := iggcon.NewIdentifier(uint32(1))
	request := SendMessages{
	StreamId: streamId,
	TopicId: topicId,
	Partitioning: iggcon.PartitionId(0),
	Messages: []iggcon.IggyMessage{message},
	Compression: iggcon.MESSAGE_COMPRESSION_S2,
	}

	if _, err := request.MarshalBinary(); err != nil {
	t.Fatal(err)
	}
	if !bytes.Equal(request.Messages[0].Payload, []byte("short")) {
	t.Fatal("a payload under 32 bytes must pass through uncompressed")
	}
	}

	func createDefaultMessageHeaders() []iggcon.HeaderEntry {
	return []iggcon.HeaderEntry{
	{Key: iggcon.HeaderKey{Kind: iggcon.String, Value: []byte("HeaderKey1")}, Value: iggcon.HeaderValue{Kind: iggcon.String, Value: []byte("Value 1")}},
	{Key: iggcon.HeaderKey{Kind: iggcon.String, Value: []byte("HeaderKey2")}, Value: iggcon.HeaderValue{Kind: iggcon.Uint32, Value: []byte{0x01, 0x02, 0x03, 0x04}}},
	}
	}

	func generateTestMessage(payload string) iggcon.IggyMessage {
	msg, _ := iggcon.NewIggyMessage(
	[]byte(payload),
	iggcon.WithID(uuid.New()),
	iggcon.WithUserHeaders(createDefaultMessageHeaders()))
	return msg
	}