fix(emqx_bridge): unwrap rpc multicall result

apps/emqx_bridge/src/emqx_bridge_api.erl

@@ -856,7 +856,7 @@ unpack_bridge_conf(Type, PackedConf) ->
 
 is_ok(ok) ->
     ok;
-is_ok({ok, _} = OkResult) ->
+is_ok(OkResult = {ok, _}) ->
     OkResult;
 is_ok(Error = {error, _}) ->
     Error;
@@ -933,7 +933,9 @@ maybe_try_restart(_, _, _) ->
     {501}.
 
 do_bpapi_call(all, Call, Args) ->
-    do_bpapi_call_vsn(emqx_bpapi:supported_version(emqx_bridge), Call, Args);
+    maybe_unwrap(
+        do_bpapi_call_vsn(emqx_bpapi:supported_version(emqx_bridge), Call, Args)
+    );
 do_bpapi_call(Node, Call, Args) ->
     do_bpapi_call_vsn(emqx_bpapi:supported_version(Node, emqx_bridge), Call, Args).
 
@@ -945,6 +947,11 @@ do_bpapi_call_vsn(SupportedVersion, Call, Args) ->
             {error, not_implemented}
     end.
 
+maybe_unwrap({error, not_implemented}) ->
+    {error, not_implemented};
+maybe_unwrap(RpcMulticallResult) ->
+    emqx_rpc:unwrap_erpc(RpcMulticallResult).
+
 supported_versions(start_bridge_to_node) -> [2];
 supported_versions(start_bridges_to_all_nodes) -> [2];
 supported_versions(_Call) -> [1, 2].

apps/emqx_bridge/test/emqx_bridge_api_SUITE.erl

@@ -38,12 +38,16 @@
     <<"type">> => TYPE,
     <<"name">> => NAME
 }).
--define(MQTT_BRIDGE(SERVER), ?BRIDGE(<<"mqtt_egress_test_bridge">>, <<"mqtt">>)#{
+
+-define(BRIDGE_TYPE_MQTT, <<"mqtt">>).
+-define(MQTT_BRIDGE(SERVER, NAME), ?BRIDGE(NAME, ?BRIDGE_TYPE_MQTT)#{
     <<"server">> => SERVER,
     <<"username">> => <<"user1">>,
     <<"password">> => <<"">>,
     <<"proto_ver">> => <<"v5">>
 }).
+-define(MQTT_BRIDGE(SERVER), ?MQTT_BRIDGE(SERVER, <<"mqtt_egress_test_bridge">>)).
+
 -define(HTTP_BRIDGE(URL, TYPE, NAME), ?BRIDGE(NAME, TYPE)#{
     <<"url">> => URL,
     <<"local_topic">> => <<"emqx_webhook/#">>,
@@ -536,7 +540,27 @@ do_start_stop_bridges(Type, Config) ->
     ?assertMatch(#{<<"status">> := <<"connected">>}, jsx:decode(Bridge4)),
     %% delete the bridge
     {ok, 204, <<>>} = request(delete, uri(["bridges", BridgeID]), []),
-    {ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []).
+    {ok, 200, <<"[]">>} = request(get, uri(["bridges"]), []),
+
+    %% Create broken bridge
+    BadServer = <<"nohost">>,
+    BadName = <<"bad_", (atom_to_binary(Type))/binary>>,
+    {ok, 201, BadBridge1} = request(
+        post,
+        uri(["bridges"]),
+        ?MQTT_BRIDGE(BadServer, BadName)
+    ),
+    #{
+        <<"type">> := ?BRIDGE_TYPE_MQTT,
+        <<"name">> := BadName,
+        <<"enable">> := true,
+        <<"server">> := BadServer,
+        <<"status">> := <<"disconnected">>,
+        <<"node_status">> := [_ | _]
+    } = jsx:decode(BadBridge1),
+    BadBridgeID = emqx_bridge_resource:bridge_id(?BRIDGE_TYPE_MQTT, BadName),
+    {ok, 500, _} = request(post, operation_path(Type, start, BadBridgeID), <<"">>),
+    ok.
 
 t_enable_disable_bridges(Config) ->
     %% assert we there's no bridges at first