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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
|
-- Copyright (C) Dejiang Zhu(doujiang24)
local ffi = require "ffi"
local bit = require "bit"
local setmetatable = setmetatable
local concat = table.concat
local rshift = bit.rshift
local band = bit.band
local char = string.char
local crc32 = ngx.crc32_long
local tonumber = tonumber
local _M = {}
local mt = { __index = _M }
local MESSAGE_VERSION_0 = 0
local MESSAGE_VERSION_1 = 1
local API_VERSION_V0 = 0
local API_VERSION_V1 = 1
local API_VERSION_V2 = 2
_M.ProduceRequest = 0
_M.FetchRequest = 1
_M.OffsetRequest = 2
_M.MetadataRequest = 3
_M.OffsetCommitRequest = 8
_M.OffsetFetchRequest = 9
_M.ConsumerMetadataRequest = 10
local function str_int8(int)
return char(band(int, 0xff))
end
local function str_int16(int)
return char(band(rshift(int, 8), 0xff),
band(int, 0xff))
end
local function str_int32(int)
-- ngx.say(debug.traceback())
return char(band(rshift(int, 24), 0xff),
band(rshift(int, 16), 0xff),
band(rshift(int, 8), 0xff),
band(int, 0xff))
end
-- XX int can be cdata: LL or lua number
local function str_int64(int)
return char(tonumber(band(rshift(int, 56), 0xff)),
tonumber(band(rshift(int, 48), 0xff)),
tonumber(band(rshift(int, 40), 0xff)),
tonumber(band(rshift(int, 32), 0xff)),
tonumber(band(rshift(int, 24), 0xff)),
tonumber(band(rshift(int, 16), 0xff)),
tonumber(band(rshift(int, 8), 0xff)),
tonumber(band(int, 0xff)))
end
function _M.new(self, apikey, correlation_id, client_id, api_version)
local c_len = #client_id
api_version = api_version or API_VERSION_V0
local req = {
0, -- request size: int32
str_int16(apikey),
str_int16(api_version),
str_int32(correlation_id),
str_int16(c_len),
client_id,
}
return setmetatable({
_req = req,
api_key = apikey,
api_version = api_version,
offset = 7,
len = c_len + 10,
}, mt)
end
function _M.int16(self, int)
local req = self._req
local offset = self.offset
req[offset] = str_int16(int)
self.offset = offset + 1
self.len = self.len + 2
end
function _M.int32(self, int)
local req = self._req
local offset = self.offset
req[offset] = str_int32(int)
self.offset = offset + 1
self.len = self.len + 4
end
function _M.int64(self, int)
local req = self._req
local offset = self.offset
req[offset] = str_int64(int)
self.offset = offset + 1
self.len = self.len + 8
end
function _M.string(self, str)
local req = self._req
local offset = self.offset
local str_len = #str
req[offset] = str_int16(str_len)
req[offset + 1] = str
self.offset = offset + 2
self.len = self.len + 2 + str_len
end
function _M.bytes(self, str)
local req = self._req
local offset = self.offset
local str_len = #str
req[offset] = str_int32(str_len)
req[offset + 1] = str
self.offset = offset + 2
self.len = self.len + 4 + str_len
end
local function message_package(key, msg, message_version)
local key = key or ""
local key_len = #key
local len = #msg
local req
local head_len
if message_version == MESSAGE_VERSION_1 then
req = {
-- MagicByte
str_int8(1),
-- XX hard code no Compression
str_int8(0),
str_int64(ffi.new("int64_t", (os.time() * 1000))), -- timestamp
str_int32(key_len),
key,
str_int32(len),
msg,
}
head_len = 22
else
req = {
-- MagicByte
str_int8(0),
-- XX hard code no Compression
str_int8(0),
str_int32(key_len),
key,
str_int32(len),
msg,
}
head_len = 14
end
local str = concat(req)
return crc32(str), str, key_len + len + head_len
end
function _M.message_set(self, messages, index)
local req = self._req
local off = self.offset
local msg_set_size = 0
local index = index or #messages
local message_version = MESSAGE_VERSION_0
if self.api_key == _M.ProduceRequest and self.api_version == API_VERSION_V2 then
message_version = MESSAGE_VERSION_1
end
for i = 1, index, 2 do
local crc32, str, msg_len = message_package(messages[i], messages[i + 1], message_version)
req[off + 1] = str_int64(0) -- offset
req[off + 2] = str_int32(msg_len) -- include the crc32 length
req[off + 3] = str_int32(crc32)
req[off + 4] = str
off = off + 4
msg_set_size = msg_set_size + msg_len + 12
end
req[self.offset] = str_int32(msg_set_size) -- MessageSetSize
self.offset = off + 1
self.len = self.len + 4 + msg_set_size
end
function _M.package(self)
local req = self._req
req[1] = str_int32(self.len)
return req
end
return _M
|
