Merge pull request #6419 from emqx/sync-from-v4.3

Sync from v4.3

.github/workflows/build_packages.yaml

@@ -486,6 +486,22 @@ jobs:
              -X POST \
              -d "{\"repo\":\"emqx/emqx\", \"tag\": \"${{ env.version }}\" }" \
              ${{ secrets.EMQX_IO_RELEASE_API }}
+    - name: push docker image to docker hub
+      if: github.event_name == 'release'
+      run: |
+        set -e -x -u
+        sudo make docker-prepare
+        cd _packages/${{ matrix.profile }} && for var in $(ls |grep docker |grep -v sha256); do unzip $var; sudo docker load < ${var%.*}; rm -f ${var%.*}; done && cd -
+        echo ${{ secrets.DOCKER_HUB_TOKEN }} |sudo docker login -u ${{ secrets.DOCKER_HUB_USER }} --password-stdin
+        sudo TARGET=emqx/${{ matrix.profile }} make docker-push
+        sudo TARGET=emqx/${{ matrix.profile }} make docker-manifest-list
+    - name: push docker image to aws ecr
+      if: github.event_name == 'release'
+      run: |
+        set -e -x -u
+        aws ecr-public get-login-password --region us-east-1 | docker login --username AWS --password-stdin public.ecr.aws
+        docker tag emqx/emqx:${version#v} public.ecr.aws/emqx/emqx:${version#v}
+        docker push public.ecr.aws/emqx/emqx:${version#v}
     - name: update repo.emqx.io
       if: github.event_name == 'release' && matrix.profile == 'emqx-ee'
       run: |

apps/emqx_web_hook/src/emqx_web_hook_actions.erl

@@ -311,15 +311,12 @@ ensure_content_type_header(Headers, _Method) ->
 
 merge_path(CommonPath, <<>>) ->
     l2b(CommonPath);
-merge_path(CommonPath, Path0) ->
+merge_path(CommonPath, Path) ->
-    case emqx_http_lib:uri_parse(Path0) of
+    Path1 = case Path of
-        {ok, #{path := Path1, 'query' := Query0}} ->
+                <<"/", Path0/binary>> -> Path0;
-            Path2 = l2b(filename:join(CommonPath, Path1)),
+                _ -> Path
-            Query = l2b(Query0),
+            end,
-            <<Path2/binary, "?", Query/binary>>;
+    l2b(filename:join(CommonPath, Path1)).
-        {ok, #{path := Path1}} ->
-            l2b(filename:join(CommonPath, Path1))
-    end.
 
 method(GET) when GET == <<"GET">>; GET == <<"get">> -> get;
 method(POST) when POST == <<"POST">>; POST == <<"post">> -> post;

rebar.config

@@ -37,7 +37,7 @@
 
 {deps,
     [ {gpb, "4.11.2"} %% gpb only used to build, but not for release, pin it here to avoid fetching a wrong version due to rebar plugins scattered in all the deps
-    , {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.1.12"}}}
+    , {ehttpc, {git, "https://github.com/emqx/ehttpc", {tag, "0.1.13"}}}
     , {eredis_cluster, {git, "https://github.com/emqx/eredis_cluster", {tag, "0.6.5"}}}
     , {gproc, {git, "https://github.com/uwiger/gproc", {tag, "0.8.0"}}}
     , {jiffy, {git, "https://github.com/emqx/jiffy", {tag, "1.0.5"}}}

src/emqx_channel.erl

@@ -721,20 +721,25 @@ maybe_update_expiry_interval(_Properties, Channel) -> Channel.
 
 -spec(handle_deliver(list(emqx_types:deliver()), channel())
       -> {ok, channel()} | {ok, replies(), channel()}).
+handle_deliver(Delivers, Channel = #channel{takeover = true,
+                                            pendings = Pendings,
+                                            session = Session,
+                                            clientinfo = #{clientid := ClientId}}) ->
+    %% NOTE: Order is important here. While the takeover is in
+    %% progress, the session cannot enqueue messages, since it already
+    %% passed on the queue to the new connection in the session state.
+    NPendings = lists:append(Pendings, ignore_local(maybe_nack(Delivers), ClientId, Session)),
+    {ok, Channel#channel{pendings = NPendings}};
+
 handle_deliver(Delivers, Channel = #channel{conn_state = disconnected,
+                                            takeover   = false,
                                             session    = Session,
                                             clientinfo = #{clientid := ClientId}}) ->
     NSession = emqx_session:enqueue(ignore_local(maybe_nack(Delivers), ClientId, Session), Session),
     {ok, Channel#channel{session = NSession}};
 
-handle_deliver(Delivers, Channel = #channel{takeover = true,
-                                            pendings = Pendings,
-                                            session = Session,
-                                            clientinfo = #{clientid := ClientId}}) ->
-    NPendings = lists:append(Pendings, ignore_local(maybe_nack(Delivers), ClientId, Session)),
-    {ok, Channel#channel{pendings = NPendings}};
-
 handle_deliver(Delivers, Channel = #channel{session = Session,
+                                            takeover = false,
                                             clientinfo = #{clientid := ClientId}}) ->
     case emqx_session:deliver(ignore_local(Delivers, ClientId, Session), Session) of
         {ok, Publishes, NSession} ->