8628139215

Committed 10 Apr 2024 08:18AM UTC coverage: 62.44% (-0.05%) from 62.489%

Build # 8628139215

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #12851 from zmstone/0327-feat-add-emqx_variform

emqx_variform for string substitution and transform

Run Details

206 of 238 new or added lines in 3 files covered. (86.55%)

28 existing lines in 16 files now uncovered.

34895 of 55886 relevant lines covered (62.44%)

6585.43 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

44.94

/apps/emqx_gateway_mqttsn/src/emqx_mqttsn_frame.erl

%%--------------------------------------------------------------------
%% Copyright (c) 2020-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

%% @doc The frame parser for MQTT-SN protocol
-module(emqx_mqttsn_frame).

-behaviour(emqx_gateway_frame).

-include("emqx_mqttsn.hrl").

-export([
    initial_parse_state/1,
    serialize_opts/0,
    parse/2,
    serialize_pkt/2,
    message_type/1,
    format/1,
    type/1,
    is_message/1
]).

-define(flag, 1 / binary).
-define(byte, 8 / big - integer).
-define(short, 16 / big - integer).

-type parse_state() :: #{}.
-type serialize_opts() :: #{}.

-export_type([
    parse_state/0,
    serialize_opts/0
]).

%%--------------------------------------------------------------------
%% Initial

initial_parse_state(_) ->
    #{}.

serialize_opts() ->
    #{}.

%%--------------------------------------------------------------------
%% Parse MQTT-SN Message
%%--------------------------------------------------------------------

