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
75
76
77
78
79
80
81
82
83
84
|
package kafka
import (
"bufio"
"encoding/binary"
"fmt"
)
type apiKey int16
const (
produceRequest apiKey = 0
fetchRequest apiKey = 1
listOffsetRequest apiKey = 2
metadataRequest apiKey = 3
offsetCommitRequest apiKey = 8
offsetFetchRequest apiKey = 9
groupCoordinatorRequest apiKey = 10
joinGroupRequest apiKey = 11
heartbeatRequest apiKey = 12
leaveGroupRequest apiKey = 13
syncGroupRequest apiKey = 14
describeGroupsRequest apiKey = 15
listGroupsRequest apiKey = 16
createTopicsRequest apiKey = 19
deleteTopicsRequest apiKey = 20
)
type apiVersion int16
const (
v0 apiVersion = 0
v1 apiVersion = 1
v2 apiVersion = 2
v3 apiVersion = 3
)
type requestHeader struct {
Size int32
ApiKey int16
ApiVersion int16
CorrelationID int32
ClientID string
}
func (h requestHeader) size() int32 {
return 4 + 2 + 2 + 4 + sizeofString(h.ClientID)
}
func (h requestHeader) writeTo(w *bufio.Writer) {
writeInt32(w, h.Size)
writeInt16(w, h.ApiKey)
writeInt16(w, h.ApiVersion)
writeInt32(w, h.CorrelationID)
writeString(w, h.ClientID)
}
type request interface {
size() int32
writeTo(*bufio.Writer)
}
func makeInt8(b []byte) int8 {
return int8(b[0])
}
func makeInt16(b []byte) int16 {
return int16(binary.BigEndian.Uint16(b))
}
func makeInt32(b []byte) int32 {
return int32(binary.BigEndian.Uint32(b))
}
func makeInt64(b []byte) int64 {
return int64(binary.BigEndian.Uint64(b))
}
func expectZeroSize(sz int, err error) error {
if err == nil && sz != 0 {
err = fmt.Errorf("reading a response left %d unread bytes", sz)
}
return err
}
|
