Merge pull request #8311 from HJianBo/support-key-dispatch

Ensuring that exhook dispatches the client events are sequential

CHANGES-4.3.md

@@ -19,6 +19,7 @@ File format:
   management HTTPS listeners. [#8129]
 - Add message republish supports using placeholder variables to specify QoS and Retain values. Set `${qos}` and `${flags.retain}` use the original QoS & Retain flag.
 - Add supports specifying the network interface address of the cluster listener & rcp call listener. Specify `0.0.0.0` use all network interfaces, or a particular network interface IP address.
+- ExHook supports to customize the socket parameters for gRPC client. [#8314]
 
 ### Bug fixes
 
@@ -29,6 +30,7 @@ File format:
   information to fill them. [#8280]
 - Fixed issue in Lua hook that didn't prevent a topic from being
   subscribed to. [#8288]
+- Ensuring that exhook dispatches the client events are sequential. [#8311]
 
 ## v4.3.15
 

apps/emqx_exhook/src/emqx_exhook.appup.src

@@ -1,6 +1,7 @@
 %% -*- mode: erlang -*-
 {VSN,
- [  {"4.3.5", [
+ [
+    {"4.3.5", [
         {load_module, emqx_exhook_server, brutal_purge, soft_purge, []}
     ]},
     {"4.3.4", [
@@ -16,7 +17,8 @@
     ]},
     {<<".*">>, []}
  ],
- [  {"4.3.5", [
+ [
+    {"4.3.5", [
         {load_module, emqx_exhook_server, brutal_purge, soft_purge, []}
     ]},
     {"4.3.4", [

apps/emqx_exhook/src/emqx_exhook_server.erl

@@ -273,7 +273,7 @@ match_topic_filter(TopicName, TopicFilter) ->
 -spec do_call(string(), atom(), map(), map()) -> {ok, map()} | {error, term()}.
 do_call(ChannName, Fun, Req, ReqOpts) ->
     NReq = Req#{meta => emqx_exhook:request_meta()},
-    Options = ReqOpts#{channel => ChannName},
+    Options = ReqOpts#{channel => ChannName, key_dispatch => key_dispatch(NReq)},
     ?LOG(debug, "Call ~0p:~0p(~0p, ~0p)", [?PB_CLIENT_MOD, Fun, NReq, Options]),
     case catch apply(?PB_CLIENT_MOD, Fun, [NReq, Options]) of
         {ok, Resp, _Metadata} ->
@@ -331,3 +331,13 @@ available_hooks() ->
      'session.created', 'session.subscribed', 'session.unsubscribed',
      'session.resumed', 'session.discarded', 'session.takeovered',
      'session.terminated' | message_hooks()].
+
+%% @doc Get dispatch_key for each request
+key_dispatch(_Req = #{clientinfo := #{clientid := ClientId}}) ->
+    ClientId;
+key_dispatch(_Req = #{conninfo := #{clientid := ClientId}}) ->
+    ClientId;
+key_dispatch(_Req = #{message := #{from := From}}) ->
+    From;
+key_dispatch(_Req) ->
+    self().

rebar.config

@@ -62,7 +62,7 @@
     , {lc, {git, "https://github.com/emqx/lc.git", {tag, "0.3.1"}}}
     , {mongodb, {git,"https://github.com/emqx/mongodb-erlang", {tag, "v3.0.13"}}}
     , {epgsql, {git, "https://github.com/emqx/epgsql.git", {tag, "4.6.0"}}}
-    , {grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.5"}}}
+    , {grpc, {git, "https://github.com/emqx/grpc-erl", {tag, "0.6.6"}}}
     ]}.
 
 {xref_ignores,