1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
// +build !go1.8
package proto
import (
"bytes"
"fmt"
"reflect"
"testing"
"time"
)
func TestProduceRequest(t *testing.T) {
req := &ProduceReq{
CorrelationID: 241,
ClientID: "test",
RequiredAcks: RequiredAcksAll,
Timeout: time.Second,
Topics: []ProduceReqTopic{
{
Name: "foo",
Partitions: []ProduceReqPartition{
{
ID: 0,
Messages: []*Message{
{
Offset: 0,
Crc: 3099221847,
Key: []byte("foo"),
Value: []byte("bar"),
},
},
},
},
},
},
}
tests := []struct {
Compression Compression
Expected []byte
}{
{
CompressionNone,
[]byte{0x0, 0x0, 0x0, 0x49, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0xf1, 0x0, 0x4, 0x74, 0x65, 0x73, 0x74, 0xff, 0xff, 0x0, 0x0, 0x3, 0xe8, 0x0, 0x0, 0x0, 0x1, 0x0, 0x3, 0x66, 0x6f, 0x6f, 0x0, 0x0, 0x0, 0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x20, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x14, 0xb8, 0xba, 0x5f, 0x57, 0x0, 0x0, 0x0, 0x0, 0x0, 0x3, 0x66, 0x6f, 0x6f, 0x0, 0x0, 0x0, 0x3, 0x62, 0x61, 0x72},
},
{
CompressionGzip,
[]byte{0x0, 0x0, 0x0, 0x6d, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0xf1, 0x0, 0x4, 0x74, 0x65, 0x73, 0x74, 0xff, 0xff, 0x0, 0x0, 0x3, 0xe8, 0x0, 0x0, 0x0, 0x1, 0x0, 0x3, 0x66, 0x6f, 0x6f, 0x0, 0x0, 0x0, 0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x44, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x38, 0x9d, 0x81, 0x74, 0xc4, 0x0, 0x1, 0xff, 0xff, 0xff, 0xff, 0x0, 0x0, 0x0, 0x2a, 0x1f, 0x8b, 0x8, 0x0, 0x0, 0x9, 0x6e, 0x88, 0x0, 0xff, 0x62, 0x40, 0x0, 0x91, 0x1d, 0xbb, 0xe2, 0xc3, 0xc1, 0x2c, 0xe6, 0xb4, 0xfc, 0x7c, 0x10, 0x95, 0x94, 0x58, 0x4, 0x8, 0x0, 0x0, 0xff, 0xff, 0xa0, 0xbc, 0x10, 0xc2, 0x20, 0x0, 0x0, 0x0},
},
{
CompressionSnappy,
[]byte{0x0, 0x0, 0x0, 0x5c, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0xf1, 0x0, 0x4, 0x74, 0x65, 0x73, 0x74, 0xff, 0xff, 0x0, 0x0, 0x3, 0xe8, 0x0, 0x0, 0x0, 0x1, 0x0, 0x3, 0x66, 0x6f, 0x6f, 0x0, 0x0, 0x0, 0x1, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x33, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x27, 0x2e, 0xd4, 0xed, 0xcd, 0x0, 0x2, 0xff, 0xff, 0xff, 0xff, 0x0, 0x0, 0x0, 0x19, 0x20, 0x0, 0x0, 0x19, 0x1, 0x10, 0x14, 0xb8, 0xba, 0x5f, 0x57, 0x5, 0xf, 0x28, 0x3, 0x66, 0x6f, 0x6f, 0x0, 0x0, 0x0, 0x3, 0x62, 0x61, 0x72},
},
}
for _, tt := range tests {
req.Compression = tt.Compression
testRequestSerialization(t, req)
b, _ := req.Bytes()
if !bytes.Equal(b, tt.Expected) {
fmt.Printf("%#v\n", tt.Expected)
fmt.Printf("%#v\n", b)
t.Fatalf("expected different bytes representation: %#v", b)
}
r, _ := ReadProduceReq(bytes.NewBuffer(tt.Expected))
req.Compression = CompressionNone // isn't set on deserialization
if !reflect.DeepEqual(r, req) {
t.Fatalf("malformed request: %#v", r)
}
}
}
|
