yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Implement MQTT Version 5.0 client

This commit is contained in:
Feng Lee 2018-05-06 17:45:18 +08:00
parent 1fe28a7aef
commit 31bc091873
21 changed files with 1510 additions and 1237 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

BIN
docs/mqtt-v5.0.pdf
View File

Binary file not shown.

37
include/emqx_client.hrl
View File

@ -1,37 +0,0 @@
%%-define(CLIENT_IN_BROKER, true).
%% Default timeout
-define(DEFAULT_KEEPALIVE, 60000).
-define(DEFAULT_ACK_TIMEOUT, 20000).
-define(DEFAULT_CONNECT_TIMEOUT, 30000).
-define(DEFAULT_TCP_OPTIONS,
[binary, {packet, raw}, {active, false},
{nodelay, true}, {reuseaddr, true}]).
-ifdef(CLIENT_IN_BROKER).
-define(LOG(Level, Msg), emqx_log:Level(Msg)).
-define(LOG(Level, Format, Args), emqx_log:Level(Format, Args)).
-else.
-define(LOG(Level, Msg),
(case Level of
debug -> error_logger:info_msg(Msg);
info -> error_logger:info_msg(Msg);
warning -> error_logger:warning_msg(Msg);
error -> error_logger:error_msg(Msg);
critical -> error_logger:error_msg(Msg)
end)).
-define(LOG(Level, Format, Args),
(case Level of
debug -> error_logger:info_msg(Format, Args);
info -> error_logger:info_msg(Format, Args);
warning -> error_logger:warning_msg(Format, Args);
error -> error_logger:error_msg(Format, Args);
critical -> error_logger:error_msg(Format, Args)
end)).
-endif.

265
include/emqx_mqtt.hrl
View File

