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
|
package sarama
import "github.com/rcrowley/go-metrics"
// PacketEncoder is the interface providing helpers for writing with Kafka's encoding rules.
// Types implementing Encoder only need to worry about calling methods like PutString,
// not about how a string is represented in Kafka.
type packetEncoder interface {
// Primitives
putInt8(in int8)
putInt16(in int16)
putInt32(in int32)
putInt64(in int64)
putVarint(in int64)
putUVarint(in uint64)
putFloat64(in float64)
putCompactArrayLength(in int)
putArrayLength(in int) error
putBool(in bool)
// Collections
putBytes(in []byte) error
putVarintBytes(in []byte) error
putCompactBytes(in []byte) error
putRawBytes(in []byte) error
putCompactString(in string) error
putNullableCompactString(in *string) error
putString(in string) error
putNullableString(in *string) error
putStringArray(in []string) error
putCompactInt32Array(in []int32) error
putNullableCompactInt32Array(in []int32) error
putInt32Array(in []int32) error
putInt64Array(in []int64) error
putEmptyTaggedFieldArray()
// Provide the current offset to record the batch size metric
offset() int
// Stacks, see PushEncoder
push(in pushEncoder)
pop() error
// To record metrics when provided
metricRegistry() metrics.Registry
}
// PushEncoder is the interface for encoding fields like CRCs and lengths where the value
// of the field depends on what is encoded after it in the packet. Start them with PacketEncoder.Push() where
// the actual value is located in the packet, then PacketEncoder.Pop() them when all the bytes they
// depend upon have been written.
type pushEncoder interface {
// Saves the offset into the input buffer as the location to actually write the calculated value when able.
saveOffset(in int)
// Returns the length of data to reserve for the output of this encoder (eg 4 bytes for a CRC32).
reserveLength() int
// Indicates that all required data is now available to calculate and write the field.
// SaveOffset is guaranteed to have been called first. The implementation should write ReserveLength() bytes
// of data to the saved offset, based on the data between the saved offset and curOffset.
run(curOffset int, buf []byte) error
}
// dynamicPushEncoder extends the interface of pushEncoder for uses cases where the length of the
// fields itself is unknown until its value was computed (for instance varint encoded length
// fields).
type dynamicPushEncoder interface {
pushEncoder
// Called during pop() to adjust the length of the field.
// It should return the difference in bytes between the last computed length and current length.
adjustLength(currOffset int) int
}
|
