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

feat: add stop after render and after render trace to pulsar action

This commit is contained in:
Kjell Winblad 2024-04-24 14:24:52 +02:00
parent 9c37c99b62
commit 74fac80e7e
1 changed files with 11 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
apps/emqx_bridge_pulsar/src/emqx_bridge_pulsar_connector.erl
View File

@ -196,6 +196,11 @@ on_query(_InstanceId, {ChannelId, Message}, State) ->
{error, channel_not_found}; {error, channel_not_found};
{ok, #{message := MessageTmpl, sync_timeout := SyncTimeout, producers := Producers}} -> {ok, #{message := MessageTmpl, sync_timeout := SyncTimeout, producers := Producers}} ->
PulsarMessage = render_message(Message, MessageTmpl), PulsarMessage = render_message(Message, MessageTmpl),
emqx_trace:rendered_action_template(ChannelId, #{
message => PulsarMessage,
sync_timeout => SyncTimeout,
is_async => false
}),
try try
pulsar:send_sync(Producers, [PulsarMessage], SyncTimeout) pulsar:send_sync(Producers, [PulsarMessage], SyncTimeout)
catch catch
@ -217,12 +222,16 @@ on_query_async(_InstanceId, {ChannelId, Message}, AsyncReplyFn, State) ->
?tp_span( ?tp_span(
pulsar_producer_on_query_async, pulsar_producer_on_query_async,
#{instance_id => _InstanceId, message => Message}, #{instance_id => _InstanceId, message => Message},
on_query_async2(Producers, Message, MessageTmpl, AsyncReplyFn) on_query_async2(ChannelId, Producers, Message, MessageTmpl, AsyncReplyFn)
) )
end. end.
on_query_async2(Producers, Message, MessageTmpl, AsyncReplyFn) -> on_query_async2(ChannelId, Producers, Message, MessageTmpl, AsyncReplyFn) ->
PulsarMessage = render_message(Message, MessageTmpl), PulsarMessage = render_message(Message, MessageTmpl),
emqx_trace:rendered_action_template(ChannelId, #{
message => PulsarMessage,
is_async => true
}),
pulsar:send(Producers, [PulsarMessage], #{callback_fn => AsyncReplyFn}). pulsar:send(Producers, [PulsarMessage], #{callback_fn => AsyncReplyFn}).
%%------------------------------------------------------------------------------------- %%-------------------------------------------------------------------------------------