Merge pull request #12203 from JimMoen/EMQX-11610-gw-OCPP-subscription-cnt

EMQX 11610 gw ocpp subscription cnt

apps/emqx_gateway/src/emqx_gateway_api_clients.erl

@@ -197,6 +197,10 @@ subscriptions(get, #{
         case emqx_gateway_http:list_client_subscriptions(GwName, ClientId) of
             {error, not_found} ->
                 return_http_error(404, "client process not found");
+            {error, ignored} ->
+                return_http_error(
+                    400, "get subscriptions failed: unsupported"
+                );
             {error, Reason} ->
                 return_http_error(400, Reason);
             {ok, Subs} ->
@@ -222,7 +226,13 @@ subscriptions(post, #{
                     )
                 of
                     {error, not_found} ->
-                        return_http_error(404, "client process not found");
+                        return_http_error(
+                            404, "client process not found"
+                        );
+                    {error, ignored} ->
+                        return_http_error(
+                            400, "subscribe failed: unsupported"
+                        );
                     {error, Reason} ->
                         return_http_error(400, Reason);
                     {ok, {NTopic, NSubOpts}} ->
@@ -241,8 +251,14 @@ subscriptions(delete, #{
     with_gateway(Name0, fun(GwName, _) ->
         case lookup_topic(GwName, ClientId, Topic) of
             {ok, _} ->
-                _ = emqx_gateway_http:client_unsubscribe(GwName, ClientId, Topic),
+                case emqx_gateway_http:client_unsubscribe(GwName, ClientId, Topic) of
-                {204};
+                    {error, ignored} ->
+                        return_http_error(
+                            400, "unsubscribe failed: unsupported"
+                        );
+                    _ ->
+                        {204}
+                end;
             {error, not_found} ->
                 return_http_error(404, "Resource not found")
         end

apps/emqx_gateway/src/emqx_gateway_http.erl

@@ -378,6 +378,8 @@ client_call(GwName, ClientId, Req) ->
     of
         undefined ->
             {error, not_found};
+        ignored ->
+            {error, ignored};
         Res ->
             Res
     catch

apps/emqx_gateway_ocpp/src/emqx_ocpp_channel.erl

@@ -186,10 +186,10 @@ info(timers, #channel{timers = Timers}) ->
 
 -spec stats(channel()) -> emqx_types:stats().
 stats(#channel{mqueue = MQueue}) ->
-    %% XXX:
+    %% XXX: A fake stats for managed by emqx_management
     SessionStats = [
-        {subscriptions_cnt, 0},
+        {subscriptions_cnt, 1},
-        {subscriptions_max, 0},
+        {subscriptions_max, 1},
         {inflight_cnt, 0},
         {inflight_max, 0},
         {mqueue_len, queue:len(MQueue)},
@@ -524,9 +524,13 @@ handle_out(Type, Data, Channel) ->
 %%--------------------------------------------------------------------
 
 apply_frame(Frames, Channel) when is_list(Frames) ->
-    {Outgoings, NChannel} = lists:foldl(fun apply_frame/2, {[], Channel}, Frames),
+    {Outgoings, NChannel} = lists:foldl(fun do_apply_frame/2, {[], Channel}, Frames),
     {lists:reverse(Outgoings), NChannel};
-apply_frame(?IS_BootNotification_RESP(Payload), {Outgoings, Channel}) ->
+apply_frame(Frames, Channel) ->
+    ?SLOG(error, #{msg => "unexpected_frame_list", frames => Frames, channel => Channel}),
+    Channel.
+
+do_apply_frame(?IS_BootNotification_RESP(Payload), {Outgoings, Channel}) ->
     case maps:get(<<"status">>, Payload) of
         <<"Accepted">> ->
             Intv = maps:get(<<"interval">>, Payload),
@@ -535,8 +539,9 @@ apply_frame(?IS_BootNotification_RESP(Payload), {Outgoings, Channel}) ->
         _ ->
             {Outgoings, Channel}
     end;
-apply_frame(_, Channel) ->
+do_apply_frame(Frame, Acc = {_Outgoings, Channel}) ->
-    Channel.
+    ?SLOG(error, #{msg => "unexpected_frame", frame => Frame, channel => Channel}),
+    Acc.
 
 %%--------------------------------------------------------------------
 %% Handle call

apps/emqx_gateway_ocpp/test/emqx_ocpp_SUITE.erl

@@ -33,27 +33,27 @@
 
 -define(HEARTBEAT, <<$\n>>).
 
--define(CONF_DEFAULT, <<
+%% erlfmt-ignore
-    "\n"
+-define(CONF_DEFAULT, <<"
-    "gateway.ocpp {\n"
+    gateway.ocpp {
-    "  mountpoint = \"ocpp/\"\n"
+      mountpoint = \"ocpp/\"
-    "  default_heartbeat_interval = \"60s\"\n"
+      default_heartbeat_interval = \"60s\"
-    "  heartbeat_checking_times_backoff = 1\n"
+      heartbeat_checking_times_backoff = 1
-    "  message_format_checking = disable\n"
+      message_format_checking = disable
-    "  upstream {\n"
+      upstream {
-    "    topic = \"cp/${clientid}\"\n"
+        topic = \"cp/${clientid}\"
-    "    reply_topic = \"cp/${clientid}/Reply\"\n"
+        reply_topic = \"cp/${clientid}/Reply\"
-    "    error_topic = \"cp/${clientid}/Reply\"\n"
+        error_topic = \"cp/${clientid}/Reply\"
-    "  }\n"
+      }
-    "  dnstream {\n"
+      dnstream {
-    "    topic = \"cs/${clientid}\"\n"
+        topic = \"cs/${clientid}\"
-    "  }\n"
+      }
-    "  listeners.ws.default {\n"
+      listeners.ws.default {
-    "      bind = \"0.0.0.0:33033\"\n"
+          bind = \"0.0.0.0:33033\"
-    "      websocket.path = \"/ocpp\"\n"
+          websocket.path = \"/ocpp\"
-    "  }\n"
+      }
-    "}\n"
+    }
->>).
+">>).
 
 all() -> emqx_common_test_helpers:all(?MODULE).