yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(bridge): Bridges should send a JSON message if payload template not set

This commit is contained in:
Shawn 2021-12-28 14:13:26 +08:00
parent 4406589980
commit 07997ab865
2 changed files with 13 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
apps/emqx_connector/src/emqx_connector_http.erl
View File

@ -266,11 +266,16 @@ process_request(#{
} = Conf, Msg) -> } = Conf, Msg) ->
Conf#{ method => make_method(emqx_plugin_libs_rule:proc_tmpl(MethodTks, Msg)) Conf#{ method => make_method(emqx_plugin_libs_rule:proc_tmpl(MethodTks, Msg))
, path => emqx_plugin_libs_rule:proc_tmpl(PathTks, Msg) , path => emqx_plugin_libs_rule:proc_tmpl(PathTks, Msg)
, body => emqx_plugin_libs_rule:proc_tmpl(BodyTks, Msg) , body => process_request_body(BodyTks, Msg)
, headers => maps:to_list(proc_headers(HeadersTks, Msg)) , headers => maps:to_list(proc_headers(HeadersTks, Msg))
, request_timeout => ReqTimeout , request_timeout => ReqTimeout
}. }.
process_request_body([], Msg) ->
emqx_json:encode(Msg);
process_request_body(BodyTks, Msg) ->
emqx_plugin_libs_rule:proc_tmpl(BodyTks, Msg).
proc_headers(HeaderTks, Msg) -> proc_headers(HeaderTks, Msg) ->
maps:fold(fun(K, V, Acc) -> maps:fold(fun(K, V, Acc) ->
Acc#{emqx_plugin_libs_rule:proc_tmpl(K, Msg) => Acc#{emqx_plugin_libs_rule:proc_tmpl(K, Msg) =>

9
apps/emqx_connector/src/mqtt/emqx_connector_mqtt_msg.erl
View File

@ -66,7 +66,7 @@ to_remote_msg(#message{flags = Flags0} = Msg, Vars) ->
to_remote_msg(MapMsg, #{remote_topic := TopicToken, payload := PayloadToken, to_remote_msg(MapMsg, #{remote_topic := TopicToken, payload := PayloadToken,
remote_qos := QoSToken, retain := RetainToken, mountpoint := Mountpoint}) when is_map(MapMsg) -> remote_qos := QoSToken, retain := RetainToken, mountpoint := Mountpoint}) when is_map(MapMsg) ->
Topic = replace_vars_in_str(TopicToken, MapMsg), Topic = replace_vars_in_str(TopicToken, MapMsg),
Payload = replace_vars_in_str(PayloadToken, MapMsg), Payload = process_payload(PayloadToken, MapMsg),
QoS = replace_simple_var(QoSToken, MapMsg), QoS = replace_simple_var(QoSToken, MapMsg),
Retain = replace_simple_var(RetainToken, MapMsg), Retain = replace_simple_var(RetainToken, MapMsg),
#mqtt_msg{qos = QoS, #mqtt_msg{qos = QoS,
@ -82,13 +82,18 @@ to_broker_msg(#{dup := Dup, properties := Props} = MapMsg,
#{local_topic := TopicToken, payload := PayloadToken, #{local_topic := TopicToken, payload := PayloadToken,
local_qos := QoSToken, retain := RetainToken, mountpoint := Mountpoint}) -> local_qos := QoSToken, retain := RetainToken, mountpoint := Mountpoint}) ->
Topic = replace_vars_in_str(TopicToken, MapMsg), Topic = replace_vars_in_str(TopicToken, MapMsg),
Payload = replace_vars_in_str(PayloadToken, MapMsg), Payload = process_payload(PayloadToken, MapMsg),
QoS = replace_simple_var(QoSToken, MapMsg), QoS = replace_simple_var(QoSToken, MapMsg),
Retain = replace_simple_var(RetainToken, MapMsg), Retain = replace_simple_var(RetainToken, MapMsg),
set_headers(Props, set_headers(Props,
emqx_message:set_flags(#{dup => Dup, retain => Retain}, emqx_message:set_flags(#{dup => Dup, retain => Retain},
emqx_message:make(bridge, QoS, topic(Mountpoint, Topic), Payload))). emqx_message:make(bridge, QoS, topic(Mountpoint, Topic), Payload))).
process_payload([], Msg) ->
emqx_json:encode(Msg);
process_payload(Tks, Msg) ->
replace_vars_in_str(Tks, Msg).
%% Replace a string contains vars to another string in which the placeholders are replace by the %% Replace a string contains vars to another string in which the placeholders are replace by the
%% corresponding values. For example, given "a: ${var}", if the var=1, the result string will be: %% corresponding values. For example, given "a: ${var}", if the var=1, the result string will be:
%% "a: 1". %% "a: 1".