blob: 7c54748c2b38e8f0e2dc32657fad2f76a387b040 [file] [log] [blame]
Scott Bakereee8dd82019-09-24 12:52:34 -07001package sarama
2
3import (
4 "fmt"
5 "time"
6)
7
Scott Baker611f6bd2019-10-18 13:45:19 -07008const (
9 //CompressionNone no compression
10 CompressionNone CompressionCodec = iota
11 //CompressionGZIP compression using GZIP
12 CompressionGZIP
13 //CompressionSnappy compression using snappy
14 CompressionSnappy
15 //CompressionLZ4 compression using LZ4
16 CompressionLZ4
17 //CompressionZSTD compression using ZSTD
18 CompressionZSTD
Scott Bakereee8dd82019-09-24 12:52:34 -070019
Scott Baker611f6bd2019-10-18 13:45:19 -070020 // The lowest 3 bits contain the compression codec used for the message
21 compressionCodecMask int8 = 0x07
22
23 // Bit 3 set for "LogAppend" timestamps
24 timestampTypeMask = 0x08
25
26 // CompressionLevelDefault is the constant to use in CompressionLevel
27 // to have the default compression level for any codec. The value is picked
28 // that we don't use any existing compression levels.
29 CompressionLevelDefault = -1000
30)
Scott Bakereee8dd82019-09-24 12:52:34 -070031
32// CompressionCodec represents the various compression codecs recognized by Kafka in messages.
33type CompressionCodec int8
34
Scott Bakereee8dd82019-09-24 12:52:34 -070035func (cc CompressionCodec) String() string {
36 return []string{
37 "none",
38 "gzip",
39 "snappy",
40 "lz4",
Scott Baker611f6bd2019-10-18 13:45:19 -070041 "zstd",
Scott Bakereee8dd82019-09-24 12:52:34 -070042 }[int(cc)]
43}
44
Scott Baker611f6bd2019-10-18 13:45:19 -070045//Message is a kafka message type
Scott Bakereee8dd82019-09-24 12:52:34 -070046type Message struct {
47 Codec CompressionCodec // codec used to compress the message contents
48 CompressionLevel int // compression level
49 LogAppendTime bool // the used timestamp is LogAppendTime
50 Key []byte // the message key, may be nil
51 Value []byte // the message contents
52 Set *MessageSet // the message set a message might wrap
53 Version int8 // v1 requires Kafka 0.10
54 Timestamp time.Time // the timestamp of the message (version 1+ only)
55
56 compressedCache []byte
57 compressedSize int // used for computing the compression ratio metrics
58}
59
60func (m *Message) encode(pe packetEncoder) error {
61 pe.push(newCRC32Field(crcIEEE))
62
63 pe.putInt8(m.Version)
64
65 attributes := int8(m.Codec) & compressionCodecMask
66 if m.LogAppendTime {
67 attributes |= timestampTypeMask
68 }
69 pe.putInt8(attributes)
70
71 if m.Version >= 1 {
72 if err := (Timestamp{&m.Timestamp}).encode(pe); err != nil {
73 return err
74 }
75 }
76
77 err := pe.putBytes(m.Key)
78 if err != nil {
79 return err
80 }
81
82 var payload []byte
83
84 if m.compressedCache != nil {
85 payload = m.compressedCache
86 m.compressedCache = nil
87 } else if m.Value != nil {
88
89 payload, err = compress(m.Codec, m.CompressionLevel, m.Value)
90 if err != nil {
91 return err
92 }
93 m.compressedCache = payload
94 // Keep in mind the compressed payload size for metric gathering
95 m.compressedSize = len(payload)
96 }
97
98 if err = pe.putBytes(payload); err != nil {
99 return err
100 }
101
102 return pe.pop()
103}
104
105func (m *Message) decode(pd packetDecoder) (err error) {
Scott Baker611f6bd2019-10-18 13:45:19 -0700106 crc32Decoder := acquireCrc32Field(crcIEEE)
107 defer releaseCrc32Field(crc32Decoder)
108
109 err = pd.push(crc32Decoder)
Scott Bakereee8dd82019-09-24 12:52:34 -0700110 if err != nil {
111 return err
112 }
113
114 m.Version, err = pd.getInt8()
115 if err != nil {
116 return err
117 }
118
119 if m.Version > 1 {
120 return PacketDecodingError{fmt.Sprintf("unknown magic byte (%v)", m.Version)}
121 }
122
123 attribute, err := pd.getInt8()
124 if err != nil {
125 return err
126 }
127 m.Codec = CompressionCodec(attribute & compressionCodecMask)
128 m.LogAppendTime = attribute&timestampTypeMask == timestampTypeMask
129
130 if m.Version == 1 {
131 if err := (Timestamp{&m.Timestamp}).decode(pd); err != nil {
132 return err
133 }
134 }
135
136 m.Key, err = pd.getBytes()
137 if err != nil {
138 return err
139 }
140
141 m.Value, err = pd.getBytes()
142 if err != nil {
143 return err
144 }
145
146 // Required for deep equal assertion during tests but might be useful
147 // for future metrics about the compression ratio in fetch requests
148 m.compressedSize = len(m.Value)
149
150 switch m.Codec {
151 case CompressionNone:
152 // nothing to do
153 default:
154 if m.Value == nil {
155 break
156 }
157
158 m.Value, err = decompress(m.Codec, m.Value)
159 if err != nil {
160 return err
161 }
162 if err := m.decodeSet(); err != nil {
163 return err
164 }
165 }
166
167 return pd.pop()
168}
169
170// decodes a message set from a previously encoded bulk-message
171func (m *Message) decodeSet() (err error) {
172 pd := realDecoder{raw: m.Value}
173 m.Set = &MessageSet{}
174 return m.Set.decode(&pd)
175}