parse(<<16#01:?byte, Len:?short, Type:?byte, Var/binary>>, State) ->
    {ok, parse(Type, Len - 4, Var), <<>>, State};
parse(<<Len:?byte, Type:?byte, Var/binary>>, State) ->
    {ok, parse(Type, Len - 2, Var), <<>>, State}.

parse(Type, Len, Var) when Len =:= size(Var) ->
    #mqtt_sn_message{type = Type, variable = parse_var(Type, Var)};
parse(_Type, _Len, _Var) ->
    error(malformed_message_len).

parse_var(?SN_ADVERTISE, <<GwId:?byte, Duration:?short>>) ->
    {GwId, Duration};
parse_var(?SN_SEARCHGW, <<Radius:?byte>>) ->
    Radius;
parse_var(?SN_GWINFO, <<GwId:?byte, GwAdd/binary>>) ->
    {GwId, GwAdd};
parse_var(?SN_CONNECT, <<Flags:?flag, ProtocolId:?byte, Duration:?short, ClientId/binary>>) ->
    {parse_flags(?SN_CONNECT, Flags), ProtocolId, Duration, ClientId};
parse_var(?SN_CONNACK, <<ReturnCode:?byte>>) ->
    ReturnCode;
parse_var(?SN_WILLTOPICREQ, <<>>) ->
    undefined;
parse_var(?SN_WILLTOPIC, <<>>) ->
    undefined;
parse_var(?SN_WILLTOPIC, <<Flags:?flag, WillTopic/binary>>) ->
    {parse_flags(?SN_WILLTOPIC, Flags), WillTopic};
parse_var(?SN_WILLMSGREQ, <<>>) ->
    undefined;
parse_var(?SN_WILLMSG, <<WillMsg/binary>>) ->
    WillMsg;
parse_var(?SN_REGISTER, <<TopicId:?short, MsgId:?short, TopicName/binary>>) ->
    {TopicId, MsgId, TopicName};
parse_var(?SN_REGACK, <<TopicId:?short, MsgId:?short, ReturnCode:?byte>>) ->
    {TopicId, MsgId, ReturnCode};
parse_var(?SN_PUBLISH, <<FlagsBin:?flag, Topic:2/binary, MsgId:?short, Data/binary>>) ->
    #mqtt_sn_flags{topic_id_type = IdType} = Flags = parse_flags(?SN_PUBLISH, FlagsBin),
    {Flags, parse_topic(IdType, Topic), MsgId, Data};
parse_var(?SN_PUBACK, <<TopicId:?short, MsgId:?short, ReturnCode:?byte>>) ->
    {TopicId, MsgId, ReturnCode};
parse_var(PubRec, <<MsgId:?short>>) when
    PubRec == ?SN_PUBREC; PubRec == ?SN_PUBREL; PubRec == ?SN_PUBCOMP
->
    MsgId;
parse_var(Sub, <<FlagsBin:?flag, MsgId:?short, Topic/binary>>) when
    Sub == ?SN_SUBSCRIBE; Sub == ?SN_UNSUBSCRIBE
->
    #mqtt_sn_flags{topic_id_type = IdType} = Flags = parse_flags(Sub, FlagsBin),
    {Flags, MsgId, parse_topic(IdType, Topic)};
parse_var(?SN_SUBACK, <<Flags:?flag, TopicId:?short, MsgId:?short, ReturnCode:?byte>>) ->
    {parse_flags(?SN_SUBACK, Flags), TopicId, MsgId, ReturnCode};
parse_var(?SN_UNSUBACK, <<MsgId:?short>>) ->
    MsgId;
parse_var(?SN_PINGREQ, ClientId) ->
    ClientId;
parse_var(?SN_PINGRESP, _) ->
    undefined;
parse_var(?SN_DISCONNECT, <<>>) ->
    undefined;
parse_var(?SN_DISCONNECT, <<Duration:?short>>) ->
    Duration;
parse_var(?SN_WILLTOPICUPD, <<>>) ->
    {undefined, undefined};
parse_var(?SN_WILLTOPICUPD, <<Flags:?flag, WillTopic/binary>>) ->
    {parse_flags(?SN_WILLTOPICUPD, Flags), WillTopic};
parse_var(?SN_WILLMSGUPD, WillMsg) ->
    WillMsg;
parse_var(?SN_WILLTOPICRESP, <<ReturnCode:?byte>>) ->
    ReturnCode;
parse_var(?SN_WILLMSGRESP, <<ReturnCode:?byte>>) ->
    ReturnCode;
parse_var(_Type, _Var) ->
    error(unkown_message_type).

parse_flags(?SN_CONNECT, <<_D:1, _Q:2, _R:1, Will:1, CleanStart:1, _IdType:2>>) ->
    #mqtt_sn_flags{will = bool(Will), clean_start = bool(CleanStart)};
parse_flags(?SN_WILLTOPIC, <<_D:1, QoS:2, Retain:1, _Will:1, _C:1, _:2>>) ->
    #mqtt_sn_flags{qos = QoS, retain = bool(Retain)};
parse_flags(?SN_PUBLISH, <<Dup:1, QoS:2, Retain:1, _Will:1, _C:1, IdType:2>>) ->
    #mqtt_sn_flags{dup = bool(Dup), qos = QoS, retain = bool(Retain), topic_id_type = IdType};
parse_flags(Sub, <<Dup:1, QoS:2, _R:1, _Will:1, _C:1, IdType:2>>) when
    Sub == ?SN_SUBSCRIBE; Sub == ?SN_UNSUBSCRIBE
->
    #mqtt_sn_flags{dup = bool(Dup), qos = QoS, topic_id_type = IdType};
parse_flags(?SN_SUBACK, <<_D:1, QoS:2, _R:1, _W:1, _C:1, _Id:2>>) ->
    #mqtt_sn_flags{qos = QoS};
parse_flags(?SN_WILLTOPICUPD, <<_D:1, QoS:2, Retain:1, _W:1, _C:1, _Id:2>>) ->
    #mqtt_sn_flags{qos = QoS, retain = bool(Retain)};
parse_flags(_Type, _) ->
    error(malformed_message_flags).

parse_topic(2#00, Topic) -> Topic;
parse_topic(2#01, <<Id:16>>) -> Id;
parse_topic(2#10, Topic) -> Topic;
parse_topic(2#11, Topic) -> Topic.

%%--------------------------------------------------------------------
%% Serialize MQTT-SN Message
%%--------------------------------------------------------------------

serialize_pkt(#mqtt_sn_message{type = Type, variable = Var}, Opts) ->
    VarBin = serialize(Type, Var, Opts),
    VarLen = size(VarBin),
    case VarLen < 254 of
        true ->
            <<(VarLen + 2), Type, VarBin/binary>>;
        false ->
            <<16#01, (VarLen + 4):?short, Type, VarBin/binary>>
    end.

serialize(?SN_ADVERTISE, {GwId, Duration}, _Opts) ->
    <<GwId, Duration:?short>>;
serialize(?SN_SEARCHGW, Radius, _Opts) ->
    <<Radius>>;
serialize(?SN_GWINFO, {GwId, GwAdd}, _Opts) ->
    <<GwId, GwAdd/binary>>;
serialize(?SN_CONNECT, {Flags, ProtocolId, Duration, ClientId}, _Opts) ->
    <<(serialize_flags(Flags))/binary, ProtocolId, Duration:?short, ClientId/binary>>;
serialize(?SN_CONNACK, ReturnCode, _Opts) ->
    <<ReturnCode>>;
serialize(?SN_WILLTOPICREQ, _, _Opts) ->
    <<>>;
serialize(?SN_WILLTOPIC, undefined, _Opts) ->
    <<>>;
serialize(?SN_WILLTOPIC, {Flags, Topic}, _Opts) ->
    %% The WillTopic must a short topic name
    <<(serialize_flags(Flags))/binary, Topic/binary>>;
serialize(?SN_WILLMSGREQ, _, _Opts) ->
    <<>>;
serialize(?SN_WILLMSG, WillMsg, _Opts) ->
    WillMsg;
serialize(?SN_REGISTER, {TopicId, MsgId, TopicName}, _Opts) ->
    <<TopicId:?short, MsgId:?short, TopicName/binary>>;
serialize(?SN_REGACK, {TopicId, MsgId, ReturnCode}, _Opts) ->
    <<TopicId:?short, MsgId:?short, ReturnCode>>;
serialize(
    ?SN_PUBLISH,
    {Flags = #mqtt_sn_flags{topic_id_type = ?SN_NORMAL_TOPIC}, TopicId, MsgId, Data},
    _Opts
) ->
    <<(serialize_flags(Flags))/binary, TopicId:?short, MsgId:?short, Data/binary>>;
serialize(
    ?SN_PUBLISH,
    {Flags = #mqtt_sn_flags{topic_id_type = ?SN_PREDEFINED_TOPIC}, TopicId, MsgId, Data},
    _Opts
) ->
    <<(serialize_flags(Flags))/binary, TopicId:?short, MsgId:?short, Data/binary>>;
serialize(
    ?SN_PUBLISH,
    {Flags = #mqtt_sn_flags{topic_id_type = ?SN_SHORT_TOPIC}, STopicName, MsgId, Data},
    _Opts
) ->
    <<(serialize_flags(Flags))/binary, STopicName:2/binary, MsgId:?short, Data/binary>>;
serialize(?SN_PUBACK, {TopicId, MsgId, ReturnCode}, _Opts) ->
    <<TopicId:?short, MsgId:?short, ReturnCode>>;
serialize(PubRec, MsgId, _Opts) when
    PubRec == ?SN_PUBREC; PubRec == ?SN_PUBREL; PubRec == ?SN_PUBCOMP
->
    <<MsgId:?short>>;
serialize(Sub, {Flags = #mqtt_sn_flags{topic_id_type = IdType}, MsgId, Topic}, _Opts) when
    Sub == ?SN_SUBSCRIBE; Sub == ?SN_UNSUBSCRIBE
->
    <<(serialize_flags(Flags))/binary, MsgId:16, (serialize_topic(IdType, Topic))/binary>>;
serialize(?SN_SUBACK, {Flags, TopicId, MsgId, ReturnCode}, _Opts) ->
    <<(serialize_flags(Flags))/binary, TopicId:?short, MsgId:?short, ReturnCode>>;
serialize(?SN_UNSUBACK, MsgId, _Opts) ->
    <<MsgId:?short>>;
serialize(?SN_PINGREQ, ClientId, _Opts) ->
    ClientId;
serialize(?SN_PINGRESP, _, _Opts) ->
    <<>>;
serialize(?SN_WILLTOPICUPD, {Flags, WillTopic}, _Opts) ->
    <<(serialize_flags(Flags))/binary, WillTopic/binary>>;
serialize(?SN_WILLMSGUPD, WillMsg, _Opts) ->
    WillMsg;
serialize(?SN_WILLTOPICRESP, ReturnCode, _Opts) ->
    <<ReturnCode>>;
serialize(?SN_WILLMSGRESP, ReturnCode, _Opts) ->
    <<ReturnCode>>;
serialize(?SN_DISCONNECT, undefined, _Opts) ->
    <<>>;
serialize(?SN_DISCONNECT, Duration, _Opts) ->
    <<Duration:?short>>.

serialize_flags(#mqtt_sn_flags{
    dup = Dup,
    qos = QoS,
    retain = Retain,
    will = Will,
    clean_start = CleanStart,
    topic_id_type = IdType
}) ->
    <<
        (bool(Dup)):1,
        (i(QoS)):2,
        (bool(Retain)):1,
        (bool(Will)):1,
        (bool(CleanStart)):1,
        (i(IdType)):2
    >>.

serialize_topic(2#00, Topic) -> Topic;
serialize_topic(2#01, Id) -> <<Id:?short>>;
serialize_topic(2#10, Topic) -> Topic;
serialize_topic(2#11, Topic) -> Topic.

bool(0) -> false;
bool(1) -> true;
bool(false) -> 0;
bool(true) -> 1;
bool(undefined) -> 0.

i(undefined) -> 0;
i(I) when is_integer(I) -> I.

message_type(16#00) ->
    "SN_ADVERTISE";
message_type(16#01) ->
    "SN_SEARCHGW";
message_type(16#02) ->
    "SN_GWINFO";
message_type(16#04) ->
    "SN_CONNECT";
message_type(16#05) ->
    "SN_CONNACK";
message_type(16#06) ->
    "SN_WILLTOPICREQ";
message_type(16#07) ->
    "SN_WILLTOPIC";
message_type(16#08) ->
    "SN_WILLMSGREQ";
message_type(16#09) ->
    "SN_WILLMSG";
message_type(16#0a) ->
    "SN_REGISTER";
message_type(16#0b) ->
    "SN_REGACK";
message_type(16#0c) ->
    "SN_PUBLISH";
message_type(16#0d) ->
    "SN_PUBACK";
message_type(16#0e) ->
    "SN_PUBCOMP";
message_type(16#0f) ->
    "SN_PUBREC";
message_type(16#10) ->
    "SN_PUBREL";
message_type(16#12) ->
    "SN_SUBSCRIBE";
message_type(16#13) ->
    "SN_SUBACK";
message_type(16#14) ->
    "SN_UNSUBSCRIBE";
message_type(16#15) ->
    "SN_UNSUBACK";
message_type(16#16) ->
    "SN_PINGREQ";
message_type(16#17) ->
    "SN_PINGRESP";
message_type(16#18) ->
    "SN_DISCONNECT";
message_type(16#1a) ->
    "SN_WILLTOPICUPD";
message_type(16#1b) ->
    "SN_WILLTOPICRESP";
message_type(16#1c) ->
    "SN_WILLMSGUPD";
message_type(16#1d) ->
    "SN_WILLMSGRESP";
message_type(Type) ->
    io_lib:format("Unknown Type ~p", [Type]).

format(?SN_CONNECT_MSG(Flags, ProtocolId, Duration, ClientId)) ->
    #mqtt_sn_flags{
        will = Will,
        clean_start = CleanStart
    } = Flags,
    io_lib:format(
        "SN_CONNECT(W~w, C~w, ProtocolId=~w, Duration=~w, "
        "ClientId=~ts)",
        [
            bool(Will),
            bool(CleanStart),
            ProtocolId,
            Duration,
            ClientId
        ]
    );
format(?SN_CONNACK_MSG(ReturnCode)) ->
    io_lib:format("SN_CONNACK(ReturnCode=~w)", [ReturnCode]);
format(?SN_WILLTOPICREQ_MSG()) ->
    "SN_WILLTOPICREQ()";
format(?SN_WILLTOPIC_MSG(Flags, Topic)) ->
    #mqtt_sn_flags{
        qos = QoS,
        retain = Retain
    } = Flags,
    io_lib:format(
        "SN_WILLTOPIC(Q~w, R~w, Topic=~s)",
        [QoS, bool(Retain), Topic]
    );
format(?SN_WILLTOPIC_EMPTY_MSG) ->
    "SN_WILLTOPIC(_)";
format(?SN_WILLMSGREQ_MSG()) ->
    "SN_WILLMSGREQ()";
format(?SN_WILLMSG_MSG(Msg)) ->
    io_lib:format("SN_WILLMSG_MSG(Msg=~p)", [Msg]);
format(?SN_PUBLISH_MSG(Flags, TopicId, MsgId, Data)) ->
    #mqtt_sn_flags{
        dup = Dup,
        qos = QoS,
        retain = Retain,
        topic_id_type = TopicIdType
    } = Flags,
    io_lib:format(
        "SN_PUBLISH(D~w, Q~w, R~w, TopicIdType=~w, TopicId=~w, "
        "MsgId=~w, Payload=~p)",
        [
            bool(Dup),
            QoS,
            bool(Retain),
            TopicIdType,
            TopicId,
            MsgId,
            Data
        ]
    );
format(?SN_PUBACK_MSG(TopicId, MsgId, ReturnCode)) ->
    io_lib:format(
        "SN_PUBACK(TopicId=~w, MsgId=~w, ReturnCode=~w)",
        [TopicId, MsgId, ReturnCode]
    );
format(?SN_PUBREC_MSG(?SN_PUBCOMP, MsgId)) ->
    io_lib:format("SN_PUBCOMP(MsgId=~w)", [MsgId]);
format(?SN_PUBREC_MSG(?SN_PUBREC, MsgId)) ->
    io_lib:format("SN_PUBREC(MsgId=~w)", [MsgId]);
format(?SN_PUBREC_MSG(?SN_PUBREL, MsgId)) ->
    io_lib:format("SN_PUBREL(MsgId=~w)", [MsgId]);
format(?SN_SUBSCRIBE_MSG(Flags, Msgid, Topic)) ->
    #mqtt_sn_flags{
        dup = Dup,
        qos = QoS,
        topic_id_type = TopicIdType
    } = Flags,
    io_lib:format(
        "SN_SUBSCRIBE(D~w, Q~w, TopicIdType=~w, MsgId=~w, "
        "TopicId=~w)",
        [bool(Dup), QoS, TopicIdType, Msgid, Topic]
    );
format(?SN_SUBACK_MSG(Flags, TopicId, MsgId, ReturnCode)) ->
    #mqtt_sn_flags{qos = QoS} = Flags,
    io_lib:format(
        "SN_SUBACK(GrantedQoS=~w, MsgId=~w, TopicId=~w, "
        "ReturnCode=~w)",
        [QoS, MsgId, TopicId, ReturnCode]
    );
format(?SN_UNSUBSCRIBE_MSG(Flags, Msgid, Topic)) ->
    #mqtt_sn_flags{topic_id_type = TopicIdType} = Flags,
    io_lib:format(
        "SN_UNSUBSCRIBE(TopicIdType=~w, MsgId=~w, TopicId=~w)",
        [TopicIdType, Msgid, Topic]
    );
format(?SN_UNSUBACK_MSG(MsgId)) ->
    io_lib:format("SN_UNSUBACK(MsgId=~w)", [MsgId]);
format(?SN_REGISTER_MSG(TopicId, MsgId, TopicName)) ->
    io_lib:format(
        "SN_REGISTER(TopicId=~w, MsgId=~w, TopicName=~s)",
        [TopicId, MsgId, TopicName]
    );
format(?SN_REGACK_MSG(TopicId, MsgId, ReturnCode)) ->
    io_lib:format(
        "SN_REGACK(TopicId=~w, MsgId=~w, ReturnCode=~w)",
        [TopicId, MsgId, ReturnCode]
    );
format(?SN_PINGREQ_MSG(ClientId)) ->
    io_lib:format("SN_PINGREQ(ClientId=~s)", [ClientId]);
format(?SN_PINGRESP_MSG()) ->
    "SN_PINGRESP()";
format(?SN_DISCONNECT_MSG(Duration)) ->
    io_lib:format("SN_DISCONNECT(Duration=~w)", [Duration]);
format(#mqtt_sn_message{type = Type, variable = Var}) ->
    io_lib:format(
        "mqtt_sn_message(type=~s, Var=~w)",
        [emqx_mqttsn_frame:message_type(Type), Var]
    ).

is_message(#mqtt_sn_message{type = Type}) when
    Type == ?SN_PUBLISH
->
    true;
is_message(_) ->
    false.

type(#mqtt_sn_message{type = Type}) ->
    type(Type);
type(?SN_ADVERTISE) ->
    advertise;
type(?SN_SEARCHGW) ->
    serachgw;
type(?SN_GWINFO) ->
    gwinfo;
type(?SN_CONNECT) ->
    connect;
type(?SN_CONNACK) ->
    connack;
type(?SN_WILLTOPICREQ) ->
    willtopicreq;
type(?SN_WILLTOPIC) ->
    willtopic;
type(?SN_WILLMSGREQ) ->
    willmsgreq;
type(?SN_WILLMSG) ->
    willmsg;
type(?SN_REGISTER) ->
    register;
type(?SN_REGACK) ->
    regack;
type(?SN_PUBLISH) ->
    publish;
type(?SN_PUBACK) ->
    puback;
type(?SN_PUBCOMP) ->
    pubcomp;
type(?SN_PUBREC) ->
    pubrec;
type(?SN_PUBREL) ->
    pubrel;
type(?SN_SUBSCRIBE) ->
    subscribe;
type(?SN_SUBACK) ->
    suback;
type(?SN_UNSUBSCRIBE) ->
    unsubscribe;
type(?SN_UNSUBACK) ->
    unsuback;
type(?SN_PINGREQ) ->
    pingreq;
type(?SN_PINGRESP) ->
    pingresp;
type(?SN_DISCONNECT) ->
    disconnect;
type(?SN_WILLTOPICUPD) ->
    willtopicupd;
type(?SN_WILLTOPICRESP) ->
    willtopicresp;
type(?SN_WILLMSGUPD) ->
    willmsgupd;
type(?SN_WILLMSGRESP) ->
    willmsgresp.

1	%%--------------------------------------------------------------------
2	%% Copyright (c) 2020-2024 EMQ Technologies Co., Ltd. All Rights Reserved.
3	%%
4	%%
5	%% Licensed under the Apache License, Version 2.0 (the "License");
6	%% you may not use this file except in compliance with the License.
7	%% You may obtain a copy of the License at
8	%%
9	%% http://www.apache.org/licenses/LICENSE-2.0
10	%%
11	%% Unless required by applicable law or agreed to in writing, software
12	%% distributed under the License is distributed on an "AS IS" BASIS,
13	%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	%% See the License for the specific language governing permissions and
15	%% limitations under the License.
16	%%--------------------------------------------------------------------
17
18	%% @doc The frame parser for MQTT-SN protocol
19	-module(emqx_mqttsn_frame).
20
21	-behaviour(emqx_gateway_frame).
22
23	-include("emqx_mqttsn.hrl").
24
25	-export([
26	initial_parse_state/1,
27	serialize_opts/0,
28	parse/2,
29	serialize_pkt/2,
30	message_type/1,
31	format/1,
32	type/1,
33	is_message/1
34	]).
35
36	-define(flag, 1 / binary).
37	-define(byte, 8 / big - integer).
38	-define(short, 16 / big - integer).
39
40	-type parse_state() :: #{}.
41	-type serialize_opts() :: #{}.
42
43	-export_type([
44	parse_state/0,
45	serialize_opts/0
46	]).
47
48	%%--------------------------------------------------------------------
49	%% Initial
50
51	initial_parse_state(_) ->
52	#{}.	×
53
54	serialize_opts() ->
55	#{}.	×
56
57	%%--------------------------------------------------------------------
58	%% Parse MQTT-SN Message
59	%%--------------------------------------------------------------------
60
61	parse(<<16#01:?byte, Len:?short, Type:?byte, Var/binary>>, State) ->
62	{ok, parse(Type, Len - 4, Var), <<>>, State};	1✔
63	parse(<<Len:?byte, Type:?byte, Var/binary>>, State) ->
64	{ok, parse(Type, Len - 2, Var), <<>>, State}.	99✔
65
66	parse(Type, Len, Var) when Len =:= size(Var) ->
67	#mqtt_sn_message{type = Type, variable = parse_var(Type, Var)};	100✔
68	parse(_Type, _Len, _Var) ->
69	error(malformed_message_len).	×
70
71	parse_var(?SN_ADVERTISE, <<GwId:?byte, Duration:?short>>) ->
72	{GwId, Duration};	2✔
73	parse_var(?SN_SEARCHGW, <<Radius:?byte>>) ->
74	Radius;	3✔
75	parse_var(?SN_GWINFO, <<GwId:?byte, GwAdd/binary>>) ->
76	{GwId, GwAdd};	2✔
77	parse_var(?SN_CONNECT, <<Flags:?flag, ProtocolId:?byte, Duration:?short, ClientId/binary>>) ->
78	{parse_flags(?SN_CONNECT, Flags), ProtocolId, Duration, ClientId};	6✔
79	parse_var(?SN_CONNACK, <<ReturnCode:?byte>>) ->
80	ReturnCode;	5✔
81	parse_var(?SN_WILLTOPICREQ, <<>>) ->
82	undefined;	5✔
83	parse_var(?SN_WILLTOPIC, <<>>) ->
84	undefined;	2✔
85	parse_var(?SN_WILLTOPIC, <<Flags:?flag, WillTopic/binary>>) ->
86	{parse_flags(?SN_WILLTOPIC, Flags), WillTopic};	6✔
87	parse_var(?SN_WILLMSGREQ, <<>>) ->
88	undefined;	5✔
89	parse_var(?SN_WILLMSG, <<WillMsg/binary>>) ->
90	WillMsg;	4✔
91	parse_var(?SN_REGISTER, <<TopicId:?short, MsgId:?short, TopicName/binary>>) ->
92	{TopicId, MsgId, TopicName};	6✔
93	parse_var(?SN_REGACK, <<TopicId:?short, MsgId:?short, ReturnCode:?byte>>) ->
94	{TopicId, MsgId, ReturnCode};	2✔
95	parse_var(?SN_PUBLISH, <<FlagsBin:?flag, Topic:2/binary, MsgId:?short, Data/binary>>) ->
96	#mqtt_sn_flags{topic_id_type = IdType} = Flags = parse_flags(?SN_PUBLISH, FlagsBin),	4✔
97	{Flags, parse_topic(IdType, Topic), MsgId, Data};	4✔
98	parse_var(?SN_PUBACK, <<TopicId:?short, MsgId:?short, ReturnCode:?byte>>) ->
99	{TopicId, MsgId, ReturnCode};	5✔
100	parse_var(PubRec, <<MsgId:?short>>) when
101	PubRec == ?SN_PUBREC; PubRec == ?SN_PUBREL; PubRec == ?SN_PUBCOMP
102	->
103	MsgId;	2✔
104	parse_var(Sub, <<FlagsBin:?flag, MsgId:?short, Topic/binary>>) when
105	Sub == ?SN_SUBSCRIBE; Sub == ?SN_UNSUBSCRIBE
106	->
107	#mqtt_sn_flags{topic_id_type = IdType} = Flags = parse_flags(Sub, FlagsBin),	5✔
108	{Flags, MsgId, parse_topic(IdType, Topic)};	5✔
109	parse_var(?SN_SUBACK, <<Flags:?flag, TopicId:?short, MsgId:?short, ReturnCode:?byte>>) ->
110	{parse_flags(?SN_SUBACK, Flags), TopicId, MsgId, ReturnCode};	3✔
111	parse_var(?SN_UNSUBACK, <<MsgId:?short>>) ->
112	MsgId;	5✔
113	parse_var(?SN_PINGREQ, ClientId) ->
114	ClientId;	1✔
115	parse_var(?SN_PINGRESP, _) ->
116	undefined;	5✔
117	parse_var(?SN_DISCONNECT, <<>>) ->
118	undefined;	4✔
119	parse_var(?SN_DISCONNECT, <<Duration:?short>>) ->
120	Duration;	4✔
121	parse_var(?SN_WILLTOPICUPD, <<>>) ->
122	{undefined, undefined};	×
123	parse_var(?SN_WILLTOPICUPD, <<Flags:?flag, WillTopic/binary>>) ->
124	{parse_flags(?SN_WILLTOPICUPD, Flags), WillTopic};	1✔
125	parse_var(?SN_WILLMSGUPD, WillMsg) ->
126	WillMsg;	5✔
127	parse_var(?SN_WILLTOPICRESP, <<ReturnCode:?byte>>) ->
128	ReturnCode;	5✔
129	parse_var(?SN_WILLMSGRESP, <<ReturnCode:?byte>>) ->
130	ReturnCode;	3✔
131	parse_var(_Type, _Var) ->
132	error(unkown_message_type).	×
133
134	parse_flags(?SN_CONNECT, <<_D:1, _Q:2, _R:1, Will:1, CleanStart:1, _IdType:2>>) ->
135	#mqtt_sn_flags{will = bool(Will), clean_start = bool(CleanStart)};	6✔
136	parse_flags(?SN_WILLTOPIC, <<_D:1, QoS:2, Retain:1, _Will:1, _C:1, _:2>>) ->
137	#mqtt_sn_flags{qos = QoS, retain = bool(Retain)};	6✔
138	parse_flags(?SN_PUBLISH, <<Dup:1, QoS:2, Retain:1, _Will:1, _C:1, IdType:2>>) ->
139	#mqtt_sn_flags{dup = bool(Dup), qos = QoS, retain = bool(Retain), topic_id_type = IdType};	4✔
140	parse_flags(Sub, <<Dup:1, QoS:2, _R:1, _Will:1, _C:1, IdType:2>>) when
141	Sub == ?SN_SUBSCRIBE; Sub == ?SN_UNSUBSCRIBE
142	->
143	#mqtt_sn_flags{dup = bool(Dup), qos = QoS, topic_id_type = IdType};	5✔
144	parse_flags(?SN_SUBACK, <<_D:1, QoS:2, _R:1, _W:1, _C:1, _Id:2>>) ->
145	#mqtt_sn_flags{qos = QoS};	3✔
146	parse_flags(?SN_WILLTOPICUPD, <<_D:1, QoS:2, Retain:1, _W:1, _C:1, _Id:2>>) ->
147	#mqtt_sn_flags{qos = QoS, retain = bool(Retain)};	1✔
148	parse_flags(_Type, _) ->
149	error(malformed_message_flags).	×
150
151	parse_topic(2#00, Topic) -> Topic;	3✔
152	parse_topic(2#01, <<Id:16>>) -> Id;	4✔
153	parse_topic(2#10, Topic) -> Topic;	2✔
154	parse_topic(2#11, Topic) -> Topic.	×
155
156	%%--------------------------------------------------------------------
157	%% Serialize MQTT-SN Message
158	%%--------------------------------------------------------------------
159
160	serialize_pkt(#mqtt_sn_message{type = Type, variable = Var}, Opts) ->
161	VarBin = serialize(Type, Var, Opts),	100✔
162	VarLen = size(VarBin),	100✔
163	case VarLen < 254 of	100✔
164	true ->
165	<<(VarLen + 2), Type, VarBin/binary>>;	99✔
166	false ->
167	<<16#01, (VarLen + 4):?short, Type, VarBin/binary>>	1✔
168	end.
169
170	serialize(?SN_ADVERTISE, {GwId, Duration}, _Opts) ->
171	<<GwId, Duration:?short>>;	2✔
172	serialize(?SN_SEARCHGW, Radius, _Opts) ->
173	<<Radius>>;	3✔
174	serialize(?SN_GWINFO, {GwId, GwAdd}, _Opts) ->
175	<<GwId, GwAdd/binary>>;	2✔
176	serialize(?SN_CONNECT, {Flags, ProtocolId, Duration, ClientId}, _Opts) ->
177	<<(serialize_flags(Flags))/binary, ProtocolId, Duration:?short, ClientId/binary>>;	6✔
178	serialize(?SN_CONNACK, ReturnCode, _Opts) ->
179	<<ReturnCode>>;	5✔
180	serialize(?SN_WILLTOPICREQ, _, _Opts) ->
181	<<>>;	5✔
182	serialize(?SN_WILLTOPIC, undefined, _Opts) ->
183	<<>>;	2✔
184	serialize(?SN_WILLTOPIC, {Flags, Topic}, _Opts) ->
185	%% The WillTopic must a short topic name
186	<<(serialize_flags(Flags))/binary, Topic/binary>>;	6✔
187	serialize(?SN_WILLMSGREQ, _, _Opts) ->
188	<<>>;	5✔
189	serialize(?SN_WILLMSG, WillMsg, _Opts) ->
190	WillMsg;	4✔
191	serialize(?SN_REGISTER, {TopicId, MsgId, TopicName}, _Opts) ->
192	<<TopicId:?short, MsgId:?short, TopicName/binary>>;	6✔
193	serialize(?SN_REGACK, {TopicId, MsgId, ReturnCode}, _Opts) ->
194	<<TopicId:?short, MsgId:?short, ReturnCode>>;	2✔
195	serialize(
196	?SN_PUBLISH,
197	{Flags = #mqtt_sn_flags{topic_id_type = ?SN_NORMAL_TOPIC}, TopicId, MsgId, Data},
198	_Opts
199	) ->
200	<<(serialize_flags(Flags))/binary, TopicId:?short, MsgId:?short, Data/binary>>;	×
201	serialize(
202	?SN_PUBLISH,
203	{Flags = #mqtt_sn_flags{topic_id_type = ?SN_PREDEFINED_TOPIC}, TopicId, MsgId, Data},
204	_Opts
205	) ->
206	<<(serialize_flags(Flags))/binary, TopicId:?short, MsgId:?short, Data/binary>>;	2✔
207	serialize(
208	?SN_PUBLISH,
209	{Flags = #mqtt_sn_flags{topic_id_type = ?SN_SHORT_TOPIC}, STopicName, MsgId, Data},
210	_Opts
211	) ->
212	<<(serialize_flags(Flags))/binary, STopicName:2/binary, MsgId:?short, Data/binary>>;	2✔
213	serialize(?SN_PUBACK, {TopicId, MsgId, ReturnCode}, _Opts) ->
214	<<TopicId:?short, MsgId:?short, ReturnCode>>;	5✔
215	serialize(PubRec, MsgId, _Opts) when
216	PubRec == ?SN_PUBREC; PubRec == ?SN_PUBREL; PubRec == ?SN_PUBCOMP
217	->
218	<<MsgId:?short>>;	2✔
219	serialize(Sub, {Flags = #mqtt_sn_flags{topic_id_type = IdType}, MsgId, Topic}, _Opts) when
220	Sub == ?SN_SUBSCRIBE; Sub == ?SN_UNSUBSCRIBE
221	->
222	<<(serialize_flags(Flags))/binary, MsgId:16, (serialize_topic(IdType, Topic))/binary>>;	5✔
223	serialize(?SN_SUBACK, {Flags, TopicId, MsgId, ReturnCode}, _Opts) ->
224	<<(serialize_flags(Flags))/binary, TopicId:?short, MsgId:?short, ReturnCode>>;	3✔
225	serialize(?SN_UNSUBACK, MsgId, _Opts) ->
226	<<MsgId:?short>>;	5✔
227	serialize(?SN_PINGREQ, ClientId, _Opts) ->
228	ClientId;	1✔
229	serialize(?SN_PINGRESP, _, _Opts) ->
230	<<>>;	5✔
231	serialize(?SN_WILLTOPICUPD, {Flags, WillTopic}, _Opts) ->
232	<<(serialize_flags(Flags))/binary, WillTopic/binary>>;	1✔
233	serialize(?SN_WILLMSGUPD, WillMsg, _Opts) ->
234	WillMsg;	5✔
235	serialize(?SN_WILLTOPICRESP, ReturnCode, _Opts) ->
236	<<ReturnCode>>;	5✔
237	serialize(?SN_WILLMSGRESP, ReturnCode, _Opts) ->
238	<<ReturnCode>>;	3✔
239	serialize(?SN_DISCONNECT, undefined, _Opts) ->
240	<<>>;	4✔
241	serialize(?SN_DISCONNECT, Duration, _Opts) ->
242	<<Duration:?short>>.	4✔
243
244	serialize_flags(#mqtt_sn_flags{
245	dup = Dup,
246	qos = QoS,
247	retain = Retain,
248	will = Will,
249	clean_start = CleanStart,
250	topic_id_type = IdType
251	}) ->
252	<<	25✔
253	(bool(Dup)):1,
254	(i(QoS)):2,
255	(bool(Retain)):1,
256	(bool(Will)):1,
257	(bool(CleanStart)):1,
258	(i(IdType)):2
259	>>.
260
261	serialize_topic(2#00, Topic) -> Topic;	3✔
262	serialize_topic(2#01, Id) -> <<Id:?short>>;	2✔
UNCOV 263	serialize_topic(2#10, Topic) -> Topic;	×
264	serialize_topic(2#11, Topic) -> Topic.	×
265
266	bool(0) -> false;	16✔
267	bool(1) -> true;	16✔
268	bool(false) -> 0;	84✔
269	bool(true) -> 1;	16✔
270	bool(undefined) -> 0.	×
271
272	i(undefined) -> 0;	×
273	i(I) when is_integer(I) -> I.	50✔
274
275	message_type(16#00) ->
276	"SN_ADVERTISE";	×
277	message_type(16#01) ->
278	"SN_SEARCHGW";	×
279	message_type(16#02) ->
280	"SN_GWINFO";	×
281	message_type(16#04) ->
282	"SN_CONNECT";	×
283	message_type(16#05) ->
284	"SN_CONNACK";	×
285	message_type(16#06) ->
286	"SN_WILLTOPICREQ";	×
287	message_type(16#07) ->
288	"SN_WILLTOPIC";	×
289	message_type(16#08) ->
290	"SN_WILLMSGREQ";	×
291	message_type(16#09) ->
292	"SN_WILLMSG";	×
293	message_type(16#0a) ->
294	"SN_REGISTER";	×
295	message_type(16#0b) ->
296	"SN_REGACK";	×
297	message_type(16#0c) ->
298	"SN_PUBLISH";	×
299	message_type(16#0d) ->
300	"SN_PUBACK";	×
301	message_type(16#0e) ->
302	"SN_PUBCOMP";	×
303	message_type(16#0f) ->
304	"SN_PUBREC";	×
305	message_type(16#10) ->
306	"SN_PUBREL";	×
307	message_type(16#12) ->
308	"SN_SUBSCRIBE";	×
309	message_type(16#13) ->
310	"SN_SUBACK";	×
311	message_type(16#14) ->
312	"SN_UNSUBSCRIBE";	×
313	message_type(16#15) ->
314	"SN_UNSUBACK";	×
315	message_type(16#16) ->
316	"SN_PINGREQ";	×
317	message_type(16#17) ->
318	"SN_PINGRESP";	×
319	message_type(16#18) ->
320	"SN_DISCONNECT";	×
321	message_type(16#1a) ->
322	"SN_WILLTOPICUPD";	×
323	message_type(16#1b) ->
324	"SN_WILLTOPICRESP";	×
325	message_type(16#1c) ->
326	"SN_WILLMSGUPD";	×
327	message_type(16#1d) ->
328	"SN_WILLMSGRESP";	×
329	message_type(Type) ->
330	io_lib:format("Unknown Type ~p", [Type]).	×
331
332	format(?SN_CONNECT_MSG(Flags, ProtocolId, Duration, ClientId)) ->
333	#mqtt_sn_flags{	×
334	will = Will,
335	clean_start = CleanStart
336	} = Flags,
337	io_lib:format(	×
338	"SN_CONNECT(W~w, C~w, ProtocolId=~w, Duration=~w, "
339	"ClientId=~ts)",
340	[
341	bool(Will),
342	bool(CleanStart),
343	ProtocolId,
344	Duration,
345	ClientId
346	]
347	);
348	format(?SN_CONNACK_MSG(ReturnCode)) ->
349	io_lib:format("SN_CONNACK(ReturnCode=~w)", [ReturnCode]);	×
350	format(?SN_WILLTOPICREQ_MSG()) ->
351	"SN_WILLTOPICREQ()";	×
352	format(?SN_WILLTOPIC_MSG(Flags, Topic)) ->
353	#mqtt_sn_flags{	×
354	qos = QoS,
355	retain = Retain
356	} = Flags,
357	io_lib:format(	×
358	"SN_WILLTOPIC(Q~w, R~w, Topic=~s)",
359	[QoS, bool(Retain), Topic]
360	);
361	format(?SN_WILLTOPIC_EMPTY_MSG) ->
362	"SN_WILLTOPIC(_)";	×
363	format(?SN_WILLMSGREQ_MSG()) ->
364	"SN_WILLMSGREQ()";	×
365	format(?SN_WILLMSG_MSG(Msg)) ->
366	io_lib:format("SN_WILLMSG_MSG(Msg=~p)", [Msg]);	×
367	format(?SN_PUBLISH_MSG(Flags, TopicId, MsgId, Data)) ->
368	#mqtt_sn_flags{	×
369	dup = Dup,
370	qos = QoS,
371	retain = Retain,
372	topic_id_type = TopicIdType
373	} = Flags,
374	io_lib:format(	×
375	"SN_PUBLISH(D~w, Q~w, R~w, TopicIdType=~w, TopicId=~w, "
376	"MsgId=~w, Payload=~p)",
377	[
378	bool(Dup),
379	QoS,
380	bool(Retain),
381	TopicIdType,
382	TopicId,
383	MsgId,
384	Data
385	]
386	);
387	format(?SN_PUBACK_MSG(TopicId, MsgId, ReturnCode)) ->
388	io_lib:format(	×
389	"SN_PUBACK(TopicId=~w, MsgId=~w, ReturnCode=~w)",
390	[TopicId, MsgId, ReturnCode]
391	);
392	format(?SN_PUBREC_MSG(?SN_PUBCOMP, MsgId)) ->
393	io_lib:format("SN_PUBCOMP(MsgId=~w)", [MsgId]);	×
394	format(?SN_PUBREC_MSG(?SN_PUBREC, MsgId)) ->
395	io_lib:format("SN_PUBREC(MsgId=~w)", [MsgId]);	×
396	format(?SN_PUBREC_MSG(?SN_PUBREL, MsgId)) ->
397	io_lib:format("SN_PUBREL(MsgId=~w)", [MsgId]);	×
398	format(?SN_SUBSCRIBE_MSG(Flags, Msgid, Topic)) ->
399	#mqtt_sn_flags{	×
400	dup = Dup,
401	qos = QoS,
402	topic_id_type = TopicIdType
403	} = Flags,
404	io_lib:format(	×
405	"SN_SUBSCRIBE(D~w, Q~w, TopicIdType=~w, MsgId=~w, "
406	"TopicId=~w)",
407	[bool(Dup), QoS, TopicIdType, Msgid, Topic]
408	);
409	format(?SN_SUBACK_MSG(Flags, TopicId, MsgId, ReturnCode)) ->
410	#mqtt_sn_flags{qos = QoS} = Flags,	×
411	io_lib:format(	×
412	"SN_SUBACK(GrantedQoS=~w, MsgId=~w, TopicId=~w, "
413	"ReturnCode=~w)",
414	[QoS, MsgId, TopicId, ReturnCode]
415	);
416	format(?SN_UNSUBSCRIBE_MSG(Flags, Msgid, Topic)) ->
417	#mqtt_sn_flags{topic_id_type = TopicIdType} = Flags,	×
418	io_lib:format(	×
419	"SN_UNSUBSCRIBE(TopicIdType=~w, MsgId=~w, TopicId=~w)",
420	[TopicIdType, Msgid, Topic]
421	);
422	format(?SN_UNSUBACK_MSG(MsgId)) ->
423	io_lib:format("SN_UNSUBACK(MsgId=~w)", [MsgId]);	×
424	format(?SN_REGISTER_MSG(TopicId, MsgId, TopicName)) ->
425	io_lib:format(	×
426	"SN_REGISTER(TopicId=~w, MsgId=~w, TopicName=~s)",
427	[TopicId, MsgId, TopicName]
428	);
429	format(?SN_REGACK_MSG(TopicId, MsgId, ReturnCode)) ->
430	io_lib:format(	×
431	"SN_REGACK(TopicId=~w, MsgId=~w, ReturnCode=~w)",
432	[TopicId, MsgId, ReturnCode]
433	);
434	format(?SN_PINGREQ_MSG(ClientId)) ->
435	io_lib:format("SN_PINGREQ(ClientId=~s)", [ClientId]);	×
436	format(?SN_PINGRESP_MSG()) ->
437	"SN_PINGRESP()";	×
438	format(?SN_DISCONNECT_MSG(Duration)) ->
439	io_lib:format("SN_DISCONNECT(Duration=~w)", [Duration]);	×
440	format(#mqtt_sn_message{type = Type, variable = Var}) ->
441	io_lib:format(	×
442	"mqtt_sn_message(type=~s, Var=~w)",
443	[emqx_mqttsn_frame:message_type(Type), Var]
444	).
445
446	is_message(#mqtt_sn_message{type = Type}) when
447	Type == ?SN_PUBLISH
448	->
449	true;	×
450	is_message(_) ->
451	false.	×
452
453	type(#mqtt_sn_message{type = Type}) ->
454	type(Type);	×
455	type(?SN_ADVERTISE) ->
456	advertise;	×
457	type(?SN_SEARCHGW) ->
458	serachgw;	×
459	type(?SN_GWINFO) ->
460	gwinfo;	×
461	type(?SN_CONNECT) ->
462	connect;	×
463	type(?SN_CONNACK) ->
464	connack;	×
465	type(?SN_WILLTOPICREQ) ->
466	willtopicreq;	×
467	type(?SN_WILLTOPIC) ->
468	willtopic;	×
469	type(?SN_WILLMSGREQ) ->
470	willmsgreq;	×
471	type(?SN_WILLMSG) ->
472	willmsg;	×
473	type(?SN_REGISTER) ->
474	register;	×
475	type(?SN_REGACK) ->
476	regack;	×
477	type(?SN_PUBLISH) ->
478	publish;	×
479	type(?SN_PUBACK) ->
480	puback;	×
481	type(?SN_PUBCOMP) ->
482	pubcomp;	×
483	type(?SN_PUBREC) ->
484	pubrec;	×
485	type(?SN_PUBREL) ->
486	pubrel;	×
487	type(?SN_SUBSCRIBE) ->
488	subscribe;	×
489	type(?SN_SUBACK) ->
490	suback;	×
491	type(?SN_UNSUBSCRIBE) ->
492	unsubscribe;	×
493	type(?SN_UNSUBACK) ->
494	unsuback;	×
495	type(?SN_PINGREQ) ->
496	pingreq;	×
497	type(?SN_PINGRESP) ->
498	pingresp;	×
499	type(?SN_DISCONNECT) ->
500	disconnect;	×
501	type(?SN_WILLTOPICUPD) ->
502	willtopicupd;	×
503	type(?SN_WILLTOPICRESP) ->
504	willtopicresp;	×
505	type(?SN_WILLMSGUPD) ->
506	willmsgupd;	×
507	type(?SN_WILLMSGRESP) ->
508	willmsgresp.	×

emqx / emqx / 8628139215

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous