585 lines
27 KiB
Erlang
585 lines
27 KiB
Erlang
%%--------------------------------------------------------------------
|
|
%% Copyright (c) 2018-2023 EMQ Technologies Co., Ltd. All Rights Reserved.
|
|
%%
|
|
%% Licensed under the Apache License, Version 2.0 (the "License");
|
|
%% you may not use this file except in compliance with the License.
|
|
%% You may obtain a copy of the License at
|
|
%%
|
|
%% http://www.apache.org/licenses/LICENSE-2.0
|
|
%%
|
|
%% Unless required by applicable law or agreed to in writing, software
|
|
%% distributed under the License is distributed on an "AS IS" BASIS,
|
|
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
%% See the License for the specific language governing permissions and
|
|
%% limitations under the License.
|
|
%%--------------------------------------------------------------------
|
|
|
|
-module(emqx_frame_SUITE).
|
|
|
|
-compile(export_all).
|
|
-compile(nowarn_export_all).
|
|
|
|
-include_lib("emqx/include/emqx_mqtt.hrl").
|
|
-include_lib("eunit/include/eunit.hrl").
|
|
-include_lib("common_test/include/ct.hrl").
|
|
-include_lib("emqx_ct_helpers/include/emqx_ct.hrl").
|
|
|
|
all() ->
|
|
[{group, parse},
|
|
{group, connect},
|
|
{group, connack},
|
|
{group, publish},
|
|
{group, puback},
|
|
{group, subscribe},
|
|
{group, suback},
|
|
{group, unsubscribe},
|
|
{group, unsuback},
|
|
{group, ping},
|
|
{group, disconnect},
|
|
{group, auth}
|
|
].
|
|
|
|
groups() ->
|
|
[{parse, [parallel],
|
|
[t_parse_cont,
|
|
t_parse_frame_too_large,
|
|
t_parse_frame_malformed_variable_byte_integer,
|
|
t_parse_frame_variable_byte_integer,
|
|
t_parse_malformed_utf8_string,
|
|
t_parse_frame_proxy_protocol %% proxy_protocol_config_disabled packet.
|
|
]},
|
|
{connect, [parallel],
|
|
[t_serialize_parse_v3_connect,
|
|
t_serialize_parse_v4_connect,
|
|
t_serialize_parse_v5_connect,
|
|
t_serialize_parse_connect_without_clientid,
|
|
t_serialize_parse_connect_with_will,
|
|
t_serialize_parse_bridge_connect
|
|
]},
|
|
{connack, [parallel],
|
|
[t_serialize_parse_connack,
|
|
t_serialize_parse_connack_v5
|
|
]},
|
|
{publish, [parallel],
|
|
[t_parse_sticky_frames,
|
|
t_serialize_parse_qos0_publish,
|
|
t_serialize_parse_qos1_publish,
|
|
t_serialize_parse_qos2_publish,
|
|
t_serialize_parse_publish_v5
|
|
]},
|
|
{puback, [parallel],
|
|
[t_serialize_parse_puback,
|
|
t_serialize_parse_puback_v3_4,
|
|
t_serialize_parse_puback_v5,
|
|
t_serialize_parse_pubrec,
|
|
t_serialize_parse_pubrec_v5,
|
|
t_serialize_parse_pubrel,
|
|
t_serialize_parse_pubrel_v5,
|
|
t_serialize_parse_pubcomp,
|
|
t_serialize_parse_pubcomp_v5
|
|
]},
|
|
{subscribe, [parallel],
|
|
[t_serialize_parse_subscribe,
|
|
t_serialize_parse_subscribe_v5
|
|
]},
|
|
{suback, [parallel],
|
|
[t_serialize_parse_suback,
|
|
t_serialize_parse_suback_v5
|
|
]},
|
|
{unsubscribe, [parallel],
|
|
[t_serialize_parse_unsubscribe,
|
|
t_serialize_parse_unsubscribe_v5
|
|
]},
|
|
{unsuback, [parallel],
|
|
[t_serialize_parse_unsuback,
|
|
t_serialize_parse_unsuback_v5
|
|
]},
|
|
{ping, [parallel],
|
|
[t_serialize_parse_pingreq,
|
|
t_serialize_parse_pingresp
|
|
]},
|
|
{disconnect, [parallel],
|
|
[t_serialize_parse_disconnect,
|
|
t_serialize_parse_disconnect_v5
|
|
]},
|
|
{auth, [parallel],
|
|
[t_serialize_parse_auth_v5]
|
|
}].
|
|
|
|
init_per_suite(Config) ->
|
|
Config.
|
|
|
|
end_per_suite(_Config) ->
|
|
ok.
|
|
|
|
init_per_group(_Group, Config) ->
|
|
Config.
|
|
|
|
end_per_group(_Group, _Config) ->
|
|
ok.
|
|
|
|
t_parse_cont(_) ->
|
|
Packet = ?CONNECT_PACKET(#mqtt_packet_connect{}),
|
|
ParseState = emqx_frame:initial_parse_state(),
|
|
<<HdrBin:1/binary, LenBin:1/binary, RestBin/binary>> = serialize_to_binary(Packet),
|
|
{more, ContParse} = emqx_frame:parse(<<>>, ParseState),
|
|
{more, ContParse1} = emqx_frame:parse(HdrBin, ContParse),
|
|
{more, ContParse2} = emqx_frame:parse(LenBin, ContParse1),
|
|
{more, ContParse3} = emqx_frame:parse(<<>>, ContParse2),
|
|
{ok, Packet, <<>>, _} = emqx_frame:parse(RestBin, ContParse3).
|
|
|
|
t_parse_frame_too_large(_) ->
|
|
Packet = ?PUBLISH_PACKET(?QOS_1, <<"t">>, 1, payload(1000)),
|
|
?catch_error(frame_too_large, parse_serialize(Packet, #{max_size => 256})),
|
|
?catch_error(frame_too_large, parse_serialize(Packet, #{max_size => 512})),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{max_size => 2048, version => ?MQTT_PROTO_V4})).
|
|
|
|
t_parse_frame_malformed_variable_byte_integer(_) ->
|
|
MalformedPayload = << <<16#80>> || _ <- lists:seq(1, 4) >>,
|
|
ParseState = emqx_frame:initial_parse_state(#{}),
|
|
?catch_error(malformed_variable_byte_integer,
|
|
emqx_frame:parse(MalformedPayload, ParseState)).
|
|
|
|
t_parse_frame_variable_byte_integer(_) ->
|
|
Bin = <<2#10010011, 2#10000000, 2#10001000, 2#10011001, 2#10101101, 2#00110010>>,
|
|
?catch_error(malformed_variable_byte_integer,
|
|
emqx_frame:parse_variable_byte_integer(Bin)).
|
|
|
|
t_parse_malformed_utf8_string(_) ->
|
|
MalformedPacket = <<16,31,0,4,
|
|
%% Specification name, should be "MQTT"
|
|
%% 77,81,84,84,
|
|
%% malformed 1-Byte UTF-8 in (U+0000 .. U+001F] && [U+007F])
|
|
16#00,16#01,16#1F,16#7F,
|
|
|
|
4,194,0,60,
|
|
0,4,101,109,
|
|
113,120,0,5,
|
|
97,100,109,105,
|
|
110,0,6,112,
|
|
117,98,108,105,
|
|
99>>,
|
|
ParseState = emqx_frame:initial_parse_state(#{strict_mode => true}),
|
|
?catch_error(utf8_string_invalid, emqx_frame:parse(MalformedPacket, ParseState)).
|
|
|
|
t_parse_frame_proxy_protocol(_) ->
|
|
BinList = [ <<"PROXY TCP4 ">>, <<"PROXY TCP6 ">>, <<"PROXY UNKNOWN">>
|
|
, <<"\r\n\r\n\0\r\nQUIT\n">>],
|
|
[?assertError( proxy_protocol_config_disabled
|
|
, emqx_frame:parse(Bin))
|
|
|| Bin <- BinList].
|
|
|
|
t_serialize_parse_v3_connect(_) ->
|
|
Bin = <<16,37,0,6,77,81,73,115,100,112,3,2,0,60,0,23,109,111,115,
|
|
113,112,117, 98,47,49,48,52,53,49,45,105,77,97,99,46,108,
|
|
111,99,97>>,
|
|
Packet = ?CONNECT_PACKET(
|
|
#mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V3,
|
|
proto_name = <<"MQIsdp">>,
|
|
clientid = <<"mosqpub/10451-iMac.loca">>,
|
|
clean_start = true,
|
|
keepalive = 60
|
|
}),
|
|
{ok, Packet, <<>>, PState} = emqx_frame:parse(Bin),
|
|
?assertMatch({none, #{version := ?MQTT_PROTO_V3}}, PState).
|
|
|
|
t_serialize_parse_v4_connect(_) ->
|
|
Bin = <<16,35,0,4,77,81,84,84,4,2,0,60,0,23,109,111,115,113,112,117,
|
|
98,47,49,48,52,53,49,45,105,77,97,99,46,108,111,99,97>>,
|
|
Packet = ?CONNECT_PACKET(
|
|
#mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V4,
|
|
proto_name = <<"MQTT">>,
|
|
clientid = <<"mosqpub/10451-iMac.loca">>,
|
|
clean_start = true,
|
|
keepalive = 60
|
|
}),
|
|
?assertEqual(Bin, serialize_to_binary(Packet)),
|
|
?assertMatch({ok, Packet, <<>>, _}, emqx_frame:parse(Bin)).
|
|
|
|
t_serialize_parse_v5_connect(_) ->
|
|
Props = #{'Session-Expiry-Interval' => 60,
|
|
'Receive-Maximum' => 100,
|
|
'Maximum-QoS' => ?QOS_2,
|
|
'Retain-Available' => 1,
|
|
'Maximum-Packet-Size' => 1024,
|
|
'Topic-Alias-Maximum' => 10,
|
|
'Request-Response-Information' => 1,
|
|
'Request-Problem-Information' => 1,
|
|
'Authentication-Method' => <<"oauth2">>,
|
|
'Authentication-Data' => <<"33kx93k">>
|
|
},
|
|
|
|
WillProps = #{'Will-Delay-Interval' => 60,
|
|
'Payload-Format-Indicator' => 1,
|
|
'Message-Expiry-Interval' => 60,
|
|
'Content-Type' => <<"text/json">>,
|
|
'Response-Topic' => <<"topic">>,
|
|
'Correlation-Data' => <<"correlateid">>,
|
|
'User-Property' => [{<<"k">>, <<"v">>}]
|
|
},
|
|
Packet = ?CONNECT_PACKET(
|
|
#mqtt_packet_connect{proto_name = <<"MQTT">>,
|
|
proto_ver = ?MQTT_PROTO_V5,
|
|
is_bridge = false,
|
|
clean_start = true,
|
|
clientid = <<>>,
|
|
will_flag = true,
|
|
will_qos = ?QOS_1,
|
|
will_retain = false,
|
|
keepalive = 60,
|
|
properties = Props,
|
|
will_props = WillProps,
|
|
will_topic = <<"topic">>,
|
|
will_payload = <<>>,
|
|
username = <<"device:1">>,
|
|
password = <<"passwd">>
|
|
}),
|
|
?assertEqual(Packet, parse_serialize(Packet)).
|
|
|
|
t_serialize_parse_connect_without_clientid(_) ->
|
|
Bin = <<16,12,0,4,77,81,84,84,4,2,0,60,0,0>>,
|
|
Packet = ?CONNECT_PACKET(#mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V4,
|
|
proto_name = <<"MQTT">>,
|
|
clientid = <<>>,
|
|
clean_start = true,
|
|
keepalive = 60
|
|
}),
|
|
?assertEqual(Bin, serialize_to_binary(Packet)),
|
|
?assertMatch({ok, Packet, <<>>, _}, emqx_frame:parse(Bin)).
|
|
|
|
t_serialize_parse_connect_with_will(_) ->
|
|
Bin = <<16,67,0,6,77,81,73,115,100,112,3,206,0,60,0,23,109,111,115,113,112,
|
|
117,98,47,49,48,52,53,50,45,105,77,97,99,46,108,111,99,97,0,5,47,119,
|
|
105,108,108,0,7,119,105,108,108,109,115,103,0,4,116,101,115,116,0,6,
|
|
112,117,98,108,105,99>>,
|
|
Packet = #mqtt_packet{header = #mqtt_packet_header{type = ?CONNECT},
|
|
variable = #mqtt_packet_connect{proto_ver = ?MQTT_PROTO_V3,
|
|
proto_name = <<"MQIsdp">>,
|
|
clientid = <<"mosqpub/10452-iMac.loca">>,
|
|
clean_start = true,
|
|
keepalive = 60,
|
|
will_retain = false,
|
|
will_qos = ?QOS_1,
|
|
will_flag = true,
|
|
will_topic = <<"/will">>,
|
|
will_payload = <<"willmsg">>,
|
|
username = <<"test">>,
|
|
password = <<"public">>
|
|
}},
|
|
?assertEqual(Bin, serialize_to_binary(Packet)),
|
|
?assertMatch({ok, Packet, <<>>, _}, emqx_frame:parse(Bin)).
|
|
|
|
t_serialize_parse_bridge_connect(_) ->
|
|
Bin = <<16,86,0,6,77,81,73,115,100,112,131,44,0,60,0,19,67,95,48,48,58,48,67,
|
|
58,50,57,58,50,66,58,55,55,58,53,50,0,48,36,83,89,83,47,98,114,111,107,
|
|
101,114,47,99,111,110,110,101,99,116,105,111,110,47,67,95,48,48,58,48,
|
|
67,58,50,57,58,50,66,58,55,55,58,53,50,47,115,116,97,116,101,0,1,48>>,
|
|
Topic = <<"$SYS/broker/connection/C_00:0C:29:2B:77:52/state">>,
|
|
Packet = #mqtt_packet{header = #mqtt_packet_header{type = ?CONNECT},
|
|
variable = #mqtt_packet_connect{clientid = <<"C_00:0C:29:2B:77:52">>,
|
|
proto_ver = 16#03,
|
|
proto_name = <<"MQIsdp">>,
|
|
is_bridge = true,
|
|
will_retain = true,
|
|
will_qos = ?QOS_1,
|
|
will_flag = true,
|
|
clean_start = false,
|
|
keepalive = 60,
|
|
will_topic = Topic,
|
|
will_payload = <<"0">>
|
|
}},
|
|
?assertEqual(Bin, serialize_to_binary(Packet)),
|
|
?assertMatch({ok, Packet, <<>>, _}, emqx_frame:parse(Bin)),
|
|
Packet1 = ?CONNECT_PACKET(#mqtt_packet_connect{is_bridge = true}),
|
|
?assertEqual(Packet1, parse_serialize(Packet1)).
|
|
|
|
t_serialize_parse_connack(_) ->
|
|
Packet = ?CONNACK_PACKET(?RC_SUCCESS),
|
|
?assertEqual(<<32,2,0,0>>, serialize_to_binary(Packet)),
|
|
?assertEqual(Packet, parse_serialize(Packet)).
|
|
|
|
t_serialize_parse_connack_v5(_) ->
|
|
Props = #{'Session-Expiry-Interval' => 60,
|
|
'Receive-Maximum' => 100,
|
|
'Maximum-QoS' => ?QOS_2,
|
|
'Retain-Available' => 1,
|
|
'Maximum-Packet-Size' => 1024,
|
|
'Assigned-Client-Identifier' => <<"id">>,
|
|
'Topic-Alias-Maximum' => 10,
|
|
'Reason-String' => <<>>,
|
|
'Wildcard-Subscription-Available' => 1,
|
|
'Subscription-Identifier-Available' => 1,
|
|
'Shared-Subscription-Available' => 1,
|
|
'Server-Keep-Alive' => 60,
|
|
'Response-Information' => <<"response">>,
|
|
'Server-Reference' => <<"192.168.1.10">>,
|
|
'Authentication-Method' => <<"oauth2">>,
|
|
'Authentication-Data' => <<"33kx93k">>
|
|
},
|
|
Packet = ?CONNACK_PACKET(?RC_SUCCESS, 0, Props),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_parse_sticky_frames(_) ->
|
|
Payload = lists:duplicate(10, 0),
|
|
P = #mqtt_packet{header = #mqtt_packet_header{type = ?PUBLISH,
|
|
dup = false,
|
|
qos = ?QOS_0,
|
|
retain = false},
|
|
variable = #mqtt_packet_publish{topic_name = <<"a/b">>,
|
|
packet_id = undefined},
|
|
payload = iolist_to_binary(Payload)
|
|
},
|
|
Bin = serialize_to_binary(P),
|
|
Size = size(Bin),
|
|
<<H:(Size-2)/binary, TailTwoBytes/binary>> = Bin,
|
|
{more, PState1} = emqx_frame:parse(H), %% needs 2 more bytes
|
|
%% feed 3 bytes as if the next 1 byte belongs to the next packet.
|
|
{ok, _, <<42>>, PState2} = emqx_frame:parse(iolist_to_binary([TailTwoBytes, 42]), PState1),
|
|
?assertMatch({none, _}, PState2).
|
|
|
|
t_serialize_parse_qos0_publish(_) ->
|
|
Bin = <<48,14,0,7,120,120,120,47,121,121,121,104,101,108,108,111>>,
|
|
Packet = #mqtt_packet{header = #mqtt_packet_header{type = ?PUBLISH,
|
|
dup = false,
|
|
qos = ?QOS_0,
|
|
retain = false},
|
|
variable = #mqtt_packet_publish{topic_name = <<"xxx/yyy">>,
|
|
packet_id = undefined},
|
|
payload = <<"hello">>},
|
|
?assertEqual(Bin, serialize_to_binary(Packet)),
|
|
?assertMatch(Packet, parse_to_packet(Bin, #{strict_mode => true})).
|
|
|
|
t_serialize_parse_qos1_publish(_) ->
|
|
Bin = <<50,13,0,5,97,47,98,47,99,0,1,104,97,104,97>>,
|
|
Packet = #mqtt_packet{header = #mqtt_packet_header{type = ?PUBLISH,
|
|
dup = false,
|
|
qos = ?QOS_1,
|
|
retain = false},
|
|
variable = #mqtt_packet_publish{topic_name = <<"a/b/c">>,
|
|
packet_id = 1},
|
|
payload = <<"haha">>},
|
|
?assertEqual(Bin, serialize_to_binary(Packet)),
|
|
?assertMatch(Packet, parse_to_packet(Bin, #{strict_mode => true})),
|
|
%% strict_mode = true
|
|
?catch_error(bad_packet_id, parse_serialize(?PUBLISH_PACKET(?QOS_1, <<"Topic">>, 0, <<>>))),
|
|
%% strict_mode = false
|
|
_ = parse_serialize(?PUBLISH_PACKET(?QOS_1, <<"Topic">>, 0, <<>>), #{strict_mode => false}).
|
|
|
|
t_serialize_parse_qos2_publish(_) ->
|
|
Packet = ?PUBLISH_PACKET(?QOS_2, <<"Topic">>, 1, <<>>),
|
|
Bin = <<52,9,0,5,84,111,112,105,99,0,1>>,
|
|
?assertEqual(Packet, parse_serialize(Packet)),
|
|
?assertEqual(Bin, serialize_to_binary(Packet)),
|
|
?assertMatch(Packet, parse_to_packet(Bin, #{strict_mode => true})),
|
|
%% strict_mode = true
|
|
?catch_error(bad_packet_id, parse_serialize(?PUBLISH_PACKET(?QOS_2, <<"Topic">>, 0, <<>>))),
|
|
%% strict_mode = false
|
|
_ = parse_serialize(?PUBLISH_PACKET(?QOS_2, <<"Topic">>, 0, <<>>), #{strict_mode => false}).
|
|
|
|
t_serialize_parse_publish_v5(_) ->
|
|
Props = #{'Payload-Format-Indicator' => 1,
|
|
'Message-Expiry-Interval' => 60,
|
|
'Topic-Alias' => 16#AB,
|
|
'Response-Topic' => <<"reply">>,
|
|
'Correlation-Data' => <<"correlation-id">>,
|
|
'Subscription-Identifier' => 1,
|
|
'Content-Type' => <<"text/json">>},
|
|
Packet = ?PUBLISH_PACKET(?QOS_1, <<"$share/group/topic">>, 1, Props, <<"payload">>),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_serialize_parse_puback(_) ->
|
|
Packet = ?PUBACK_PACKET(1),
|
|
?assertEqual(<<64,2,0,1>>, serialize_to_binary(Packet)),
|
|
?assertEqual(Packet, parse_serialize(Packet)),
|
|
%% strict_mode = true
|
|
?catch_error(bad_packet_id, parse_serialize(?PUBACK_PACKET(0))),
|
|
%% strict_mode = false
|
|
?PUBACK_PACKET(0) = parse_serialize(?PUBACK_PACKET(0), #{strict_mode => false}).
|
|
|
|
t_serialize_parse_puback_v3_4(_) ->
|
|
Bin = <<64,2,0,1>>,
|
|
Packet = #mqtt_packet{header = #mqtt_packet_header{type = ?PUBACK}, variable = 1},
|
|
?assertEqual(Bin, serialize_to_binary(Packet, ?MQTT_PROTO_V3)),
|
|
?assertEqual(Bin, serialize_to_binary(Packet, ?MQTT_PROTO_V4)),
|
|
?assertEqual(?PUBACK_PACKET(1), parse_to_packet(Bin, #{version => ?MQTT_PROTO_V3})),
|
|
?assertEqual(?PUBACK_PACKET(1), parse_to_packet(Bin, #{version => ?MQTT_PROTO_V4})).
|
|
|
|
t_serialize_parse_puback_v5(_) ->
|
|
Packet = ?PUBACK_PACKET(16, ?RC_SUCCESS, #{'Reason-String' => <<"success">>}),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_serialize_parse_pubrec(_) ->
|
|
Packet = ?PUBREC_PACKET(1),
|
|
?assertEqual(<<5:4,0:4,2,0,1>>, serialize_to_binary(Packet)),
|
|
?assertEqual(Packet, parse_serialize(Packet)),
|
|
%% strict_mode = true
|
|
?catch_error(bad_packet_id, parse_serialize(?PUBREC_PACKET(0))),
|
|
%% strict_mode = false
|
|
?PUBREC_PACKET(0) = parse_serialize(?PUBREC_PACKET(0), #{strict_mode => false}).
|
|
|
|
t_serialize_parse_pubrec_v5(_) ->
|
|
Packet = ?PUBREC_PACKET(16, ?RC_SUCCESS, #{'Reason-String' => <<"success">>}),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_serialize_parse_pubrel(_) ->
|
|
Packet = ?PUBREL_PACKET(1),
|
|
Bin = serialize_to_binary(Packet),
|
|
?assertEqual(<<6:4,2:4,2,0,1>>, Bin),
|
|
?assertEqual(Packet, parse_serialize(Packet)),
|
|
%% PUBREL with bad qos 0
|
|
Bin0 = <<6:4,0:4,2,0,1>>,
|
|
?assertMatch(Packet, parse_to_packet(Bin0, #{strict_mode => false})),
|
|
?catch_error(bad_frame_header, parse_to_packet(Bin0, #{strict_mode => true})),
|
|
%% strict_mode = false
|
|
?PUBREL_PACKET(0) = parse_serialize(?PUBREL_PACKET(0), #{strict_mode => false}),
|
|
%% strict_mode = true
|
|
?catch_error(bad_packet_id, parse_serialize(?PUBREL_PACKET(0))).
|
|
|
|
t_serialize_parse_pubrel_v5(_) ->
|
|
Packet = ?PUBREL_PACKET(16, ?RC_SUCCESS, #{'Reason-String' => <<"success">>}),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_serialize_parse_pubcomp(_) ->
|
|
Packet = ?PUBCOMP_PACKET(1),
|
|
Bin = serialize_to_binary(Packet),
|
|
?assertEqual(<<7:4,0:4,2,0,1>>, Bin),
|
|
?assertEqual(Packet, parse_serialize(Packet)),
|
|
%% strict_mode = false
|
|
?PUBCOMP_PACKET(0) = parse_serialize(?PUBCOMP_PACKET(0), #{strict_mode => false}),
|
|
%% strict_mode = true
|
|
?catch_error(bad_packet_id, parse_serialize(?PUBCOMP_PACKET(0))).
|
|
|
|
t_serialize_parse_pubcomp_v5(_) ->
|
|
Packet = ?PUBCOMP_PACKET(16, ?RC_SUCCESS, #{'Reason-String' => <<"success">>}),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_serialize_parse_subscribe(_) ->
|
|
%% SUBSCRIBE(Q1, R0, D0, PacketId=2, TopicTable=[{<<"TopicA">>,2}])
|
|
Bin = <<?SUBSCRIBE:4,2:4,11,0,2,0,6,84,111,112,105,99,65,2>>,
|
|
TopicOpts = #{nl => 0 , rap => 0, rh => 0, qos => 2},
|
|
TopicFilters = [{<<"TopicA">>, TopicOpts}],
|
|
Packet = ?SUBSCRIBE_PACKET(2, TopicFilters),
|
|
?assertEqual(Bin, serialize_to_binary(Packet)),
|
|
?assertMatch(Packet, parse_to_packet(Bin, #{strict_mode => true})),
|
|
%% SUBSCRIBE with bad qos 0
|
|
Bin0 = <<?SUBSCRIBE:4,0:4,11,0,2,0,6,84,111,112,105,99,65,2>>,
|
|
?assertMatch(Packet, parse_to_packet(Bin0, #{strict_mode => false})),
|
|
%% strict_mode = false
|
|
_ = parse_to_packet(Bin0, #{strict_mode => false}),
|
|
?catch_error(bad_frame_header, parse_to_packet(Bin0, #{strict_mode => true})),
|
|
%% strict_mode = false
|
|
_ = parse_serialize(?SUBSCRIBE_PACKET(0, TopicFilters), #{strict_mode => false}),
|
|
%% strict_mode = true
|
|
?catch_error(bad_packet_id, parse_serialize(?SUBSCRIBE_PACKET(0, TopicFilters))),
|
|
?catch_error(bad_subqos, parse_serialize(?SUBSCRIBE_PACKET(1, [{<<"t">>, #{qos => 3}}]))).
|
|
|
|
t_serialize_parse_subscribe_v5(_) ->
|
|
TopicFilters = [{<<"TopicQos0">>, #{rh => 1, qos => ?QOS_2, rap => 0, nl => 0}},
|
|
{<<"TopicQos1">>, #{rh => 1, qos => ?QOS_2, rap => 0, nl => 0}}],
|
|
Packet = ?SUBSCRIBE_PACKET(3, #{'Subscription-Identifier' => 16#FFFFFFF}, TopicFilters),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_serialize_parse_suback(_) ->
|
|
Packet = ?SUBACK_PACKET(10, [?QOS_0, ?QOS_1, 128]),
|
|
?assertEqual(Packet, parse_serialize(Packet)),
|
|
%% strict_mode = false
|
|
_ = parse_serialize(?SUBACK_PACKET(0, [?QOS_0]), #{strict_mode => false}),
|
|
%% strict_mode = true
|
|
?catch_error(bad_packet_id, parse_serialize(?SUBACK_PACKET(0, [?QOS_0]))).
|
|
|
|
t_serialize_parse_suback_v5(_) ->
|
|
Packet = ?SUBACK_PACKET(1, #{'Reason-String' => <<"success">>,
|
|
'User-Property' => [{<<"key">>, <<"value">>}]},
|
|
[?QOS_0, ?QOS_1, 128]),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_serialize_parse_unsubscribe(_) ->
|
|
%% UNSUBSCRIBE(Q1, R1, D0, PacketId=2, TopicTable=[<<"TopicA">>])
|
|
Bin = <<?UNSUBSCRIBE:4,2:4,10,0,2,0,6,84,111,112,105,99,65>>,
|
|
Packet = ?UNSUBSCRIBE_PACKET(2, [<<"TopicA">>]),
|
|
?assertEqual(Bin, serialize_to_binary(Packet)),
|
|
?assertMatch(Packet, parse_to_packet(Bin, #{strict_mode => true})),
|
|
%% UNSUBSCRIBE with bad qos
|
|
%% UNSUBSCRIBE(Q1, R0, D0, PacketId=2, TopicTable=[<<"TopicA">>])
|
|
Bin0 = <<?UNSUBSCRIBE:4,0:4,10,0,2,0,6,84,111,112,105,99,65>>,
|
|
?assertMatch(Packet, parse_to_packet(Bin0, #{strict_mode => false})),
|
|
?catch_error(bad_frame_header, parse_to_packet(Bin0, #{strict_mode => true})),
|
|
%% strict_mode = false
|
|
_ = parse_serialize(?UNSUBSCRIBE_PACKET(0, [<<"TopicA">>]), #{strict_mode => false}),
|
|
%% strict_mode = true
|
|
?catch_error(bad_packet_id, parse_serialize(?UNSUBSCRIBE_PACKET(0, [<<"TopicA">>]))).
|
|
|
|
t_serialize_parse_unsubscribe_v5(_) ->
|
|
Props = #{'User-Property' => [{<<"key">>, <<"val">>}]},
|
|
Packet = ?UNSUBSCRIBE_PACKET(10, Props, [<<"Topic1">>, <<"Topic2">>]),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_serialize_parse_unsuback(_) ->
|
|
Packet = ?UNSUBACK_PACKET(10),
|
|
?assertEqual(Packet, parse_serialize(Packet)).
|
|
|
|
t_serialize_parse_unsuback_v5(_) ->
|
|
Packet = ?UNSUBACK_PACKET(10, #{'Reason-String' => <<"Not authorized">>,
|
|
'User-Property' => [{<<"key">>, <<"val">>}]},
|
|
[16#87, 16#87, 16#87]),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_serialize_parse_pingreq(_) ->
|
|
PingReq = ?PACKET(?PINGREQ),
|
|
?assertEqual(PingReq, parse_serialize(PingReq)).
|
|
|
|
t_serialize_parse_pingresp(_) ->
|
|
PingResp = ?PACKET(?PINGRESP),
|
|
?assertEqual(PingResp, parse_serialize(PingResp)).
|
|
|
|
t_parse_disconnect(_) ->
|
|
Packet = ?DISCONNECT_PACKET(?RC_SUCCESS),
|
|
?assertMatch({ok, Packet, <<>>, _}, emqx_frame:parse(<<224, 0>>)).
|
|
|
|
t_serialize_parse_disconnect(_) ->
|
|
Packet = ?DISCONNECT_PACKET(?RC_SUCCESS),
|
|
?assertEqual(Packet, parse_serialize(Packet)).
|
|
|
|
t_serialize_parse_disconnect_v5(_) ->
|
|
Packet = ?DISCONNECT_PACKET(?RC_SUCCESS,
|
|
#{'Session-Expiry-Interval' => 60,
|
|
'Reason-String' => <<"server_moved">>,
|
|
'Server-Reference' => <<"192.168.1.10">>
|
|
}),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})).
|
|
|
|
t_serialize_parse_auth_v5(_) ->
|
|
Packet = ?AUTH_PACKET(?RC_SUCCESS,
|
|
#{'Authentication-Method' => <<"oauth2">>,
|
|
'Authentication-Data' => <<"3zekkd">>,
|
|
'Reason-String' => <<"success">>,
|
|
'User-Property' => [{<<"key1">>, <<"val1">>},
|
|
{<<"key2">>, <<"val2">>}]
|
|
}),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5})),
|
|
?assertEqual(Packet, parse_serialize(Packet, #{version => ?MQTT_PROTO_V5,
|
|
strict_mode => true})).
|
|
|
|
parse_serialize(Packet) ->
|
|
parse_serialize(Packet, #{strict_mode => true}).
|
|
|
|
parse_serialize(Packet, Opts) when is_map(Opts) ->
|
|
Ver = maps:get(version, Opts, ?MQTT_PROTO_V4),
|
|
Bin = iolist_to_binary(emqx_frame:serialize(Packet, Ver)),
|
|
ParseState = emqx_frame:initial_parse_state(Opts),
|
|
{ok, NPacket, <<>>, _} = emqx_frame:parse(Bin, ParseState),
|
|
NPacket.
|
|
|
|
serialize_to_binary(Packet) ->
|
|
iolist_to_binary(emqx_frame:serialize(Packet)).
|
|
|
|
serialize_to_binary(Packet, Ver) ->
|
|
iolist_to_binary(emqx_frame:serialize(Packet, Ver)).
|
|
|
|
parse_to_packet(Bin, Opts) ->
|
|
PState = emqx_frame:initial_parse_state(Opts),
|
|
{ok, Packet, <<>>, _} = emqx_frame:parse(Bin, PState),
|
|
Packet.
|
|
|
|
payload(Len) -> iolist_to_binary(lists:duplicate(Len, 1)).
|