@ -22,7 +22,7 @@
{backlog, 512}, {nodelay, true}]).
%%--------------------------------------------------------------------
%% MQTT Protocol Version and Levels
%% MQTT Protocol Version and Names
%%--------------------------------------------------------------------
-define(MQTT_PROTO_V3, 3).
@ -34,10 +34,10 @@
{?MQTT_PROTO_V4, <<"MQTT">>},
{?MQTT_PROTO_V5, <<"MQTT">>}]).
-type(mqtt_vsn() :: ?MQTT_PROTO_V3 | ?MQTT_PROTO_V4 | ?MQTT_PROTO_V5).
-type(mqtt_version() :: ?MQTT_PROTO_V3 | ?MQTT_PROTO_V4 | ?MQTT_PROTO_V5).
%%--------------------------------------------------------------------
%% MQTT QoS Level
%% MQTT QoS Levels
%%--------------------------------------------------------------------
-define(QOS_0, 0). %% At most once
@ -71,8 +71,13 @@
end)
end).
-define(IS_QOS_NAME(I),
(I =:= qos0; I =:= at_most_once;
I =:= qos1; I =:= at_least_once;
I =:= qos2; I =:= exactly_once)).
%%--------------------------------------------------------------------
%% Max ClientId Length. Why 1024?
%% Maximum ClientId Length. Why 1024?
%%--------------------------------------------------------------------
-define(MAX_CLIENTID_LEN, 1024).
@ -87,8 +92,8 @@
username :: binary() | undefined,
peername :: {inet:ip_address(), inet:port_number()},
clean_sess :: boolean(),
proto_ver :: 3 | 4,
keepalive = 0,
proto_ver :: mqtt_version(),
keepalive = 0 :: non_neg_integer(),
will_topic :: undefined | binary(),
mountpoint :: undefined | binary(),
connected_at :: erlang:timestamp(),
@ -119,39 +124,76 @@
-define(AUTH, 15). %% Authentication exchange
-define(TYPE_NAMES, [
'CONNECT',
'CONNACK',
'PUBLISH',
'PUBACK',
'PUBREC',
'PUBREL',
'PUBCOMP',
'SUBSCRIBE',
'SUBACK',
'UNSUBSCRIBE',
'UNSUBACK',
'PINGREQ',
'PINGRESP',
'DISCONNECT',
'AUTH']).
'CONNECT',
'CONNACK',
'PUBLISH',
'PUBACK',
'PUBREC',
'PUBREL',
'PUBCOMP',
'SUBSCRIBE',
'SUBACK',
'UNSUBSCRIBE',
'UNSUBACK',
'PINGREQ',
'PINGRESP',
'DISCONNECT',
'AUTH']).
-type(mqtt_packet_type() :: ?RESERVED..?AUTH).
%%--------------------------------------------------------------------
%% MQTT Connect Return Codes
%% MQTT Reason Codes
%%--------------------------------------------------------------------
-define(CONNACK_ACCEPT, 0). %% Connection accepted
-define(CONNACK_PROTO_VER, 1). %% Unacceptable protocol version
-define(CONNACK_INVALID_ID, 2). %% Client Identifier is correct UTF-8 but not allowed by the Server
-define(CONNACK_SERVER, 3). %% Server unavailable
-define(CONNACK_CREDENTIALS, 4). %% Username or password is malformed
-define(CONNACK_AUTH, 5). %% Client is not authorized to connect
-type(mqtt_connack() :: ?CONNACK_ACCEPT..?CONNACK_AUTH).
-define(RC_SUCCESS, 16#00).
-define(RC_NORMAL_DISCONNECTION, 16#00).
-define(RC_GRANTED_QOS_0, 16#00).
-define(RC_GRANTED_QOS_1, 16#01).
-define(RC_GRANTED_QOS_2, 16#02).
-define(RC_DISCONNECT_WITH_WILL_MESSAGE, 16#04).
-define(RC_NO_MATCHING_SUBSCRIBERS, 16#10).
-define(RC_NO_SUBSCRIPTION_EXISTED, 16#11).
-define(RC_CONTINUE_AUTHENTICATION, 16#18).
-define(RC_RE_AUTHENTICATE, 16#19).
-define(RC_UNSPECIFIED_ERROR, 16#80).
-define(RC_MALFORMED_PACKET, 16#81).
-define(RC_PROTOCOL_ERROR, 16#82).
-define(RC_IMPLEMENTATION_SPECIFIC_ERROR, 16#83).
-define(RC_UNSUPPORTED_PROTOCOL_VERSION, 16#84).
-define(RC_CLIENT_IDENTIFIER_NOT_VALID, 16#85).
-define(RC_BAD_USER_NAME_OR_PASSWORD, 16#86).
-define(RC_NOT_AUTHORIZED, 16#87).
-define(RC_SERVER_UNAVAILABLE, 16#88).
-define(RC_SERVER_BUSY, 16#89).
-define(RC_BANNED, 16#8A).
-define(RC_SERVER_SHUTTING_DOWN, 16#8B).
-define(RC_BAD_AUTHENTICATION_METHOD, 16#8C).
-define(RC_KEEP_ALIVE_TIMEOUT, 16#8D).
-define(RC_SESSION_TAKEN_OVER, 16#8E).
-define(RC_TOPIC_FILTER_INVALID, 16#8F).
-define(RC_TOPIC_NAME_INVALID, 16#90).
-define(RC_PACKET_IDENTIFIER_IN_USE, 16#91).
-define(RC_PACKET_IDENTIFIER_NOT_FOUND, 16#92).
-define(RC_RECEIVE_MAXIMUM_EXCEEDED, 16#93).
-define(RC_TOPIC_ALIAS_INVALID, 16#94).
-define(RC_PACKET_TOO_LARGE, 16#95).
-define(RC_MESSAGE_RATE_TOO_HIGH, 16#96).
-define(RC_QUOTA_EXCEEDED, 16#97).
-define(RC_ADMINISTRATIVE_ACTION, 16#98).
-define(RC_PAYLOAD_FORMAT_INVALID, 16#99).
-define(RC_RETAIN_NOT_SUPPORTED, 16#9A).
-define(RC_QOS_NOT_SUPPORTED, 16#9B).
-define(RC_USE_ANOTHER_SERVER, 16#9C).
-define(RC_SERVER_MOVED, 16#9D).
-define(RC_SHARED_SUBSCRIPTIONS_NOT_SUPPORTED, 16#9E).
-define(RC_CONNECTION_RATE_EXCEEDED, 16#9F).
-define(RC_MAXIMUM_CONNECT_TIME, 16#A0).
-define(RC_SUBSCRIPTION_IDENTIFIERS_NOT_SUPPORTED, 16#A1).
-define(RC_WILDCARD_SUBSCRIPTIONS_NOT_SUPPORTED, 16#A2).
%%--------------------------------------------------------------------
%% Max MQTT Packet Length
%% Maximum MQTT Packet Length
%%--------------------------------------------------------------------
-define(MAX_PACKET_SIZE, 16#fffffff).
@ -184,44 +226,43 @@
-type(mqtt_username() :: binary() | undefined).
-type(mqtt_packet_id() :: 1..16#ffff | undefined).
-type(mqtt_packet_id() :: 1..16#FFFF | undefined).
-type(mqtt_reason_code() :: 1..16#ff | undefined).
-type(mqtt_reason_code() :: 0..16#FF | undefined).
-type(mqtt_properties() :: undefined | map()).
-type(mqtt_properties() :: #{atom() => term()} | undefined).
-type(mqtt_subopt() :: {qos, mqtt_qos()}
| {retain_handling, boolean()}
| {keep_retain, boolean()}
| {no_local, boolean()}).
%% nl: no local, rap: retain as publish, rh: retain handling
-record(mqtt_subopts, {rh = 0, rap = 0, nl = 0, qos = ?QOS_0}).
-type(mqtt_subopts() :: #mqtt_subopts{}).
-record(mqtt_packet_connect,
{ client_id = <<>> :: mqtt_client_id(),
proto_ver = ?MQTT_PROTO_V4 :: mqtt_vsn(),
proto_name = <<"MQTT">> :: binary(),
will_retain = false :: boolean(),
will_qos = ?QOS_1 :: mqtt_qos(),
will_flag = false :: boolean(),
clean_sess = false :: boolean(),
clean_start = true :: boolean(),
keep_alive = 60 :: non_neg_integer(),
will_props = undefined :: undefined | map(),
will_topic = undefined :: undefined | binary(),
will_msg = undefined :: undefined | binary(),
username = undefined :: undefined | binary(),
password = undefined :: undefined | binary(),
is_bridge = false :: boolean(),
properties = undefined :: mqtt_properties() %% MQTT Version 5.0
{ proto_name = <<"MQTT">> :: binary(),
proto_ver = ?MQTT_PROTO_V4 :: mqtt_version(),
is_bridge = false :: boolean(),
clean_start = true :: boolean(),
will_flag = false :: boolean(),
will_qos = ?QOS_1 :: mqtt_qos(),
will_retain = false :: boolean(),
keepalive = 0 :: non_neg_integer(),
properties = undefined :: mqtt_properties(),
client_id = <<>> :: mqtt_client_id(),
will_props = undefined :: undefined | map(),
will_topic = undefined :: undefined | binary(),
will_payload = undefined :: undefined | binary(),
username = undefined :: undefined | binary(),
password = undefined :: undefined | binary()
}).
-record(mqtt_packet_connack,
{ ack_flags = ?RESERVED :: 0 | 1,
reason_code :: mqtt_connack(),
properties :: map()
{ ack_flags :: 0 | 1,
reason_code :: mqtt_reason_code(),
properties :: mqtt_properties()
}).
-record(mqtt_packet_publish,
{ topic_name :: binary(),
{ topic_name :: mqtt_topic(),
packet_id :: mqtt_packet_id(),
properties :: mqtt_properties()
}).
@ -235,13 +276,7 @@
-record(mqtt_packet_subscribe,
{ packet_id :: mqtt_packet_id(),
properties :: mqtt_properties(),
topic_filters :: list({binary(), mqtt_subopt()})
}).
-record(mqtt_packet_unsubscribe,
{ packet_id :: mqtt_packet_id(),
properties :: mqtt_properties(),
topics :: list(binary())
topic_filters :: [{mqtt_topic(), mqtt_subopts()}]
}).
-record(mqtt_packet_suback,
@ -250,9 +285,15 @@
reason_codes :: list(mqtt_reason_code())
}).
-record(mqtt_packet_unsubscribe,
{ packet_id :: mqtt_packet_id(),
properties :: mqtt_properties(),
topic_filters :: [mqtt_topic()]
}).
-record(mqtt_packet_unsuback,
{ packet_id :: mqtt_packet_id(),
properties :: mqtt_properties(),
{ packet_id :: mqtt_packet_id(),
properties :: mqtt_properties(),
reason_codes :: list(mqtt_reason_code())
}).
@ -311,6 +352,19 @@
reason_code = ReasonCode,
properties = Properties}}).
-define(AUTH_PACKET(),
#mqtt_packet{header = #mqtt_packet_header{type = ?AUTH},
variable = #mqtt_packet_auth{reason_code = 0}}).
-define(AUTH_PACKET(ReasonCode),
#mqtt_packet{header = #mqtt_packet_header{type = ?AUTH},
variable = #mqtt_packet_auth{reason_code = ReasonCode}}).
-define(AUTH_PACKET(ReasonCode, Properties),
#mqtt_packet{header = #mqtt_packet_header{type = ?AUTH},
variable = #mqtt_packet_auth{reason_code = ReasonCode,
properties = Properties}}).
-define(PUBLISH_PACKET(Qos, PacketId),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBLISH,
qos = Qos},
@ -323,31 +377,64 @@
packet_id = PacketId},
payload = Payload}).
-define(PUBLISH_PACKET(Header, Topic, PacketId, Properties, Payload),
#mqtt_packet{header = Header = #mqtt_packet_header{type = ?PUBLISH},
variable = #mqtt_packet_publish{topic_name = Topic,
packet_id = PacketId,
properties = Properties},
payload = Payload}).
-define(PUBACK_PACKET(PacketId),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBACK},
variable = #mqtt_packet_puback{packet_id = PacketId}}).
-define(PUBACK_PACKET(Type, PacketId),
#mqtt_packet{header = #mqtt_packet_header{type = Type},
variable = #mqtt_packet_puback{packet_id = PacketId}}).
-define(PUBACK_PACKET(PacketId, ReasonCode, Properties),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBACK},
variable = #mqtt_packet_puback{packet_id = PacketId,
reason_code = ReasonCode,
properties = Properties}}).
-define(PUBREC_PACKET(PacketId),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBREC},
variable = #mqtt_packet_puback{packet_id = PacketId}}).
-define(PUBREC_PACKET(PacketId, ReasonCode, Properties),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBREC},
variable = #mqtt_packet_puback{packet_id = PacketId,
reason_code = ReasonCode,
properties = Properties}}).
-define(PUBREL_PACKET(PacketId),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBREL, qos = ?QOS_1},
variable = #mqtt_packet_puback{packet_id = PacketId}}).
-define(PUBREL_PACKET(PacketId, ReasonCode, Properties),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBREL, qos = ?QOS_1},
variable = #mqtt_packet_puback{packet_id = PacketId,
reason_code = ReasonCode,
properties = Properties}}).
-define(PUBCOMP_PACKET(PacketId),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBCOMP},
variable = #mqtt_packet_puback{packet_id = PacketId}}).
-define(PUBCOMP_PACKET(PacketId, ReasonCode, Properties),
#mqtt_packet{header = #mqtt_packet_header{type = ?PUBCOMP},
variable = #mqtt_packet_puback{packet_id = PacketId,
reason_code = ReasonCode,
properties = Properties}}).
-define(SUBSCRIBE_PACKET(PacketId, TopicFilters),
#mqtt_packet{header = #mqtt_packet_header{type = ?SUBSCRIBE, qos = ?QOS_1},
variable = #mqtt_packet_subscribe{packet_id = PacketId,
topic_filters = TopicFilters}}).
-define(SUBSCRIBE_PACKET(PacketId, Properties, TopicFilters),
#mqtt_packet{header = #mqtt_packet_header{type = ?SUBSCRIBE, qos = ?QOS_1},
variable = #mqtt_packet_subscribe{packet_id = PacketId,
properties = Properties,
topic_filters = TopicFilters}}).
-define(SUBACK_PACKET(PacketId, ReasonCodes),
#mqtt_packet{header = #mqtt_packet_header{type = ?SUBACK},
variable = #mqtt_packet_suback{packet_id = PacketId,
@ -358,14 +445,39 @@
variable = #mqtt_packet_suback{packet_id = PacketId,
properties = Properties,
reason_codes = ReasonCodes}}).
-define(UNSUBSCRIBE_PACKET(PacketId, Topics),
-define(UNSUBSCRIBE_PACKET(PacketId, TopicFilters),
#mqtt_packet{header = #mqtt_packet_header{type = ?UNSUBSCRIBE, qos = ?QOS_1},
variable = #mqtt_packet_unsubscribe{packet_id = PacketId,
topics = Topics}}).
variable = #mqtt_packet_unsubscribe{packet_id = PacketId,
topic_filters = TopicFilters}}).
-define(UNSUBSCRIBE_PACKET(PacketId, Properties, TopicFilters),
#mqtt_packet{header = #mqtt_packet_header{type = ?UNSUBSCRIBE, qos = ?QOS_1},
variable = #mqtt_packet_unsubscribe{packet_id = PacketId,
properties = Properties,
topic_filters = TopicFilters}}).
-define(UNSUBACK_PACKET(PacketId),
#mqtt_packet{header = #mqtt_packet_header{type = ?UNSUBACK},
#mqtt_packet{header = #mqtt_packet_header{type = ?UNSUBACK},
variable = #mqtt_packet_unsuback{packet_id = PacketId}}).
-define(UNSUBACK_PACKET(PacketId, Properties, ReasonCodes),
#mqtt_packet{header = #mqtt_packet_header{type = ?UNSUBACK},
variable = #mqtt_packet_unsuback{packet_id = PacketId,
properties = Properties,
reason_codes = ReasonCodes}}).
-define(DISCONNECT_PACKET(),
#mqtt_packet{header = #mqtt_packet_header{type = ?DISCONNECT}}).
-define(DISCONNECT_PACKET(ReasonCode),
#mqtt_packet{header = #mqtt_packet_header{type = ?DISCONNECT},
variable = #mqtt_packet_disconnect{reason_code = ReasonCode}}).
-define(DISCONNECT_PACKET(ReasonCode, Properties),
#mqtt_packet{header = #mqtt_packet_header{type = ?DISCONNECT},
variable = #mqtt_packet_disconnect{reason_code = ReasonCode,
properties = Properties}}).
-define(PACKET(Type),
#mqtt_packet{header = #mqtt_packet_header{type = Type}}).
@ -406,6 +518,11 @@
-type(mqtt_message() :: #mqtt_message{}).
-define(WILL_MSG(Qos, Retain, Topic, Props, Payload),
#mqtt_message{qos = WillQos, retain = WillRetain,
topic = WillTopic, properties = Props,
payload = WillPayload}).
%%--------------------------------------------------------------------
%% MQTT Delivery
%%--------------------------------------------------------------------

2
src/emqx_broker.erl
View File

@ -56,7 +56,7 @@
-spec(start_link(atom(), pos_integer())
-> {ok, pid()} | ignore | {error, term()}).
start_link(Pool, Id) ->
gen_server:start_link(emqx_misc:proc_name(?MODULE, Id),
gen_server:start_link({local, emqx_misc:proc_name(?MODULE, Id)},
?MODULE, [Pool, Id], [{hibernate_after, 2000}]).
%%--------------------------------------------------------------------

794
src/emqx_client.erl
View File

File diff suppressed because it is too large Load Diff

100
src/emqx_client_sock.erl
View File

@ -16,15 +16,10 @@
-module(emqx_client_sock).
-include("emqx_client.hrl").
-export([connect/4, connect/5, send/2, close/1, stop/1]).
-export([connect/4, send/2, close/1]).
-export([sockname/1, setopts/2, getstat/2]).
%% Internal export
-export([receiver/2, receiver_loop/3]).
-record(ssl_socket, {tcp, ssl}).
-type(socket() :: inet:socket() | #ssl_socket{}).
@ -32,37 +27,23 @@
-type(sockname() :: {inet:ip_address(), inet:port_number()}).
-type(option() :: gen_tcp:connect_option()
| {ssl_options, [ssl:ssl_option()]}).
| {ssl_opts, [ssl:ssl_option()]}).
-export_type([socket/0, option/0]).
%%--------------------------------------------------------------------
%% Socket API
%%--------------------------------------------------------------------
-define(DEFAULT_TCP_OPTIONS, [binary, {packet, raw}, {active, false},
{nodelay, true}, {reuseaddr, true}]).
-spec(connect(pid(), inet:ip_address() | inet:hostname(),
inet:port_number(), [option()])
-spec(connect(inet:ip_address() | inet:hostname(),
inet:port_number(), [option()], timeout())
-> {ok, socket()} | {error, term()}).
connect(ClientPid, Host, Port, SockOpts) ->
connect(ClientPid, Host, Port, SockOpts, ?DEFAULT_CONNECT_TIMEOUT).
connect(ClientPid, Host, Port, SockOpts, Timeout) ->
case do_connect(Host, Port, SockOpts, Timeout) of
{ok, Sock} ->
Receiver = spawn_link(?MODULE, receiver, [ClientPid, Sock]),
ok = controlling_process(Sock, Receiver),
{ok, Sock, Receiver};
Error ->
Error
end.
do_connect(Host, Port, SockOpts, Timeout) ->
connect(Host, Port, SockOpts, Timeout) ->
TcpOpts = emqx_misc:merge_opts(?DEFAULT_TCP_OPTIONS,
lists:keydelete(ssl_options, 1, SockOpts)),
lists:keydelete(ssl_opts, 1, SockOpts)),
case gen_tcp:connect(Host, Port, TcpOpts, Timeout) of
{ok, Sock} ->
case lists:keyfind(ssl_options, 1, SockOpts) of
{ssl_options, SslOpts} ->
case lists:keyfind(ssl_opts, 1, SockOpts) of
{ssl_opts, SslOpts} ->
ssl_upgrade(Sock, SslOpts, Timeout);
false -> {ok, Sock}
end;
@ -73,23 +54,17 @@ do_connect(Host, Port, SockOpts, Timeout) ->
ssl_upgrade(Sock, SslOpts, Timeout) ->
case ssl:connect(Sock, SslOpts, Timeout) of
{ok, SslSock} ->
ok = ssl:controlling_process(SslSock, self()),
{ok, #ssl_socket{tcp = Sock, ssl = SslSock}};
{error, Reason} -> {error, Reason}
end.
-spec(controlling_process(socket(), pid()) -> ok).
controlling_process(Sock, Pid) when is_port(Sock) ->
gen_tcp:controlling_process(Sock, Pid);
controlling_process(#ssl_socket{ssl = SslSock}, Pid) ->
ssl:controlling_process(SslSock, Pid).
-spec(send(socket(), iodata()) -> ok | {error, einval | closed}).
send(Sock, Data) when is_port(Sock) ->
try erlang:port_command(Sock, Data) of
true -> ok
catch
error:badarg ->
{error, einval}
error:badarg -> {error, einval}
end;
send(#ssl_socket{ssl = SslSock}, Data) ->
ssl:send(SslSock, Data).
@ -100,10 +75,6 @@ close(Sock) when is_port(Sock) ->
close(#ssl_socket{ssl = SslSock}) ->
ssl:close(SslSock).
-spec(stop(Receiver :: pid()) -> stop).
stop(Receiver) ->
Receiver ! stop.
-spec(setopts(socket(), [gen_tcp:option() | ssl:socketoption()]) -> ok).
setopts(Sock, Opts) when is_port(Sock) ->
inet:setopts(Sock, Opts);
@ -123,50 +94,3 @@ sockname(Sock) when is_port(Sock) ->
sockname(#ssl_socket{ssl = SslSock}) ->
ssl:sockname(SslSock).
%%--------------------------------------------------------------------
%% Receiver
%%--------------------------------------------------------------------
receiver(ClientPid, Sock) ->
receiver_activate(ClientPid, Sock, emqx_parser:initial_state()).
receiver_activate(ClientPid, Sock, ParseState) ->
setopts(Sock, [{active, once}]),
erlang:hibernate(?MODULE, receiver_loop, [ClientPid, Sock, ParseState]).
receiver_loop(ClientPid, Sock, ParseState) ->
receive
{TcpOrSsL, _Sock, Data} when TcpOrSsL =:= tcp;
TcpOrSsL =:= ssl ->
case parse_received_bytes(ClientPid, Data, ParseState) of
{ok, NewParseState} ->
receiver_activate(ClientPid, Sock, NewParseState);
{error, Error} ->
exit({frame_error, Error})
end;
{Error, _Sock, Reason} when Error =:= tcp_error;
Error =:= ssl_error ->
exit({Error, Reason});
{Closed, _Sock} when Closed =:= tcp_closed;
Closed =:= ssl_closed ->
exit(Closed);
stop ->
close(Sock)
end.
parse_received_bytes(_ClientPid, <<>>, ParseState) ->
{ok, ParseState};
parse_received_bytes(ClientPid, Data, ParseState) ->
io:format("RECV Data: ~p~n", [Data]),
case emqx_parser:parse(Data, ParseState) of
{more, ParseState1} ->
{ok, ParseState1};
{ok, Packet, Rest} ->
io:format("RECV Packet: ~p~n", [Packet]),
gen_statem:cast(ClientPid, Packet),
parse_received_bytes(ClientPid, Rest, emqx_parser:initial_state());
{error, Error} ->
{error, Error}
end.

73
src/emqx_connection.erl
View File

@ -1,18 +1,18 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2013-2018 EMQ Inc. 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.
%%--------------------------------------------------------------------
%%%===================================================================
%%% Copyright (c) 2013-2018 EMQ Inc. 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_connection).
@ -49,7 +49,7 @@
%% Unused fields: connname, peerhost, peerport
-record(state, {connection, peername, conn_state, await_recv,
rate_limit, packet_size, parser, proto_state,
rate_limit, max_packet_size, proto_state, parse_state,
keepalive, enable_stats, idle_timeout, force_gc_count}).
-define(INFO_KEYS, [peername, conn_state, await_recv]).
@ -109,24 +109,22 @@ do_init(Conn, Env, Peername) ->
SendFun = send_fun(Conn, Peername),
RateLimit = get_value(rate_limit, Conn:opts()),
PacketSize = get_value(max_packet_size, Env, ?MAX_PACKET_SIZE),
Parser = emqx_parser:initial_state(PacketSize),
ProtoState = emqx_protocol:init(Conn, Peername, SendFun, Env),
EnableStats = get_value(client_enable_stats, Env, false),
IdleTimout = get_value(client_idle_timeout, Env, 30000),
ForceGcCount = emqx_gc:conn_max_gc_count(),
State = run_socket(#state{connection = Conn,
peername = Peername,
await_recv = false,
conn_state = running,
rate_limit = RateLimit,
packet_size = PacketSize,
parser = Parser,
proto_state = ProtoState,
enable_stats = EnableStats,
idle_timeout = IdleTimout,
force_gc_count = ForceGcCount}),
State = run_socket(#state{connection = Conn,
peername = Peername,
await_recv = false,
conn_state = running,
rate_limit = RateLimit,
max_packet_size = PacketSize,
proto_state = ProtoState,
enable_stats = EnableStats,
idle_timeout = IdleTimout,
force_gc_count = ForceGcCount}),
gen_server:enter_loop(?MODULE, [{hibernate_after, 10000}],
State, self(), IdleTimout).
init_parse_state(State), self(), IdleTimout).
send_fun(Conn, Peername) ->
Self = self(),
@ -143,6 +141,11 @@ send_fun(Conn, Peername) ->
end
end.
init_parse_state(State = #state{max_packet_size = Size, proto_state = ProtoState}) ->
emqx_parser:initial_state([{max_len, Size},
{ver, emqx_protocol:get(proto_ver, ProtoState)}]),
State.
handle_pre_hibernate(State) ->
{hibernate, emqx_gc:reset_conn_gc_count(#state.force_gc_count, emit_stats(State))}.
@ -308,19 +311,17 @@ code_change(_OldVsn, State, _Extra) ->
received(<<>>, State) ->
{noreply, gc(State)};
received(Bytes, State = #state{parser = Parser,
packet_size = PacketSize,
received(Bytes, State = #state{parse_state = ParseState,
proto_state = ProtoState,
idle_timeout = IdleTimeout}) ->
case catch emqx_parser:parse(Bytes, Parser) of
{more, NewParser} ->
{noreply, run_socket(State#state{parser = NewParser}), IdleTimeout};
case catch emqx_parser:parse(Bytes, ParseState) of
{more, NewParseState} ->
{noreply, State#state{parse_state = NewParseState}, IdleTimeout};
{ok, Packet, Rest} ->
emqx_metrics:received(Packet),
case emqx_protocol:received(Packet, ProtoState) of
{ok, ProtoState1} ->
received(Rest, State#state{parser = emqx_parser:initial_state(PacketSize),
proto_state = ProtoState1});
received(Rest, init_parse_state(State#state{proto_state = ProtoState1}));
{error, Error} ->
?LOG(error, "Protocol error - ~p", [Error], State),
shutdown(Error, State);

8
src/emqx_mod_subscription.erl
View File

@ -33,10 +33,10 @@
load(Topics) ->
emqx:hook('client.connected', fun ?MODULE:on_client_connected/3, [Topics]).
on_client_connected(?CONNACK_ACCEPT, Client = #client{client_id = ClientId,
client_pid = ClientPid,
username = Username}, Topics) ->
on_client_connected(RC, Client = #client{client_id = ClientId,
client_pid = ClientPid,
username = Username}, Topics)
when RC < 16#80 ->
Replace = fun(Topic) -> rep(<<"%u">>, Username, rep(<<"%c">>, ClientId, Topic)) end,
TopicTable = [{Replace(Topic), Qos} || {Topic, Qos} <- Topics],
ClientPid ! {subscribe, TopicTable},

117
src/emqx_mqtt_properties.erl
View File

@ -16,71 +16,67 @@
-module(emqx_mqtt_properties).
-export([name/1, id/1, validate/1]).
-include("emqx_mqtt.hrl").
%%--------------------------------------------------------------------
%% Property id to name
%%--------------------------------------------------------------------
-export([id/1, name/1, filter/2, validate/1]).
-define(PROPS_TABLE,
#{16#01 => {'Payload-Format-Indicator', 'Byte', [?PUBLISH]},
16#02 => {'Message-Expiry-Interval', 'Four-Byte-Integer', [?PUBLISH]},
16#03 => {'Content-Type', 'UTF8-Encoded-String', [?PUBLISH]},
16#08 => {'Response-Topic', 'UTF8-Encoded-String', [?PUBLISH]},
16#09 => {'Correlation-Data', 'Binary-Data', [?PUBLISH]},
16#0B => {'Subscription-Identifier', 'Variable-Byte-Integer', [?PUBLISH, ?SUBSCRIBE]},
16#11 => {'Session-Expiry-Interval', 'Four-Byte-Integer', [?CONNECT, ?CONNACK, ?DISCONNECT]},
16#12 => {'Assigned-Client-Identifier', 'UTF8-Encoded-String', [?CONNACK]},
16#13 => {'Server-Keep-Alive', 'Two-Byte-Integer', [?CONNACK]},
16#15 => {'Authentication-Method', 'UTF8-Encoded-String', [?CONNECT, ?CONNACK, ?AUTH]},
16#16 => {'Authentication-Data', 'Binary-Data', [?CONNECT, ?CONNACK, ?AUTH]},
16#17 => {'Request-Problem-Information', 'Byte', [?CONNECT]},
16#18 => {'Will-Delay-Interval', 'Four-Byte-Integer', ['WILL']},
16#19 => {'Request-Response-Information', 'Byte', [?CONNECT]},
16#1A => {'Response-Information', 'UTF8-Encoded-String', [?CONNACK]},
16#1C => {'Server-Reference', 'UTF8-Encoded-String', [?CONNACK, ?DISCONNECT]},
16#1F => {'Reason-String', 'UTF8-Encoded-String', 'ALL'},
16#21 => {'Receive-Maximum', 'Two-Byte-Integer', [?CONNECT, ?CONNACK]},
16#22 => {'Topic-Alias-Maximum', 'Two-Byte-Integer', [?CONNECT, ?CONNACK]},
16#23 => {'Topic-Alias', 'Two-Byte-Integer', [?PUBLISH]},
16#24 => {'Maximum-QoS', 'Byte', [?CONNACK]},
16#25 => {'Retain-Available', 'Byte', [?CONNACK]},
16#26 => {'User-Property', 'UTF8-String-Pair', 'ALL'},
16#27 => {'Maximum-Packet-Size', 'Four-Byte-Integer', [?CONNECT, ?CONNACK]},
16#28 => {'Wildcard-Subscription-Available', 'Byte', [?CONNACK]},
16#29 => {'Subscription-Identifier-Available', 'Byte', [?CONNACK]},
16#2A => {'Shared-Subscription-Available', 'Byte', [?CONNACK]}}).
%% 01: Byte; PUBLISH, Will Properties
name(16#01) -> 'Payload-Format-Indicator';
%% 02: Four Byte Integer; PUBLISH, Will Properties
name(16#02) -> 'Message-Expiry-Interval';
%% 03: UTF-8 Encoded String; PUBLISH, Will Properties
name(16#03) -> 'Content-Type';
%% 08: UTF-8 Encoded String; PUBLISH, Will Properties
name(16#08) -> 'Response-Topic';
%% 09: Binary Data; PUBLISH, Will Properties
name(16#09) -> 'Correlation-Data';
%% 11: Variable Byte Integer; PUBLISH, SUBSCRIBE
name(16#0B) -> 'Subscription-Identifier';
%% 17: Four Byte Integer; CONNECT, CONNACK, DISCONNECT
name(16#11) -> 'Session-Expiry-Interval';
%% 18: UTF-8 Encoded String; CONNACK
name(16#12) -> 'Assigned-Client-Identifier';
%% 19: Two Byte Integer; CONNACK
name(16#13) -> 'Server-Keep-Alive';
%% 21: UTF-8 Encoded String; CONNECT, CONNACK, AUTH
name(16#15) -> 'Authentication-Method';
%% 22: Binary Data; CONNECT, CONNACK, AUTH
name(16#16) -> 'Authentication-Data';
%% 23: Byte; CONNECT
name(16#17) -> 'Request-Problem-Information';
%% 24: Four Byte Integer; Will Properties
name(16#18) -> 'Will-Delay-Interval';
%% 25: Byte; CONNECT
name(16#19) -> 'Request-Response-Information';
%% 26: UTF-8 Encoded String; CONNACK
name(16#1A) -> 'Response Information';
%% 28: UTF-8 Encoded String; CONNACK, DISCONNECT
name(16#1A) -> 'Response-Information';
name(16#1C) -> 'Server-Reference';
%% 31: UTF-8 Encoded String; CONNACK, PUBACK, PUBREC, PUBREL, PUBCOMP, SUBACK, UNSUBACK, DISCONNECT, AUTH
name(16#1F) -> 'Reason-String';
%% 33: Two Byte Integer; CONNECT, CONNACK
name(16#21) -> 'Receive-Maximum';
%% 34: Two Byte Integer; CONNECT, CONNACK
name(16#22) -> 'Topic-Alias-Maximum';
%% 35: Two Byte Integer; PUBLISH
name(16#23) -> 'Topic Alias';
%% 36: Byte; CONNACK
name(16#23) -> 'Topic-Alias';
name(16#24) -> 'Maximum-QoS';
%% 37: Byte; CONNACK
name(16#25) -> 'Retain-Available';
%% 38: UTF-8 String Pair; ALL
name(16#26) -> 'User-Property';
%% 39: Four Byte Integer; CONNECT, CONNACK
name(16#27) -> 'Maximum-Packet-Size';
%% 40: Byte; CONNACK
name(16#28) -> 'Wildcard-Subscription-Available';
%% 41: Byte; CONNACK
name(16#29) -> 'Subscription-Identifier-Available';
%% 42: Byte; CONNACK
name(16#2A) -> 'Shared-Subscription-Available'.
%%--------------------------------------------------------------------
%% Property name to id
%%--------------------------------------------------------------------
id('Payload-Format-Indicator') -> 16#01;
id('Message-Expiry-Interval') -> 16#02;
id('Content-Type') -> 16#03;
@ -91,16 +87,16 @@ id('Session-Expiry-Interval') -> 16#11;
id('Assigned-Client-Identifier') -> 16#12;
id('Server-Keep-Alive') -> 16#13;
id('Authentication-Method') -> 16#15;
id('Authentication Data') -> 16#16;
id('Authentication-Data') -> 16#16;
id('Request-Problem-Information') -> 16#17;
id('Will-Delay-Interval') -> 16#18;
id('Request-Response-Information') -> 16#19;
id('Response Information') -> 16#1A;
id('Response-Information') -> 16#1A;
id('Server-Reference') -> 16#1C;
id('Reason-String') -> 16#1F;
id('Receive-Maximum') -> 16#21;
id('Topic-Alias-Maximum') -> 16#22;
id('Topic Alias') -> 16#23;
id('Topic-Alias') -> 16#23;
id('Maximum-QoS') -> 16#24;
id('Retain-Available') -> 16#25;
id('User-Property') -> 16#26;
@ -109,5 +105,42 @@ id('Wildcard-Subscription-Available') -> 16#28;
id('Subscription-Identifier-Available') -> 16#29;
id('Shared-Subscription-Available') -> 16#2A.
%%TODO:
validate(Props) when is_list(Props) -> ok.
filter(Packet, Props) when ?CONNECT =< Packet, Packet =< ?AUTH ->
Fun = fun(Name) ->
case maps:find(id(Name), ?PROPS_TABLE) of
{ok, {Name, _Type, 'ALL'}} ->
true;
{ok, {Name, _Type, Packets}} ->
lists:member(Packet, Packets);
error -> false
end
end,
[Prop || Prop = {Name, _} <- Props, Fun(Name)].
validate(Props) when is_map(Props) ->
lists:foreach(fun validate_prop/1, maps:to_list(Props)).
validate_prop(Prop = {Name, Val}) ->
case maps:find(id(Name), ?PROPS_TABLE) of
{ok, {Name, Type, _}} ->
validate_value(Type, Val)
orelse error(bad_property, Prop);
error ->
error({bad_property, Prop})
end.
validate_value('Byte', Val) ->
is_integer(Val);
validate_value('Two-Byte-Integer', Val) ->
is_integer(Val);
validate_value('Four-Byte-Integer', Val) ->
is_integer(Val);
validate_value('Variable-Byte-Integer', Val) ->
is_integer(Val);
validate_value('UTF8-Encoded-String', Val) ->
is_binary(Val);
validate_value('Binary-Data', Val) ->
is_binary(Val);
validate_value('User-Property', Val) ->
is_tuple(Val) orelse is_list(Val).

115
src/emqx_mqtt_rscode.erl
View File

@ -1,115 +0,0 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2013-2018 EMQ Inc. 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_mqtt_rscode).
-export([value/1]).
%%--------------------------------------------------------------------
%% Reason code to name
%%--------------------------------------------------------------------
%% 00: Success; CONNACK, PUBACK, PUBREC, PUBREL, PUBCOMP, UNSUBACK, AUTH
value('Success') -> 16#00;
%% 00: Normal disconnection; DISCONNECT
value('Normal-Disconnection') -> 16#00;
%% 00: Granted QoS 0; SUBACK
value('Granted-QoS0') -> 16#00;
%% 01: Granted QoS 1; SUBACK
value('Granted-QoS1') -> 16#01;
%% 02: Granted QoS 2; SUBACK
value('Granted-QoS2') -> 16#02;
%% 04: Disconnect with Will Message; DISCONNECT
value('Disconnect-With-Will-Message') -> 16#04;
%% 16: No matching subscribers; PUBACK, PUBREC
value('No-Matching-Subscribers') -> 16#10;
%% 17: No subscription existed; UNSUBACK
value('No-Subscription-Existed') -> 16#11;
%% 24: Continue authentication; AUTH
value('Continue-Authentication') -> 16#18;
%% 25: Re-Authenticate; AUTH
value('Re-Authenticate') -> 16#19;
%% 128: Unspecified error; CONNACK, PUBACK, PUBREC, SUBACK, UNSUBACK, DISCONNECT
value('Unspecified-Error') -> 16#80;
%% 129: Malformed Packet; CONNACK, DISCONNECT
value('Malformed-Packet') -> 16#81;
%% 130: Protocol Error; CONNACK, DISCONNECT
value('Protocol-Error') -> 16#82;
%% 131: Implementation specific error; CONNACK, PUBACK, PUBREC, SUBACK, UNSUBACK, DISCONNECT
value('Implementation-Specific-Error') -> 16#83;
%% 132: Unsupported Protocol Version; CONNACK
value('Unsupported-Protocol-Version') -> 16#84;
%% 133: Client Identifier not valid; CONNACK
value('Client-Identifier-not-Valid') -> 16#85;
%% 134: Bad User Name or Password; CONNACK
value('Bad-Username-or-Password') -> 16#86;
%% 135: Not authorized; CONNACK, PUBACK, PUBREC, SUBACK, UNSUBACK, DISCONNECT
value('Not-Authorized') -> 16#87;
%% 136: Server unavailable; CONNACK
value('Server-Unavailable') -> 16#88;
%% 137: Server busy; CONNACK, DISCONNECT
value('Server-Busy') -> 16#89;
%% 138: Banned; CONNACK
value('Banned') -> 16#8A;
%% 139: Server shutting down; DISCONNECT
value('Server-Shutting-Down') -> 16#8B;
%% 140: Bad authentication method; CONNACK, DISCONNECT
value('Bad-Authentication-Method') -> 16#8C;
%% 141: Keep Alive timeout; DISCONNECT
value('Keep-Alive-Timeout') -> 16#8D;
%% 142: Session taken over; DISCONNECT
value('Session-Taken-Over') -> 16#8E;
%% 143: Topic Filter invalid; SUBACK, UNSUBACK, DISCONNECT
value('Topic-Filter-Invalid') -> 16#8F;
%% 144: Topic Name invalid; CONNACK, PUBACK, PUBREC, DISCONNECT
value('Topic-Name-Invalid') -> 16#90;
%% 145: Packet Identifier in use; PUBACK, PUBREC, SUBACK, UNSUBACK
value('Packet-Identifier-Inuse') -> 16#91;
%% 146: Packet Identifier not found; PUBREL, PUBCOMP
value('Packet-Identifier-Not-Found') -> 16#92;
%% 147: Receive Maximum exceeded; DISCONNECT
value('Receive-Maximum-Exceeded') -> 16#93;
%% 148: Topic Alias invalid; DISCONNECT
value('Topic-Alias-Invalid') -> 16#94;
%% 149: Packet too large; CONNACK, DISCONNECT
value('Packet-Too-Large') -> 16#95;
%% 150: Message rate too high; DISCONNECT
value('Message-Rate-Too-High') -> 16#96;
%% 151: Quota exceeded; CONNACK, PUBACK, PUBREC, SUBACK, DISCONNECT
value('Quota-Exceeded') -> 16#97;
%% 152: Administrative action; DISCONNECT
value('Administrative-Action') -> 16#98;
%% 153: Payload format invalid; CONNACK, PUBACK, PUBREC, DISCONNECT
value('Payload-Format-Invalid') -> 16#99;
%% 154: Retain not supported; CONNACK, DISCONNECT
value('Retain-Not-Supported') -> 16#9A;
%% 155: QoS not supported; CONNACK, DISCONNECT
value('QoS-Not-Supported') -> 16#9B;
%% 156: Use another server; CONNACK, DISCONNECT
value('Use-Another-Server') -> 16#9C;
%% 157: Server moved; CONNACK, DISCONNECT
value('Server-Moved') -> 16#9D;
%% 158: Shared Subscriptions not supported; SUBACK, DISCONNECT
value('Shared-Subscriptions-Not-Supported') -> 16#9E;
%% 159: Connection rate exceeded; CONNACK, DISCONNECT
value('Connection-Rate-Exceeded') -> 16#9F;
%% 160: Maximum connect time; DISCONNECT
value('Maximum-Connect-Time') -> 16#A0;
%% 161: Subscription Identifiers not supported; SUBACK, DISCONNECT
value('Subscription-Identifiers-Not-Supported') -> 16#A1;
%% 162: Wildcard-Subscriptions-Not-Supported; SUBACK, DISCONNECT
value('Wildcard-Subscriptions-Not-Supported') -> 16#A2.

62
src/emqx_packet.erl
View File

@ -20,14 +20,14 @@
-include("emqx_mqtt.hrl").
-export([protocol_name/1, type_name/1, connack_error/1]).
-export([protocol_name/1, type_name/1]).
-export([format/1]).
-export([to_message/1, from_message/1]).
%% @doc Protocol name of version
-spec(protocol_name(mqtt_vsn()) -> binary()).
-spec(protocol_name(mqtt_version()) -> binary()).
protocol_name(?MQTT_PROTO_V3) -> <<"MQIsdp">>;
protocol_name(?MQTT_PROTO_V4) -> <<"MQTT">>;
protocol_name(?MQTT_PROTO_V5) -> <<"MQTT">>.
@ -37,16 +37,6 @@ protocol_name(?MQTT_PROTO_V5) -> <<"MQTT">>.
type_name(Type) when Type > ?RESERVED andalso Type =< ?AUTH ->
lists:nth(Type, ?TYPE_NAMES).
%% @doc Connack Error
-spec(connack_error(mqtt_connack()) -> atom()).
connack_error(?CONNACK_ACCEPT) -> 'CONNACK_ACCEPT';
connack_error(?CONNACK_PROTO_VER) -> 'CONNACK_PROTO_VER';
connack_error(?CONNACK_INVALID_ID) -> 'CONNACK_INVALID_ID';
connack_error(?CONNACK_SERVER) -> 'CONNACK_SERVER';
connack_error(?CONNACK_CREDENTIALS) -> 'CONNACK_CREDENTIALS';
connack_error(?CONNACK_AUTH) -> 'CONNACK_AUTH';
connack_error(_ReasonCode) -> 'CONNACK_UNKNOWN_ERR'.
%% @doc From Message to Packet
-spec(from_message(message()) -> mqtt_packet()).
from_message(Msg = #message{qos = Qos,
@ -80,11 +70,11 @@ to_message(#mqtt_packet{header = #mqtt_packet_header{type = ?PUBLISH,
properties = Props};
to_message(#mqtt_packet_connect{will_flag = false}) ->
undefined;
to_message(#mqtt_packet_connect{will_retain = Retain,
will_qos = Qos,
will_topic = Topic,
will_props = Props,
will_msg = Payload}) ->
to_message(#mqtt_packet_connect{will_retain = Retain,
will_qos = Qos,
will_topic = Topic,
will_props = Props,
will_payload = Payload}) ->
Msg = emqx_message:make(undefined, Topic, Payload),
Msg#message{flags = #{retain => Retain},
headers = #{qos => Qos},
@ -113,23 +103,23 @@ format_variable(Variable, Payload) ->
io_lib:format("~s, Payload=~p", [format_variable(Variable), Payload]).
format_variable(#mqtt_packet_connect{
proto_ver = ProtoVer,
proto_name = ProtoName,
will_retain = WillRetain,
will_qos = WillQoS,
will_flag = WillFlag,
clean_sess = CleanSess,
keep_alive = KeepAlive,
client_id = ClientId,
will_topic = WillTopic,
will_msg = WillMsg,
username = Username,
password = Password}) ->
Format = "ClientId=~s, ProtoName=~s, ProtoVsn=~p, CleanSess=~s, KeepAlive=~p, Username=~s, Password=~s",
Args = [ClientId, ProtoName, ProtoVer, CleanSess, KeepAlive, Username, format_password(Password)],
proto_ver = ProtoVer,
proto_name = ProtoName,
will_retain = WillRetain,
will_qos = WillQoS,
will_flag = WillFlag,
clean_start = CleanStart,
keepalive = KeepAlive,
client_id = ClientId,
will_topic = WillTopic,
will_payload = WillPayload,
username = Username,
password = Password}) ->
Format = "ClientId=~s, ProtoName=~s, ProtoVsn=~p, CleanStart=~s, KeepAlive=~p, Username=~s, Password=~s",
Args = [ClientId, ProtoName, ProtoVer, CleanStart, KeepAlive, Username, format_password(Password)],
{Format1, Args1} = if
WillFlag -> { Format ++ ", Will(Q~p, R~p, Topic=~s, Msg=~s)",
Args ++ [WillQoS, i(WillRetain), WillTopic, WillMsg] };
WillFlag -> { Format ++ ", Will(Q~p, R~p, Topic=~s, Payload=~p)",
Args ++ [WillQoS, i(WillRetain), WillTopic, WillPayload] };
true -> {Format, Args}
end,
io_lib:format(Format1, Args1);
@ -149,9 +139,9 @@ format_variable(#mqtt_packet_subscribe{packet_id = PacketId,
topic_filters = TopicFilters}) ->
io_lib:format("PacketId=~p, TopicFilters=~p", [PacketId, TopicFilters]);
format_variable(#mqtt_packet_unsubscribe{packet_id = PacketId,
topics = Topics}) ->
io_lib:format("PacketId=~p, Topics=~p", [PacketId, Topics]);
format_variable(#mqtt_packet_unsubscribe{packet_id = PacketId,
topic_filters = Topics}) ->
io_lib:format("PacketId=~p, TopicFilters=~p", [PacketId, Topics]);
format_variable(#mqtt_packet_suback{packet_id = PacketId,
reason_codes = ReasonCodes}) ->

442
src/emqx_parser.erl
View File

@ -20,180 +20,198 @@
-include("emqx_mqtt.hrl").
%% API
-export([initial_state/0, initial_state/1, parse/2]).
-type(max_packet_size() :: 1..?MAX_PACKET_SIZE).
-type(state() :: #{maxlen := max_packet_size(), vsn := mqtt_vsn()}).
-type(option() :: {max_len, max_packet_size()}
| {version, mqtt_version()}).
-spec(initial_state() -> {none, state()}).
initial_state() ->
initial_state(?MAX_PACKET_SIZE).
-type(state() :: {none, map()} | {more, fun()}).
-export_type([option/0, state/0]).
%% @doc Initialize a parser
-spec(initial_state(max_packet_size()) -> {none, state()}).
initial_state(MaxSize) ->
{none, #{maxlen => MaxSize, vsn => ?MQTT_PROTO_V4}}.
-spec(initial_state() -> {none, map()}).
initial_state() -> initial_state([]).
-spec(initial_state([option()]) -> {none, map()}).
initial_state(Options) when is_list(Options) ->
{none, parse_opt(Options, #{max_len => ?MAX_PACKET_SIZE,
version => ?MQTT_PROTO_V4})}.
parse_opt([], Map) ->
Map;
parse_opt([{version, Ver}|Opts], Map) ->
parse_opt(Opts, Map#{version := Ver});
parse_opt([{max_len, Len}|Opts], Map) ->
parse_opt(Opts, Map#{max_len := Len});
parse_opt([_|Opts], Map) ->
parse_opt(Opts, Map).
%% @doc Parse MQTT Packet
-spec(parse(binary(), {none, state()} | fun())
-> {ok, mqtt_packet()} | {error, term()} | {more, fun()}).
parse(<<>>, {none, State}) ->
{more, fun(Bin) -> parse(Bin, {none, State}) end};
parse(<<Type:4, Dup:1, QoS:2, Retain:1, Rest/binary>>, {none, State}) ->
-spec(parse(binary(), {none, map()} | fun())
-> {ok, mqtt_packet()} | {error, term()} | {more, fun()}).
parse(<<>>, {none, Options}) ->
{more, fun(Bin) -> parse(Bin, {none, Options}) end};
parse(<<Type:4, Dup:1, QoS:2, Retain:1, Rest/binary>>, {none, Options}) ->
parse_remaining_len(Rest, #mqtt_packet_header{type = Type,
dup = bool(Dup),
qos = fixqos(Type, QoS),
retain = bool(Retain)}, State);
retain = bool(Retain)}, Options);
parse(Bin, Cont) -> Cont(Bin).
parse_remaining_len(<<>>, Header, State) ->
{more, fun(Bin) -> parse_remaining_len(Bin, Header, State) end};
parse_remaining_len(Rest, Header, State) ->
parse_remaining_len(Rest, Header, 1, 0, State).
parse_remaining_len(<<>>, Header, Options) ->
{more, fun(Bin) -> parse_remaining_len(Bin, Header, Options) end};
parse_remaining_len(Rest, Header, Options) ->
parse_remaining_len(Rest, Header, 1, 0, Options).
parse_remaining_len(_Bin, _Header, _Multiplier, Length, #{maxlen := MaxLen})
parse_remaining_len(_Bin, _Header, _Multiplier, Length, #{max_len := MaxLen})
when Length > MaxLen ->
{error, invalid_mqtt_frame_len};
parse_remaining_len(<<>>, Header, Multiplier, Length, State) ->
{more, fun(Bin) -> parse_remaining_len(Bin, Header, Multiplier, Length, State) end};
{error, mqtt_frame_too_long};
parse_remaining_len(<<>>, Header, Multiplier, Length, Options) ->
{more, fun(Bin) -> parse_remaining_len(Bin, Header, Multiplier, Length, Options) end};
%% Optimize: match PUBACK, PUBREC, PUBREL, PUBCOMP, UNSUBACK...
parse_remaining_len(<<0:1, 2:7, Rest/binary>>, Header, 1, 0, State) ->
parse_frame(Rest, Header, 2, State);
parse_remaining_len(<<0:1, 2:7, Rest/binary>>, Header, 1, 0, Options) ->
parse_frame(Rest, Header, 2, Options);
%% optimize: match PINGREQ...
parse_remaining_len(<<0:8, Rest/binary>>, Header, 1, 0, State) ->
parse_frame(Rest, Header, 0, State);
parse_remaining_len(<<1:1, Len:7, Rest/binary>>, Header, Multiplier, Value, State) ->
parse_remaining_len(Rest, Header, Multiplier * ?HIGHBIT, Value + Len * Multiplier, State);
parse_remaining_len(<<0:1, Len:7, Rest/binary>>, Header, Multiplier, Value, State = #{maxlen := MaxLen}) ->
parse_remaining_len(<<0:8, Rest/binary>>, Header, 1, 0, Options) ->
parse_frame(Rest, Header, 0, Options);
parse_remaining_len(<<1:1, Len:7, Rest/binary>>, Header, Multiplier, Value, Options) ->
parse_remaining_len(Rest, Header, Multiplier * ?HIGHBIT, Value + Len * Multiplier, Options);
parse_remaining_len(<<0:1, Len:7, Rest/binary>>, Header, Multiplier, Value,
Options = #{max_len := MaxLen}) ->
FrameLen = Value + Len * Multiplier,
if
FrameLen > MaxLen -> {error, invalid_mqtt_frame_len};
true -> parse_frame(Rest, Header, FrameLen, State)
FrameLen > MaxLen -> error(mqtt_frame_too_long);
true -> parse_frame(Rest, Header, FrameLen, Options)
end.
parse_frame(Bin, #mqtt_packet_header{type = Type, qos = Qos} = Header, Length, State = #{vsn := Vsn}) ->
case {Type, Bin} of
{?CONNECT, <<FrameBin:Length/binary, Rest/binary>>} ->
{ProtoName, Rest1} = parse_utf(FrameBin),
%% Fix mosquitto bridge: 0x83, 0x84
<<BridgeTag:4, ProtoVer:4, Rest2/binary>> = Rest1,
<<UsernameFlag : 1,
PasswordFlag : 1,
WillRetain : 1,
WillQos : 2,
WillFlag : 1,
CleanSess : 1,
_Reserved : 1,
KeepAlive : 16/big,
Rest3/binary>> = Rest2,
{Properties, Rest4} = parse_properties(ProtoVer, Rest3),
{ClientId, Rest5} = parse_utf(Rest4),
{WillProps, Rest6} = parse_will_props(Rest5, ProtoVer, WillFlag),
{WillTopic, Rest7} = parse_utf(Rest6, WillFlag),
{WillMsg, Rest8} = parse_msg(Rest7, WillFlag),
{UserName, Rest9} = parse_utf(Rest8, UsernameFlag),
{PasssWord, <<>>} = parse_utf(Rest9, PasswordFlag),
case protocol_name_approved(ProtoVer, ProtoName) of
true ->
wrap(Header,
#mqtt_packet_connect{
proto_ver = ProtoVer,
proto_name = ProtoName,
will_retain = bool(WillRetain),
will_qos = WillQos,
will_flag = bool(WillFlag),
clean_sess = bool(CleanSess),
keep_alive = KeepAlive,
client_id = ClientId,
will_props = WillProps,
will_topic = WillTopic,
will_msg = WillMsg,
username = UserName,
password = PasssWord,
is_bridge = (BridgeTag =:= 8),
properties = Properties}, Rest);
false ->
{error, protocol_header_corrupt}
parse_frame(Bin, Header, 0, _Options) ->
wrap(Header, Bin);
parse_frame(Bin, Header, Length, Options) ->
case Bin of
<<FrameBin:Length/binary, Rest/binary>> ->
case parse_packet(Header, FrameBin, Options) of
{Variable, Payload} ->
wrap(Header, Variable, Payload, Rest);
Variable ->
wrap(Header, Variable, Rest)
end;
{?CONNACK, <<FrameBin:Length/binary, Rest/binary>>} ->
<<_Reserved:7, SP:1, ReasonCode:8>> = FrameBin,
wrap(Header, #mqtt_packet_connack{ack_flags = SP,
reason_code = ReasonCode}, Rest);
{?PUBLISH, <<FrameBin:Length/binary, Rest/binary>>} ->
{TopicName, Rest1} = parse_utf(FrameBin),
{PacketId, Rest2} = case Qos of
0 -> {undefined, Rest1};
_ -> <<Id:16/big, R/binary>> = Rest1,
{Id, R}
end,
{Properties, Payload} = parse_properties(Vsn, Rest2),
wrap(fixdup(Header), #mqtt_packet_publish{topic_name = TopicName,
packet_id = PacketId,
properties = Properties},
Payload, Rest);
{PubAck, <<FrameBin:Length/binary, Rest/binary>>}
when PubAck == ?PUBACK; PubAck == ?PUBREC; PubAck == ?PUBREL; PubAck == ?PUBCOMP ->
<<PacketId:16/big, Rest1/binary>> = FrameBin,
case Vsn == ?MQTT_PROTO_V5 of
true ->
<<ReasonCode, Rest2/binary>> = Rest1,
{Properties, Rest3} = parse_properties(Vsn, Rest2),
wrap(Header, #mqtt_packet_puback{packet_id = PacketId,
reason_code = ReasonCode,
properties = Properties}, Rest3);
false ->
wrap(Header, #mqtt_packet_puback{packet_id = PacketId}, Rest)
end;
{?SUBSCRIBE, <<FrameBin:Length/binary, Rest/binary>>} ->
%% 1 = Qos,
<<PacketId:16/big, Rest1/binary>> = FrameBin,
{Properties, Rest2} = parse_properties(Vsn, Rest1),
TopicFilters = parse_topics(?SUBSCRIBE, Rest2, []),
wrap(Header, #mqtt_packet_subscribe{packet_id = PacketId,
properties = Properties,
topic_filters = TopicFilters}, Rest);
{?SUBACK, <<FrameBin:Length/binary, Rest/binary>>} ->
<<PacketId:16/big, Rest1/binary>> = FrameBin,
{Properties, Rest2} = parse_properties(Vsn, Rest1),
wrap(Header, #mqtt_packet_suback{packet_id = PacketId, properties = Properties,
reason_codes = parse_qos(Rest2, [])}, Rest);
{?UNSUBSCRIBE, <<FrameBin:Length/binary, Rest/binary>>} ->
%% 1 = Qos,
<<PacketId:16/big, Rest1/binary>> = FrameBin,
{Properties, Rest2} = parse_properties(Vsn, Rest1),
Topics = parse_topics(?UNSUBSCRIBE, Rest2, []),
wrap(Header, #mqtt_packet_unsubscribe{packet_id = PacketId,
properties = Properties,
topics = Topics}, Rest);
{?UNSUBACK, <<FrameBin:Length/binary, Rest/binary>>} ->
<<PacketId:16/big, Rest1/binary>> = FrameBin,
{Properties, _Rest2} = parse_properties(Vsn, Rest1),
wrap(Header, #mqtt_packet_unsuback{packet_id = PacketId,
properties = Properties}, Rest);
{?PINGREQ, Rest} ->
Length = 0,
wrap(Header, Rest);
{?PINGRESP, Rest} ->
Length = 0,
wrap(Header, Rest);
{?DISCONNECT, <<FrameBin:Length/binary, Rest/binary>>} ->
if
Vsn == ?MQTT_PROTO_V5 ->
<<ReasonCode, Rest1/binary>> = FrameBin,
{Properties, Rest2} = parse_properties(Vsn, Rest1),
wrap(Header, #mqtt_packet_disconnect{reason_code = ReasonCode,
properties = Properties}, Rest2);
true ->
Length = 0, wrap(Header, Rest)
end;
{_, TooShortBin} ->
TooShortBin ->
{more, fun(BinMore) ->
parse_frame(<<TooShortBin/binary, BinMore/binary>>, Header, Length, State)
parse_frame(<<TooShortBin/binary, BinMore/binary>>, Header, Length, Options)
end}
end.
parse_packet(#mqtt_packet_header{type = ?CONNECT}, FrameBin, _Options) ->
{ProtoName, Rest} = parse_utf8_string(FrameBin),
<<BridgeTag:4, ProtoVer:4, Rest1/binary>> = Rest,
<<UsernameFlag : 1,
PasswordFlag : 1,
WillRetain : 1,
WillQos : 2,
WillFlag : 1,
CleanStart : 1,
_Reserved : 1,
KeepAlive : 16/big,
Rest2/binary>> = Rest1,
case protocol_name_approved(ProtoVer, ProtoName) of
true -> ok;
false -> error(protocol_name_unapproved)
end,
{Properties, Rest3} = parse_properties(Rest2, ProtoVer),
{ClientId, Rest4} = parse_utf8_string(Rest3),
ConnPacket = #mqtt_packet_connect{proto_name = ProtoName,
proto_ver = ProtoVer,
is_bridge = (BridgeTag =:= 8),
clean_start = bool(CleanStart),
will_flag = bool(WillFlag),
will_qos = WillQos,
will_retain = bool(WillRetain),
keepalive = KeepAlive,
properties = Properties,
client_id = ClientId},
{ConnPacket1, Rest5} = parse_will_message(ConnPacket, Rest4),
{Username, Rest6} = parse_utf8_string(Rest5, bool(UsernameFlag)),
{Passsword, <<>>} = parse_utf8_string(Rest6, bool(PasswordFlag)),
ConnPacket1#mqtt_packet_connect{username = Username, password = Passsword};
parse_packet(#mqtt_packet_header{type = ?CONNACK},
<<AckFlags:8, ReasonCode:8, Rest/binary>>, #{version := Ver}) ->
{Properties, <<>>} = parse_properties(Rest, Ver),
#mqtt_packet_connack{ack_flags = AckFlags,
reason_code = ReasonCode,
properties = Properties};
parse_packet(#mqtt_packet_header{type = ?PUBLISH, qos = QoS}, Bin,
#{version := Ver}) ->
{TopicName, Rest} = parse_utf8_string(Bin),
{PacketId, Rest1} = case QoS of
?QOS_0 -> {undefined, Rest};
_ -> parse_packet_id(Rest)
end,
{Properties, Payload} = parse_properties(Rest1, Ver),
{#mqtt_packet_publish{topic_name = TopicName,
packet_id = PacketId,
properties = Properties}, Payload};
parse_packet(#mqtt_packet_header{type = PubAck}, <<PacketId:16/big>>, _Options)
when ?PUBACK =< PubAck, PubAck =< ?PUBCOMP ->
#mqtt_packet_puback{packet_id = PacketId, reason_code = 0};
parse_packet(#mqtt_packet_header{type = PubAck}, <<PacketId:16/big, ReasonCode, Rest/binary>>,
#{version := Ver = ?MQTT_PROTO_V5})
when ?PUBACK =< PubAck, PubAck =< ?PUBCOMP ->
{Properties, <<>>} = parse_properties(Rest, Ver),
#mqtt_packet_puback{packet_id = PacketId,
reason_code = ReasonCode,
properties = Properties};
parse_packet(#mqtt_packet_header{type = ?SUBSCRIBE}, <<PacketId:16/big, Rest/binary>>,
#{version := Ver}) ->
{Properties, Rest1} = parse_properties(Rest, Ver),
TopicFilters = parse_topic_filters(subscribe, Rest1),
#mqtt_packet_subscribe{packet_id = PacketId,
properties = Properties,
topic_filters = TopicFilters};
parse_packet(#mqtt_packet_header{type = ?SUBACK}, <<PacketId:16/big, Rest/binary>>,
#{version := Ver}) ->
{Properties, Rest1} = parse_properties(Rest, Ver),
#mqtt_packet_suback{packet_id = PacketId,
properties = Properties,
reason_codes = parse_reason_codes(Rest1)};
parse_packet(#mqtt_packet_header{type = ?UNSUBSCRIBE}, <<PacketId:16/big, Rest/binary>>,
#{version := Ver}) ->
{Properties, Rest1} = parse_properties(Rest, Ver),
TopicFilters = parse_topic_filters(unsubscribe, Rest1),
#mqtt_packet_unsubscribe{packet_id = PacketId,
properties = Properties,
topic_filters = TopicFilters};
parse_packet(#mqtt_packet_header{type = ?UNSUBACK}, <<PacketId:16/big>>, _Options) ->
#mqtt_packet_unsuback{packet_id = PacketId};
parse_packet(#mqtt_packet_header{type = ?UNSUBACK}, <<PacketId:16/big, Rest/binary>>,
#{version := Ver}) ->
{Properties, Rest1} = parse_properties(Rest, Ver),
ReasonCodes = parse_reason_codes(Rest1),
#mqtt_packet_unsuback{packet_id = PacketId,
properties = Properties,
reason_codes = ReasonCodes};
parse_packet(#mqtt_packet_header{type = ?DISCONNECT}, <<ReasonCode, Rest/binary>>,
#{version := ?MQTT_PROTO_V5}) ->
{Properties, <<>>} = parse_properties(Rest, ?MQTT_PROTO_V5),
#mqtt_packet_disconnect{reason_code = ReasonCode,
properties = Properties};
parse_packet(#mqtt_packet_header{type = ?AUTH}, <<ReasonCode, Rest/binary>>,
#{version := ?MQTT_PROTO_V5}) ->
{Properties, <<>>} = parse_properties(Rest, ?MQTT_PROTO_V5),
#mqtt_packet_auth{reason_code = ReasonCode, properties = Properties}.
wrap(Header, Variable, Payload, Rest) ->
{ok, #mqtt_packet{header = Header, variable = Variable, payload = Payload}, Rest}.
wrap(Header, Variable, Rest) ->
@ -201,111 +219,100 @@ wrap(Header, Variable, Rest) ->
wrap(Header, Rest) ->
{ok, #mqtt_packet{header = Header}, Rest}.
parse_will_props(Bin, ProtoVer = ?MQTT_PROTO_V5, 1) ->
parse_properties(ProtoVer, Bin);
parse_will_props(Bin, _ProtoVer, _WillFlag) ->
{#{}, Bin}.
protocol_name_approved(Ver, Name) ->
lists:member({Ver, Name}, ?PROTOCOL_NAMES).
parse_properties(?MQTT_PROTO_V5, Bin) ->
parse_will_message(Packet = #mqtt_packet_connect{will_flag = true,
proto_ver = Ver}, Bin) ->
{Props, Rest} = parse_properties(Bin, Ver),
{Topic, Rest1} = parse_utf8_string(Rest),
{Payload, Rest2} = parse_binary_data(Rest1),
{Packet#mqtt_packet_connect{will_props = Props,
will_topic = Topic,
will_payload = Payload}, Rest2};
parse_will_message(Packet, Bin) ->
{Packet, Bin}.
parse_packet_id(<<PacketId:16/big, Rest/binary>>) ->
{PacketId, Rest}.
parse_properties(Bin, Ver) when Ver =/= ?MQTT_PROTO_V5 ->
{undefined, Bin};
parse_properties(<<0, Rest/binary>>, ?MQTT_PROTO_V5) ->
{#{}, Rest};
parse_properties(Bin, ?MQTT_PROTO_V5) ->
{Len, Rest} = parse_variable_byte_integer(Bin),
<<PropsBin:Len/binary, Rest1/binary>> = Rest,
{parse_property(PropsBin, #{}), Rest1};
parse_properties(_MQTT_PROTO_V3, Bin) ->
{#{}, Bin}. %% No properties.
{parse_property(PropsBin, #{}), Rest1}.
parse_property(<<>>, Props) ->
Props;
%% 01: 'Payload-Format-Indicator', Byte;
parse_property(<<16#01, Val, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Payload-Format-Indicator' => Val});
%% 02: 'Message-Expiry-Interval', Four Byte Integer;
parse_property(<<16#02, Val:32/big, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Message-Expiry-Interval' => Val});
%% 03: 'Content-Type', UTF-8 Encoded String;
parse_property(<<16#03, Bin/binary>>, Props) ->
{Val, Rest} = parse_utf(Bin),
{Val, Rest} = parse_utf8_string(Bin),
parse_property(Rest, Props#{'Content-Type' => Val});
%% 08: 'Response-Topic', UTF-8 Encoded String;
parse_property(<<16#08, Bin/binary>>, Props) ->
{Val, Rest} = parse_utf(Bin),
{Val, Rest} = parse_utf8_string(Bin),
parse_property(Rest, Props#{'Response-Topic' => Val});
%% 09: 'Correlation-Data', Binary Data;
parse_property(<<16#09, Len:16/big, Val:Len/binary, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Correlation-Data' => Val});
%% 11: 'Subscription-Identifier', Variable Byte Integer;
parse_property(<<16#0B, Bin/binary>>, Props) ->
{Val, Rest} = parse_variable_byte_integer(Bin),
parse_property(Rest, Props#{'Subscription-Identifier' => Val});
%% 17: 'Session-Expiry-Interval', Four Byte Integer;
parse_property(<<16#11, Val:32/big, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Session-Expiry-Interval' => Val});
%% 18: 'Assigned-Client-Identifier', UTF-8 Encoded String;
parse_property(<<16#12, Bin/binary>>, Props) ->
{Val, Rest} = parse_utf(Bin),
{Val, Rest} = parse_utf8_string(Bin),
parse_property(Rest, Props#{'Assigned-Client-Identifier' => Val});
%% 19: 'Server-Keep-Alive', Two Byte Integer;
parse_property(<<16#13, Val:16, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Server-Keep-Alive' => Val});
%% 21: 'Authentication-Method', UTF-8 Encoded String;
parse_property(<<16#15, Bin/binary>>, Props) ->
{Val, Rest} = parse_utf(Bin),
{Val, Rest} = parse_utf8_string(Bin),
parse_property(Rest, Props#{'Authentication-Method' => Val});
%% 22: 'Authentication-Data', Binary Data;
parse_property(<<16#16, Len:16/big, Val:Len/binary, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Authentication-Data' => Val});
%% 23: 'Request-Problem-Information', Byte;
parse_property(<<16#17, Val, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Request-Problem-Information' => Val});
%% 24: 'Will-Delay-Interval', Four Byte Integer;
parse_property(<<16#18, Val:32, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Will-Delay-Interval' => Val});
%% 25: 'Request-Response-Information', Byte;
parse_property(<<16#19, Val, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Request-Response-Information' => Val});
%% 26: 'Response Information', UTF-8 Encoded String;
parse_property(<<16#1A, Bin/binary>>, Props) ->
{Val, Rest} = parse_utf(Bin),
{Val, Rest} = parse_utf8_string(Bin),
parse_property(Rest, Props#{'Response-Information' => Val});
%% 28: 'Server-Reference', UTF-8 Encoded String;
parse_property(<<16#1C, Bin/binary>>, Props) ->
{Val, Rest} = parse_utf(Bin),
{Val, Rest} = parse_utf8_string(Bin),
parse_property(Rest, Props#{'Server-Reference' => Val});
%% 31: 'Reason-String', UTF-8 Encoded String;
parse_property(<<16#1F, Bin/binary>>, Props) ->
{Val, Rest} = parse_utf(Bin),
{Val, Rest} = parse_utf8_string(Bin),
parse_property(Rest, Props#{'Reason-String' => Val});
%% 33: 'Receive-Maximum', Two Byte Integer;
parse_property(<<16#21, Val:16/big, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Receive-Maximum' => Val});
%% 34: 'Topic-Alias-Maximum', Two Byte Integer;
parse_property(<<16#22, Val:16/big, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Topic-Alias-Maximum' => Val});
%% 35: 'Topic-Alias', Two Byte Integer;
parse_property(<<16#23, Val:16/big, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Topic-Alias' => Val});
%% 36: 'Maximum-QoS', Byte;
parse_property(<<16#24, Val, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Maximum-QoS' => Val});
%% 37: 'Retain-Available', Byte;
parse_property(<<16#25, Val, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Retain-Available' => Val});
%% 38: 'User-Property', UTF-8 String Pair;
parse_property(<<16#26, Bin/binary>>, Props) ->
{Pair, Rest} = parse_utf_pair(Bin),
parse_property(Rest, case maps:find('User-Property', Props) of
{ok, UserProps} -> Props#{'User-Property' := [Pair | UserProps]};
error -> Props#{'User-Property' := [Pair]}
end);
%% 39: 'Maximum-Packet-Size', Four Byte Integer;
{Pair, Rest} = parse_utf8_pair(Bin),
case maps:find('User-Property', Props) of
{ok, UserProps} ->
parse_property(Rest,Props#{'User-Property' := [Pair|UserProps]});
error ->
parse_property(Rest, Props#{'User-Property' => [Pair]})
end;
parse_property(<<16#27, Val:32, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Maximum-Packet-Size' => Val});
%% 40: 'Wildcard-Subscription-Available', Byte;
parse_property(<<16#28, Val, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Wildcard-Subscription-Available' => Val});
%% 41: 'Subscription-Identifier-Available', Byte;
parse_property(<<16#29, Val, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Subscription-Identifier-Available' => Val});
%% 42: 'Shared-Subscription-Available', Byte;
parse_property(<<16#2A, Val, Bin/binary>>, Props) ->
parse_property(Bin, Props#{'Shared-Subscription-Available' => Val}).
@ -316,53 +323,36 @@ parse_variable_byte_integer(<<1:1, Len:7, Rest/binary>>, Multiplier, Value) ->
parse_variable_byte_integer(<<0:1, Len:7, Rest/binary>>, Multiplier, Value) ->
{Value + Len * Multiplier, Rest}.
parse_topics(_Packet, <<>>, Topics) ->
lists:reverse(Topics);
parse_topics(?SUBSCRIBE = Sub, Bin, Topics) ->
{Name, <<_Reserved:2, RetainHandling:2, KeepRetain:1, NoLocal:1, QoS:2, Rest/binary>>} = parse_utf(Bin),
SubOpts = [{qos, QoS}, {retain_handling, RetainHandling}, {keep_retain, KeepRetain}, {no_local, NoLocal}],
parse_topics(Sub, Rest, [{Name, SubOpts}| Topics]);
parse_topics(?UNSUBSCRIBE = Sub, Bin, Topics) ->
{Name, <<Rest/binary>>} = parse_utf(Bin),
parse_topics(Sub, Rest, [Name | Topics]).
parse_topic_filters(subscribe, Bin) ->
[{Topic, #mqtt_subopts{rh = Rh, rap = Rap, nl = Nl, qos = QoS}}
|| <<Len:16/big, Topic:Len/binary, _:2, Rh:2, Rap:1, Nl:1, QoS:2>> <= Bin];
parse_qos(<<>>, Acc) ->
lists:reverse(Acc);
parse_qos(<<QoS:8/unsigned, Rest/binary>>, Acc) ->
parse_qos(Rest, [QoS | Acc]).
parse_topic_filters(unsubscribe, Bin) ->
[Topic || <<Len:16/big, Topic:Len/binary>> <= Bin].
parse_utf_pair(Bin) ->
parse_reason_codes(Bin) ->
[Code || <<Code>> <= Bin].
parse_utf8_pair(Bin) ->
[{Name, Value} || <<Len:16/big, Name:Len/binary, Len2:16/big, Value:Len2/binary>> <= Bin].
parse_utf(Bin, 0) ->
parse_utf8_string(Bin, false) ->
{undefined, Bin};
parse_utf(Bin, _) ->
parse_utf(Bin).
parse_utf8_string(Bin, true) ->
parse_utf8_string(Bin).
parse_utf(<<Len:16/big, Str:Len/binary, Rest/binary>>) ->
parse_utf8_string(<<Len:16/big, Str:Len/binary, Rest/binary>>) ->
{Str, Rest}.
parse_msg(Bin, 0) ->
{undefined, Bin};
parse_msg(<<Len:16/big, Msg:Len/binary, Rest/binary>>, _) ->
{Msg, Rest}.
parse_binary_data(<<Len:16/big, Data:Len/binary, Rest/binary>>) ->
{Data, Rest}.
bool(0) -> false;
bool(1) -> true.
protocol_name_approved(Ver, Name) ->
lists:member({Ver, Name}, ?PROTOCOL_NAMES).
%% Fix Issue#575
fixqos(?PUBREL, 0) -> 1;
fixqos(?SUBSCRIBE, 0) -> 1;
fixqos(?UNSUBSCRIBE, 0) -> 1;
fixqos(_Type, QoS) -> QoS.
%% Fix Issue#1319
fixdup(Header = #mqtt_packet_header{qos = ?QOS0, dup = true}) ->
Header#mqtt_packet_header{dup = false};
fixdup(Header = #mqtt_packet_header{qos = ?QOS2, dup = true}) ->
Header#mqtt_packet_header{dup = false};
fixdup(Header) -> Header.

76
src/emqx_protocol.erl
View File

@ -25,7 +25,7 @@
-import(proplists, [get_value/2, get_value/3]).
%% API
-export([init/3, init/4, info/1, stats/1, clientid/1, client/1, session/1]).
-export([init/3, init/4, get/2, info/1, stats/1, clientid/1, client/1, session/1]).
-export([subscribe/2, unsubscribe/2, pubrel/2, shutdown/2]).
@ -43,14 +43,14 @@
%% Protocol State
%% ws_initial_headers: Headers from first HTTP request for WebSocket Client.
-record(proto_state, {peername, sendfun, connected = false, client_id, client_pid,
clean_sess, proto_ver, proto_name, username, is_superuser,
clean_start, proto_ver, proto_name, username, is_superuser,
will_msg, keepalive, keepalive_backoff, max_clientid_len,
session, stats_data, mountpoint, ws_initial_headers,
peercert_username, is_bridge, connected_at}).
-type(proto_state() :: #proto_state{}).
-define(INFO_KEYS, [client_id, username, clean_sess, proto_ver, proto_name,
-define(INFO_KEYS, [client_id, username, clean_start, proto_ver, proto_name,
keepalive, will_msg, ws_initial_headers, mountpoint,
peercert_username, connected_at]).
@ -98,6 +98,12 @@ repl_username_with_peercert(State = #proto_state{peercert_username = undefined})
repl_username_with_peercert(State = #proto_state{peercert_username = PeerCert}) ->
State#proto_state{username = PeerCert}.
%%TODO::
get(proto_ver, #proto_state{proto_ver = Ver}) ->
Ver;
get(_, _ProtoState) ->
undefined.
info(ProtoState) ->
?record_to_proplist(proto_state, ProtoState, ?INFO_KEYS).
@ -111,7 +117,7 @@ client(#proto_state{client_id = ClientId,
client_pid = ClientPid,
peername = Peername,
username = Username,
clean_sess = CleanSess,
clean_start = CleanStart,
proto_ver = ProtoVer,
keepalive = Keepalive,
will_msg = WillMsg,
@ -188,8 +194,8 @@ process(?CONNECT_PACKET(Var), State0) ->
proto_name = ProtoName,
username = Username,
password = Password,
clean_sess = CleanSess,
keep_alive = KeepAlive,
clean_start= CleanStart,
keepalive = KeepAlive,
client_id = ClientId,
is_bridge = IsBridge} = Var,
@ -198,7 +204,7 @@ process(?CONNECT_PACKET(Var), State0) ->
proto_name = ProtoName,
username = Username,
client_id = ClientId,
clean_sess = CleanSess,
clean_start = CleanStart,
keepalive = KeepAlive,
will_msg = willmsg(Var, State0),
is_bridge = IsBridge,
@ -206,14 +212,14 @@ process(?CONNECT_PACKET(Var), State0) ->
{ReturnCode1, SessPresent, State3} =
case validate_connect(Var, State1) of
?CONNACK_ACCEPT ->
?RC_SUCCESS ->
case authenticate(client(State1), Password) of
{ok, IsSuperuser} ->
%% Generate clientId if null
State2 = maybe_set_clientid(State1),
%% Start session
case emqx_sm:open_session(#{clean_start => CleanSess,
case emqx_sm:open_session(#{clean_start => CleanStart,
client_id => clientid(State2),
username => Username,
client_pid => self()}) of
@ -227,13 +233,13 @@ process(?CONNECT_PACKET(Var), State0) ->
%% Emit Stats
self() ! emit_stats,
%% ACCEPT
{?CONNACK_ACCEPT, SP, State2#proto_state{session = Session, is_superuser = IsSuperuser}};
{?RC_SUCCESS, SP, State2#proto_state{session = Session, is_superuser = IsSuperuser}};
{error, Error} ->
{stop, {shutdown, Error}, State2}
end;
{error, Reason}->
?LOG(error, "Username '~s' login failed for ~p", [Username, Reason], State1),
{?CONNACK_CREDENTIALS, false, State1}
{?RC_BAD_USER_NAME_OR_PASSWORD, false, State1}
end;
ReturnCode ->
{ReturnCode, false, State1}
@ -252,19 +258,19 @@ process(Packet = ?PUBLISH_PACKET(_Qos, Topic, _PacketId, _Payload), State = #pro
end,
{ok, State};
process(?PUBACK_PACKET(?PUBACK, PacketId), State = #proto_state{session = Session}) ->
process(?PUBACK_PACKET(PacketId), State = #proto_state{session = Session}) ->
emqx_session:puback(Session, PacketId),
{ok, State};
process(?PUBACK_PACKET(?PUBREC, PacketId), State = #proto_state{session = Session}) ->
process(?PUBREC_PACKET(PacketId), State = #proto_state{session = Session}) ->
emqx_session:pubrec(Session, PacketId),
send(?PUBREL_PACKET(PacketId), State);
process(?PUBACK_PACKET(?PUBREL, PacketId), State = #proto_state{session = Session}) ->
process(?PUBREL_PACKET(PacketId), State = #proto_state{session = Session}) ->
emqx_session:pubrel(Session, PacketId),
send(?PUBACK_PACKET(?PUBCOMP, PacketId), State);
send(?PUBCOMP_PACKET(PacketId), State);
process(?PUBACK_PACKET(?PUBCOMP, PacketId), State = #proto_state{session = Session})->
process(?PUBCOMP_PACKET(PacketId), State = #proto_state{session = Session})->
emqx_session:pubcomp(Session, PacketId), {ok, State};
%% Protect from empty topic table
@ -346,7 +352,10 @@ with_puback(Type, Packet = ?PUBLISH_PACKET(_Qos, PacketId),
Msg1 = Msg#message{from = #client{client_id = ClientId, username = Username}},
case emqx_session:publish(Session, mount(replvar(MountPoint, State), Msg1)) of
ok ->
send(?PUBACK_PACKET(Type, PacketId), State);
case Type of
?PUBACK -> send(?PUBACK_PACKET(PacketId), State);
?PUBREC -> send(?PUBREC_PACKET(PacketId), State)
end;
{error, Error} ->
?LOG(error, "PUBLISH ~p error: ~p", [PacketId, Error], State)
end.
@ -390,11 +399,10 @@ inc_stats(Type, PktPos, PktCnt, MsgPos, MsgCnt, Stats) ->
false -> Stats1
end.
stop_if_auth_failure(RC, State) when RC == ?CONNACK_CREDENTIALS; RC == ?CONNACK_AUTH ->
{stop, {shutdown, auth_failure}, State};
stop_if_auth_failure(_RC, State) ->
{ok, State}.
stop_if_auth_failure(?RC_SUCCESS, State) ->
{ok, State};
stop_if_auth_failure(RC, State) when RC =/= ?RC_SUCCESS ->
{stop, {shutdown, auth_failure}, State}.
shutdown(_Error, #proto_state{client_id = undefined}) ->
ignore;
@ -452,13 +460,11 @@ validate_connect(Connect = #mqtt_packet_connect{}, ProtoState) ->
case validate_protocol(Connect) of
true ->
case validate_clientid(Connect, ProtoState) of
true ->
?CONNACK_ACCEPT;
false ->
?CONNACK_INVALID_ID
true -> ?RC_SUCCESS;
false -> ?RC_CLIENT_IDENTIFIER_NOT_VALID
end;
false ->
?CONNACK_PROTO_VER
?RC_UNSUPPORTED_PROTOCOL_VERSION
end.
validate_protocol(#mqtt_packet_connect{proto_ver = Ver, proto_name = Name}) ->
@ -469,10 +475,10 @@ validate_clientid(#mqtt_packet_connect{client_id = ClientId},
when (byte_size(ClientId) >= 1) andalso (byte_size(ClientId) =< MaxLen) ->
true;
%% Issue#599: Null clientId and clean_sess = false
validate_clientid(#mqtt_packet_connect{client_id = ClientId,
clean_sess = CleanSess}, _ProtoState)
when byte_size(ClientId) == 0 andalso (not CleanSess) ->
%% Issue#599: Null clientId and clean_start = false
validate_clientid(#mqtt_packet_connect{client_id = ClientId,
clean_start = CleanStart}, _ProtoState)
when byte_size(ClientId) == 0 andalso (not CleanStart) ->
false;
%% MQTT3.1.1 allow null clientId.
@ -481,10 +487,10 @@ validate_clientid(#mqtt_packet_connect{proto_ver =?MQTT_PROTO_V4,
when byte_size(ClientId) =:= 0 ->
true;
validate_clientid(#mqtt_packet_connect{proto_ver = ProtoVer,
clean_sess = CleanSess}, ProtoState) ->
?LOG(warning, "Invalid clientId. ProtoVer: ~p, CleanSess: ~s",
[ProtoVer, CleanSess], ProtoState),
validate_clientid(#mqtt_packet_connect{proto_ver = ProtoVer,
clean_start = CleanStart}, ProtoState) ->
?LOG(warning, "Invalid clientId. ProtoVer: ~p, CleanStart: ~s",
[ProtoVer, CleanStart], ProtoState),
false.
validate_packet(?PUBLISH_PACKET(_Qos, Topic, _PacketId, _Payload)) ->

110
src/emqx_reason_codes.erl Normal file
View File

@ -0,0 +1,110 @@
%%%===================================================================
%%% Copyright (c) 2013-2018 EMQ Inc. 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_reason_codes).
-export([name/1, text/1]).
name(16#00) -> success;
name(16#01) -> granted_qos1;
name(16#02) -> granted_qos2;
name(16#04) -> disconnect_with_will_message;
name(16#10) -> no_matching_subscribers;
name(16#11) -> no_subscription_existed;
name(16#18) -> continue_authentication;
name(16#19) -> re_authenticate;
name(16#80) -> unspecified_error;
name(16#81) -> malformed_Packet;
name(16#82) -> protocol_error;
name(16#83) -> implementation_specific_error;
name(16#84) -> unsupported_protocol_version;
name(16#85) -> client_identifier_not_valid;
name(16#86) -> bad_username_or_password;
name(16#87) -> not_authorized;
name(16#88) -> server_unavailable;
name(16#89) -> server_busy;
name(16#8A) -> banned;
name(16#8B) -> server_shutting_down;
name(16#8C) -> bad_authentication_method;
name(16#8D) -> keepalive_timeout;
name(16#8E) -> session_taken_over;
name(16#8F) -> topic_filter_invalid;
name(16#90) -> topic_name_invalid;
name(16#91) -> packet_identifier_inuse;
name(16#92) -> packet_identifier_not_found;
name(16#93) -> receive_maximum_exceeded;
name(16#94) -> topic_alias_invalid;
name(16#95) -> packet_too_large;
name(16#96) -> message_rate_too_high;
name(16#97) -> quota_exceeded;
name(16#98) -> administrative_action;
name(16#99) -> payload_format_invalid;
name(16#9A) -> retain_not_supported;
name(16#9B) -> qos_not_supported;
name(16#9C) -> use_another_server;
name(16#9D) -> server_moved;
name(16#9E) -> shared_subscriptions_not_supported;
name(16#9F) -> connection_rate_exceeded;
name(16#A0) -> maximum_connect_time;
name(16#A1) -> subscription_identifiers_not_supported;
name(16#A2) -> wildcard_subscriptions_not_supported;
name(Code) -> list_to_atom("unkown_" ++ integer_to_list(Code)).
text(16#00) -> <<"Success">>;
text(16#01) -> <<"Granted QoS 1">>;
text(16#02) -> <<"Granted QoS 2">>;
text(16#04) -> <<"Disconnect with Will Message">>;
text(16#10) -> <<"No matching subscribers">>;
text(16#11) -> <<"No subscription existed">>;
text(16#18) -> <<"Continue authentication">>;
text(16#19) -> <<"Re-authenticate">>;
text(16#80) -> <<"Unspecified error">>;
text(16#81) -> <<"Malformed Packet">>;
text(16#82) -> <<"Protocol Error">>;
text(16#83) -> <<"Implementation specific error">>;
text(16#84) -> <<"Unsupported Protocol Version">>;
text(16#85) -> <<"Client Identifier not valid">>;
text(16#86) -> <<"Bad User Name or Password">>;
text(16#87) -> <<"Not authorized">>;
text(16#88) -> <<"Server unavailable">>;
text(16#89) -> <<"Server busy">>;
text(16#8A) -> <<"Banned">>;
text(16#8B) -> <<"Server shutting down">>;
text(16#8C) -> <<"Bad authentication method">>;
text(16#8D) -> <<"Keep Alive timeout">>;
text(16#8E) -> <<"Session taken over">>;
text(16#8F) -> <<"Topic Filter invalid">>;
text(16#90) -> <<"Topic Name invalid">>;
text(16#91) -> <<"Packet Identifier in use">>;
text(16#92) -> <<"Packet Identifier not found">>;
text(16#93) -> <<"Receive Maximum exceeded">>;
text(16#94) -> <<"Topic Alias invalid">>;
text(16#95) -> <<"Packet too large">>;
text(16#96) -> <<"Message rate too high">>;
text(16#97) -> <<"Quota exceeded">>;
text(16#98) -> <<"Administrative action">>;
text(16#99) -> <<"Payload format invalid">>;
text(16#9A) -> <<"Retain not supported">>;
text(16#9B) -> <<"QoS not supported">>;
text(16#9C) -> <<"Use another server">>;
text(16#9D) -> <<"Server moved">>;
text(16#9E) -> <<"Shared Subscriptions not supported">>;
text(16#9F) -> <<"Connection rate exceeded">>;
text(16#A0) -> <<"Maximum connect time">>;
text(16#A1) -> <<"Subscription Identifiers not supported">>;
text(16#A2) -> <<"Wildcard Subscriptions not supported">>;
text(Code) -> iolist_to_binary(["Unkown", integer_to_list(Code)]).

2
src/emqx_router.erl
View File

@ -67,7 +67,7 @@ mnesia(copy) ->
-spec(start_link(atom(), pos_integer())
-> {ok, pid()} | ignore | {error, term()}).
start_link(Pool, Id) ->
gen_server:start_link(emqx_misc:proc_name(?MODULE, Id),
gen_server:start_link({local, emqx_misc:proc_name(?MODULE, Id)},
?MODULE, [Pool, Id], [{hibernate_after, 10000}]).
%%--------------------------------------------------------------------

356
src/emqx_serializer.erl
View File

@ -1,18 +1,18 @@
%%--------------------------------------------------------------------
%% Copyright (c) 2013-2018 EMQ Inc. 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.
%%--------------------------------------------------------------------
%%%===================================================================
%%% Copyright (c) 2013-2018 EMQ Inc. 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_serializer).
@ -20,224 +20,266 @@
-include("emqx_mqtt.hrl").
%% API
-export([serialize/1]).
-type(option() :: {version, mqtt_version()}).
%% @doc Serialise MQTT Packet
-spec(serialize(mqtt_packet()) -> iolist()).
serialize(#mqtt_packet{header = Header = #mqtt_packet_header{type = Type},
-export_type([option/0]).
-export([serialize/1, serialize/2]).
-spec(serialize(mqtt_packet()) -> iodata()).
serialize(Packet) -> serialize(Packet, []).
-spec(serialize(mqtt_packet(), [option()]) -> iodata()).
serialize(#mqtt_packet{header = Header,
variable = Variable,
payload = Payload}) ->
serialize_header(Header,
serialize_variable(Type, Variable,
serialize_payload(Payload))).
payload = Payload}, Opts) when is_list(Opts) ->
Opts1 = parse_opt(Opts, #{version => ?MQTT_PROTO_V4}),
serialize(Header, serialize_variable(Variable, Opts1), serialize_payload(Payload)).
serialize_header(#mqtt_packet_header{type = Type,
dup = Dup,
qos = Qos,
retain = Retain},
{VariableBin, PayloadBin})
when ?CONNECT =< Type andalso Type =< ?DISCONNECT ->
Len = byte_size(VariableBin) + byte_size(PayloadBin),
parse_opt([], Map) ->
Map;
parse_opt([{version, Ver}|Opts], Map) ->
parse_opt(Opts, Map#{version := Ver});
parse_opt([_|Opts], Map) ->
parse_opt(Opts, Map).
serialize(#mqtt_packet_header{type = Type,
dup = Dup,
qos = Qos,
retain = Retain}, VariableData, PayloadData)
when ?CONNECT =< Type andalso Type =< ?AUTH ->
Len = iolist_size(VariableData) + iolist_size(PayloadData),
true = (Len =< ?MAX_PACKET_SIZE),
[<<Type:4, (opt(Dup)):1, (opt(Qos)):2, (opt(Retain)):1>>,
serialize_len(Len), VariableBin, PayloadBin].
serialize_remaining_len(Len), VariableData, PayloadData].
serialize_variable(?CONNECT, #mqtt_packet_connect{client_id = ClientId,
proto_ver = ProtoVer,
proto_name = ProtoName,
will_retain = WillRetain,
will_qos = WillQos,
will_flag = WillFlag,
clean_sess = CleanSess,
keep_alive = KeepAlive,
will_topic = WillTopic,
will_msg = WillMsg,
username = Username,
password = Password}, undefined) ->
VariableBin = <<(byte_size(ProtoName)):16/big-unsigned-integer,
ProtoName/binary,
ProtoVer:8,
(opt(Username)):1,
(opt(Password)):1,
(opt(WillRetain)):1,
WillQos:2,
(opt(WillFlag)):1,
(opt(CleanSess)):1,
0:1,
KeepAlive:16/big-unsigned-integer>>,
PayloadBin = serialize_utf(ClientId),
PayloadBin1 = case WillFlag of
true -> <<PayloadBin/binary,
(serialize_utf(WillTopic))/binary,
(byte_size(WillMsg)):16/big-unsigned-integer,
WillMsg/binary>>;
false -> PayloadBin
end,
UserPasswd = << <<(serialize_utf(B))/binary>> || B <- [Username, Password], B =/= undefined >>,
{VariableBin, <<PayloadBin1/binary, UserPasswd/binary>>};
serialize_variable(#mqtt_packet_connect{proto_name = ProtoName,
proto_ver = ProtoVer,
is_bridge = IsBridge,
clean_start = CleanStart,
will_flag = WillFlag,
will_qos = WillQos,
will_retain = WillRetain,
keepalive = KeepAlive,
properties = Properties,
client_id = ClientId,
will_props = WillProps,
will_topic = WillTopic,
will_payload = WillPayload,
username = Username,
password = Password}, _Opts) ->
[serialize_binary_data(ProtoName),
<<(case IsBridge of
true -> 16#80 + ProtoVer;
false -> ProtoVer
end):8,
(opt(Username)):1,
(opt(Password)):1,
(opt(WillRetain)):1,
WillQos:2,
(opt(WillFlag)):1,
(opt(CleanStart)):1,
0:1,
KeepAlive:16/big-unsigned-integer>>,
serialize_properties(Properties, ProtoVer),
serialize_utf8_string(ClientId),
case WillFlag of
true -> [serialize_properties(WillProps, ProtoVer),
serialize_utf8_string(WillTopic),
serialize_binary_data(WillPayload)];
false -> <<>>
end,
serialize_utf8_string(Username, true),
serialize_utf8_string(Password, true)];
serialize_variable(?CONNACK, #mqtt_packet_connack{ack_flags = AckFlags,
reason_code = ReasonCode,
properties = Properties}, undefined) ->
PropsBin = serialize_properties(Properties),
{<<AckFlags:8, ReasonCode:8, PropsBin/binary>>, <<>>};
serialize_variable(#mqtt_packet_connack{ack_flags = AckFlags,
reason_code = ReasonCode,
properties = Properties}, #{version := Ver}) ->
[AckFlags, ReasonCode, serialize_properties(Properties, Ver)];
serialize_variable(?SUBSCRIBE, #mqtt_packet_subscribe{packet_id = PacketId,
topic_filters = TopicFilters}, undefined) ->
{<<PacketId:16/big>>, serialize_topics(TopicFilters)};
serialize_variable(#mqtt_packet_publish{topic_name = TopicName,
packet_id = PacketId,
properties = Properties}, #{version := Ver}) ->
[serialize_utf8_string(TopicName),
if
PacketId =:= undefined -> <<>>;
true -> <<PacketId:16/big-unsigned-integer>>
end,
serialize_properties(Properties, Ver)];
serialize_variable(?SUBACK, #mqtt_packet_suback{packet_id = PacketId,
properties = Properties,
reason_codes = ReasonCodes}, undefined) ->
io:format("SubAck ReasonCodes: ~p~n", [ReasonCodes]),
{<<PacketId:16/big, (serialize_properties(Properties))/binary>>, << <<Code>> || Code <- ReasonCodes >>};
serialize_variable(#mqtt_packet_puback{packet_id = PacketId}, #{version := Ver})
when Ver == ?MQTT_PROTO_V3; Ver == ?MQTT_PROTO_V4 ->
<<PacketId:16/big-unsigned-integer>>;
serialize_variable(#mqtt_packet_puback{packet_id = PacketId,
reason_code = ReasonCode,
properties = Properties},
#{version := ?MQTT_PROTO_V5}) ->
[<<PacketId:16/big-unsigned-integer>>, ReasonCode,
serialize_properties(Properties, ?MQTT_PROTO_V5)];
serialize_variable(?UNSUBSCRIBE, #mqtt_packet_unsubscribe{packet_id = PacketId,
topics = Topics }, undefined) ->
{<<PacketId:16/big>>, serialize_topics(Topics)};
serialize_variable(#mqtt_packet_subscribe{packet_id = PacketId,
properties = Properties,
topic_filters = TopicFilters},
#{version := Ver}) ->
[<<PacketId:16/big-unsigned-integer>>, serialize_properties(Properties, Ver),
serialize_topic_filters(subscribe, TopicFilters, Ver)];
serialize_variable(?UNSUBACK, #mqtt_packet_unsuback{packet_id = PacketId,
properties = Properties,
reason_codes = ReasonCodes}, undefined) ->
{<<PacketId:16/big, (serialize_properties(Properties))/binary>>, << <<Code>> || Code <- ReasonCodes >>};
serialize_variable(#mqtt_packet_suback{packet_id = PacketId,
properties = Properties,
reason_codes = ReasonCodes},
#{version := Ver}) ->
[<<PacketId:16/big-unsigned-integer>>, serialize_properties(Properties, Ver),
<< <<Code>> || Code <- ReasonCodes >>];
serialize_variable(?PUBLISH, #mqtt_packet_publish{topic_name = TopicName,
packet_id = PacketId }, PayloadBin) ->
TopicBin = serialize_utf(TopicName),
PacketIdBin = if
PacketId =:= undefined -> <<>>;
true -> <<PacketId:16/big>>
end,
{<<TopicBin/binary, PacketIdBin/binary>>, PayloadBin};
serialize_variable(#mqtt_packet_unsubscribe{packet_id = PacketId,
properties = Properties,
topic_filters = TopicFilters},
#{version := Ver}) ->
[<<PacketId:16/big-unsigned-integer>>, serialize_properties(Properties, Ver),
serialize_topic_filters(unsubscribe, TopicFilters, Ver)];
serialize_variable(PubAck, #mqtt_packet_puback{packet_id = PacketId}, _Payload)
when PubAck =:= ?PUBACK; PubAck =:= ?PUBREC; PubAck =:= ?PUBREL; PubAck =:= ?PUBCOMP ->
{<<PacketId:16/big>>, <<>>};
serialize_variable(#mqtt_packet_unsuback{packet_id = PacketId,
properties = Properties,
reason_codes = ReasonCodes},
#{version := Ver}) ->
[<<PacketId:16/big-unsigned-integer>>, serialize_properties(Properties, Ver),
<< <<Code>> || Code <- ReasonCodes >>];
serialize_variable(?PINGREQ, undefined, undefined) ->
{<<>>, <<>>};
serialize_variable(#mqtt_packet_disconnect{}, #{version := Ver})
when Ver == ?MQTT_PROTO_V3; Ver == ?MQTT_PROTO_V4 ->
<<>>;
serialize_variable(?PINGRESP, undefined, undefined) ->
{<<>>, <<>>};
serialize_variable(#mqtt_packet_disconnect{reason_code = ReasonCode,
properties = Properties},
#{version := Ver = ?MQTT_PROTO_V5}) ->
[ReasonCode, serialize_properties(Properties, Ver)];
serialize_variable(#mqtt_packet_disconnect{}, _Ver) ->
<<>>;
serialize_variable(?DISCONNECT, #mqtt_packet_disconnect{reason_code = ReasonCode,
properties = Properties}, undefined) ->
{<<ReasonCode, (serialize_properties(Properties))/binary>>, <<>>};
serialize_variable(#mqtt_packet_auth{reason_code = ReasonCode,
properties = Properties},
#{version := Ver = ?MQTT_PROTO_V5}) ->
[ReasonCode, serialize_properties(Properties, Ver)];
serialize_variable(?AUTH, #mqtt_packet_auth{reason_code = ReasonCode,
properties = Properties}, undefined) ->
{<<ReasonCode, (serialize_properties(Properties))/binary>>, <<>>}.
serialize_variable(PacketId, ?MQTT_PROTO_V3) when is_integer(PacketId) ->
<<PacketId:16/big-unsigned-integer>>;
serialize_variable(PacketId, ?MQTT_PROTO_V4) when is_integer(PacketId) ->
<<PacketId:16/big-unsigned-integer>>;
serialize_variable(undefined, _Ver) ->
<<>>.
serialize_payload(undefined) ->
undefined;
serialize_payload(Bin) when is_binary(Bin) ->
<<>>;
serialize_payload(Bin) when is_binary(Bin); is_list(Bin) ->
Bin.
serialize_properties(undefined) ->
serialize_properties(_Props, Ver) when Ver =/= ?MQTT_PROTO_V5 ->
<<>>;
serialize_properties(Props) ->
<< <<(serialize_property(Prop, Val))/binary>> || {Prop, Val} <- maps:to_list(Props) >>.
serialize_properties(Props, ?MQTT_PROTO_V5) ->
serialize_properties(Props).
%% 01: Byte;
serialize_properties(undefined) ->
<<0>>;
serialize_properties(Props) when map_size(Props) == 0 ->
<<0>>;
serialize_properties(Props) when is_map(Props) ->
Bin = << <<(serialize_property(Prop, Val))/binary>> || {Prop, Val} <- maps:to_list(Props) >>,
[serialize_variable_byte_integer(byte_size(Bin)), Bin].
%% Ignore undefined
serialize_property(_, undefined) ->
<<>>;
serialize_property('Payload-Format-Indicator', Val) ->
<<16#01, Val>>;
%% 02: Four Byte Integer;
serialize_property('Message-Expiry-Interval', Val) ->
<<16#02, Val:32/big>>;
%% 03: UTF-8 Encoded String;
serialize_property('Content-Type', Val) ->
<<16#03, (serialize_utf(Val))/binary>>;
%% 08: UTF-8 Encoded String;
<<16#03, (serialize_utf8_string(Val))/binary>>;
serialize_property('Response-Topic', Val) ->
<<16#08, (serialize_utf(Val))/binary>>;
%% 09: Binary Data;
<<16#08, (serialize_utf8_string(Val))/binary>>;
serialize_property('Correlation-Data', Val) ->
<<16#09, (iolist_size(Val)):16, Val/binary>>;
%% 11: Variable Byte Integer;
<<16#09, (byte_size(Val)):16, Val/binary>>;
serialize_property('Subscription-Identifier', Val) ->
<<16#0B, (serialize_variable_byte_integer(Val))/binary>>;
%% 17: Four Byte Integer;
serialize_property('Session-Expiry-Interval', Val) ->
<<16#11, Val:32/big>>;
%% 18: UTF-8 Encoded String;
serialize_property('Assigned-Client-Identifier', Val) ->
<<16#12, (serialize_utf(Val))/binary>>;
%% 19: Two Byte Integer;
<<16#12, (serialize_utf8_string(Val))/binary>>;
serialize_property('Server-Keep-Alive', Val) ->
<<16#13, Val:16/big>>;
%% 21: UTF-8 Encoded String;
serialize_property('Authentication-Method', Val) ->
<<16#15, (serialize_utf(Val))/binary>>;
%% 22: Binary Data;
<<16#15, (serialize_utf8_string(Val))/binary>>;
serialize_property('Authentication-Data', Val) ->
<<16#16, (iolist_size(Val)):16, Val/binary>>;
%% 23: Byte;
serialize_property('Request-Problem-Information', Val) ->
<<16#17, Val>>;
%% 24: Four Byte Integer;
serialize_property('Will-Delay-Interval', Val) ->
<<16#18, Val:32/big>>;
%% 25: Byte;
serialize_property('Request-Response-Information', Val) ->
<<16#19, Val>>;
%% 26: UTF-8 Encoded String;
serialize_property('Response-Information', Val) ->
<<16#1A, (serialize_utf(Val))/binary>>;
%% 28: UTF-8 Encoded String;
<<16#1A, (serialize_utf8_string(Val))/binary>>;
serialize_property('Server-Reference', Val) ->
<<16#1C, (serialize_utf(Val))/binary>>;
%% 31: UTF-8 Encoded String;
<<16#1C, (serialize_utf8_string(Val))/binary>>;
serialize_property('Reason-String', Val) ->
<<16#1F, (serialize_utf(Val))/binary>>;
%% 33: Two Byte Integer;
<<16#1F, (serialize_utf8_string(Val))/binary>>;
serialize_property('Receive-Maximum', Val) ->
<<16#21, Val:16/big>>;
%% 34: Two Byte Integer;
serialize_property('Topic-Alias-Maximum', Val) ->
<<16#22, Val:16/big>>;
%% 35: Two Byte Integer;
serialize_property('Topic-Alias', Val) ->
<<16#23, Val:16/big>>;
%% 36: Byte;
serialize_property('Maximum-QoS', Val) ->
<<16#24, Val>>;
%% 37: Byte;
serialize_property('Retain-Available', Val) ->
<<16#25, Val>>;
%% 38: UTF-8 String Pair;
serialize_property('User-Property', Val) ->
<<16#26, (serialize_utf_pair(Val))/binary>>;
%% 39: Four Byte Integer;
serialize_property('User-Property', {Key, Val}) ->
<<16#26, (serialize_utf8_pair({Key, Val}))/binary>>;
serialize_property('User-Property', Props) when is_list(Props) ->
<< <<(serialize_property('User-Property', {Key, Val}))/binary>>
|| {Key, Val} <- Props >>;
serialize_property('Maximum-Packet-Size', Val) ->
<<16#27, Val:32/big>>;
%% 40: Byte;
serialize_property('Wildcard-Subscription-Available', Val) ->
<<16#28, Val>>;
%% 41: Byte;
serialize_property('Subscription-Identifier-Available', Val) ->
<<16#29, Val>>;
%% 42: Byte;
serialize_property('Shared-Subscription-Available', Val) ->
<<16#2A, Val>>.
serialize_topics([{_Topic, _Qos}|_] = Topics) ->
<< <<(serialize_utf(Topic))/binary, ?RESERVED:6, Qos:2>> || {Topic, Qos} <- Topics >>;
serialize_topic_filters(subscribe, TopicFilters, ?MQTT_PROTO_V5) ->
<< <<(serialize_utf8_string(Topic))/binary, ?RESERVED:2, Rh:2, (opt(Rap)):1, (opt(Nl)):1, Qos:2>>
|| {Topic, #mqtt_subopts{rh = Rh, rap = Rap, nl = Nl, qos = Qos}} <- TopicFilters >>;
serialize_topics([H|_] = Topics) when is_binary(H) ->
<< <<(serialize_utf(Topic))/binary>> || Topic <- Topics >>.
serialize_topic_filters(subscribe, TopicFilters, _Ver) ->
<< <<(serialize_utf8_string(Topic))/binary, ?RESERVED:6, Qos:2>>
|| {Topic, #mqtt_subopts{qos = Qos}} <- TopicFilters >>;
serialize_utf_pair({Name, Value}) ->
<< <<(serialize_utf(S))/binary, (serialize_utf(S))/binary>> || S <- [Name, Value] >>.
serialize_topic_filters(unsubscribe, TopicFilters, _Ver) ->
<< <<(serialize_utf8_string(Topic))/binary>> || Topic <- TopicFilters >>.
serialize_utf(String) ->
serialize_utf8_pair({Name, Value}) ->
<< <<(serialize_utf8_string(S))/binary,
(serialize_utf8_string(S))/binary>> || S <- [Name, Value] >>.
serialize_binary_data(Bin) ->
[<<(byte_size(Bin)):16/big-unsigned-integer>>, Bin].
serialize_utf8_string(undefined, false) ->
error(utf8_string_undefined);
serialize_utf8_string(undefined, true) ->
<<>>;
serialize_utf8_string(String, _AllowNull) ->
serialize_utf8_string(String).
serialize_utf8_string(String) ->
StringBin = unicode:characters_to_binary(String),
Len = byte_size(StringBin),
true = (Len =< 16#ffff),
<<Len:16/big, StringBin/binary>>.
serialize_len(I) ->
serialize_variable_byte_integer(I). %%TODO: refactor later.
serialize_remaining_len(I) ->
serialize_variable_byte_integer(I).
serialize_variable_byte_integer(N) when N =< ?LOWBITS ->
<<0:1, N:7>>;

11
src/emqx_session.erl
View File

@ -725,10 +725,10 @@ acked(puback, PacketId, State = #state{client_id = ClientId,
username = Username,
inflight = Inflight}) ->
case Inflight:lookup(PacketId) of
{publish, Msg, _Ts} ->
{value, {publish, Msg, _Ts}} ->
emqx_hooks:run('message.acked', [ClientId, Username], Msg),
State#state{inflight = Inflight:delete(PacketId)};
_ ->
none ->
?LOG(warning, "Duplicated PUBACK Packet: ~p", [PacketId], State),
State
end;
@ -737,11 +737,14 @@ acked(pubrec, PacketId, State = #state{client_id = ClientId,
username = Username,
inflight = Inflight}) ->
case Inflight:lookup(PacketId) of
{publish, Msg, _Ts} ->
{value, {publish, Msg, _Ts}} ->
emqx_hooks:run('message.acked', [ClientId, Username], Msg),
State#state{inflight = Inflight:update(PacketId, {pubrel, PacketId, os:timestamp()})};
{pubrel, PacketId, _Ts} ->
{value, {pubrel, PacketId, _Ts}} ->
?LOG(warning, "Duplicated PUBREC Packet: ~p", [PacketId], State),
State;
none ->
?LOG(warning, "Unexpected PUBREC Packet: ~p", [PacketId], State),
State
end;

2
src/emqx_sm_sup.erl
View File

@ -38,5 +38,5 @@ init([]) ->
Manager = {manager, {emqx_sm, start_link, []},
permanent, 5000, worker, [emqx_sm]},
{ok, {{one_for_rest, 10, 3600}, [Locker, Registry, Manager]}}.
{ok, {{rest_for_one, 10, 3600}, [Locker, Registry, Manager]}}.

10
test/emqx_access_SUITE.erl
View File

@ -118,8 +118,8 @@ unregister_mod(_) ->
[] = ?AC:lookup_mods(auth).
check_acl(_) ->
User1 = #mqtt_client{client_id = <<"client1">>, username = <<"testuser">>},
User2 = #mqtt_client{client_id = <<"client2">>, username = <<"xyz">>},
User1 = #client{client_id = <<"client1">>, username = <<"testuser">>},
User2 = #client{client_id = <<"client2">>, username = <<"xyz">>},
allow = ?AC:check_acl(User1, subscribe, <<"users/testuser/1">>),
allow = ?AC:check_acl(User1, subscribe, <<"clients/client1">>),
allow = ?AC:check_acl(User1, subscribe, <<"clients/client1/x/y">>),
@ -158,8 +158,8 @@ compile_rule(_) ->
{deny, all} = compile({deny, all}).
match_rule(_) ->
User = #mqtt_client{peername = {{127,0,0,1}, 2948}, client_id = <<"testClient">>, username = <<"TestUser">>},
User2 = #mqtt_client{peername = {{192,168,0,10}, 3028}, client_id = <<"testClient">>, username = <<"TestUser">>},
User = #client{peername = {{127,0,0,1}, 2948}, client_id = <<"testClient">>, username = <<"TestUser">>},
User2 = #client{peername = {{192,168,0,10}, 3028}, client_id = <<"testClient">>, username = <<"TestUser">>},
{matched, allow} = match(User, <<"Test/Topic">>, {allow, all}),
{matched, deny} = match(User, <<"Test/Topic">>, {deny, all}),
@ -169,7 +169,7 @@ match_rule(_) ->
nomatch = match(User, <<"d/e/f/x">>, compile({allow, {user, "admin"}, pubsub, ["d/e/f/#"]})),
{matched, allow} = match(User, <<"testTopics/testClient">>, compile({allow, {client, "testClient"}, publish, ["testTopics/testClient"]})),
{matched, allow} = match(User, <<"clients/testClient">>, compile({allow, all, pubsub, ["clients/%c"]})),
{matched, allow} = match(#mqtt_client{username = <<"user2">>}, <<"users/user2/abc/def">>,
{matched, allow} = match(#client{username = <<"user2">>}, <<"users/user2/abc/def">>,
compile({allow, all, subscribe, ["users/%u/#"]})),
{matched, deny} = match(User, <<"d/e/f">>, compile({deny, all, subscribe, ["$SYS/#", "#"]})),
Rule = compile({allow, {'and', [{ipaddr, "127.0.0.1"}, {user, <<"WrongUser">>}]}, publish, <<"Topic">>}),

56
test/emqx_protocol_SUITE.erl
View File

@ -244,62 +244,6 @@ parse_disconnect(_) ->
qos = 0,
retain = false}}, <<>>} = emqx_parser:parse(Bin, Parser).
%%--------------------------------------------------------------------
%% Serialize Cases
%%--------------------------------------------------------------------
serialize_connect(_) ->
serialize(?CONNECT_PACKET(#mqtt_packet_connect{})),
serialize(?CONNECT_PACKET(#mqtt_packet_connect{
client_id = <<"clientId">>,
will_qos = ?QOS1,
will_flag = true,
will_retain = true,
will_topic = <<"will">>,
will_msg = <<"haha">>,
clean_sess = true})).
serialize_connack(_) ->
ConnAck = #mqtt_packet{header = #mqtt_packet_header{type = ?CONNACK},
variable = #mqtt_packet_connack{ack_flags = 0, return_code = 0}},
?assertEqual(<<32,2,0,0>>, iolist_to_binary(serialize(ConnAck))).
serialize_publish(_) ->
serialize(?PUBLISH_PACKET(?QOS_0, <<"Topic">>, undefined, <<"Payload">>)),
serialize(?PUBLISH_PACKET(?QOS_1, <<"Topic">>, 938, <<"Payload">>)),
serialize(?PUBLISH_PACKET(?QOS_2, <<"Topic">>, 99, long_payload())).
serialize_puback(_) ->
serialize(?PUBACK_PACKET(?PUBACK, 10384)).
serialize_pubrel(_) ->
serialize(?PUBREL_PACKET(10384)).
serialize_subscribe(_) ->
TopicTable = [{<<"TopicQos0">>, ?QOS_0}, {<<"TopicQos1">>, ?QOS_1}, {<<"TopicQos2">>, ?QOS_2}],
serialize(?SUBSCRIBE_PACKET(10, TopicTable)).
serialize_suback(_) ->
serialize(?SUBACK_PACKET(10, [?QOS_0, ?QOS_1, 128])).
serialize_unsubscribe(_) ->
serialize(?UNSUBSCRIBE_PACKET(10, [<<"Topic1">>, <<"Topic2">>])).
serialize_unsuback(_) ->
serialize(?UNSUBACK_PACKET(10)).
serialize_pingreq(_) ->
serialize(?PACKET(?PINGREQ)).
serialize_pingresp(_) ->
serialize(?PACKET(?PINGRESP)).
serialize_disconnect(_) ->
serialize(?PACKET(?DISCONNECT)).
long_payload() ->
iolist_to_binary(["payload." || _I <- lists:seq(1, 100)]).
%%--------------------------------------------------------------------
%% Packet Cases
%%--------------------------------------------------------------------

91
test/emqx_serializer_SUITE.erl Normal file
View File

@ -0,0 +1,91 @@
%%%===================================================================
%%% Copyright (c) 2013-2018 EMQ Inc. 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_serializer_SUITE).
-compile(export_all).
-include("emqx_mqtt.hrl").
-include_lib("eunit/include/eunit.hrl").
-import(emqx_serializer, [serialize/1]).
all() ->
[serialize_connect,
serialize_connack,
serialize_publish,
serialize_puback,
serialize_pubrel,
serialize_subscribe,
serialize_suback,
serialize_unsubscribe,
serialize_unsuback,
serialize_pingreq,
serialize_pingresp,
serialize_disconnect].
serialize_connect(_) ->
serialize(?CONNECT_PACKET(#mqtt_packet_connect{})),
serialize(?CONNECT_PACKET(#mqtt_packet_connect{
client_id = <<"clientId">>,
will_qos = ?QOS1,
will_flag = true,
will_retain = true,
will_topic = <<"will">>,
will_payload = <<"haha">>,
clean_sess = true})).
serialize_connack(_) ->
ConnAck = #mqtt_packet{header = #mqtt_packet_header{type = ?CONNACK},
variable = #mqtt_packet_connack{ack_flags = 0, return_code = 0}},
?assertEqual(<<32,2,0,0>>, iolist_to_binary(serialize(ConnAck))).
serialize_publish(_) ->
serialize(?PUBLISH_PACKET(?QOS_0, <<"Topic">>, undefined, <<"Payload">>)),
serialize(?PUBLISH_PACKET(?QOS_1, <<"Topic">>, 938, <<"Payload">>)),
serialize(?PUBLISH_PACKET(?QOS_2, <<"Topic">>, 99, long_payload())).
serialize_puback(_) ->
serialize(?PUBACK_PACKET(?PUBACK, 10384)).
serialize_pubrel(_) ->
serialize(?PUBREL_PACKET(10384)).
serialize_subscribe(_) ->
TopicTable = [{<<"TopicQos0">>, ?QOS_0}, {<<"TopicQos1">>, ?QOS_1}, {<<"TopicQos2">>, ?QOS_2}],
serialize(?SUBSCRIBE_PACKET(10, TopicTable)).
serialize_suback(_) ->
serialize(?SUBACK_PACKET(10, [?QOS_0, ?QOS_1, 128])).
serialize_unsubscribe(_) ->
serialize(?UNSUBSCRIBE_PACKET(10, [<<"Topic1">>, <<"Topic2">>])).
serialize_unsuback(_) ->
serialize(?UNSUBACK_PACKET(10)).
serialize_pingreq(_) ->
serialize(?PACKET(?PINGREQ)).
serialize_pingresp(_) ->
serialize(?PACKET(?PINGRESP)).
serialize_disconnect(_) ->
serialize(?PACKET(?DISCONNECT)).
long_payload() ->
iolist_to_binary(["payload." || _I <- lists:seq(1, 100)]).