mqtt.luaのコードリスト
以下のコードである mqtt.lua は、プロトコル拡張を使用して Citrix ADC で MQTT プロトコルを実装するためのコードを示します。このコードには、TCP クライアントデータコールバック関数 (client.on_data ()) のみが定義されています。サーバーデータの場合、コールバック関数を追加せず、サーバーからクライアントへの高速ネイティブパスを取得します。クライアントデータの場合、コードは CONNECT MQTT プロトコルのメッセージを解析し、ClientID を抽出します。次に、user_token の ClientID を使用します。この値は、LB 仮想サーバーの LB メソッドを USER_TOKEN に設定することで、クライアント ID に基づいて接続のすべてのクライアントトラフィックの負荷分散に使用されます。ユーザーセッション値にもクライアント ID を使用します。これは、LB 仮想サーバーの永続性タイプを USERSESSION に設定することで、LB 永続性に使用できます。このコードでは、ns.send () を使用して LB を行い、初期データを送信します。ns.pipe () API を使用して、残りのクライアントトラフィックをサーバー接続に直接送信し、拡張コールバックハンドラへの呼び出しをバイパスします。
--[[
MQTT event handler for TCP client data
ctxt - TCP client side App processing context.
data - TCP Data stream received.
- parse the client ID from the connect message - the first message should be connect
- send the data to LB with ClientID as user token and session
- pipe the subsequent data to LB directly. This way the subsequent MQTT traffic will
bypass the tcp client on_data handler
- if a parse error is seen, throw an error so the connection is reset
--]]
function client.on_data(ctxt, payload)
local data = payload.data
local data_len = data:len()
local offset = 1
local byte = nil
local utf8_str_len = 0
local msg_type = 0
local multiplier = 1
local max_multiplier = 128 * 128 * 128
local rem_length = 0
local clientID = nil
-- check if MQTT fixed header is present (fixed header length is atleast 2 bytes)
if (data_len < 2) then
goto need_more_data
end
byte = data:byte(offset)
offset = offset + 1
-- check for connect packet - type value 1
msg_type = bit32.rshift(byte, 4)
if (msg_type ~= 1) then
error("Missing MQTT Connect packet.")
end
-- parse the remaining length
repeat
if (multiplier > max_multiplier) then
error("MQTT CONNECT packet parse error - invalid Remaining Length.")
end
if (data_len < offset) then
goto need_more_data
end
byte = data:byte(offset)
offset = offset + 1
rem_length = rem_length + (bit32.band(byte, 0x7F) * multiplier)
multiplier = multiplier * 128
until (bit32.band(byte, 0x80) == 0)
-- protocol name
-- check if protocol name length is present
if (data_len < offset + 1) then
goto need_more_data
end
-- protocol name length MSB
byte = data:byte(offset)
offset = offset + 1
utf8_str_len = byte * 256
-- length LSB
byte = data:byte(offset)
offset = offset + 1
utf8_str_len = utf8_str_len + byte
-- skip the variable header for connect message
-- the four required fields (protocol name, protocol level, connect flags, keep alive)
offset = offset + utf8_str_len + 4
-- parse the client ID
--
-- check if client ID len is present
if (data_len < offset + 1) then
goto need_more_data
end
-- client ID length MSB
byte = data:byte(offset)
offset = offset + 1
utf8_str_len = byte * 256
-- length LSB
byte = data:byte(offset)
offset = offset + 1
utf8_str_len = utf8_str_len + byte
if (data_len < (offset + utf8_str_len - 1)) then
goto need_more_data
end
clientID = data:sub(offset, offset + utf8_str_len - 1)
-- send the data so far to lb, user_token is set to do LB based on clientID
-- user_session is set to clientID as well (it will be used to persist session)
ns.send(ctxt.output, "DATA", {data = data,
user_token = clientID,
user_session = clientID})
-- pipe the subsequent traffic to the lb - to bypass the extension handler
ns.pipe(ctxt.input, ctxt.output)
goto parse_done
::need_more_data::
ctxt:hold(data)
::parse_done::
return
end
<!--NeedCopy-->
mqtt.luaのコードリスト
この記事の概要
コピー完了
コピー失敗