fix(mqtt_bridge): return value of sending messages was discarded

apps/emqx_connector/src/emqx_connector_mqtt.erl

@@ -187,8 +187,7 @@ on_stop(_InstId, #{name := InstanceId}) ->
 
 on_query(_InstId, {send_message, Msg}, #{name := InstanceId}) ->
     ?TRACE("QUERY", "send_msg_to_remote_node", #{message => Msg, connector => InstanceId}),
-    emqx_connector_mqtt_worker:send_to_remote(InstanceId, Msg),
-    ok.
+    emqx_connector_mqtt_worker:send_to_remote(InstanceId, Msg).
 
 on_query_async(
     _InstId,
@@ -197,8 +196,7 @@ on_query_async(
     #{name := InstanceId}
 ) ->
     ?TRACE("QUERY", "async_send_msg_to_remote_node", #{message => Msg, connector => InstanceId}),
-    emqx_connector_mqtt_worker:send_to_remote_async(InstanceId, Msg, {ReplayFun, Args}),
-    ok.
+    emqx_connector_mqtt_worker:send_to_remote_async(InstanceId, Msg, {ReplayFun, Args}).
 
 on_get_status(_InstId, #{name := InstanceId, bridge_conf := Conf}) ->
     AutoReconn = maps:get(auto_reconnect, Conf, true),

apps/emqx_connector/src/mqtt/emqx_connector_mqtt_worker.erl

@@ -342,7 +342,7 @@ common(_StateName, {call, From}, get_forwards, #{connect_opts := #{forwards := F
 common(_StateName, {call, From}, get_subscriptions, #{connection := Connection}) ->
     {keep_state_and_data, [{reply, From, maps:get(subscriptions, Connection, #{})}]};
 common(_StateName, {call, From}, Req, _State) ->
-    {keep_state_and_data, [{reply, From, {unsuppored_request, Req}}]};
+    {keep_state_and_data, [{reply, From, {error, {unsuppored_request, Req}}}]};
 common(_StateName, info, {'EXIT', _, _}, State) ->
     {keep_state, State};
 common(StateName, Type, Content, #{name := Name} = State) ->

apps/emqx_resource/src/emqx_resource_worker.erl

@@ -585,7 +585,12 @@ assert_ok_result(ok) ->
 assert_ok_result({async_return, R}) ->
     assert_ok_result(R);
 assert_ok_result(R) when is_tuple(R) ->
-    ok = erlang:element(1, R);
+    try
+        ok = erlang:element(1, R)
+    catch
+        error:{badmatch, _} ->
+            error({not_ok_result, R})
+    end;
 assert_ok_result(R) ->
     error({not_ok_result, R}).