Merge pull request #6077 from zmstone/sync-exproto-code

Sync exproto code from ee to ce

apps/emqx_auth_pgsql/rebar.config

@@ -1,5 +1,5 @@
 {deps,
- [{epgsql, {git, "https://github.com/epgsql/epgsql", {tag, "4.4.0"}}}
+ [{epgsql, {git, "https://github.com/epgsql/epgsql.git", {tag, "4.4.0"}}}
  ]}.
 
 {erl_opts, [warn_unused_vars,

apps/emqx_exproto/src/emqx_exproto.app.src

@@ -1,6 +1,6 @@
 {application, emqx_exproto,
  [{description, "EMQ X Extension for Protocol"},
-  {vsn, "4.3.2"}, %% strict semver
+  {vsn, "4.3.4"}, %% 4.3.3 is used by ee
   {modules, []},
   {registered, []},
   {mod, {emqx_exproto_app, []}},

apps/emqx_exproto/src/emqx_exproto.appup.src

@@ -1,31 +1,26 @@
-%% -*-: erlang -*-
+%% -*- mode: erlang -*-
 {VSN,
- [
-    {"4.3.1", [
-      {load_module, emqx_exproto_gsvr, brutal_purge, soft_purge, []},
-      {load_module, emqx_exproto_gcli, brutal_purge, soft_purge, []},
-      {load_module, emqx_exproto_channel, brutal_purge, soft_purge, []}
-    ]},
-    {"4.3.0", [
-      {load_module, emqx_exproto_gsvr, brutal_purge, soft_purge, []},
-      {load_module, emqx_exproto_gcli, brutal_purge, soft_purge, []},
-      {load_module, emqx_exproto_conn, brutal_purge, soft_purge, []},
-      {load_module, emqx_exproto_channel, brutal_purge, soft_purge, []}
-    ]},
-    {<<".*">>, []}
- ],
- [
-    {"4.3.1", [
-      {load_module, emqx_exproto_gsvr, brutal_purge, soft_purge, []},
-      {load_module, emqx_exproto_gcli, brutal_purge, soft_purge, []},
-      {load_module, emqx_exproto_channel, brutal_purge, soft_purge, []}
-    ]},
-    {"4.3.0", [
-      {load_module, emqx_exproto_gsvr, brutal_purge, soft_purge, []},
-      {load_module, emqx_exproto_gcli, brutal_purge, soft_purge, []},
-      {load_module, emqx_exproto_conn, brutal_purge, soft_purge, []},
-      {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",
+    [{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
+   {<<"4.3.[0-1]">>,
+    [{load_module,emqx_exproto_gsvr,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_gcli,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
+     {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",
+    [{load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
+   {<<"4.3.[0-1]">>,
+    [{load_module,emqx_exproto_gsvr,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_gcli,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_conn,brutal_purge,soft_purge,[]},
+     {load_module,emqx_exproto_channel,brutal_purge,soft_purge,[]}]},
+   {<<".*">>,[]}]}.

apps/emqx_exproto/src/emqx_exproto_channel.erl

@@ -94,6 +94,9 @@
          awaiting_rel_max
         ]).
 
+-define(CHANMOCK(P), {exproto_anonymous_client, P}).
+-define(CHAN_CONN_TAB, emqx_channel_conn).
+
 %%--------------------------------------------------------------------
 %% Info, Attrs and Caps
 %%--------------------------------------------------------------------
@@ -103,7 +106,7 @@
 info(Channel) ->
     maps:from_list(info(?INFO_KEYS, Channel)).
 
--spec(info(list(atom())|atom(), channel()) -> term()).
+-spec(info(list(atom()) | atom(), channel()) -> term()).
 info(Keys, Channel) when is_list(Keys) ->
     [{Key, info(Key, Channel)} || Key <- Keys];
 info(conninfo, #channel{conninfo = ConnInfo}) ->
@@ -155,13 +158,19 @@ init(ConnInfo = #{socktype := Socktype,
                        conn_state = connecting,
                        timers = #{}
                       },
-
-    Req = #{conninfo =>
-            peercert(Peercert,
-                     #{socktype => socktype(Socktype),
-                       peername => address(Peername),
-                       sockname => address(Sockname)})},
-    try_dispatch(on_socket_created, wrap(Req), Channel).
+    case emqx_hooks:run_fold('client.connect', [NConnInfo], #{}) of
+        {error, _Reason} ->
+            throw(nopermission);
+        _ ->
+            ConnMod = maps:get(conn_mod, NConnInfo),
+            true = ets:insert(?CHAN_CONN_TAB, {?CHANMOCK(self()), ConnMod}),
+            Req = #{conninfo =>
+                    peercert(Peercert,
+                             #{socktype => socktype(Socktype),
+                               peername => address(Peername),
+                               sockname => address(Sockname)})},
+            try_dispatch(on_socket_created, wrap(Req), Channel)
+    end.
 
 %% @private
 peercert(NoSsl, ConnInfo) when NoSsl == nossl;
@@ -283,6 +292,7 @@ handle_call({auth, ClientInfo0, Password},
                 emqx_metrics:inc('client.auth.anonymous'),
             NClientInfo = maps:merge(ClientInfo1, AuthResult),
             NChannel = Channel1#channel{clientinfo = NClientInfo},
+            clean_anonymous_clients(),
             case emqx_cm:open_session(true, NClientInfo, NConnInfo) of
                 {ok, _Session} ->
                     ?LOG(debug, "Client ~s (Username: '~s') authorized successfully!",
@@ -399,12 +409,16 @@ handle_info(Info, Channel) ->
 
 -spec(terminate(any(), channel()) -> channel()).
 terminate(Reason, Channel) ->
+    clean_anonymous_clients(),
     Req = #{reason => stringfy(Reason)},
     try_dispatch(on_socket_closed, wrap(Req), Channel).
 
 is_anonymous(#{anonymous := true}) -> true;
 is_anonymous(_AuthResult)          -> false.
 
+clean_anonymous_clients() ->
+    ets:delete(?CHAN_CONN_TAB, ?CHANMOCK(self())).
+
 %%--------------------------------------------------------------------
 %% Sub/UnSub
 %%--------------------------------------------------------------------
@@ -577,7 +591,6 @@ default_conninfo(ConnInfo) ->
     ConnInfo#{clean_start => true,
               clientid => undefined,
               username => undefined,
-              conn_mod => undefined,
               conn_props => #{},
               connected => true,
               connected_at => erlang:system_time(millisecond),

apps/emqx_exproto/src/emqx_exproto_conn.erl

@@ -115,7 +115,7 @@ start_link(esockd_transport, Sock, Options) ->
 %%--------------------------------------------------------------------
 
 %% @doc Get infos of the connection/channel.
--spec(info(pid()|state()) -> emqx_types:infos()).
+-spec(info(pid() | state()) -> emqx_types:infos()).
 info(CPid) when is_pid(CPid) ->
     call(CPid, info);
 info(State = #state{channel = Channel}) ->
@@ -137,7 +137,7 @@ info(sockstate, #state{sockstate = SockSt}) ->
 info(active_n, #state{active_n = ActiveN}) ->
     ActiveN.
 
--spec(stats(pid()|state()) -> emqx_types:stats()).
+-spec(stats(pid() | state()) -> emqx_types:stats()).
 stats(CPid) when is_pid(CPid) ->
     call(CPid, stats);
 stats(#state{socket  = Socket,
@@ -233,7 +233,11 @@ init(Parent, WrappedSock, Peername0, Options) ->
     case esockd_wait(WrappedSock) of
         {ok, NWrappedSock} ->
             Peername = esockd_peername(NWrappedSock, Peername0),
-            run_loop(Parent, init_state(NWrappedSock, Peername, Options));
+            try
+                run_loop(Parent, init_state(NWrappedSock, Peername, Options))
+            catch
+                throw : nopermission -> erlang:exit(normal)
+            end;
         {error, Reason} ->
             ok = esockd_close(WrappedSock),
             exit_on_sock_error(Reason)
@@ -337,7 +341,7 @@ cancel_stats_timer(State) -> State.
 
 process_msg([], Parent, State) -> recvloop(Parent, State);
 
-process_msg([Msg|More], Parent, State) ->
+process_msg([Msg | More], Parent, State) ->
     case catch handle_msg(Msg, State) of
         ok ->
             process_msg(More, Parent, State);
@@ -413,7 +417,7 @@ handle_msg({Passive, _Sock}, State)
 
 handle_msg(Deliver = {deliver, _Topic, _Msg},
            State = #state{active_n = ActiveN}) ->
-    Delivers = [Deliver|emqx_misc:drain_deliver(ActiveN)],
+    Delivers = [Deliver | emqx_misc:drain_deliver(ActiveN)],
     with_channel(handle_deliver, [Delivers], State);
 
 %% Something sent
@@ -601,9 +605,9 @@ handle_outgoing(IoData, State = #state{socket = Socket}) ->
 handle_info(activate_socket, State = #state{sockstate = OldSst}) ->
     case activate_socket(State) of
         {ok, NState = #state{sockstate = NewSst}} ->
-            if OldSst =/= NewSst ->
-                   {ok, {event, NewSst}, NState};
-               true -> {ok, NState}
+            case OldSst =/= NewSst of
+                true -> {ok, {event, NewSst}, NState};
+                false -> {ok, NState}
             end;
         {error, Reason} ->
             handle_info({sock_error, Reason}, State)

apps/emqx_rule_engine/src/emqx_rule_engine.app.src

@@ -1,6 +1,6 @@
 {application, emqx_rule_engine,
  [{description, "EMQ X Rule Engine"},
-  {vsn, "4.3.5"}, % strict semver, bump manually!
+  {vsn, "4.3.6"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_rule_engine_sup, emqx_rule_registry]},
   {applications, [kernel,stdlib,rulesql,getopt]},

apps/emqx_rule_engine/src/emqx_rule_engine.appup.src

@@ -1,64 +1,52 @@
-%% -*-: erlang -*-
-{"4.3.5",
- [ {"4.3.0",
-    [ {load_module, emqx_rule_funcs, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
-    , {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
-    , {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
-    ]},
-   {"4.3.1",
-    [ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
-    , {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
-    , {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
-    ]},
-   {"4.3.2",
-    [ {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
-    , {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
-    , {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
-    ]},
-   {"4.3.3",
-    [ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
-    ]},
-   {"4.3.4",
-    [ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
-    ]},
-   {<<".*">>, []}
- ],
- [
+%% -*- mode: erlang -*-
+{VSN,
+  [{"4.3.5",[{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.3.0",
-    [ {load_module, emqx_rule_funcs, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
-    , {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
-    , {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
-    ]},
+    [{load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
    {"4.3.1",
-    [ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
-    , {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
-    , {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
-    ]},
+    [{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
    {"4.3.2",
-    [ {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
-    , {apply, {emqx_stats, cancel_update, [rule_registery_stats]}}
-    , {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
-    ]},
+    [{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
+     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
    {"4.3.3",
-    [ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_actions, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
-    ]},
+    [{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
    {"4.3.4",
-    [ {load_module, emqx_rule_engine, brutal_purge, soft_purge, []}
-    , {load_module, emqx_rule_registry, brutal_purge, soft_purge, []}
-    ]},
-   {<<".*">>, []}
- ]
-}.
+    [{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+   {<<".*">>,[]}],
+  [{"4.3.5",[{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+   {"4.3.0",
+    [{load_module,emqx_rule_funcs,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
+   {"4.3.1",
+    [{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
+   {"4.3.2",
+    [{load_module,emqx_rule_registry,brutal_purge,soft_purge,[]},
+     {apply,{emqx_stats,cancel_update,[rule_registery_stats]}},
+     {load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]}]},
+   {"4.3.3",
+    [{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_actions,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+   {"4.3.4",
+    [{load_module,emqx_rule_engine,brutal_purge,soft_purge,[]},
+     {load_module,emqx_rule_registry,brutal_purge,soft_purge,[]}]},
+   {<<".*">>,[]}]}.

apps/emqx_stomp/src/emqx_stomp.app.src

@@ -1,6 +1,6 @@
 {application, emqx_stomp,
  [{description, "EMQ X Stomp Protocol Plugin"},
-  {vsn, "4.3.1"}, % strict semver, bump manually!
+  {vsn, "4.3.2"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_stomp_sup]},
   {applications, [kernel,stdlib]},

apps/emqx_stomp/src/emqx_stomp.appup.src

@@ -1,8 +1,10 @@
 %% -*- mode: erlang -*-
-{"4.3.1",
-  [{"4.3.0",
+{VSN,
+  [{"4.3.1",[{load_module,emqx_stomp_connection,brutal_purge,soft_purge,[]}]},
+   {"4.3.0",
     [{restart_application,emqx_stomp}]},
    {<<".*">>,[]}],
-  [{"4.3.0",
-     [{restart_application,emqx_stomp}]},
+  [{"4.3.1",[{load_module,emqx_stomp_connection,brutal_purge,soft_purge,[]}]},
+   {"4.3.0",
+    [{restart_application,emqx_stomp}]},
    {<<".*">>,[]}]}.

apps/emqx_stomp/src/emqx_stomp_connection.erl

@@ -91,6 +91,8 @@
 
 -define(ENABLED(X), (X =/= undefined)).
 
+-elvis([{elvis_style, invalid_dynamic_call, #{ignore => [emqx_stomp_connection]}}]).
+
 -dialyzer({nowarn_function, [ ensure_stats_timer/2
                             ]}).
 
@@ -101,7 +103,7 @@
 start_link(Transport, Sock, ProtoEnv) ->
     {ok, proc_lib:spawn_link(?MODULE, init, [[Transport, Sock, ProtoEnv]])}.
 
--spec info(pid()|state()) -> emqx_types:infos().
+-spec info(pid() | state()) -> emqx_types:infos().
 info(CPid) when is_pid(CPid) ->
     call(CPid, info);
 info(State = #state{pstate = PState}) ->
@@ -123,7 +125,7 @@ info(sockstate, #state{sockstate = SockSt}) ->
 info(active_n, #state{active_n = ActiveN}) ->
     ActiveN.
 
--spec stats(pid()|state()) -> emqx_types:stats().
+-spec stats(pid() | state()) -> emqx_types:stats().
 stats(CPid) when is_pid(CPid) ->
     call(CPid, stats);
 stats(#state{transport = Transport,

lib-ce/emqx_dashboard/src/emqx_dashboard.app.src

@@ -1,6 +1,6 @@
 {application, emqx_dashboard,
  [{description, "EMQ X Web Dashboard"},
-  {vsn, "4.3.5"}, % strict semver, bump manually!
+  {vsn, "4.3.6"}, % strict semver, bump manually!
   {modules, []},
   {registered, [emqx_dashboard_sup]},
   {applications, [kernel,stdlib,mnesia,minirest]},