From 6fb3ff1f9f183c444c9126dc06e025d4d8fe9fd2 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Fri, 5 Nov 2021 18:32:43 +0800 Subject: [PATCH 01/24] feat(exhook): expose headers for on_messages_publish hook --- apps/emqx_exhook/priv/protos/exhook.proto | 25 ++++++++++ apps/emqx_exhook/src/emqx_exhook_handler.erl | 48 +++++++++++++++++-- .../emqx_exhook/test/emqx_exhook_demo_svr.erl | 11 +++-- src/emqx_types.erl | 1 + 4 files changed, 78 insertions(+), 7 deletions(-) diff --git a/apps/emqx_exhook/priv/protos/exhook.proto b/apps/emqx_exhook/priv/protos/exhook.proto index 72ba26581..639066c6a 100644 --- a/apps/emqx_exhook/priv/protos/exhook.proto +++ b/apps/emqx_exhook/priv/protos/exhook.proto @@ -358,6 +358,31 @@ message Message { bytes payload = 6; uint64 timestamp = 7; + + // The key of header can be: + // - username: + // * Readonly + // * The username of sender client + // * Value type: utf8 string + // - protocol: + // * Readonly + // * The protocol name of sender client + // * Value type: string enum with "mqtt", "mqtt-sn", ... + // - peerhost: + // * Readonly + // * The peerhost of sender client + // * Value type: ip address string + // - allow_publish: + // * Writable + // * Whether to allow the message to be published by emqx + // * Value type: string enum with "true", "false", default is "true" + // + // Notes: All header may be missing, which means that the message does not + // carry these headers. We can guarantee that clients coming from MQTT, + // MQTT-SN, CoAP, LwM2M and other natively supported protocol clients will + // carry these headers, but there is no guarantee that messages published + // by other means will do, e.g. messages published by HTTP-API + map headers = 8; } message Property { diff --git a/apps/emqx_exhook/src/emqx_exhook_handler.erl b/apps/emqx_exhook/src/emqx_exhook_handler.erl index f3964dc42..efdf7f8b9 100644 --- a/apps/emqx_exhook/src/emqx_exhook_handler.erl +++ b/apps/emqx_exhook/src/emqx_exhook_handler.erl @@ -258,17 +258,57 @@ clientinfo(ClientInfo = cn => maybe(maps:get(cn, ClientInfo, undefined)), dn => maybe(maps:get(dn, ClientInfo, undefined))}. -message(#message{id = Id, qos = Qos, from = From, topic = Topic, payload = Payload, timestamp = Ts}) -> +message(#message{id = Id, qos = Qos, from = From, topic = Topic, + payload = Payload, timestamp = Ts, headers = Headers}) -> #{node => stringfy(node()), id => emqx_guid:to_hexstr(Id), qos => Qos, from => stringfy(From), topic => Topic, payload => Payload, - timestamp => Ts}. + timestamp => Ts, + headers => headers(Headers) + }. -assign_to_message(#{qos := Qos, topic := Topic, payload := Payload}, Message) -> - Message#message{qos = Qos, topic = Topic, payload = Payload}. +headers(undefined) -> + #{}; +headers(Headers) -> + Ls = [username, protocol, peerhost, allow_publish], + maps:fold( + fun + (_, undefined, Acc) -> + Acc; %% Ignore undefined value + (K, V, Acc) -> + case lists:member(K, Ls) of + true -> + Acc#{atom_to_binary(K) => bin(K, V)}; + _ -> + Acc + end + end, #{}, Headers). + +bin(K, V) when K == username; + K == protocol; + K == allow_publish -> + bin(V); +bin(peerhost, V) -> + bin(inet:ntoa(V)). + +bin(V) when is_binary(V) -> V; +bin(V) when is_atom(V) -> atom_to_binary(V); +bin(V) when is_list(V) -> iolist_to_binary(V). + +assign_to_message(InMessage = #{qos := Qos, topic := Topic, + payload := Payload}, Message) -> + NMsg = Message#message{qos = Qos, topic = Topic, payload = Payload}, + enrich_header(maps:get(headers, InMessage, #{}), NMsg). + +enrich_header(Headers, Message) -> + AllowPub = case maps:get(<<"allow_publish">>, Headers, <<"true">>) of + <<"false">> -> false; + _ -> true + end, + emqx_message:set_header(allow_publish, AllowPub, Message). topicfilters(Tfs) when is_list(Tfs) -> [#{name => Topic, qos => Qos} || {Topic, #{qos := Qos}} <- Tfs]. diff --git a/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl b/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl index c2db04dd4..fd8a5f9a3 100644 --- a/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl +++ b/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl @@ -295,14 +295,14 @@ on_session_terminated(Req, Md) -> | {error, grpc_cowboy_h:error_response()}. on_message_publish(#{message := #{from := From} = Msg} = Req, Md) -> ?MODULE:in({?FUNCTION_NAME, Req}), - %io:format("fun: ~p, req: ~0p~n", [?FUNCTION_NAME, Req]), + io:format(standard_error, "fun: ~p, req: ~0p~n", [?FUNCTION_NAME, Req]), %% some cases for testing case From of <<"baduser">> -> - NMsg = Msg#{qos => 0, + NMsg = deny(Msg#{qos => 0, topic => <<"">>, payload => <<"">> - }, + }), {ok, #{type => 'STOP_AND_RETURN', value => {message, NMsg}}, Md}; <<"gooduser">> -> @@ -314,6 +314,11 @@ on_message_publish(#{message := #{from := From} = Msg} = Req, Md) -> {ok, #{type => 'IGNORE'}, Md} end. +deny(Msg) -> + NHeader = maps:put(<<"allow_publish">>, <<"false">>, + maps:get(headers, Msg, #{})), + maps:put(headers, NHeader, Msg). + -spec on_message_delivered(emqx_exhook_pb:message_delivered_request(), grpc:metadata()) -> {ok, emqx_exhook_pb:empty_success(), grpc:metadata()} | {error, grpc_cowboy_h:error_response()}. diff --git a/src/emqx_types.erl b/src/emqx_types.erl index 3e53eafd1..410ed5a27 100644 --- a/src/emqx_types.erl +++ b/src/emqx_types.erl @@ -193,6 +193,7 @@ username => username(), peerhost => peerhost(), properties => properties(), + allow_publish => boolean(), atom() => term()}). -type(banned() :: #banned{}). From ef2a5c1dc71722d35aafc307d516a99503c7e5cd Mon Sep 17 00:00:00 2001 From: JianBo He Date: Mon, 8 Nov 2021 11:11:16 +0800 Subject: [PATCH 02/24] chore(exhook): fix diaylzer warnings --- apps/emqx_exhook/src/emqx_exhook_handler.erl | 2 -- 1 file changed, 2 deletions(-) diff --git a/apps/emqx_exhook/src/emqx_exhook_handler.erl b/apps/emqx_exhook/src/emqx_exhook_handler.erl index efdf7f8b9..45e6bf082 100644 --- a/apps/emqx_exhook/src/emqx_exhook_handler.erl +++ b/apps/emqx_exhook/src/emqx_exhook_handler.erl @@ -270,8 +270,6 @@ message(#message{id = Id, qos = Qos, from = From, topic = Topic, headers => headers(Headers) }. -headers(undefined) -> - #{}; headers(Headers) -> Ls = [username, protocol, peerhost, allow_publish], maps:fold( From 012c7415926d2cf11cc1e52917c4774dbb89021e Mon Sep 17 00:00:00 2001 From: JianBo He Date: Mon, 8 Nov 2021 11:11:37 +0800 Subject: [PATCH 03/24] chore(exhook): update appup.src --- apps/emqx_exhook/src/emqx_exhook.app.src | 2 +- apps/emqx_exhook/src/emqx_exhook.appup.src | 22 ++++++++++++---------- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/apps/emqx_exhook/src/emqx_exhook.app.src b/apps/emqx_exhook/src/emqx_exhook.app.src index 46223d212..b386bcaca 100644 --- a/apps/emqx_exhook/src/emqx_exhook.app.src +++ b/apps/emqx_exhook/src/emqx_exhook.app.src @@ -1,6 +1,6 @@ {application, emqx_exhook, [{description, "EMQ X Extension for Hook"}, - {vsn, "4.3.4"}, + {vsn, "4.3.5"}, {modules, []}, {registered, []}, {mod, {emqx_exhook_app, []}}, diff --git a/apps/emqx_exhook/src/emqx_exhook.appup.src b/apps/emqx_exhook/src/emqx_exhook.appup.src index d6a699c33..e78afef1e 100644 --- a/apps/emqx_exhook/src/emqx_exhook.appup.src +++ b/apps/emqx_exhook/src/emqx_exhook.appup.src @@ -1,15 +1,17 @@ %% -*-: erlang -*- {VSN, - [ - {<<"4.3.[0-3]">>, [ - {restart_application, emqx_exhook} - ]}, - {<<".*">>, []} + [{"4.3.4", + [{load_module,emqx_exhook_pb,brutal_purge,soft_purge,[]}, + {load_module,emqx_exhook_handler,brutal_purge,soft_purge,[emqx_exhook_pb]}]}, + {<<"4\\.3\\.[0-3]+">>, + [{restart_application,emqx_exhook}]}, + {<<".*">>, []} ], - [ - {<<"4.3.[0-3]">>, [ - {restart_application, emqx_exhook} - ]}, - {<<".*">>, []} + [{"4.3.4", + [{load_module,emqx_exhook_pb,brutal_purge,soft_purge,[]}, + {load_module,emqx_exhook_handler,brutal_purge,soft_purge,[emqx_exhook_pb]}]}, + {<<"4\\.3\\.[0-3]+">>, + [{restart_application,emqx_exhook}]}, + {<<".*">>, []} ] }. From b756e7d17a4e73213329d518cd5e0217355cf268 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Mon, 8 Nov 2021 14:15:32 +0800 Subject: [PATCH 04/24] chore: upgrade grpc to 0.6.4 --- apps/emqx_exhook/rebar.config | 2 +- apps/emqx_exproto/rebar.config | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/emqx_exhook/rebar.config b/apps/emqx_exhook/rebar.config index 3529b6314..d1cc4d778 100644 --- a/apps/emqx_exhook/rebar.config +++ b/apps/emqx_exhook/rebar.config @@ -5,7 +5,7 @@ ]}. {deps, - [{grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.3"}}} + [{grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.4"}}} ]}. {grpc, diff --git a/apps/emqx_exproto/rebar.config b/apps/emqx_exproto/rebar.config index 4ad1aa192..da868de82 100644 --- a/apps/emqx_exproto/rebar.config +++ b/apps/emqx_exproto/rebar.config @@ -13,7 +13,7 @@ ]}. {deps, - [{grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.3"}}} + [{grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.4"}}} ]}. {grpc, From c170d076e32ad72db408a85a41644db2c8df94e2 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Mon, 8 Nov 2021 14:16:26 +0800 Subject: [PATCH 05/24] feat(exhook): expose process pool_size for grpc client --- apps/emqx_exhook/etc/emqx_exhook.conf | 5 +++++ apps/emqx_exhook/priv/emqx_exhook.schema | 4 ++++ apps/emqx_exhook/src/emqx_exhook_mngr.erl | 11 +++++++++++ apps/emqx_exhook/src/emqx_exhook_server.erl | 3 ++- apps/emqx_exhook/src/emqx_exhook_sup.erl | 5 +++-- 5 files changed, 25 insertions(+), 3 deletions(-) diff --git a/apps/emqx_exhook/etc/emqx_exhook.conf b/apps/emqx_exhook/etc/emqx_exhook.conf index ffb71e43b..23895f902 100644 --- a/apps/emqx_exhook/etc/emqx_exhook.conf +++ b/apps/emqx_exhook/etc/emqx_exhook.conf @@ -24,6 +24,11 @@ ## Value: false | Duration #exhook.auto_reconnect = 60s +## The process pool size for gRPC client +## +## Default: Equals cpu cores +## Value: Integer +#exhook.pool_size = 16 ##-------------------------------------------------------------------- ## The Hook callback servers diff --git a/apps/emqx_exhook/priv/emqx_exhook.schema b/apps/emqx_exhook/priv/emqx_exhook.schema index d11001c0d..f55913d72 100644 --- a/apps/emqx_exhook/priv/emqx_exhook.schema +++ b/apps/emqx_exhook/priv/emqx_exhook.schema @@ -26,6 +26,10 @@ end end}. +{mapping, "exhook.pool_size", "emqx_exhook.pool_size", [ + {datatype, integer} +]}. + {mapping, "exhook.server.$name.url", "emqx_exhook.servers", [ {datatype, string} ]}. diff --git a/apps/emqx_exhook/src/emqx_exhook_mngr.erl b/apps/emqx_exhook/src/emqx_exhook_mngr.erl index cadd5eb37..54e106f13 100644 --- a/apps/emqx_exhook/src/emqx_exhook_mngr.erl +++ b/apps/emqx_exhook/src/emqx_exhook_mngr.erl @@ -36,6 +36,8 @@ , server/1 , put_request_failed_action/1 , get_request_failed_action/0 + , put_pool_size/1 + , get_pool_size/0 ]). %% gen_server callbacks @@ -117,6 +119,9 @@ init([Servers, AutoReconnect, ReqOpts0]) -> put_request_failed_action( maps:get(request_failed_action, ReqOpts0, deny) ), + put_pool_size( + maps:get(pool_size, ReqOpts0, erlang:system_info(schedulers)) + ), %% Load the hook servers ReqOpts = maps:without([request_failed_action], ReqOpts0), @@ -286,6 +291,12 @@ put_request_failed_action(Val) -> get_request_failed_action() -> persistent_term:get({?APP, request_failed_action}). +put_pool_size(Val) -> + persistent_term:put({?APP, pool_size}, Val). + +get_pool_size() -> + persistent_term:get({?APP, pool_size}). + save(Name, ServerState) -> Saved = persistent_term:get(?APP, []), persistent_term:put(?APP, lists:reverse([Name | Saved])), diff --git a/apps/emqx_exhook/src/emqx_exhook_server.erl b/apps/emqx_exhook/src/emqx_exhook_server.erl index 7df5b643c..123cbb558 100644 --- a/apps/emqx_exhook/src/emqx_exhook_server.erl +++ b/apps/emqx_exhook/src/emqx_exhook_server.erl @@ -131,7 +131,8 @@ channel_opts(Opts) -> transport_opts => SslOpts}}; _ -> #{} end, - {SvrAddr, ClientOpts}. + NClientOpts = ClientOpts#{pool_size => emqx_exhook_mngr:get_pool_size()}, + {SvrAddr, NClientOpts}. format_http_uri(Scheme, Host0, Port) -> Host = case is_tuple(Host0) of diff --git a/apps/emqx_exhook/src/emqx_exhook_sup.erl b/apps/emqx_exhook/src/emqx_exhook_sup.erl index e9c405de0..c92fd6ca4 100644 --- a/apps/emqx_exhook/src/emqx_exhook_sup.erl +++ b/apps/emqx_exhook/src/emqx_exhook_sup.erl @@ -54,7 +54,8 @@ auto_reconnect() -> request_options() -> #{timeout => env(request_timeout, 5000), - request_failed_action => env(request_failed_action, deny) + request_failed_action => env(request_failed_action, deny), + pool_size => env(pool_size, erlang:system_info(schedulers)) }. env(Key, Def) -> @@ -67,7 +68,7 @@ env(Key, Def) -> -spec start_grpc_client_channel( string(), uri_string:uri_string(), - grpc_client:options()) -> {ok, pid()} | {error, term()}. + grpc_client_sup:options()) -> {ok, pid()} | {error, term()}. start_grpc_client_channel(Name, SvrAddr, Options) -> grpc_client_sup:create_channel_pool(Name, SvrAddr, Options). From 44008b9a6da0aa39354be5c7a96197297942023e Mon Sep 17 00:00:00 2001 From: JianBo He Date: Mon, 8 Nov 2021 15:05:29 +0800 Subject: [PATCH 06/24] chore: fix compiling warnings --- apps/emqx_sn/test/emqx_sn_protocol_SUITE.erl | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/emqx_sn/test/emqx_sn_protocol_SUITE.erl b/apps/emqx_sn/test/emqx_sn_protocol_SUITE.erl index 0947bdaca..794eedd6b 100644 --- a/apps/emqx_sn/test/emqx_sn_protocol_SUITE.erl +++ b/apps/emqx_sn/test/emqx_sn_protocol_SUITE.erl @@ -171,7 +171,7 @@ t_subscribe_case02(_) -> {ok, Socket} = gen_udp:open(0, [binary]), ClientId = ?CLIENTID, - send_connect_msg(Socket, ?CLIENTID), + send_connect_msg(Socket, ClientId), ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), Topic1 = ?PREDEF_TOPIC_NAME1, From 5922521e499b8bb2b551e5ca2b8564daefd49310 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Mon, 8 Nov 2021 15:05:54 +0800 Subject: [PATCH 07/24] test(props): cover messages headers --- apps/emqx_exhook/src/emqx_exhook_handler.erl | 14 +++++++++----- .../emqx_exhook/test/emqx_exhook_demo_svr.erl | 9 +++++++-- .../test/props/prop_exhook_hooks.erl | 19 +++++++++++++------ 3 files changed, 29 insertions(+), 13 deletions(-) diff --git a/apps/emqx_exhook/src/emqx_exhook_handler.erl b/apps/emqx_exhook/src/emqx_exhook_handler.erl index 45e6bf082..ea6cdf5bc 100644 --- a/apps/emqx_exhook/src/emqx_exhook_handler.erl +++ b/apps/emqx_exhook/src/emqx_exhook_handler.erl @@ -50,6 +50,7 @@ %% Utils -export([ message/1 + , headers/1 , stringfy/1 , merge_responsed_bool/2 , merge_responsed_message/2 @@ -302,11 +303,14 @@ assign_to_message(InMessage = #{qos := Qos, topic := Topic, enrich_header(maps:get(headers, InMessage, #{}), NMsg). enrich_header(Headers, Message) -> - AllowPub = case maps:get(<<"allow_publish">>, Headers, <<"true">>) of - <<"false">> -> false; - _ -> true - end, - emqx_message:set_header(allow_publish, AllowPub, Message). + case maps:get(<<"allow_publish">>, Headers, undefined) of + <<"false">> -> + emqx_message:set_header(allow_publish, false, Message); + <<"true">> -> + emqx_message:set_header(allow_publish, true, Message); + _ -> + Message + end. topicfilters(Tfs) when is_list(Tfs) -> [#{name => Topic, qos => Qos} || {Topic, #{qos := Qos}} <- Tfs]. diff --git a/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl b/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl index fd8a5f9a3..6e9691c3e 100644 --- a/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl +++ b/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl @@ -306,8 +306,8 @@ on_message_publish(#{message := #{from := From} = Msg} = Req, Md) -> {ok, #{type => 'STOP_AND_RETURN', value => {message, NMsg}}, Md}; <<"gooduser">> -> - NMsg = Msg#{topic => From, - payload => From}, + NMsg = allow(Msg#{topic => From, + payload => From}), {ok, #{type => 'STOP_AND_RETURN', value => {message, NMsg}}, Md}; _ -> @@ -319,6 +319,11 @@ deny(Msg) -> maps:get(headers, Msg, #{})), maps:put(headers, NHeader, Msg). +allow(Msg) -> + NHeader = maps:put(<<"allow_publish">>, <<"true">>, + maps:get(headers, Msg, #{})), + maps:put(headers, NHeader, Msg). + -spec on_message_delivered(emqx_exhook_pb:message_delivered_request(), grpc:metadata()) -> {ok, emqx_exhook_pb:empty_success(), grpc:metadata()} | {error, grpc_cowboy_h:error_response()}. diff --git a/apps/emqx_exhook/test/props/prop_exhook_hooks.erl b/apps/emqx_exhook/test/props/prop_exhook_hooks.erl index 24f45c8b0..88eba8f11 100644 --- a/apps/emqx_exhook/test/props/prop_exhook_hooks.erl +++ b/apps/emqx_exhook/test/props/prop_exhook_hooks.erl @@ -299,19 +299,24 @@ prop_message_publish() -> _ -> ExpectedOutMsg = case emqx_message:from(Msg) of <<"baduser">> -> - MsgMap = emqx_message:to_map(Msg), + MsgMap = #{headers := Headers} + = emqx_message:to_map(Msg), emqx_message:from_map( MsgMap#{qos => 0, topic => <<"">>, - payload => <<"">> + payload => <<"">>, + headers => maps:put(allow_publish, false, Headers) }); <<"gooduser">> = From -> - MsgMap = emqx_message:to_map(Msg), + MsgMap = #{headers := Headers} + = emqx_message:to_map(Msg), emqx_message:from_map( MsgMap#{topic => From, - payload => From + payload => From, + headers => maps:put(allow_publish, true, Headers) }); - _ -> Msg + _ -> + Msg end, ?assertEqual(ExpectedOutMsg, OutMsg), @@ -464,7 +469,9 @@ from_message(Msg) -> from => stringfy(emqx_message:from(Msg)), topic => emqx_message:topic(Msg), payload => emqx_message:payload(Msg), - timestamp => emqx_message:timestamp(Msg) + timestamp => emqx_message:timestamp(Msg), + headers => emqx_exhook_handler:headers( + emqx_message:get_headers(Msg)) }. %%-------------------------------------------------------------------- From 641f36514ff6292705e31e89de2c503f6a24db22 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Tue, 9 Nov 2021 11:52:47 +0800 Subject: [PATCH 08/24] chore: fix elvis warnings --- apps/emqx_exhook/src/emqx_exhook_handler.erl | 17 +- apps/emqx_exhook/src/emqx_exhook_mngr.erl | 6 +- apps/emqx_exhook/src/emqx_exhook_server.erl | 36 +- apps/emqx_sn/test/emqx_sn_protocol_SUITE.erl | 375 ++++++++++++++----- src/emqx_types.erl | 2 +- 5 files changed, 307 insertions(+), 129 deletions(-) diff --git a/apps/emqx_exhook/src/emqx_exhook_handler.erl b/apps/emqx_exhook/src/emqx_exhook_handler.erl index ea6cdf5bc..63d41d8eb 100644 --- a/apps/emqx_exhook/src/emqx_exhook_handler.erl +++ b/apps/emqx_exhook/src/emqx_exhook_handler.erl @@ -63,6 +63,8 @@ , call_fold/3 ]). +-elvis([{elvis_style, god_modules, disable}]). + %%-------------------------------------------------------------------- %% Clients %%-------------------------------------------------------------------- @@ -341,11 +343,7 @@ merge_responsed_bool(_Req, #{type := 'IGNORE'}) -> ignore; merge_responsed_bool(Req, #{type := Type, value := {bool_result, NewBool}}) when is_boolean(NewBool) -> - NReq = Req#{result => NewBool}, - case Type of - 'CONTINUE' -> {ok, NReq}; - 'STOP_AND_RETURN' -> {stop, NReq} - end; + {ret(Type), Req#{result => NewBool}}; merge_responsed_bool(_Req, Resp) -> ?LOG(warning, "Unknown responsed value ~0p to merge to callback chain", [Resp]), ignore. @@ -353,11 +351,10 @@ merge_responsed_bool(_Req, Resp) -> merge_responsed_message(_Req, #{type := 'IGNORE'}) -> ignore; merge_responsed_message(Req, #{type := Type, value := {message, NMessage}}) -> - NReq = Req#{message => NMessage}, - case Type of - 'CONTINUE' -> {ok, NReq}; - 'STOP_AND_RETURN' -> {stop, NReq} - end; + {ret(Type), Req#{message => NMessage}}; merge_responsed_message(_Req, Resp) -> ?LOG(warning, "Unknown responsed value ~0p to merge to callback chain", [Resp]), ignore. + +ret('CONTINUE') -> ok; +ret('STOP_AND_RETURN') -> stop. diff --git a/apps/emqx_exhook/src/emqx_exhook_mngr.erl b/apps/emqx_exhook/src/emqx_exhook_mngr.erl index 54e106f13..0f3c8b8ab 100644 --- a/apps/emqx_exhook/src/emqx_exhook_mngr.erl +++ b/apps/emqx_exhook/src/emqx_exhook_mngr.erl @@ -86,11 +86,11 @@ start_link(Servers, AutoReconnect, ReqOpts) -> gen_server:start_link(?MODULE, [Servers, AutoReconnect, ReqOpts], []). --spec enable(pid(), atom()|string()) -> ok | {error, term()}. +-spec enable(pid(), atom() | string()) -> ok | {error, term()}. enable(Pid, Name) -> call(Pid, {load, Name}). --spec disable(pid(), atom()|string()) -> ok | {error, term()}. +-spec disable(pid(), atom() | string()) -> ok | {error, term()}. disable(Pid, Name) -> call(Pid, {unload, Name}). @@ -141,7 +141,7 @@ load_all_servers(Servers, ReqOpts) -> load_all_servers(Servers, ReqOpts, #{}, #{}). load_all_servers([], _Request, Waiting, Running) -> {Waiting, Running}; -load_all_servers([{Name, Options}|More], ReqOpts, Waiting, Running) -> +load_all_servers([{Name, Options} | More], ReqOpts, Waiting, Running) -> {NWaiting, NRunning} = case emqx_exhook_server:load(Name, Options, ReqOpts) of {ok, ServerState} -> diff --git a/apps/emqx_exhook/src/emqx_exhook_server.erl b/apps/emqx_exhook/src/emqx_exhook_server.erl index 123cbb558..9e88d774d 100644 --- a/apps/emqx_exhook/src/emqx_exhook_server.erl +++ b/apps/emqx_exhook/src/emqx_exhook_server.erl @@ -77,6 +77,8 @@ -dialyzer({nowarn_function, [inc_metrics/2]}). +-elvis([{elvis_style, dont_repeat_yourself, disable}]). + %%-------------------------------------------------------------------- %% Load/Unload APIs %%-------------------------------------------------------------------- @@ -125,7 +127,11 @@ channel_opts(Opts) -> SvrAddr = format_http_uri(Scheme, Host, Port), ClientOpts = case Scheme of https -> - SslOpts = lists:keydelete(ssl, 1, proplists:get_value(ssl_options, Opts, [])), + SslOpts = lists:keydelete( + ssl, + 1, + proplists:get_value(ssl_options, Opts, []) + ), #{gun_opts => #{transport => ssl, transport_opts => SslOpts}}; @@ -175,16 +181,18 @@ resovle_hookspec(HookSpecs) when is_list(HookSpecs) -> case maps:get(name, HookSpec, undefined) of undefined -> Acc; Name0 -> - Name = try binary_to_existing_atom(Name0, utf8) catch T:R:_ -> {T,R} end, - case lists:member(Name, AvailableHooks) of - true -> - case lists:member(Name, MessageHooks) of - true -> - Acc#{Name => #{topics => maps:get(topics, HookSpec, [])}}; - _ -> - Acc#{Name => #{}} - end; - _ -> error({unknown_hookpoint, Name}) + Name = try + binary_to_existing_atom(Name0, utf8) + catch T:R:_ -> {T,R} + end, + case {lists:member(Name, AvailableHooks), + lists:member(Name, MessageHooks)} of + {false, _} -> + error({unknown_hookpoint, Name}); + {true, false} -> + Acc#{Name => #{}}; + {true, true} -> + Acc#{Name => #{topics => maps:get(topics, HookSpec, [])}} end end end, #{}, HookSpecs). @@ -256,7 +264,7 @@ call(Hookpoint, Req, #server{name = ChannName, options = ReqOpts, %% @private inc_metrics(IncFun, Name) when is_function(IncFun) -> %% BACKW: e4.2.0-e4.2.2 - {env, [Prefix|_]} = erlang:fun_info(IncFun, env), + {env, [Prefix | _]} = erlang:fun_info(IncFun, env), inc_metrics(Prefix, Name); inc_metrics(Prefix, Name) when is_list(Prefix) -> emqx_metrics:inc(list_to_atom(Prefix ++ atom_to_list(Name))). @@ -272,8 +280,8 @@ do_call(ChannName, Fun, Req, ReqOpts) -> Options = ReqOpts#{channel => ChannName}, ?LOG(debug, "Call ~0p:~0p(~0p, ~0p)", [?PB_CLIENT_MOD, Fun, Req, Options]), case catch apply(?PB_CLIENT_MOD, Fun, [Req, Options]) of - {ok, Resp, _Metadata} -> - ?LOG(debug, "Response {ok, ~0p, ~0p}", [Resp, _Metadata]), + {ok, Resp, Metadata} -> + ?LOG(debug, "Response {ok, ~0p, ~0p}", [Resp, Metadata]), {ok, Resp}; {error, {Code, Msg}, _Metadata} -> ?LOG(error, "CALL ~0p:~0p(~0p, ~0p) response errcode: ~0p, errmsg: ~0p", diff --git a/apps/emqx_sn/test/emqx_sn_protocol_SUITE.erl b/apps/emqx_sn/test/emqx_sn_protocol_SUITE.erl index 794eedd6b..27215cd4f 100644 --- a/apps/emqx_sn/test/emqx_sn_protocol_SUITE.erl +++ b/apps/emqx_sn/test/emqx_sn_protocol_SUITE.erl @@ -50,6 +50,9 @@ %% erlang:system_time should be unique and random enough -define(CLIENTID, iolist_to_binary([atom_to_list(?FUNCTION_NAME), "-", integer_to_list(erlang:system_time())])). + +-elvis([{elvis_style, dont_repeat_yourself, disable}]). + %%-------------------------------------------------------------------- %% Setups %%-------------------------------------------------------------------- @@ -66,8 +69,10 @@ end_per_suite(_) -> emqx_ct_helpers:stop_apps([emqx_sn]). set_special_confs(emqx) -> - application:set_env(emqx, plugins_loaded_file, - emqx_ct_helpers:deps_path(emqx, "test/emqx_SUITE_data/loaded_plugins")); + application:set_env( + emqx, + plugins_loaded_file, + emqx_ct_helpers:deps_path(emqx, "test/emqx_SUITE_data/loaded_plugins")); set_special_confs(emqx_sn) -> application:set_env(emqx_sn, enable_qos3, ?ENABLE_QOS3), application:set_env(emqx_sn, enable_stats, true), @@ -113,7 +118,8 @@ t_subscribe(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), TopicName1 = <<"abcD">>, send_register_msg(Socket, TopicName1, MsgId), - ?assertEqual(<<7, ?SN_REGACK, TopicId:16, MsgId:16, 0:8>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_REGACK, TopicId:16, MsgId:16, 0:8>>, + receive_response(Socket)), send_subscribe_msg_normal_topic(Socket, QoS, TopicName1, MsgId), ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId:16, @@ -145,7 +151,8 @@ t_subscribe_case01(_) -> TopicName1 = <<"abcD">>, send_register_msg(Socket, TopicName1, MsgId), - ?assertEqual(<<7, ?SN_REGACK, TopicId:16, MsgId:16, 0:8>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_REGACK, TopicId:16, MsgId:16, 0:8>>, + receive_response(Socket)), send_subscribe_msg_normal_topic(Socket, QoS, TopicName1, MsgId), ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId:16, MsgId:16, ReturnCode>>, @@ -166,7 +173,7 @@ t_subscribe_case02(_) -> Will = 0, CleanSession = 0, MsgId = 1, - TopicId = ?PREDEF_TOPIC_ID1, %this TopicId is the predefined topic id corresponding to ?PREDEF_TOPIC_NAME1 + TopicId = ?PREDEF_TOPIC_ID1, ReturnCode = 0, {ok, Socket} = gen_udp:open(0, [binary]), @@ -176,7 +183,8 @@ t_subscribe_case02(_) -> Topic1 = ?PREDEF_TOPIC_NAME1, send_register_msg(Socket, Topic1, MsgId), - ?assertEqual(<<7, ?SN_REGACK, TopicId:16, MsgId:16, 0:8>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_REGACK, TopicId:16, MsgId:16, 0:8>>, + receive_response(Socket)), send_subscribe_msg_predefined_topic(Socket, QoS, TopicId, MsgId), ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId:16, MsgId:16, ReturnCode>>, @@ -206,9 +214,11 @@ t_subscribe_case03(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_short_topic(Socket, QoS, <<"te">>, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, - ?SN_NORMAL_TOPIC:2, TopicId:16, MsgId:16, ReturnCode>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, + CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId:16, MsgId:16, ReturnCode>>, + receive_response(Socket) + ), send_unsubscribe_msg_short_topic(Socket, <<"te">>, MsgId), ?assertEqual(<<4, ?SN_UNSUBACK, MsgId:16>>, receive_response(Socket)), @@ -217,8 +227,12 @@ t_subscribe_case03(_) -> ?assertEqual(<<2, ?SN_DISCONNECT>>, receive_response(Socket)), gen_udp:close(Socket). -%%In this case We use predefined topic name to register and subcribe, and expect to receive the corresponding predefined topic id but not a new generated topic id from broker. We design this case to illustrate -%% emqx_sn_gateway's compatibility of dealing with predefined and normal topics. Once we give more restrictions to different topic id type, this case would be deleted or modified. +%% In this case We use predefined topic name to register and subcribe, and +%% expect to receive the corresponding predefined topic id but not a new +%% generated topic id from broker. We design this case to illustrate +%% emqx_sn_gateway's compatibility of dealing with predefined and normal topics. +%% Once we give more restrictions to different topic id type, this case would +%% be deleted or modified. t_subscribe_case04(_) -> Dup = 0, QoS = 0, @@ -226,7 +240,7 @@ t_subscribe_case04(_) -> Will = 0, CleanSession = 0, MsgId = 1, - TopicId = ?PREDEF_TOPIC_ID1, %this TopicId is the predefined topic id corresponding to ?PREDEF_TOPIC_NAME1 + TopicId = ?PREDEF_TOPIC_ID1, ReturnCode = 0, {ok, Socket} = gen_udp:open(0, [binary]), ClientId = ?CLIENTID, @@ -234,10 +248,14 @@ t_subscribe_case04(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), Topic1 = ?PREDEF_TOPIC_NAME1, send_register_msg(Socket, Topic1, MsgId), - ?assertEqual(<<7, ?SN_REGACK, TopicId:16, MsgId:16, 0:8>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_REGACK, TopicId:16, MsgId:16, 0:8>>, + receive_response(Socket)), send_subscribe_msg_normal_topic(Socket, QoS, Topic1, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId:16, MsgId:16, ReturnCode>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId:16, MsgId:16, ReturnCode>>, + receive_response(Socket) + ), send_unsubscribe_msg_normal_topic(Socket, Topic1, MsgId), ?assertEqual(<<4, ?SN_UNSUBACK, MsgId:16>>, receive_response(Socket)), @@ -264,19 +282,30 @@ t_subscribe_case05(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_register_msg(Socket, <<"abcD">>, MsgId), - ?assertEqual(<<7, ?SN_REGACK, TopicId1:16, MsgId:16, 0:8>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_REGACK, TopicId1:16, MsgId:16, 0:8>>, + receive_response(Socket) + ), send_subscribe_msg_normal_topic(Socket, QoS, <<"abcD">>, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId1:16, MsgId:16, ReturnCode>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, MsgId:16, ReturnCode>>, + receive_response(Socket) + ), send_subscribe_msg_normal_topic(Socket, QoS, <<"/sport/#">>, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId0:16, MsgId:16, ReturnCode>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId0:16, MsgId:16, ReturnCode>>, + receive_response(Socket) + ), send_subscribe_msg_normal_topic(Socket, QoS, <<"/a/+/water">>, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId0:16, MsgId:16, ReturnCode>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId0:16, MsgId:16, ReturnCode>>, + receive_response(Socket) + ), send_subscribe_msg_normal_topic(Socket, QoS, <<"/Tom/Home">>, MsgId), ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, @@ -306,19 +335,32 @@ t_subscribe_case06(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_register_msg(Socket, <<"abc">>, MsgId), - ?assertEqual(<<7, ?SN_REGACK, TopicId1:16, MsgId:16, 0:8>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_REGACK, TopicId1:16, MsgId:16, 0:8>>, + receive_response(Socket) + ), send_register_msg(Socket, <<"/blue/#">>, MsgId), - ?assertEqual(<<7, ?SN_REGACK, TopicId0:16, MsgId:16, ?SN_RC_NOT_SUPPORTED:8>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_REGACK, TopicId0:16, + MsgId:16, ?SN_RC_NOT_SUPPORTED:8>>, + receive_response(Socket) + ), send_register_msg(Socket, <<"/blue/+/white">>, MsgId), - ?assertEqual(<<7, ?SN_REGACK, TopicId0:16, MsgId:16, ?SN_RC_NOT_SUPPORTED:8>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_REGACK, TopicId0:16, + MsgId:16, ?SN_RC_NOT_SUPPORTED:8>>, + receive_response(Socket) + ), send_register_msg(Socket, <<"/$sys/rain">>, MsgId), - ?assertEqual(<<7, ?SN_REGACK, TopicId2:16, MsgId:16, 0:8>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_REGACK, TopicId2:16, MsgId:16, 0:8>>, + receive_response(Socket) + ), send_subscribe_msg_short_topic(Socket, QoS, <<"Q2">>, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId0:16, MsgId:16, ReturnCode>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId0:16, MsgId:16, ReturnCode>>, + receive_response(Socket) + ), send_unsubscribe_msg_normal_topic(Socket, <<"Q2">>, MsgId), ?assertEqual(<<4, ?SN_UNSUBACK, MsgId:16>>, receive_response(Socket)), @@ -342,8 +384,11 @@ t_subscribe_case07(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_predefined_topic(Socket, QoS, TopicId1, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId1:16, MsgId:16, ?SN_RC_INVALID_TOPIC_ID>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, MsgId:16, ?SN_RC_INVALID_TOPIC_ID>>, + receive_response(Socket) + ), send_unsubscribe_msg_predefined_topic(Socket, TopicId2, MsgId), ?assertEqual(<<4, ?SN_UNSUBACK, MsgId:16>>, receive_response(Socket)), @@ -365,8 +410,11 @@ t_subscribe_case08(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_reserved_topic(Socket, QoS, TopicId2, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, ?SN_INVALID_TOPIC_ID:16, MsgId:16, ?SN_RC_INVALID_TOPIC_ID>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + ?SN_INVALID_TOPIC_ID:16, MsgId:16, ?SN_RC_INVALID_TOPIC_ID>>, + receive_response(Socket) + ), send_disconnect_msg(Socket, undefined), ?assertEqual(<<2, ?SN_DISCONNECT>>, receive_response(Socket)), @@ -390,15 +438,20 @@ t_publish_negqos_case09(_) -> send_subscribe_msg_normal_topic(Socket, QoS, Topic, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId1:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), MsgId1 = 3, Payload1 = <<20, 21, 22, 23>>, send_publish_msg_normal_topic(Socket, NegQoS, MsgId1, TopicId1, Payload1), timer:sleep(100), case ?ENABLE_QOS3 of true -> - Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId1:16, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, + Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, What = receive_response(Socket), ?assertEqual(Eexp, What) end, @@ -431,7 +484,9 @@ t_publish_qos0_case01(_) -> send_publish_msg_normal_topic(Socket, QoS, MsgId1, TopicId1, Payload1), timer:sleep(100), - Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId1:16, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, + Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, What = receive_response(Socket), ?assertEqual(Eexp, What), @@ -453,15 +508,20 @@ t_publish_qos0_case02(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_predefined_topic(Socket, QoS, PredefTopicId, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, PredefTopicId:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + PredefTopicId:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), MsgId1 = 3, Payload1 = <<20, 21, 22, 23>>, send_publish_msg_predefined_topic(Socket, QoS, MsgId1, PredefTopicId, Payload1), timer:sleep(100), - Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_PREDEFINED_TOPIC:2, PredefTopicId:16, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, + Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_PREDEFINED_TOPIC:2, + PredefTopicId:16, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, What = receive_response(Socket), ?assertEqual(Eexp, What), @@ -484,15 +544,20 @@ t_publish_qos0_case3(_) -> Topic = <<"/a/b/c">>, send_subscribe_msg_normal_topic(Socket, QoS, Topic, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), MsgId1 = 3, Payload1 = <<20, 21, 22, 23>>, send_publish_msg_predefined_topic(Socket, QoS, MsgId1, TopicId, Payload1), timer:sleep(100), - Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId:16, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, + Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId:16, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, What = receive_response(Socket), ?assertEqual(Eexp, What), @@ -514,8 +579,11 @@ t_publish_qos0_case04(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_normal_topic(Socket, QoS, <<"#">>, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId0:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId0:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), MsgId1 = 2, Payload1 = <<20, 21, 22, 23>>, @@ -523,7 +591,9 @@ t_publish_qos0_case04(_) -> send_publish_msg_short_topic(Socket, QoS, MsgId1, Topic, Payload1), timer:sleep(100), - Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_SHORT_TOPIC:2, Topic/binary, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, + Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_SHORT_TOPIC:2, + Topic/binary, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, What = receive_response(Socket), ?assertEqual(Eexp, What), @@ -544,8 +614,11 @@ t_publish_qos0_case05(_) -> send_connect_msg(Socket, ClientId), ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_short_topic(Socket, QoS, <<"/#">>, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId0:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId0:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), send_disconnect_msg(Socket, undefined), ?assertEqual(<<2, ?SN_DISCONNECT>>, receive_response(Socket)), @@ -567,15 +640,20 @@ t_publish_qos0_case06(_) -> Topic = <<"abc">>, send_subscribe_msg_normal_topic(Socket, QoS, Topic, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId1:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), MsgId1 = 3, Payload1 = <<20, 21, 22, 23>>, send_publish_msg_normal_topic(Socket, QoS, MsgId1, TopicId1, Payload1), timer:sleep(100), - Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId1:16, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, + Eexp = <<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, (mid(0)):16, <<20, 21, 22, 23>>/binary>>, What = receive_response(Socket), ?assertEqual(Eexp, What), @@ -597,16 +675,25 @@ t_publish_qos1_case01(_) -> send_connect_msg(Socket, ClientId), ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_normal_topic(Socket, QoS, Topic, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, - ?SN_NORMAL_TOPIC:2, TopicId1:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), Payload1 = <<20, 21, 22, 23>>, send_publish_msg_normal_topic(Socket, QoS, MsgId, TopicId1, Payload1), - ?assertEqual(<<7, ?SN_PUBACK, TopicId1:16, MsgId:16, ?SN_RC_ACCEPTED>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_PUBACK, TopicId1:16, + MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), timer:sleep(100), - ?assertEqual(<<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId1:16, MsgId:16, <<20, 21, 22, 23>>/binary>>, receive_response(Socket)), + ?assertEqual(<<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, MsgId:16, <<20, 21, 22, 23>>/binary>>, + receive_response(Socket) + ), send_disconnect_msg(Socket, undefined), gen_udp:close(Socket). @@ -625,12 +712,18 @@ t_publish_qos1_case02(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_predefined_topic(Socket, QoS, PredefTopicId, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, PredefTopicId:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + PredefTopicId:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), Payload1 = <<20, 21, 22, 23>>, send_publish_msg_predefined_topic(Socket, QoS, MsgId, PredefTopicId, Payload1), - ?assertEqual(<<7, ?SN_PUBACK, PredefTopicId:16, MsgId:16, ?SN_RC_ACCEPTED>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_PUBACK, PredefTopicId:16, + MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), timer:sleep(100), send_disconnect_msg(Socket, undefined), @@ -645,7 +738,10 @@ t_publish_qos1_case03(_) -> send_connect_msg(Socket, ClientId), ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_publish_msg_predefined_topic(Socket, QoS, MsgId, tid(5), <<20, 21, 22, 23>>), - ?assertEqual(<<7, ?SN_PUBACK, TopicId5:16, MsgId:16, ?SN_RC_INVALID_TOPIC_ID>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_PUBACK, TopicId5:16, + MsgId:16, ?SN_RC_INVALID_TOPIC_ID>>, + receive_response(Socket) + ), send_disconnect_msg(Socket, undefined), ?assertEqual(<<2, ?SN_DISCONNECT>>, receive_response(Socket)), @@ -664,15 +760,20 @@ t_publish_qos1_case04(_) -> send_connect_msg(Socket, ClientId), ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_short_topic(Socket, QoS, <<"ab">>, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, - ?SN_NORMAL_TOPIC:2, TopicId0:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId0:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), Topic = <<"ab">>, Payload1 = <<20, 21, 22, 23>>, send_publish_msg_short_topic(Socket, QoS, MsgId, Topic, Payload1), <> = Topic, - ?assertEqual(<<7, ?SN_PUBACK, TopicIdShort:16, MsgId:16, ?SN_RC_ACCEPTED>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_PUBACK, TopicIdShort:16, + MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), timer:sleep(100), send_disconnect_msg(Socket, undefined), @@ -692,13 +793,18 @@ t_publish_qos1_case05(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_normal_topic(Socket, QoS, <<"ab">>, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, - ?SN_NORMAL_TOPIC:2, TopicId1:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), send_publish_msg_short_topic(Socket, QoS, MsgId, <<"/#">>, <<20, 21, 22, 23>>), <> = <<"/#">>, - ?assertEqual(<<7, ?SN_PUBACK, TopicIdShort:16, MsgId:16, ?SN_RC_NOT_SUPPORTED>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_PUBACK, TopicIdShort:16, + MsgId:16, ?SN_RC_NOT_SUPPORTED>>, + receive_response(Socket) + ), send_disconnect_msg(Socket, undefined), ?assertEqual(<<2, ?SN_DISCONNECT>>, receive_response(Socket)), @@ -724,7 +830,10 @@ t_publish_qos1_case06(_) -> send_publish_msg_short_topic(Socket, QoS, MsgId, <<"/+">>, <<20, 21, 22, 23>>), <> = <<"/+">>, - ?assertEqual(<<7, ?SN_PUBACK, TopicIdShort:16, MsgId:16, ?SN_RC_NOT_SUPPORTED>>, receive_response(Socket)), + ?assertEqual(<<7, ?SN_PUBACK, TopicIdShort:16, + MsgId:16, ?SN_RC_NOT_SUPPORTED>>, + receive_response(Socket) + ), send_disconnect_msg(Socket, undefined), ?assertEqual(<<2, ?SN_DISCONNECT>>, receive_response(Socket)), @@ -751,7 +860,11 @@ t_publish_qos2_case01(_) -> send_publish_msg_normal_topic(Socket, QoS, MsgId, TopicId1, Payload1), ?assertEqual(<<4, ?SN_PUBREC, MsgId:16>>, receive_response(Socket)), send_pubrel_msg(Socket, MsgId), - ?assertEqual(<<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId1:16, 1:16, <<20, 21, 22, 23>>/binary>>, receive_response(Socket)), + ?assertEqual(<<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, 1:16, <<20, 21, 22, 23>>/binary>>, + receive_response(Socket) + ), ?assertEqual(<<4, ?SN_PUBCOMP, MsgId:16>>, receive_response(Socket)), timer:sleep(100), @@ -773,15 +886,21 @@ t_publish_qos2_case02(_) -> ?assertEqual(<<3, ?SN_CONNACK, 0>>, receive_response(Socket)), send_subscribe_msg_predefined_topic(Socket, QoS, PredefTopicId, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, ?FNU:1, QoS:2, ?FNU:5, PredefTopicId:16, MsgId:16, ?SN_RC_ACCEPTED>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, ?FNU:1, QoS:2, ?FNU:5, + PredefTopicId:16, MsgId:16, ?SN_RC_ACCEPTED>>, + receive_response(Socket) + ), Payload1 = <<20, 21, 22, 23>>, send_publish_msg_predefined_topic(Socket, QoS, MsgId, PredefTopicId, Payload1), ?assertEqual(<<4, ?SN_PUBREC, MsgId:16>>, receive_response(Socket)), send_pubrel_msg(Socket, MsgId), - ?assertEqual(<<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_PREDEFINED_TOPIC :2, PredefTopicId:16, 1:16, <<20, 21, 22, 23>>/binary>>, receive_response(Socket)), + ?assertEqual(<<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_PREDEFINED_TOPIC:2, + PredefTopicId:16, 1:16, <<20, 21, 22, 23>>/binary>>, + receive_response(Socket) + ), ?assertEqual(<<4, ?SN_PUBCOMP, MsgId:16>>, receive_response(Socket)), timer:sleep(100), @@ -812,7 +931,11 @@ t_publish_qos2_case03(_) -> ?assertEqual(<<4, ?SN_PUBREC, MsgId:16>>, receive_response(Socket)), send_pubrel_msg(Socket, MsgId), - ?assertEqual(<<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, Will:1, CleanSession:1, ?SN_SHORT_TOPIC :2, <<"/a">>/binary, 1:16, <<20, 21, 22, 23>>/binary>>, receive_response(Socket)), + ?assertEqual(<<11, ?SN_PUBLISH, Dup:1, QoS:2, Retain:1, + Will:1, CleanSession:1, ?SN_SHORT_TOPIC:2, + "/a", 1:16, <<20, 21, 22, 23>>/binary>>, + receive_response(Socket) + ), ?assertEqual(<<4, ?SN_PUBCOMP, MsgId:16>>, receive_response(Socket)), timer:sleep(100), @@ -1083,7 +1206,11 @@ t_asleep_test03_to_awake_qos1_dl_msg(_) -> send_register_msg(Socket, TopicName1, MsgId1), ?assertEqual(<<7, ?SN_REGACK, TopicId1:16, MsgId1:16, 0:8>>, receive_response(Socket)), send_subscribe_msg_predefined_topic(Socket, QoS, TopicId1, MsgId), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, WillBit:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId1:16, MsgId:16, ReturnCode>>, receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + WillBit:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId1:16, MsgId:16, ReturnCode>>, + receive_response(Socket) + ), % goto asleep state send_disconnect_msg(Socket, 1), @@ -1109,7 +1236,10 @@ t_asleep_test03_to_awake_qos1_dl_msg(_) -> %% the broker should sent dl msgs to the awake client before sending the pingresp UdpData = receive_response(Socket), - MsgId_udp = check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, CleanSession, ?SN_NORMAL_TOPIC, TopicId1, Payload1}, UdpData), + MsgId_udp = check_publish_msg_on_udp( + {Dup, QoS, Retain, WillBit, + CleanSession, ?SN_NORMAL_TOPIC, + TopicId1, Payload1}, UdpData), send_puback_msg(Socket, TopicId1, MsgId_udp), %% check the pingresp is received at last @@ -1141,8 +1271,11 @@ t_asleep_test04_to_awake_qos1_dl_msg(_) -> CleanSession = 0, ReturnCode = 0, send_subscribe_msg_normal_topic(Socket, QoS, TopicName1, MsgId1), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, WillBit:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId0:16, MsgId1:16, ReturnCode>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + WillBit:1,CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId0:16, MsgId1:16, ReturnCode>>, + receive_response(Socket) + ), % goto asleep state send_disconnect_msg(Socket, 1), @@ -1176,11 +1309,17 @@ t_asleep_test04_to_awake_qos1_dl_msg(_) -> send_regack_msg(Socket, TopicIdNew, MsgId3), UdpData2 = receive_response(Socket), - MsgId_udp2 = check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, CleanSession, ?SN_NORMAL_TOPIC, TopicIdNew, Payload1}, UdpData2), + MsgId_udp2 = check_publish_msg_on_udp( + {Dup, QoS, Retain, WillBit, + CleanSession, ?SN_NORMAL_TOPIC, + TopicIdNew, Payload1}, UdpData2), send_puback_msg(Socket, TopicIdNew, MsgId_udp2), UdpData3 = receive_response(Socket), - MsgId_udp3 = check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, CleanSession, ?SN_NORMAL_TOPIC, TopicIdNew, Payload2}, UdpData3), + MsgId_udp3 = check_publish_msg_on_udp( + {Dup, QoS, Retain, WillBit, + CleanSession, ?SN_NORMAL_TOPIC, + TopicIdNew, Payload2}, UdpData3), send_puback_msg(Socket, TopicIdNew, MsgId_udp3), ?assertEqual(<<2, ?SN_PINGRESP>>, receive_response(Socket)), @@ -1216,8 +1355,11 @@ t_asleep_test05_to_awake_qos1_dl_msg(_) -> CleanSession = 0, ReturnCode = 0, send_subscribe_msg_normal_topic(Socket, QoS, TopicName1, MsgId1), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, WillBit:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId0:16, MsgId1:16, ReturnCode>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + WillBit:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId0:16, MsgId1:16, ReturnCode>>, + receive_response(Socket) + ), % goto asleep state SleepDuration = 30, @@ -1250,21 +1392,28 @@ t_asleep_test05_to_awake_qos1_dl_msg(_) -> send_regack_msg(Socket, TopicIdNew, MsgId_reg), UdpData2 = receive_response(Socket), - MsgId2 = check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, CleanSession, ?SN_NORMAL_TOPIC, TopicIdNew, Payload2}, UdpData2), + MsgId2 = check_publish_msg_on_udp( + {Dup, QoS, Retain, WillBit, + CleanSession, ?SN_NORMAL_TOPIC, + TopicIdNew, Payload2}, UdpData2), send_puback_msg(Socket, TopicIdNew, MsgId2), timer:sleep(50), UdpData3 = wrap_receive_response(Socket), - MsgId3 = check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, CleanSession, ?SN_NORMAL_TOPIC, TopicIdNew, Payload3}, UdpData3), + MsgId3 = check_publish_msg_on_udp( + {Dup, QoS, Retain, WillBit, + CleanSession, ?SN_NORMAL_TOPIC, + TopicIdNew, Payload3}, UdpData3), send_puback_msg(Socket, TopicIdNew, MsgId3), timer:sleep(50), case receive_response(Socket) of <<2,23>> -> ok; UdpData4 -> - MsgId4 = check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, - CleanSession, ?SN_NORMAL_TOPIC, - TopicIdNew, Payload4}, UdpData4), + MsgId4 = check_publish_msg_on_udp( + {Dup, QoS, Retain, WillBit, + CleanSession, ?SN_NORMAL_TOPIC, + TopicIdNew, Payload4}, UdpData4), send_puback_msg(Socket, TopicIdNew, MsgId4) end, ?assertEqual(<<2, ?SN_PINGRESP>>, receive_response(Socket)), @@ -1322,7 +1471,10 @@ t_asleep_test06_to_awake_qos2_dl_msg(_) -> send_pingreq_msg(Socket, ClientId), UdpData = wrap_receive_response(Socket), - MsgId_udp = check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, CleanSession, ?SN_NORMAL_TOPIC, TopicId_tom, Payload1}, UdpData), + MsgId_udp = check_publish_msg_on_udp( + {Dup, QoS, Retain, WillBit, + CleanSession, ?SN_NORMAL_TOPIC, + TopicId_tom, Payload1}, UdpData), send_pubrec_msg(Socket, MsgId_udp), ?assertMatch(<<_:8, ?SN_PUBREL:8, _/binary>>, receive_response(Socket)), send_pubcomp_msg(Socket, MsgId_udp), @@ -1357,8 +1509,11 @@ t_asleep_test07_to_connected(_) -> send_register_msg(Socket, TopicName_tom, MsgId1), TopicId_tom = check_regack_msg_on_udp(MsgId1, receive_response(Socket)), send_subscribe_msg_predefined_topic(Socket, QoS, TopicId_tom, MsgId1), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, WillBit:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId_tom:16, MsgId1:16, ReturnCode>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + WillBit:1,CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId_tom:16, MsgId1:16, ReturnCode>>, + receive_response(Socket) + ), % goto asleep state send_disconnect_msg(Socket, SleepDuration), @@ -1436,8 +1591,11 @@ t_asleep_test09_to_awake_again_qos1_dl_msg(_) -> CleanSession = 0, ReturnCode = 0, send_subscribe_msg_normal_topic(Socket, QoS, TopicName1, MsgId1), - ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, WillBit:1, CleanSession:1, ?SN_NORMAL_TOPIC:2, TopicId0:16, MsgId1:16, ReturnCode>>, - receive_response(Socket)), + ?assertEqual(<<8, ?SN_SUBACK, Dup:1, QoS:2, Retain:1, + WillBit:1,CleanSession:1, ?SN_NORMAL_TOPIC:2, + TopicId0:16, MsgId1:16, ReturnCode>>, + receive_response(Socket) + ), % goto asleep state SleepDuration = 30, send_disconnect_msg(Socket, SleepDuration), @@ -1471,7 +1629,10 @@ t_asleep_test09_to_awake_again_qos1_dl_msg(_) -> udp_receive_timeout -> ok; UdpData2 -> - MsgId2 = check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, CleanSession, ?SN_NORMAL_TOPIC, TopicIdNew, Payload2}, UdpData2), + MsgId2 = check_publish_msg_on_udp( + {Dup, QoS, Retain, WillBit, + CleanSession, ?SN_NORMAL_TOPIC, + TopicIdNew, Payload2}, UdpData2), send_puback_msg(Socket, TopicIdNew, MsgId2) end, timer:sleep(100), @@ -1480,7 +1641,10 @@ t_asleep_test09_to_awake_again_qos1_dl_msg(_) -> udp_receive_timeout -> ok; UdpData3 -> - MsgId3 = check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, CleanSession, ?SN_NORMAL_TOPIC, TopicIdNew, Payload3}, UdpData3), + MsgId3 = check_publish_msg_on_udp( + {Dup, QoS, Retain, WillBit, + CleanSession, ?SN_NORMAL_TOPIC, + TopicIdNew, Payload3}, UdpData3), send_puback_msg(Socket, TopicIdNew, MsgId3) end, timer:sleep(100), @@ -1489,16 +1653,18 @@ t_asleep_test09_to_awake_again_qos1_dl_msg(_) -> udp_receive_timeout -> ok; UdpData4 -> - MsgId4 = check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, - CleanSession, ?SN_NORMAL_TOPIC, - TopicIdNew, Payload4}, UdpData4), + MsgId4 = check_publish_msg_on_udp( + {Dup, QoS, Retain, WillBit, + CleanSession, ?SN_NORMAL_TOPIC, + TopicIdNew, Payload4}, UdpData4), send_puback_msg(Socket, TopicIdNew, MsgId4) end, ?assertEqual(<<2, ?SN_PINGRESP>>, receive_response(Socket)), %% send PINGREQ again to enter awake state send_pingreq_msg(Socket, ClientId), - %% will not receive any buffered PUBLISH messages buffered before last awake, only receive PINGRESP here + %% will not receive any buffered PUBLISH messages buffered before last + %% awake, only receive PINGRESP here ?assertEqual(<<2, ?SN_PINGRESP>>, receive_response(Socket)), gen_udp:close(Socket). @@ -1901,8 +2067,12 @@ check_dispatched_message(Dup, QoS, Retain, TopicIdType, TopicId, Payload, Socket PubMsg = receive_response(Socket), Length = 7 + byte_size(Payload), ?LOG("check_dispatched_message ~p~n", [PubMsg]), - ?LOG("expected ~p xx ~p~n", [<>, Payload]), - <> = PubMsg, + ?LOG("expected ~p xx ~p~n", + [<>, Payload]), + <> = PubMsg, case QoS of 0 -> ok; 1 -> send_puback_msg(Socket, TopicId, MsgId); @@ -1914,11 +2084,14 @@ check_dispatched_message(Dup, QoS, Retain, TopicIdType, TopicId, Payload, Socket get_udp_broadcast_address() -> "255.255.255.255". -check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, CleanSession, TopicType, TopicId, Payload}, UdpData) -> +check_publish_msg_on_udp({Dup, QoS, Retain, WillBit, + CleanSession, TopicType, TopicId, Payload}, UdpData) -> <> = UdpData, ct:pal("UdpData: ~p, Payload: ~p, PayloadIn: ~p", [UdpData, Payload, PayloadIn]), Size9 = byte_size(Payload) + 7, - Eexp = <>, + Eexp = <>, ?assertEqual(Eexp, HeaderUdp), % mqtt-sn header should be same ?assertEqual(Payload, PayloadIn), % payload should be same MsgId. diff --git a/src/emqx_types.erl b/src/emqx_types.erl index 410ed5a27..3d5fb66a5 100644 --- a/src/emqx_types.erl +++ b/src/emqx_types.erl @@ -147,7 +147,7 @@ dn => binary(), atom() => term() }). --type(clientid() :: binary()|atom()). +-type(clientid() :: binary() | atom()). -type(username() :: maybe(binary())). -type(password() :: maybe(binary())). -type(peerhost() :: inet:ip_address()). From f2d99017a04bfb54d07fb30f646ab3cfcf064e59 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Tue, 9 Nov 2021 15:25:58 +0800 Subject: [PATCH 09/24] chore(exhook): update appup.src --- apps/emqx_exhook/src/emqx_exhook.appup.src | 10 ++++++++-- apps/emqx_exhook/test/emqx_exhook_demo_svr.erl | 2 +- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/apps/emqx_exhook/src/emqx_exhook.appup.src b/apps/emqx_exhook/src/emqx_exhook.appup.src index e78afef1e..ea8aacb7a 100644 --- a/apps/emqx_exhook/src/emqx_exhook.appup.src +++ b/apps/emqx_exhook/src/emqx_exhook.appup.src @@ -2,14 +2,20 @@ {VSN, [{"4.3.4", [{load_module,emqx_exhook_pb,brutal_purge,soft_purge,[]}, - {load_module,emqx_exhook_handler,brutal_purge,soft_purge,[emqx_exhook_pb]}]}, + {load_module,emqx_exhook_handler,brutal_purge,soft_purge,[emqx_exhook_pb]}, + {load_module,emqx_exhook_mngr,brutal_purge,soft_purge,[]}, + {load_module,emqx_exhook_server,brutal_purge,soft_purge,[]}, + {load_module,emqx_exhook_sup,brutal_purge,soft_purge,[]}]}, {<<"4\\.3\\.[0-3]+">>, [{restart_application,emqx_exhook}]}, {<<".*">>, []} ], [{"4.3.4", [{load_module,emqx_exhook_pb,brutal_purge,soft_purge,[]}, - {load_module,emqx_exhook_handler,brutal_purge,soft_purge,[emqx_exhook_pb]}]}, + {load_module,emqx_exhook_handler,brutal_purge,soft_purge,[emqx_exhook_pb]}, + {load_module,emqx_exhook_mngr,brutal_purge,soft_purge,[]}, + {load_module,emqx_exhook_server,brutal_purge,soft_purge,[]}, + {load_module,emqx_exhook_sup,brutal_purge,soft_purge,[]}]}, {<<"4\\.3\\.[0-3]+">>, [{restart_application,emqx_exhook}]}, {<<".*">>, []} diff --git a/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl b/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl index 6e9691c3e..1d3896e14 100644 --- a/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl +++ b/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl @@ -295,7 +295,7 @@ on_session_terminated(Req, Md) -> | {error, grpc_cowboy_h:error_response()}. on_message_publish(#{message := #{from := From} = Msg} = Req, Md) -> ?MODULE:in({?FUNCTION_NAME, Req}), - io:format(standard_error, "fun: ~p, req: ~0p~n", [?FUNCTION_NAME, Req]), + %io:format(standard_error, "fun: ~p, req: ~0p~n", [?FUNCTION_NAME, Req]), %% some cases for testing case From of <<"baduser">> -> From 7d06e48b4bfdf28964a2f03f469819316bd78535 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Tue, 9 Nov 2021 17:46:59 +0800 Subject: [PATCH 10/24] chore: remove needless catch Co-authored-by: Zaiming (Stone) Shi --- apps/emqx_exhook/src/emqx_exhook_server.erl | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/emqx_exhook/src/emqx_exhook_server.erl b/apps/emqx_exhook/src/emqx_exhook_server.erl index 9e88d774d..276f5a638 100644 --- a/apps/emqx_exhook/src/emqx_exhook_server.erl +++ b/apps/emqx_exhook/src/emqx_exhook_server.erl @@ -183,7 +183,7 @@ resovle_hookspec(HookSpecs) when is_list(HookSpecs) -> Name0 -> Name = try binary_to_existing_atom(Name0, utf8) - catch T:R:_ -> {T,R} + catch T:R -> {T,R} end, case {lists:member(Name, AvailableHooks), lists:member(Name, MessageHooks)} of From 8dfc8ed96b133776235cccdef13b534f502f6bd1 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Wed, 10 Nov 2021 16:24:42 +0800 Subject: [PATCH 11/24] chore: fill message headers --- apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl | 18 +++++++--- .../emqx_exproto/src/emqx_exproto_channel.erl | 29 ++++++++++++---- apps/emqx_lwm2m/src/emqx_lwm2m_protocol.erl | 18 ++++++++-- apps/emqx_stomp/src/emqx_stomp_protocol.erl | 34 +++++++++++++------ 4 files changed, 76 insertions(+), 23 deletions(-) diff --git a/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl b/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl index d465f9ca3..89fb411b2 100644 --- a/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl +++ b/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl @@ -244,15 +244,26 @@ chann_publish(Topic, Payload, State = #state{clientid = ClientId}) -> case emqx_access_control:check_acl(clientinfo(State), publish, Topic) of allow -> _ = emqx_broker:publish( - emqx_message:set_flag(retain, false, - emqx_message:make(ClientId, ?QOS_0, Topic, Payload))), - ok; + packet_to_message(Topic, Payload, State)), ok; deny -> ?LOG(warning, "publish to ~p by clientid ~p failed due to acl check.", [Topic, ClientId]), {error, forbidden} end. +packet_to_message(Topic, Payload, + #state{clientid = ClientId, + username = Username, + peername = {PeerHost, _}}) -> + Message = emqx_message:set_flag( + retain, false, + emqx_message:make(ClientId, ?QOS_0, Topic, Payload) + ), + emqx_message:set_headers( + #{ proto_ver => 1 + , protocol => coap + , username => Username + , peerhost => PeerHost}, Message). %%-------------------------------------------------------------------- %% Deliver @@ -384,4 +395,3 @@ clientinfo(#state{peername = {PeerHost, _}, mountpoint => undefined, ws_cookie => undefined }. - diff --git a/apps/emqx_exproto/src/emqx_exproto_channel.erl b/apps/emqx_exproto/src/emqx_exproto_channel.erl index 67f85f932..d8ceae4bd 100644 --- a/apps/emqx_exproto/src/emqx_exproto_channel.erl +++ b/apps/emqx_exproto/src/emqx_exproto_channel.erl @@ -340,17 +340,14 @@ handle_call({unsubscribe, TopicFilter}, handle_call({publish, Topic, Qos, Payload}, Channel = #channel{ conn_state = connected, - clientinfo = ClientInfo - = #{clientid := From, - mountpoint := Mountpoint}}) -> + clientinfo = ClientInfo}) -> case is_acl_enabled(ClientInfo) andalso emqx_access_control:check_acl(ClientInfo, publish, Topic) of deny -> {reply, {error, ?RESP_PERMISSION_DENY, <<"ACL deny">>}, Channel}; _ -> - Msg = emqx_message:make(From, Qos, Topic, Payload), - NMsg = emqx_mountpoint:mount(Mountpoint, Msg), - _ = emqx:publish(NMsg), + Msg = packet_to_message(Topic, Qos, Payload, Channel), + _ = emqx:publish(Msg), {reply, ok, Channel} end; @@ -419,6 +416,24 @@ is_anonymous(_AuthResult) -> false. clean_anonymous_clients() -> ets:delete(?CHAN_CONN_TAB, ?CHANMOCK(self())). +packet_to_message(Topic, Qos, Payload, + #channel{ + conninfo = #{proto_ver := ProtoVer}, + clientinfo = #{ + protocol := Protocol, + clientid := ClientId, + username := Username, + peerhost := PeerHost, + mountpoint := Mountpoint}}) -> + Msg = emqx_message:make( + ClientId, Qos, + Topic, Payload, #{}, + #{proto_ver => ProtoVer, + protocol => Protocol, + username => Username, + peerhost => PeerHost}), + emqx_mountpoint:mount(Mountpoint, Msg). + %%-------------------------------------------------------------------- %% Sub/UnSub %%-------------------------------------------------------------------- @@ -591,6 +606,8 @@ default_conninfo(ConnInfo) -> ConnInfo#{clean_start => true, clientid => undefined, username => undefined, + proto_name => undefined, + proto_ver => undefined, conn_props => #{}, connected => true, connected_at => erlang:system_time(millisecond), diff --git a/apps/emqx_lwm2m/src/emqx_lwm2m_protocol.erl b/apps/emqx_lwm2m/src/emqx_lwm2m_protocol.erl index 34c72dcca..4a5fafdb0 100644 --- a/apps/emqx_lwm2m/src/emqx_lwm2m_protocol.erl +++ b/apps/emqx_lwm2m/src/emqx_lwm2m_protocol.erl @@ -235,8 +235,20 @@ unsubscribe(Topic, Lwm2mState = #lwm2m_state{endpoint_name = _EndpointName}) -> emqx_broker:unsubscribe(Topic), emqx_hooks:run('session.unsubscribed', [clientinfo(Lwm2mState), Topic, Opts]). -publish(Topic, Payload, Qos, EndpointName) -> - emqx_broker:publish(emqx_message:set_flag(retain, false, emqx_message:make(EndpointName, Qos, Topic, Payload))). +publish(Topic, Payload, Qos, + #lwm2m_state{ + version = ProtoVer, + peername = {PeerHost, _}, + endpoint_name = EndpointName}) -> + Message = emqx_message:set_flag( + retain, false, + emqx_message:make(EndpointName, Qos, Topic, Payload) + ), + NMessage = emqx_message:set_headers( + #{proto_ver => ProtoVer, + protocol => lwm2m, + peerhost => PeerHost}, Message), + emqx_broker:publish(NMessage). time_now() -> erlang:system_time(millisecond). @@ -281,7 +293,7 @@ do_send_to_broker(EventType, #{<<"data">> := Data} = Payload, #lwm2m_state{endpo emqx_lwm2m_cm:register_cmd(EndpointName, ReqPath, EventType, {Code, CodeMsg, Content}), NewPayload = maps:put(<<"msgType">>, EventType, Payload), Topic = uplink_topic(EventType, Lwm2mState), - publish(Topic, emqx_json:encode(NewPayload), _Qos = 0, Lwm2mState#lwm2m_state.endpoint_name). + publish(Topic, emqx_json:encode(NewPayload), _Qos = 0, Lwm2mState). %%-------------------------------------------------------------------- %% Auto Observe diff --git a/apps/emqx_stomp/src/emqx_stomp_protocol.erl b/apps/emqx_stomp/src/emqx_stomp_protocol.erl index fc211be10..3dcc6052f 100644 --- a/apps/emqx_stomp/src/emqx_stomp_protocol.erl +++ b/apps/emqx_stomp/src/emqx_stomp_protocol.erl @@ -588,15 +588,29 @@ next_ackid() -> put(ackid, AckId + 1), AckId. -make_mqtt_message(Topic, Headers, Body) -> - Msg = emqx_message:make(stomp, Topic, Body), - Headers1 = lists:foldl(fun(Key, Headers0) -> - proplists:delete(Key, Headers0) - end, Headers, [<<"destination">>, - <<"content-length">>, - <<"content-type">>, - <<"transaction">>, - <<"receipt">>]), +make_mqtt_message(Topic, Headers, Body, + #pstate{ + conninfo = #{proto_ver := ProtoVer}, + clientinfo = #{ + protocol := Protocol, + clientid := ClientId, + username := Username, + peerhost := PeerHost}}) -> + Msg = emqx_message:make( + ClientId, ?QOS_0, + Topic, Body, #{}, + #{proto_ver => ProtoVer, + protocol => Protocol, + username => Username, + peerhost => PeerHost}), + Headers1 = lists:foldl( + fun(Key, Headers0) -> + proplists:delete(Key, Headers0) + end, Headers, [<<"destination">>, + <<"content-length">>, + <<"content-type">>, + <<"transaction">>, + <<"receipt">>]), emqx_message:set_headers(#{stomp_headers => Headers1}, Msg). receipt_id(Headers) -> @@ -611,7 +625,7 @@ handle_recv_send_frame(#stomp_frame{command = <<"SEND">>, headers = Headers, bod allow -> _ = maybe_send_receipt(receipt_id(Headers), State), _ = emqx_broker:publish( - make_mqtt_message(Topic, Headers, iolist_to_binary(Body)) + make_mqtt_message(Topic, Headers, iolist_to_binary(Body), State) ), State; deny -> From f194ae65d299347a098f637a78c0141e61e7f011 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Wed, 10 Nov 2021 16:53:03 +0800 Subject: [PATCH 12/24] chore: update appup.src --- apps/emqx_coap/src/emqx_coap.app.src | 2 +- apps/emqx_coap/src/emqx_coap.appup.src | 9 +++++++++ apps/emqx_exproto/src/emqx_exproto.app.src | 2 +- apps/emqx_exproto/src/emqx_exproto.appup.src | 8 ++++++-- apps/emqx_lwm2m/src/emqx_lwm2m.app.src | 2 +- apps/emqx_lwm2m/src/emqx_lwm2m.appup.src | 6 ++++-- apps/emqx_stomp/src/emqx_stomp.app.src | 2 +- apps/emqx_stomp/src/emqx_stomp.appup.src | 10 ++++++++-- 8 files changed, 31 insertions(+), 10 deletions(-) create mode 100644 apps/emqx_coap/src/emqx_coap.appup.src diff --git a/apps/emqx_coap/src/emqx_coap.app.src b/apps/emqx_coap/src/emqx_coap.app.src index 2b5fcbb6a..dce5ef4f4 100644 --- a/apps/emqx_coap/src/emqx_coap.app.src +++ b/apps/emqx_coap/src/emqx_coap.app.src @@ -1,6 +1,6 @@ {application, emqx_coap, [{description, "EMQ X CoAP Gateway"}, - {vsn, "4.3.0"}, % strict semver, bump manually! + {vsn, "4.3.1"}, % strict semver, bump manually! {modules, []}, {registered, []}, {applications, [kernel,stdlib,gen_coap]}, diff --git a/apps/emqx_coap/src/emqx_coap.appup.src b/apps/emqx_coap/src/emqx_coap.appup.src new file mode 100644 index 000000000..b73e26be6 --- /dev/null +++ b/apps/emqx_coap/src/emqx_coap.appup.src @@ -0,0 +1,9 @@ +%% -*-: erlang -*- +{VSN, + [{"4.3.0",[ + {load_module, emqx_coap_mqtt_adapter, brutal_purge, soft_purge, []}]}, + {<<".*">>, []}], + [{"4.3.0",[ + {load_module, emqx_coap_mqtt_adapter, brutal_purge, soft_purge, []}]}, + {<<".*">>, []}] +}. diff --git a/apps/emqx_exproto/src/emqx_exproto.app.src b/apps/emqx_exproto/src/emqx_exproto.app.src index f7cab4c2e..0ed54e6fd 100644 --- a/apps/emqx_exproto/src/emqx_exproto.app.src +++ b/apps/emqx_exproto/src/emqx_exproto.app.src @@ -1,6 +1,6 @@ {application, emqx_exproto, [{description, "EMQ X Extension for Protocol"}, - {vsn, "4.3.4"}, %% 4.3.3 is used by ee + {vsn, "4.3.5"}, %% 4.3.3 is used by ee {modules, []}, {registered, []}, {mod, {emqx_exproto_app, []}}, diff --git a/apps/emqx_exproto/src/emqx_exproto.appup.src b/apps/emqx_exproto/src/emqx_exproto.appup.src index e0a021af5..7da28e773 100644 --- a/apps/emqx_exproto/src/emqx_exproto.appup.src +++ b/apps/emqx_exproto/src/emqx_exproto.appup.src @@ -1,6 +1,8 @@ %% -*- mode: erlang -*- {VSN, - [{"4.3.3", + [{"4.3.4", + [{load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]}, + {"4.3.3", [{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]}, {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]}, {"4.3.2", @@ -12,7 +14,9 @@ {load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]}, {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]}, {<<".*">>,[]}], - [{"4.3.3", + [{"4.3.4", + [{load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]}, + {"4.3.3", [{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]}, {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]}, {"4.3.2", diff --git a/apps/emqx_lwm2m/src/emqx_lwm2m.app.src b/apps/emqx_lwm2m/src/emqx_lwm2m.app.src index 551cf8d07..b929ad854 100644 --- a/apps/emqx_lwm2m/src/emqx_lwm2m.app.src +++ b/apps/emqx_lwm2m/src/emqx_lwm2m.app.src @@ -1,6 +1,6 @@ {application,emqx_lwm2m, [{description,"EMQ X LwM2M Gateway"}, - {vsn, "4.3.4"}, % strict semver, bump manually! + {vsn, "4.3.5"}, % strict semver, bump manually! {modules,[]}, {registered,[emqx_lwm2m_sup]}, {applications,[kernel,stdlib,lwm2m_coap]}, diff --git a/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src b/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src index 600cf236b..0cd98db0d 100644 --- a/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src +++ b/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src @@ -1,5 +1,5 @@ %% -*-: erlang -*- -{"4.3.4", +{VSN, [ {<<"4\\.3\\.[0-1]">>, [ {restart_application, emqx_lwm2m} @@ -7,7 +7,9 @@ {"4.3.2", [ {load_module, emqx_lwm2m_message, brutal_purge, soft_purge, []} ]}, - {"4.3.3", []} %% only config change + {<<"4\\.3\\.[3-4]">>, [ + {load_module, emqx_lwm2m_protocol, brutal_purge, soft_purge, []} + ]} ], [ {<<"4\\.3\\.[0-1]">>, [ diff --git a/apps/emqx_stomp/src/emqx_stomp.app.src b/apps/emqx_stomp/src/emqx_stomp.app.src index d2ecae53b..c2f4b57d3 100644 --- a/apps/emqx_stomp/src/emqx_stomp.app.src +++ b/apps/emqx_stomp/src/emqx_stomp.app.src @@ -1,6 +1,6 @@ {application, emqx_stomp, [{description, "EMQ X Stomp Protocol Plugin"}, - {vsn, "4.3.2"}, % strict semver, bump manually! + {vsn, "4.3.3"}, % strict semver, bump manually! {modules, []}, {registered, [emqx_stomp_sup]}, {applications, [kernel,stdlib]}, diff --git a/apps/emqx_stomp/src/emqx_stomp.appup.src b/apps/emqx_stomp/src/emqx_stomp.appup.src index bf4603e52..dce441b3c 100644 --- a/apps/emqx_stomp/src/emqx_stomp.appup.src +++ b/apps/emqx_stomp/src/emqx_stomp.appup.src @@ -1,10 +1,16 @@ %% -*- mode: erlang -*- {VSN, - [{"4.3.1",[{load_module,emqx_stomp_connection,brutal_purge,soft_purge,[]}]}, + [{"4.3.2",[{load_module,emqx_stomp_protocol,brutal_purge,soft_purge,[]}]}, + {"4.3.1",[ + {load_module,emqx_stomp_protocol,brutal_purge,soft_purge,[]}, + {load_module,emqx_stomp_connection,brutal_purge,soft_purge,[]}]}, {"4.3.0", [{restart_application,emqx_stomp}]}, {<<".*">>,[]}], - [{"4.3.1",[{load_module,emqx_stomp_connection,brutal_purge,soft_purge,[]}]}, + [{"4.3.2",[{load_module,emqx_stomp_protocol,brutal_purge,soft_purge,[]}]}, + {"4.3.1",[ + {load_module,emqx_stomp_protocol,brutal_purge,soft_purge,[]}, + {load_module,emqx_stomp_connection,brutal_purge,soft_purge,[]}]}, {"4.3.0", [{restart_application,emqx_stomp}]}, {<<".*">>,[]}]}. From 2be33b33e3d9ef36bab0ab2e87985b62d2c0fd48 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Wed, 10 Nov 2021 17:05:27 +0800 Subject: [PATCH 13/24] chore: fix elvis warnings --- apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl | 4 +- .../emqx_exproto/src/emqx_exproto_channel.erl | 2 - apps/emqx_lwm2m/src/emqx_lwm2m_protocol.erl | 47 ++++++++++++------- apps/emqx_stomp/src/emqx_stomp_protocol.erl | 34 ++++++++++---- 4 files changed, 57 insertions(+), 30 deletions(-) diff --git a/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl b/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl index 89fb411b2..f7f0b4eda 100644 --- a/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl +++ b/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl @@ -139,7 +139,7 @@ handle_call({subscribe, Topic, CoapPid}, _From, State=#state{sub_topics = TopicL NewTopics = proplists:delete(Topic, TopicList), IsWild = emqx_topic:wildcard(Topic), {reply, chann_subscribe(Topic, State), State#state{sub_topics = - [{Topic, {IsWild, CoapPid}}|NewTopics]}, hibernate}; + [{Topic, {IsWild, CoapPid}} | NewTopics]}, hibernate}; handle_call({unsubscribe, Topic, _CoapPid}, _From, State=#state{sub_topics = TopicList}) -> NewTopics = proplists:delete(Topic, TopicList), @@ -281,7 +281,7 @@ do_deliver({Topic, Payload}, Subscribers) -> deliver_to_coap(_TopicName, _Payload, []) -> ok; -deliver_to_coap(TopicName, Payload, [{TopicFilter, {IsWild, CoapPid}}|T]) -> +deliver_to_coap(TopicName, Payload, [{TopicFilter, {IsWild, CoapPid}} | T]) -> Matched = case IsWild of true -> emqx_topic:match(TopicName, TopicFilter); false -> TopicName =:= TopicFilter diff --git a/apps/emqx_exproto/src/emqx_exproto_channel.erl b/apps/emqx_exproto/src/emqx_exproto_channel.erl index d8ceae4bd..44b8b64d3 100644 --- a/apps/emqx_exproto/src/emqx_exproto_channel.erl +++ b/apps/emqx_exproto/src/emqx_exproto_channel.erl @@ -606,8 +606,6 @@ default_conninfo(ConnInfo) -> ConnInfo#{clean_start => true, clientid => undefined, username => undefined, - proto_name => undefined, - proto_ver => undefined, conn_props => #{}, connected => true, connected_at => erlang:system_time(millisecond), diff --git a/apps/emqx_lwm2m/src/emqx_lwm2m_protocol.erl b/apps/emqx_lwm2m/src/emqx_lwm2m_protocol.erl index 4a5fafdb0..7f40041a5 100644 --- a/apps/emqx_lwm2m/src/emqx_lwm2m_protocol.erl +++ b/apps/emqx_lwm2m/src/emqx_lwm2m_protocol.erl @@ -74,7 +74,8 @@ call(Pid, Msg, Timeout) -> Error -> {error, Error} end. -init(CoapPid, EndpointName, Peername = {_Peerhost, _Port}, RegInfo = #{<<"lt">> := LifeTime, <<"lwm2m">> := Ver}) -> +init(CoapPid, EndpointName, Peername = {_Peerhost, _Port}, + RegInfo = #{<<"lt">> := LifeTime, <<"lwm2m">> := Ver}) -> Mountpoint = proplists:get_value(mountpoint, lwm2m_coap_responder:options(), ""), Lwm2mState = #lwm2m_state{peername = Peername, endpoint_name = EndpointName, @@ -103,9 +104,10 @@ init(CoapPid, EndpointName, Peername = {_Peerhost, _Port}, RegInfo = #{<<"lt">> emqx_cm:register_channel(EndpointName, CoapPid, conninfo(Lwm2mState1)) end), emqx_cm:insert_channel_info(EndpointName, info(Lwm2mState1), stats(Lwm2mState1)), - emqx_lwm2m_cm:register_channel(EndpointName, RegInfo, LifeTime, Ver, Peername), + emqx_lwm2m_cm:register_channel(EndpointName, RegInfo, LifeTime, Ver, Peername), - {ok, Lwm2mState1#lwm2m_state{life_timer = emqx_lwm2m_timer:start_timer(LifeTime, {life_timer, expired})}}; + NTimer = emqx_lwm2m_timer:start_timer(LifeTime, {life_timer, expired}), + {ok, Lwm2mState1#lwm2m_state{life_timer = NTimer}}; {error, Error} -> _ = run_hooks('client.connack', [conninfo(Lwm2mState), not_authorized], undefined), {error, Error} @@ -133,7 +135,7 @@ update_reg_info(NewRegInfo, Lwm2mState=#lwm2m_state{life_timer = LifeTimer, regi %% - report the registration info update, but only when objectList is updated. case NewRegInfo of #{<<"objectList">> := _} -> - emqx_lwm2m_cm:update_reg_info(Epn, NewRegInfo), + emqx_lwm2m_cm:update_reg_info(Epn, NewRegInfo), send_to_broker(<<"update">>, #{<<"data">> => UpdatedRegInfo}, Lwm2mState); _ -> ok end @@ -186,7 +188,8 @@ deliver(#message{topic = Topic, payload = Payload}, started_at = StartedAt, endpoint_name = EndpointName}) -> IsCacheMode = is_cache_mode(RegInfo, StartedAt), - ?LOG(debug, "Get MQTT message from broker, IsCacheModeNow?: ~p, Topic: ~p, Payload: ~p", [IsCacheMode, Topic, Payload]), + ?LOG(debug, "Get MQTT message from broker, IsCacheModeNow?: ~p, " + "Topic: ~p, Payload: ~p", [IsCacheMode, Topic, Payload]), AlternatePath = maps:get(<<"alternatePath">>, RegInfo, <<"/">>), deliver_to_coap(AlternatePath, Payload, CoapPid, IsCacheMode, EndpointName), Lwm2mState. @@ -256,7 +259,8 @@ time_now() -> erlang:system_time(millisecond). %% Deliver downlink message to coap %%-------------------------------------------------------------------- -deliver_to_coap(AlternatePath, JsonData, CoapPid, CacheMode, EndpointName) when is_binary(JsonData)-> +deliver_to_coap(AlternatePath, JsonData, + CoapPid, CacheMode, EndpointName) when is_binary(JsonData)-> try TermData = emqx_json:decode(JsonData, [return_maps]), deliver_to_coap(AlternatePath, TermData, CoapPid, CacheMode, EndpointName) @@ -285,7 +289,8 @@ deliver_to_coap(AlternatePath, TermData, CoapPid, CacheMode, EndpointName) when send_to_broker(EventType, Payload = #{}, Lwm2mState) -> do_send_to_broker(EventType, Payload, Lwm2mState). -do_send_to_broker(EventType, #{<<"data">> := Data} = Payload, #lwm2m_state{endpoint_name = EndpointName} = Lwm2mState) -> +do_send_to_broker(EventType, #{<<"data">> := Data} = Payload, + #lwm2m_state{endpoint_name = EndpointName} = Lwm2mState) -> ReqPath = maps:get(<<"reqPath">>, Data, undefined), Code = maps:get(<<"code">>, Data, undefined), CodeMsg = maps:get(<<"codeMsg">>, Data, undefined), @@ -327,18 +332,27 @@ auto_observe(AlternatePath, ObjectList, CoapPid, EndpointName) -> observe_object_list(AlternatePath, ObjectList, CoapPid, EndpointName) -> lists:foreach(fun(ObjectPath) -> - [ObjId| LastPath] = emqx_lwm2m_cmd_handler:path_list(ObjectPath), + [ObjId | LastPath] = emqx_lwm2m_cmd_handler:path_list(ObjectPath), case ObjId of <<"19">> -> [ObjInsId | _LastPath1] = LastPath, case ObjInsId of <<"0">> -> - observe_object_slowly(AlternatePath, <<"/19/0/0">>, CoapPid, 100, EndpointName); + observe_object_slowly( + AlternatePath, <<"/19/0/0">>, + CoapPid, 100, EndpointName + ); _ -> - observe_object_slowly(AlternatePath, ObjectPath, CoapPid, 100, EndpointName) + observe_object_slowly( + AlternatePath, ObjectPath, + CoapPid, 100, EndpointName + ) end; _ -> - observe_object_slowly(AlternatePath, ObjectPath, CoapPid, 100, EndpointName) + observe_object_slowly( + AlternatePath, ObjectPath, + CoapPid, 100, EndpointName + ) end end, ObjectList). @@ -392,11 +406,12 @@ get_cached_downlink_messages() -> is_cache_mode(RegInfo, StartedAt) -> case is_psm(RegInfo) orelse is_qmode(RegInfo) of true -> - QModeTimeWind = proplists:get_value(qmode_time_window, lwm2m_coap_responder:options(), 22), - Now = time_now(), - if (Now - StartedAt) >= QModeTimeWind -> true; - true -> false - end; + QModeTimeWind = proplists:get_value( + qmode_time_window, + lwm2m_coap_responder:options(), + 22 + ), + (time_now() - StartedAt) >= QModeTimeWind; false -> false end. diff --git a/apps/emqx_stomp/src/emqx_stomp_protocol.erl b/apps/emqx_stomp/src/emqx_stomp_protocol.erl index 3dcc6052f..4e371d9b0 100644 --- a/apps/emqx_stomp/src/emqx_stomp_protocol.erl +++ b/apps/emqx_stomp/src/emqx_stomp_protocol.erl @@ -108,6 +108,8 @@ , init/2 ]}). +-elvis([{elvis_style, dont_repeat_yourself, disable}]). + -type(pstate() :: #pstate{}). %% @doc Init protocol @@ -132,8 +134,7 @@ init(ConnInfo = #{peername := {PeerHost, _Port}, AllowAnonymous = get_value(allow_anonymous, Opts, false), DefaultUser = get_value(default_user, Opts), - - #pstate{ + #pstate{ conninfo = NConnInfo, clientinfo = ClientInfo, heartfun = HeartFun, @@ -165,7 +166,7 @@ default_conninfo(ConnInfo) -> info(State) -> maps:from_list(info(?INFO_KEYS, State)). --spec info(list(atom())|atom(), pstate()) -> term(). +-spec info(list(atom()) | atom(), pstate()) -> term(). info(Keys, State) when is_list(Keys) -> [{Key, info(Key, State)} || Key <- Keys]; info(conninfo, #pstate{conninfo = ConnInfo}) -> @@ -288,7 +289,12 @@ received(#stomp_frame{command = <<"CONNECT">>}, State = #pstate{connected = true received(Frame = #stomp_frame{command = <<"SEND">>, headers = Headers}, State) -> case header(<<"transaction">>, Headers) of undefined -> {ok, handle_recv_send_frame(Frame, State)}; - TransactionId -> add_action(TransactionId, {fun ?MODULE:handle_recv_send_frame/2, [Frame]}, receipt_id(Headers), State) + TransactionId -> + add_action(TransactionId, + {fun ?MODULE:handle_recv_send_frame/2, [Frame]}, + receipt_id(Headers), + State + ) end; received(#stomp_frame{command = <<"SUBSCRIBE">>, headers = Headers}, @@ -346,7 +352,11 @@ received(#stomp_frame{command = <<"UNSUBSCRIBE">>, headers = Headers}, received(Frame = #stomp_frame{command = <<"ACK">>, headers = Headers}, State) -> case header(<<"transaction">>, Headers) of undefined -> {ok, handle_recv_ack_frame(Frame, State)}; - TransactionId -> add_action(TransactionId, {fun ?MODULE:handle_recv_ack_frame/2, [Frame]}, receipt_id(Headers), State) + TransactionId -> + add_action(TransactionId, + {fun ?MODULE:handle_recv_ack_frame/2, [Frame]}, + receipt_id(Headers), + State) end; %% NACK @@ -357,7 +367,11 @@ received(Frame = #stomp_frame{command = <<"ACK">>, headers = Headers}, State) -> received(Frame = #stomp_frame{command = <<"NACK">>, headers = Headers}, State) -> case header(<<"transaction">>, Headers) of undefined -> {ok, handle_recv_nack_frame(Frame, State)}; - TransactionId -> add_action(TransactionId, {fun ?MODULE:handle_recv_nack_frame/2, [Frame]}, receipt_id(Headers), State) + TransactionId -> + add_action(TransactionId, + {fun ?MODULE:handle_recv_nack_frame/2, [Frame]}, + receipt_id(Headers), + State) end; %% BEGIN @@ -516,9 +530,9 @@ negotiate_version(Accepts) -> negotiate_version(Ver, []) -> {error, <<"Supported protocol versions < ", Ver/binary>>}; -negotiate_version(Ver, [AcceptVer|_]) when Ver >= AcceptVer -> +negotiate_version(Ver, [AcceptVer | _]) when Ver >= AcceptVer -> {ok, AcceptVer}; -negotiate_version(Ver, [_|T]) -> +negotiate_version(Ver, [_ | T]) -> negotiate_version(Ver, T). check_login(Login, _, AllowAnonymous, _) @@ -537,7 +551,7 @@ check_login(Login, Passcode, _, DefaultUser) -> add_action(Id, Action, ReceiptId, State = #pstate{transaction = Trans}) -> case maps:get(Id, Trans, undefined) of {Ts, Actions} -> - NTrans = Trans#{Id => {Ts, [Action|Actions]}}, + NTrans = Trans#{Id => {Ts, [Action | Actions]}}, {ok, State#pstate{transaction = NTrans}}; _ -> send(error_frame(ReceiptId, ["Transaction ", Id, " not found"]), State) @@ -713,7 +727,7 @@ find_sub_by_id(Id, Subs) -> end, Subs), case maps:to_list(Found) of [] -> undefined; - [Sub|_] -> Sub + [Sub | _] -> Sub end. is_acl_enabled(_) -> From 6b40048d297f8f6d4728eef4dcc89e7bb8344565 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Wed, 10 Nov 2021 18:40:40 +0800 Subject: [PATCH 14/24] chore: put the pool_size default value to avoid hot upgrade failure --- apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl | 6 ++++-- apps/emqx_exhook/src/emqx_exhook_mngr.erl | 4 +++- apps/emqx_exhook/test/emqx_exhook_demo_svr.erl | 2 +- 3 files changed, 8 insertions(+), 4 deletions(-) diff --git a/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl b/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl index f7f0b4eda..a6ea09881 100644 --- a/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl +++ b/apps/emqx_coap/src/emqx_coap_mqtt_adapter.erl @@ -58,6 +58,8 @@ -define(SUBOPTS, #{rh => 0, rap => 0, nl => 0, qos => ?QOS_0, is_new => false}). +-define(PROTO_VER, 1). + %%-------------------------------------------------------------------- %% API %%-------------------------------------------------------------------- @@ -260,7 +262,7 @@ packet_to_message(Topic, Payload, emqx_message:make(ClientId, ?QOS_0, Topic, Payload) ), emqx_message:set_headers( - #{ proto_ver => 1 + #{ proto_ver => ?PROTO_VER , protocol => coap , username => Username , peerhost => PeerHost}, Message). @@ -335,7 +337,7 @@ conninfo(#state{peername = Peername, peercert => nossl, %% TODO: dtls conn_mod => ?MODULE, proto_name => <<"CoAP">>, - proto_ver => 1, + proto_ver => ?PROTO_VER, clean_start => true, clientid => ClientId, username => undefined, diff --git a/apps/emqx_exhook/src/emqx_exhook_mngr.erl b/apps/emqx_exhook/src/emqx_exhook_mngr.erl index 0f3c8b8ab..d4c493cb8 100644 --- a/apps/emqx_exhook/src/emqx_exhook_mngr.erl +++ b/apps/emqx_exhook/src/emqx_exhook_mngr.erl @@ -295,7 +295,9 @@ put_pool_size(Val) -> persistent_term:put({?APP, pool_size}, Val). get_pool_size() -> - persistent_term:get({?APP, pool_size}). + %% Avoid the scenario that the parameter is not set after + %% the hot upgrade completed. + persistent_term:get({?APP, pool_size}, erlang:system_info(schedulers)). save(Name, ServerState) -> Saved = persistent_term:get(?APP, []), diff --git a/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl b/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl index 1d3896e14..b05748856 100644 --- a/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl +++ b/apps/emqx_exhook/test/emqx_exhook_demo_svr.erl @@ -295,7 +295,7 @@ on_session_terminated(Req, Md) -> | {error, grpc_cowboy_h:error_response()}. on_message_publish(#{message := #{from := From} = Msg} = Req, Md) -> ?MODULE:in({?FUNCTION_NAME, Req}), - %io:format(standard_error, "fun: ~p, req: ~0p~n", [?FUNCTION_NAME, Req]), + %io:format("fun: ~p, req: ~0p~n", [?FUNCTION_NAME, Req]), %% some cases for testing case From of <<"baduser">> -> From 08cf0326b318a589fb082fab0f0bc975548bcfe8 Mon Sep 17 00:00:00 2001 From: JianBo He Date: Wed, 10 Nov 2021 19:18:42 +0800 Subject: [PATCH 15/24] chore(exhook): bump version to 4.4.0 --- apps/emqx_exhook/src/emqx_exhook.app.src | 2 +- apps/emqx_exhook/src/emqx_exhook.appup.src | 20 ++------------------ 2 files changed, 3 insertions(+), 19 deletions(-) diff --git a/apps/emqx_exhook/src/emqx_exhook.app.src b/apps/emqx_exhook/src/emqx_exhook.app.src index b386bcaca..715060df4 100644 --- a/apps/emqx_exhook/src/emqx_exhook.app.src +++ b/apps/emqx_exhook/src/emqx_exhook.app.src @@ -1,6 +1,6 @@ {application, emqx_exhook, [{description, "EMQ X Extension for Hook"}, - {vsn, "4.3.5"}, + {vsn, "4.4.0"}, {modules, []}, {registered, []}, {mod, {emqx_exhook_app, []}}, diff --git a/apps/emqx_exhook/src/emqx_exhook.appup.src b/apps/emqx_exhook/src/emqx_exhook.appup.src index ea8aacb7a..0fe0ea78f 100644 --- a/apps/emqx_exhook/src/emqx_exhook.appup.src +++ b/apps/emqx_exhook/src/emqx_exhook.appup.src @@ -1,23 +1,7 @@ %% -*-: erlang -*- {VSN, - [{"4.3.4", - [{load_module,emqx_exhook_pb,brutal_purge,soft_purge,[]}, - {load_module,emqx_exhook_handler,brutal_purge,soft_purge,[emqx_exhook_pb]}, - {load_module,emqx_exhook_mngr,brutal_purge,soft_purge,[]}, - {load_module,emqx_exhook_server,brutal_purge,soft_purge,[]}, - {load_module,emqx_exhook_sup,brutal_purge,soft_purge,[]}]}, - {<<"4\\.3\\.[0-3]+">>, - [{restart_application,emqx_exhook}]}, - {<<".*">>, []} + [{<<".*">>, []} ], - [{"4.3.4", - [{load_module,emqx_exhook_pb,brutal_purge,soft_purge,[]}, - {load_module,emqx_exhook_handler,brutal_purge,soft_purge,[emqx_exhook_pb]}, - {load_module,emqx_exhook_mngr,brutal_purge,soft_purge,[]}, - {load_module,emqx_exhook_server,brutal_purge,soft_purge,[]}, - {load_module,emqx_exhook_sup,brutal_purge,soft_purge,[]}]}, - {<<"4\\.3\\.[0-3]+">>, - [{restart_application,emqx_exhook}]}, - {<<".*">>, []} + [{<<".*">>, []} ] }. From f7bdd6defe43d3df4d553c9fc321e467f1bc423e Mon Sep 17 00:00:00 2001 From: JianBo He Date: Fri, 12 Nov 2021 15:35:08 +0800 Subject: [PATCH 16/24] chore(lwm2m): fix bad appup.src --- apps/emqx_lwm2m/src/emqx_lwm2m.appup.src | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src b/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src index 0cd98db0d..6656eb149 100644 --- a/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src +++ b/apps/emqx_lwm2m/src/emqx_lwm2m.appup.src @@ -7,7 +7,8 @@ {"4.3.2", [ {load_module, emqx_lwm2m_message, brutal_purge, soft_purge, []} ]}, - {<<"4\\.3\\.[3-4]">>, [ + {"4.3.3", []}, %% only config change + {"4.3.4", [ {load_module, emqx_lwm2m_protocol, brutal_purge, soft_purge, []} ]} ], @@ -18,6 +19,9 @@ {"4.3.2", [ {load_module, emqx_lwm2m_message, brutal_purge, soft_purge, []} ]}, - {"4.3.3", []} %% only config change + {"4.3.3", []}, %% only config change + {"4.3.4", [ + {load_module, emqx_lwm2m_protocol, brutal_purge, soft_purge, []} + ]} ] }. From 25f504c90ac27d7685e598aab179c2885b033efe Mon Sep 17 00:00:00 2001 From: zhouzb Date: Fri, 12 Nov 2021 15:59:35 +0800 Subject: [PATCH 17/24] feat(mongo srv): support srv for mongodb authentication --- apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf | 9 +- .../priv/emqx_auth_mongo.schema | 37 +++++--- .../src/emqx_auth_mongo.app.src | 2 +- .../src/emqx_auth_mongo_sup.erl | 93 ++++++++++++++++++- 4 files changed, 121 insertions(+), 20 deletions(-) diff --git a/apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf b/apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf index 2a3d038f0..3d3d0ee5b 100644 --- a/apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf +++ b/apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf @@ -5,7 +5,13 @@ ## MongoDB Topology Type. ## ## Value: single | unknown | sharded | rs -auth.mongo.type = single +auth.mongo.type = + +## Whether to use SRV and TXT records. +## +## Value: true | false +## Default: false +auth.mongo.srv_record = false ## The set name if type is rs. ## @@ -37,7 +43,6 @@ auth.mongo.pool = 8 ## MongoDB AuthSource ## ## Value: String -## Default: mqtt ## auth.mongo.auth_source = admin ## MongoDB database diff --git a/apps/emqx_auth_mongo/priv/emqx_auth_mongo.schema b/apps/emqx_auth_mongo/priv/emqx_auth_mongo.schema index 8a2ff98b3..17a83c37c 100644 --- a/apps/emqx_auth_mongo/priv/emqx_auth_mongo.schema +++ b/apps/emqx_auth_mongo/priv/emqx_auth_mongo.schema @@ -6,8 +6,12 @@ {datatype, {enum, [single, unknown, sharded, rs]}} ]}. +{mapping, "auth.mongo.srv_record", "emqx_auth_mongo.server", [ + {default, false}, + {datatype, {enum, [true, false]}} +]}. + {mapping, "auth.mongo.rs_set_name", "emqx_auth_mongo.server", [ - {default, "mqtt"}, {datatype, string} ]}. @@ -41,7 +45,6 @@ ]}. {mapping, "auth.mongo.auth_source", "emqx_auth_mongo.server", [ - {default, "mqtt"}, {datatype, string} ]}. @@ -101,9 +104,9 @@ ]}. {translation, "emqx_auth_mongo.server", fun(Conf) -> - H = cuttlefish:conf_get("auth.mongo.server", Conf), - Hosts = string:tokens(H, ","), - Type0 = cuttlefish:conf_get("auth.mongo.type", Conf), + SrvRecord = cuttlefish:conf_get("auth.mongo.srv_record", Conf, false), + Server = cuttlefish:conf_get("auth.mongo.server", Conf), + Type = cuttlefish:conf_get("auth.mongo.type", Conf), Pool = cuttlefish:conf_get("auth.mongo.pool", Conf), %% FIXME: compatible with 4.0-4.2 version format, plan to delete in 5.0 Login = cuttlefish:conf_get("auth.mongo.username", Conf, @@ -111,7 +114,10 @@ ), Passwd = cuttlefish:conf_get("auth.mongo.password", Conf), DB = cuttlefish:conf_get("auth.mongo.database", Conf), - AuthSrc = cuttlefish:conf_get("auth.mongo.auth_source", Conf), + AuthSource = case cuttlefish:conf_get("auth.mongo.auth_source", Conf, undefined) of + undefined -> []; + AuthSource0 -> [{auth_source, list_to_binary(AuthSource0)}] + end, R = cuttlefish:conf_get("auth.mongo.w_mode", Conf), W = cuttlefish:conf_get("auth.mongo.r_mode", Conf), Login0 = case Login =:= [] of @@ -156,8 +162,8 @@ false -> [] end, - WorkerOptions = [{database, list_to_binary(DB)}, {auth_source, list_to_binary(AuthSrc)}] - ++ Login0 ++ Passwd0 ++ W0 ++ R0 ++ Ssl, + WorkerOptions = [{database, list_to_binary(DB)}] + ++ Login0 ++ Passwd0 ++ W0 ++ R0 ++ Ssl ++ AuthSource, Vars = cuttlefish_variable:fuzzy_matches(["auth", "mongo", "topology", "$name"], Conf), Options = lists:map(fun({_, Name}) -> @@ -174,16 +180,17 @@ {list_to_atom(Name2), cuttlefish:conf_get("auth.mongo.topology."++Name, Conf)} end, Vars), - Type = case Type0 =:= rs of - true -> {Type0, list_to_binary(cuttlefish:conf_get("auth.mongo.rs_set_name", Conf))}; - false -> Type0 - end, - [{type, Type}, - {hosts, Hosts}, + ReplicaSet = case cuttlefish:conf_get("auth.mongo.rs_set_name", Conf, undefined) of + undefined -> []; + ReplicaSet0 -> [{rs_set_name, list_to_binary(ReplicaSet0)}] + end, + [{srv_record, SrvRecord}, + {type, Type}, + {server, Server}, {options, Options}, {worker_options, WorkerOptions}, {auto_reconnect, 1}, - {pool_size, Pool}] + {pool_size, Pool}] ++ ReplicaSet end}. %% The mongodb operation timeout is specified by the value of `cursor_timeout` from application config, diff --git a/apps/emqx_auth_mongo/src/emqx_auth_mongo.app.src b/apps/emqx_auth_mongo/src/emqx_auth_mongo.app.src index cc4e72ef3..ab0b4ff56 100644 --- a/apps/emqx_auth_mongo/src/emqx_auth_mongo.app.src +++ b/apps/emqx_auth_mongo/src/emqx_auth_mongo.app.src @@ -1,6 +1,6 @@ {application, emqx_auth_mongo, [{description, "EMQ X Authentication/ACL with MongoDB"}, - {vsn, "4.3.0"}, % strict semver, bump manually! + {vsn, "4.4.0"}, % strict semver, bump manually! {modules, []}, {registered, [emqx_auth_mongo_sup]}, {applications, [kernel,stdlib,mongodb,ecpool]}, diff --git a/apps/emqx_auth_mongo/src/emqx_auth_mongo_sup.erl b/apps/emqx_auth_mongo/src/emqx_auth_mongo_sup.erl index 3f27cb1dd..55263494a 100644 --- a/apps/emqx_auth_mongo/src/emqx_auth_mongo_sup.erl +++ b/apps/emqx_auth_mongo/src/emqx_auth_mongo_sup.erl @@ -28,7 +28,96 @@ start_link() -> supervisor:start_link({local, ?MODULE}, ?MODULE, []). init([]) -> - {ok, PoolEnv} = application:get_env(?APP, server), - PoolSpec = ecpool:pool_spec(?APP, ?APP, ?APP, PoolEnv), + {ok, Opts} = application:get_env(?APP, server), + NOpts = may_parse_srv_and_txt_records(Opts), + PoolSpec = ecpool:pool_spec(?APP, ?APP, ?APP, NOpts), {ok, {{one_for_all, 10, 100}, [PoolSpec]}}. +may_parse_srv_and_txt_records(Opts) when is_list(Opts) -> + maps:to_list(may_parse_srv_and_txt_records(maps:from_list(Opts))); + +may_parse_srv_and_txt_records(#{type := Type, + srv_record := false, + server := Server} = Opts) -> + Hosts = to_hosts(Server), + case Type =:= rs of + true -> + case maps:get(rs_set_name, Opts, undefined) of + undefined -> + error({missing_parameter, rs_set_name}); + ReplicaSet -> + Opts#{type => {rs, ReplicaSet}, + hosts => Hosts} + end; + false -> + Opts#{hosts => Hosts} + end; + +may_parse_srv_and_txt_records(#{type := Type, + srv_record := true, + server := Server, + worker_options := WorkerOptions} = Opts) -> + Hosts = parse_srv_records(Server), + Opts0 = parse_txt_records(Type, Server), + NWorkerOptions = maps:to_list(maps:merge(maps:from_list(WorkerOptions), maps:with([auth_source], Opts0))), + NOpts = Opts#{hosts => Hosts, worker_options => NWorkerOptions}, + case Type =:= rs of + true -> + case maps:get(rs_set_name, Opts0, maps:get(rs_set_name, NOpts, undefined)) of + undefined -> + error({missing_parameter, rs_set_name}); + ReplicaSet -> + NOpts#{type => {Type, ReplicaSet}} + end; + false -> + NOpts + end. + +to_hosts(Server) -> + [string:trim(H) || H <- string:tokens(Server, ",")]. + +parse_srv_records(Server) -> + case inet_res:lookup("_mongodb._tcp." ++ Server, in, srv) of + [] -> + error(service_not_found); + Services -> + [Host ++ ":" ++ integer_to_list(Port) || {_, _, Port, Host} <- Services] + end. + +parse_txt_records(Type, Server) -> + case inet_res:lookup(Server, in, txt) of + [] -> + #{}; + [[QueryString]] -> + case uri_string:dissect_query(QueryString) of + {error, _, _} -> + error({invalid_txt_record, invalid_query_string}); + Options -> + Fields = case Type of + rs -> ["authSource", "replicaSet"]; + _ -> ["authSource"] + end, + take_and_convert(Fields, Options) + end; + _ -> + error({invalid_txt_record, multiple_records}) + end. + +take_and_convert(Fields, Options) -> + take_and_convert(Fields, Options, #{}). + +take_and_convert([], [_ | _], _Acc) -> + error({invalid_txt_record, invalid_option}); +take_and_convert([], [], Acc) -> + Acc; +take_and_convert([Field | More], Options, Acc) -> + case lists:keytake(Field, 1, Options) of + {value, {"authSource", V}, NOptions} -> + take_and_convert(More, NOptions, Acc#{auth_source => list_to_binary(V)}); + {value, {"replicaSet", V}, NOptions} -> + take_and_convert(More, NOptions, Acc#{rs_set_name => list_to_binary(V)}); + {value, _, _} -> + error({invalid_txt_record, invalid_option}); + false -> + take_and_convert(More, Options, Acc) + end. From cb185389577263c17af0f3348b82c4f3fc50464a Mon Sep 17 00:00:00 2001 From: zhouzb Date: Mon, 15 Nov 2021 09:41:03 +0800 Subject: [PATCH 18/24] fix(mong srv): fix wrong configuration --- apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf b/apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf index 3d3d0ee5b..8baddae19 100644 --- a/apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf +++ b/apps/emqx_auth_mongo/etc/emqx_auth_mongo.conf @@ -5,7 +5,7 @@ ## MongoDB Topology Type. ## ## Value: single | unknown | sharded | rs -auth.mongo.type = +auth.mongo.type = single ## Whether to use SRV and TXT records. ## From f46084438b1994084137dd07a4322ebf20a1084e Mon Sep 17 00:00:00 2001 From: zhanghongtong Date: Thu, 28 Oct 2021 09:19:30 +0800 Subject: [PATCH 19/24] chore(cluster): add new type for dns auto cluster Signed-off-by: zhanghongtong --- etc/emqx.conf | 5 +++++ priv/emqx.schema | 8 +++++++- 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/etc/emqx.conf b/etc/emqx.conf index 7b81e40cc..78d383930 100644 --- a/etc/emqx.conf +++ b/etc/emqx.conf @@ -101,6 +101,11 @@ cluster.autoclean = 5m ## Value: String ## cluster.dns.app = emqx +## Type of dns record. +## +## Value: Value: a | srv +## cluster.dns.type = a + ##-------------------------------------------------------------------- ## Cluster using etcd diff --git a/priv/emqx.schema b/priv/emqx.schema index 602de56b9..f8df59da9 100644 --- a/priv/emqx.schema +++ b/priv/emqx.schema @@ -96,6 +96,11 @@ {datatype, string} ]}. +{mapping, "cluster.dns.type", "ekka.cluster_discovery", [ + {datatype, {enum, [a, srv]}}, + {default, a} +]}. + %%-------------------------------------------------------------------- %% Cluster using etcd @@ -171,7 +176,8 @@ {loop, cuttlefish:conf_get("cluster.mcast.loop", Conf, true)}]; (dns) -> [{name, cuttlefish:conf_get("cluster.dns.name", Conf)}, - {app, cuttlefish:conf_get("cluster.dns.app", Conf)}]; + {app, cuttlefish:conf_get("cluster.dns.app", Conf)}, + {type, cuttlefish:conf_get("cluster.dns.type", Conf)}]; (etcd) -> SslOpts = fun(Conf) -> Options = cuttlefish_variable:filter_by_prefix("cluster.etcd.ssl", Conf), From 87a2667e35ec6ee62b8918f4d66c04f4f2b2e938 Mon Sep 17 00:00:00 2001 From: lafirest Date: Mon, 15 Nov 2021 10:57:59 +0800 Subject: [PATCH 20/24] fix(emqx_retainer): fix timer message error (#6156) * fix(emqx_retainer): fix timer message error --- apps/emqx_retainer/src/emqx_retainer.app.src | 2 +- apps/emqx_retainer/src/emqx_retainer.erl | 10 ++++++---- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/apps/emqx_retainer/src/emqx_retainer.app.src b/apps/emqx_retainer/src/emqx_retainer.app.src index 4ef423b78..a423fb9b7 100644 --- a/apps/emqx_retainer/src/emqx_retainer.app.src +++ b/apps/emqx_retainer/src/emqx_retainer.app.src @@ -1,6 +1,6 @@ {application, emqx_retainer, [{description, "EMQ X Retainer"}, - {vsn, "4.4.1"}, % strict semver, bump manually! + {vsn, "4.4.0"}, % strict semver, bump manually! {modules, []}, {registered, [emqx_retainer_sup]}, {applications, [kernel,stdlib]}, diff --git a/apps/emqx_retainer/src/emqx_retainer.erl b/apps/emqx_retainer/src/emqx_retainer.erl index fcc5652cc..b6de9d9f6 100644 --- a/apps/emqx_retainer/src/emqx_retainer.erl +++ b/apps/emqx_retainer/src/emqx_retainer.erl @@ -156,7 +156,7 @@ start_expire_timer(0, State) -> start_expire_timer(undefined, State) -> State; start_expire_timer(Ms, State) -> - Timer = erlang:send_after(Ms, self(), stats), + Timer = erlang:send_after(Ms, self(), {expire, Ms}), State#state{expiry_timer = Timer}. handle_call(Req, _From, State) -> @@ -168,12 +168,14 @@ handle_cast(Msg, State) -> {noreply, State}. handle_info(stats, State = #state{stats_fun = StatsFun}) -> + StatsTimer = erlang:send_after(timer:seconds(1), self(), stats), StatsFun(retained_count()), - {noreply, State, hibernate}; + {noreply, State#state{stats_timer = StatsTimer}, hibernate}; -handle_info(expire, State) -> +handle_info({expire, Ms} = Expire, State) -> + Timer = erlang:send_after(Ms, self(), Expire), ok = expire_messages(), - {noreply, State, hibernate}; + {noreply, State#state{expiry_timer = Timer}, hibernate}; handle_info(Info, State) -> ?LOG(error, "Unexpected info: ~p", [Info]), From 0357f7ad853db6b83b1c92978885dacd3cbbb2e4 Mon Sep 17 00:00:00 2001 From: lafirest Date: Mon, 15 Nov 2021 11:00:04 +0800 Subject: [PATCH 21/24] improve(emqx_st_statistics): optimize the parameters of on_publish_done (#6151) * fix(emqx_st_statistics): optimize the parameters of on_publish_done --- .../emqx_st_statistics/emqx_st_statistics.erl | 35 +++++++++++++------ src/emqx_session.erl | 9 +++-- 2 files changed, 30 insertions(+), 14 deletions(-) diff --git a/apps/emqx_plugin_libs/src/emqx_st_statistics/emqx_st_statistics.erl b/apps/emqx_plugin_libs/src/emqx_st_statistics/emqx_st_statistics.erl index 21cf2692e..f22aec41c 100644 --- a/apps/emqx_plugin_libs/src/emqx_st_statistics/emqx_st_statistics.erl +++ b/apps/emqx_plugin_libs/src/emqx_st_statistics/emqx_st_statistics.erl @@ -24,7 +24,7 @@ -logger_header("[SLOW TOPICS]"). --export([ start_link/1, on_publish_done/5, enable/0 +-export([ start_link/1, on_publish_done/3, enable/0 , disable/0, clear_history/0 ]). @@ -42,7 +42,7 @@ -type state() :: #{ config := proplist:proplist() , period := pos_integer() , last_tick_at := pos_integer() - , counter := counters:counter_ref() + , counter := counters:counters_ref() , enable := boolean() }. @@ -70,6 +70,13 @@ -type slow_log() :: #slow_log{}. -type top_k_map() :: #{emqx_types:topic() => top_k()}. +-type publish_done_env() :: #{ ignore_before_create := boolean() + , threshold := pos_integer() + , counter := counters:counters_ref() + }. + +-type publish_done_args() :: #{session_rebirth_time => pos_integer()}. + -ifdef(TEST). -define(TOPK_ACCESS, public). -else. @@ -90,13 +97,16 @@ start_link(Env) -> gen_server:start_link({local, ?MODULE}, ?MODULE, [Env], []). --spec on_publish_done(message(), - pos_integer(), boolean(), pos_integer(), counters:counters_ref()) -> ok. -on_publish_done(#message{timestamp = Timestamp}, Created, IgnoreBeforeCreate, _, _) +-spec on_publish_done(message(), publish_done_args(), publish_done_env()) -> ok. +on_publish_done(#message{timestamp = Timestamp}, + #{session_rebirth_time := Created}, + #{ignore_before_create := IgnoreBeforeCreate}) when IgnoreBeforeCreate, Timestamp < Created -> ok; -on_publish_done(#message{timestamp = Timestamp} = Msg, _, _, Threshold, Counter) -> +on_publish_done(#message{timestamp = Timestamp} = Msg, + _, + #{threshold := Threshold, counter := Counter}) -> case ?NOW - Timestamp of Elapsed when Elapsed > Threshold -> case get_log_quota(Counter) of @@ -202,7 +212,7 @@ init_topk_tab(_) -> , {read_concurrency, true} ]). --spec get_log_quota(counter:counter_ref()) -> boolean(). +-spec get_log_quota(counters:counters_ref()) -> boolean(). get_log_quota(Counter) -> case counters:get(Counter, ?QUOTA_IDX) of Quota when Quota > 0 -> @@ -212,7 +222,7 @@ get_log_quota(Counter) -> false end. --spec set_log_quota(proplists:proplist(), counter:counter_ref()) -> ok. +-spec set_log_quota(proplists:proplist(), counters:counters_ref()) -> ok. set_log_quota(Cfg, Counter) -> MaxLogNum = get_value(max_log_num, Cfg), counters:put(Counter, ?QUOTA_IDX, MaxLogNum). @@ -328,12 +338,15 @@ publish(TickTime, Cfg, Notices) -> load(IgnoreBeforeCreate, Threshold, Counter) -> _ = emqx:hook('message.publish_done', - fun ?MODULE:on_publish_done/5, - [IgnoreBeforeCreate, Threshold, Counter]), + fun ?MODULE:on_publish_done/3, + [#{ignore_before_create => IgnoreBeforeCreate, + threshold => Threshold, + counter => Counter} + ]), ok. unload() -> - emqx:unhook('message.publish_done', fun ?MODULE:on_publish_done/5). + emqx:unhook('message.publish_done', fun ?MODULE:on_publish_done/3). -spec get_topic(proplists:proplist()) -> binary(). get_topic(Cfg) -> diff --git a/src/emqx_session.erl b/src/emqx_session.erl index d4b671e71..6122982ae 100644 --- a/src/emqx_session.erl +++ b/src/emqx_session.erl @@ -320,7 +320,8 @@ is_awaiting_full(#session{awaiting_rel = AwaitingRel, puback(PacketId, Session = #session{inflight = Inflight, created_at = CreatedAt}) -> case emqx_inflight:lookup(PacketId, Inflight) of {value, {Msg, _Ts}} when is_record(Msg, message) -> - emqx:run_hook('message.publish_done', [Msg, CreatedAt]), + emqx:run_hook('message.publish_done', + [Msg, #{session_rebirth_time => CreatedAt}]), Inflight1 = emqx_inflight:delete(PacketId, Inflight), return_with(Msg, dequeue(Session#session{inflight = Inflight1})); {value, {_Pubrel, _Ts}} -> @@ -346,7 +347,8 @@ pubrec(PacketId, Session = #session{inflight = Inflight, created_at = CreatedAt} case emqx_inflight:lookup(PacketId, Inflight) of {value, {Msg, _Ts}} when is_record(Msg, message) -> %% execute hook here, because message record will be replaced by pubrel - emqx:run_hook('message.publish_done', [Msg, CreatedAt]), + emqx:run_hook('message.publish_done', + [Msg, #{session_rebirth_time => CreatedAt}]), Inflight1 = emqx_inflight:update(PacketId, with_ts(pubrel), Inflight), {ok, Msg, Session#session{inflight = Inflight1}}; {value, {pubrel, _Ts}} -> @@ -443,7 +445,8 @@ deliver([Msg | More], Acc, Session) -> end. deliver_msg(Msg = #message{qos = ?QOS_0}, Session) -> - emqx:run_hook('message.publish_done', [Msg, Session#session.created_at]), + emqx:run_hook('message.publish_done', + [Msg, #{session_rebirth_time => Session#session.created_at}]), {ok, [{undefined, maybe_ack(Msg)}], Session}; deliver_msg(Msg = #message{qos = QoS}, Session = From 6bd5fd218ec421b144f7c75211f46e12ca4b1da2 Mon Sep 17 00:00:00 2001 From: zhongwencool Date: Mon, 15 Nov 2021 11:02:45 +0800 Subject: [PATCH 22/24] chore: limit/page to position/bytes (#6161) --- .../src/emqx_trace/emqx_trace_api.erl | 25 ++++++++++++------- .../test/emqx_mod_trace_api_SUITE.erl | 8 +++--- 2 files changed, 20 insertions(+), 13 deletions(-) diff --git a/apps/emqx_plugin_libs/src/emqx_trace/emqx_trace_api.erl b/apps/emqx_plugin_libs/src/emqx_trace/emqx_trace_api.erl index a5e5b2906..3af272776 100644 --- a/apps/emqx_plugin_libs/src/emqx_trace/emqx_trace_api.erl +++ b/apps/emqx_plugin_libs/src/emqx_trace/emqx_trace_api.erl @@ -16,6 +16,7 @@ -module(emqx_trace_api). -include_lib("emqx/include/logger.hrl"). +-include_lib("kernel/include/file.hrl"). %% API -export([ list_trace/2 @@ -75,7 +76,7 @@ download_zip_log(#{name := Name}, _Param) -> ZipDir = emqx_trace:zip_dir(), Zips = group_trace_file(ZipDir, TraceLog, TraceFiles), ZipFileName = ZipDir ++ TraceLog, - {ok, ZipFile} = zip:zip(ZipFileName, Zips), + {ok, ZipFile} = zip:zip(ZipFileName, Zips, [{cwd, ZipDir}]), emqx_trace:delete_files_after_send(ZipFileName, Zips), {ok, #{}, {sendfile, 0, filelib:file_size(ZipFile), ZipFile}}; {error, Reason} -> @@ -88,7 +89,7 @@ group_trace_file(ZipDir, TraceLog, TraceFiles) -> {ok, Node, Bin} -> ZipName = ZipDir ++ Node ++ "-" ++ TraceLog, ok = file:write_file(ZipName, Bin), - [ZipName | Acc]; + [Node ++ "-" ++ TraceLog | Acc]; {error, Node, Reason} -> ?LOG(error, "download trace log error:~p", [{Node, TraceLog, Reason}]), Acc @@ -101,20 +102,19 @@ collect_trace_file(TraceLog) -> BadNodes =/= [] andalso ?LOG(error, "download log rpc failed on ~p", [BadNodes]), Files. -%% _page as position and _limit as bytes for front-end reusing components stream_log_file(#{name := Name}, Params) -> Node0 = proplists:get_value(<<"node">>, Params, atom_to_binary(node())), - Position0 = proplists:get_value(<<"_page">>, Params, <<"0">>), - Bytes0 = proplists:get_value(<<"_limit">>, Params, <<"500">>), + Position0 = proplists:get_value(<<"position">>, Params, <<"0">>), + Bytes0 = proplists:get_value(<<"bytes">>, Params, <<"1000">>), Node = binary_to_existing_atom(Node0), Position = binary_to_integer(Position0), Bytes = binary_to_integer(Bytes0), case rpc:call(Node, ?MODULE, read_trace_file, [Name, Position, Bytes]) of {ok, Bin} -> - Meta = #{<<"page">> => Position + byte_size(Bin), <<"limit">> => Bytes}, + Meta = #{<<"position">> => Position + byte_size(Bin), <<"bytes">> => Bytes}, {ok, #{meta => Meta, items => Bin}}; - eof -> - Meta = #{<<"page">> => Position, <<"limit">> => Bytes}, + {eof, Size} -> + Meta = #{<<"position">> => Size, <<"bytes">> => Bytes}, {ok, #{meta => Meta, items => <<"">>}}; {error, Reason} -> logger:log(error, "read_file_failed by ~p", [{Name, Reason, Position, Bytes}]), @@ -134,6 +134,7 @@ read_trace_file(Name, Position, Limit) -> [] -> {error, not_found} end. +-dialyzer({nowarn_function, read_file/3}). read_file(Path, Offset, Bytes) -> {ok, IoDevice} = file:open(Path, [read, raw, binary]), try @@ -141,7 +142,13 @@ read_file(Path, Offset, Bytes) -> 0 -> ok; _ -> file:position(IoDevice, {bof, Offset}) end, - file:read(IoDevice, Bytes) + case file:read(IoDevice, Bytes) of + {ok, Bin} -> {ok, Bin}; + {error, Reason} -> {error, Reason}; + eof -> + #file_info{size = Size} = file:read_file_info(IoDevice), + {eof, Size} + end after file:close(IoDevice) end. diff --git a/lib-ce/emqx_modules/test/emqx_mod_trace_api_SUITE.erl b/lib-ce/emqx_modules/test/emqx_mod_trace_api_SUITE.erl index 609a2d93c..fc786dbd0 100644 --- a/lib-ce/emqx_modules/test/emqx_mod_trace_api_SUITE.erl +++ b/lib-ce/emqx_modules/test/emqx_mod_trace_api_SUITE.erl @@ -124,14 +124,14 @@ t_stream_log(_Config) -> {ok, FileBin} = file:read_file(File), ct:pal("FileBin: ~p ~s", [byte_size(FileBin), FileBin]), Header = auth_header_(), - {ok, Binary} = request_api(get, api_path("trace/test_stream_log/log?_limit=10"), Header), + {ok, Binary} = request_api(get, api_path("trace/test_stream_log/log?bytes=10"), Header), #{<<"code">> := 0, <<"data">> := #{<<"meta">> := Meta, <<"items">> := Bin}} = json(Binary), ?assertEqual(10, byte_size(Bin)), - ?assertEqual(#{<<"page">> => 10, <<"limit">> => 10}, Meta), - Path = api_path("trace/test_stream_log/log?_page=20&_limit=10"), + ?assertEqual(#{<<"position">> => 10, <<"bytes">> => 10}, Meta), + Path = api_path("trace/test_stream_log/log?position=20&bytes=10"), {ok, Binary1} = request_api(get, Path, Header), #{<<"code">> := 0, <<"data">> := #{<<"meta">> := Meta1, <<"items">> := Bin1}} = json(Binary1), - ?assertEqual(#{<<"page">> => 30, <<"limit">> => 10}, Meta1), + ?assertEqual(#{<<"position">> => 30, <<"bytes">> => 10}, Meta1), ?assertEqual(10, byte_size(Bin1)), unload(), ok. From 23e2bd62c59a65d4f38ae661cb3e166c49839692 Mon Sep 17 00:00:00 2001 From: zhongwencool Date: Mon, 15 Nov 2021 15:53:49 +0800 Subject: [PATCH 23/24] fix: there should not be multiple layers of directories when download trace zip file (#6165) --- apps/emqx_plugin_libs/src/emqx_trace/emqx_trace_api.erl | 4 ++-- apps/emqx_plugin_libs/test/emqx_trace_SUITE.erl | 5 ++--- lib-ce/emqx_modules/src/emqx_mod_trace_api.erl | 2 +- 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/apps/emqx_plugin_libs/src/emqx_trace/emqx_trace_api.erl b/apps/emqx_plugin_libs/src/emqx_trace/emqx_trace_api.erl index 3af272776..e6c87d69f 100644 --- a/apps/emqx_plugin_libs/src/emqx_trace/emqx_trace_api.erl +++ b/apps/emqx_plugin_libs/src/emqx_trace/emqx_trace_api.erl @@ -75,10 +75,10 @@ download_zip_log(#{name := Name}, _Param) -> TraceFiles = collect_trace_file(TraceLog), ZipDir = emqx_trace:zip_dir(), Zips = group_trace_file(ZipDir, TraceLog, TraceFiles), - ZipFileName = ZipDir ++ TraceLog, + ZipFileName = ZipDir ++ binary_to_list(Name) ++ ".zip", {ok, ZipFile} = zip:zip(ZipFileName, Zips, [{cwd, ZipDir}]), emqx_trace:delete_files_after_send(ZipFileName, Zips), - {ok, #{}, {sendfile, 0, filelib:file_size(ZipFile), ZipFile}}; + {ok, ZipFile}; {error, Reason} -> {error, Reason} end. diff --git a/apps/emqx_plugin_libs/test/emqx_trace_SUITE.erl b/apps/emqx_plugin_libs/test/emqx_trace_SUITE.erl index 1bd28d888..169fd50bc 100644 --- a/apps/emqx_plugin_libs/test/emqx_trace_SUITE.erl +++ b/apps/emqx_plugin_libs/test/emqx_trace_SUITE.erl @@ -336,9 +336,8 @@ t_download_log(_Config) -> {ok, _} = emqtt:connect(Client), [begin _ = emqtt:ping(Client) end ||_ <- lists:seq(1, 5)], ct:sleep(100), - {ok, #{}, {sendfile, 0, ZipFileSize, _ZipFile}} = - emqx_trace_api:download_zip_log(#{name => Name}, []), - ?assert(ZipFileSize > 0), + {ok, ZipFile} = emqx_trace_api:download_zip_log(#{name => Name}, []), + ?assert(filelib:file_size(ZipFile) > 0), ok = emqtt:disconnect(Client), unload(), ok. diff --git a/lib-ce/emqx_modules/src/emqx_mod_trace_api.erl b/lib-ce/emqx_modules/src/emqx_mod_trace_api.erl index a3abbedf7..99fb97796 100644 --- a/lib-ce/emqx_modules/src/emqx_mod_trace_api.erl +++ b/lib-ce/emqx_modules/src/emqx_mod_trace_api.erl @@ -87,7 +87,7 @@ update_trace(Path, Params) -> download_zip_log(Path, Params) -> case emqx_trace_api:download_zip_log(Path, Params) of - {ok, _Header, _File}= Return -> Return; + {ok, File} -> minirest:return_file(File); {error, _Reason} = Err -> return(Err) end. From 96d2615cc84137368902b0ea09db21e71c0cb2b5 Mon Sep 17 00:00:00 2001 From: zhongwencool Date: Mon, 15 Nov 2021 17:41:22 +0800 Subject: [PATCH 24/24] fix: return error code when trace log not foundd (#6168) --- lib-ce/emqx_modules/src/emqx_mod_trace_api.erl | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/lib-ce/emqx_modules/src/emqx_mod_trace_api.erl b/lib-ce/emqx_modules/src/emqx_mod_trace_api.erl index 99fb97796..1daab1520 100644 --- a/lib-ce/emqx_modules/src/emqx_mod_trace_api.erl +++ b/lib-ce/emqx_modules/src/emqx_mod_trace_api.erl @@ -88,8 +88,11 @@ update_trace(Path, Params) -> download_zip_log(Path, Params) -> case emqx_trace_api:download_zip_log(Path, Params) of {ok, File} -> minirest:return_file(File); - {error, _Reason} = Err -> return(Err) + {error, Reason} -> return({error, 'NOT_FOUND', Reason}) end. stream_log_file(Path, Params) -> - return(emqx_trace_api:stream_log_file(Path, Params)). + case emqx_trace_api:stream_log_file(Path, Params) of + {ok, File} -> return({ok, File}); + {error, Reason} -> return({error, 'NOT_FOUND', Reason}) + end.