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

refactor: make api more generic

This commit is contained in:
Thales Macedo Garitezi 2024-05-03 17:25:02 -03:00
parent a6508e2fca
commit 157e2c2535
2 changed files with 16 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
apps/emqx_bridge_s3/src/emqx_bridge_s3_connector.erl
View File

@ -409,18 +409,19 @@ init_transfer_state(Buffer, Opts) ->
mk_object_key(Buffer, #{action := Name, key := Template}) -> mk_object_key(Buffer, #{action := Name, key := Template}) ->
emqx_template:render_strict(Template, {?MODULE, {Name, Buffer}}). emqx_template:render_strict(Template, {?MODULE, {Name, Buffer}}).
process_append(Writes, Upload) -> process_append(Writes, Upload0) ->
emqx_s3_upload:append(Writes, Upload). {ok, Upload} = emqx_s3_upload:append(Writes, Upload0),
Upload.
process_write(Upload0) -> process_write(Upload0) ->
case emqx_s3_upload:write(Upload0) of case emqx_s3_upload:write(Upload0) of
{ok, Upload} -> {ok, Upload} ->
Upload; {ok, Upload};
{cont, Upload} -> {cont, Upload} ->
process_write(Upload); process_write(Upload);
{error, Reason} -> {error, Reason} ->
_ = emqx_s3_upload:abort(Upload0), _ = emqx_s3_upload:abort(Upload0),
exit({upload_failed, Reason}) {error, Reason}
end. end.
process_complete(Upload) -> process_complete(Upload) ->

17
apps/emqx_connector_aggregator/src/emqx_connector_aggreg_delivery.erl
View File

@ -44,9 +44,9 @@
-callback init_transfer_state(buffer(), map()) -> transfer_state(). -callback init_transfer_state(buffer(), map()) -> transfer_state().
-callback process_append(iodata(), transfer_state()) -> {ok, transfer_state()}. -callback process_append(iodata(), transfer_state()) -> transfer_state().
-callback process_write(transfer_state()) -> transfer_state(). -callback process_write(transfer_state()) -> {ok, transfer_state()} | {error, term()}.
-callback process_complete(transfer_state()) -> {ok, term()}. -callback process_complete(transfer_state()) -> {ok, term()}.
@ -134,7 +134,7 @@ process_append_records(
} }
) -> ) ->
{Writes, Container} = emqx_connector_aggreg_csv:fill(Records, Container0), {Writes, Container} = emqx_connector_aggreg_csv:fill(Records, Container0),
{ok, Transfer} = Mod:process_append(Writes, Transfer0), Transfer = Mod:process_append(Writes, Transfer0),
Delivery#delivery{ Delivery#delivery{
container = Container, container = Container,
transfer = Transfer, transfer = Transfer,
@ -142,8 +142,13 @@ process_append_records(
}. }.
process_write(Delivery = #delivery{callback_module = Mod, transfer = Transfer0}) -> process_write(Delivery = #delivery{callback_module = Mod, transfer = Transfer0}) ->
Transfer = Mod:process_write(Transfer0), case Mod:process_write(Transfer0) of
Delivery#delivery{transfer = Transfer}. {ok, Transfer} ->
Delivery#delivery{transfer = Transfer};
{error, Reason} ->
%% Todo: handle more gracefully? Retry?
error({transfer_failed, Reason})
end.
process_complete(#delivery{name = Name, empty = true}) -> process_complete(#delivery{name = Name, empty = true}) ->
?tp(connector_aggreg_delivery_completed, #{action => Name, transfer => empty}), ?tp(connector_aggreg_delivery_completed, #{action => Name, transfer => empty}),
@ -152,7 +157,7 @@ process_complete(#delivery{
name = Name, callback_module = Mod, container = Container, transfer = Transfer0 name = Name, callback_module = Mod, container = Container, transfer = Transfer0
}) -> }) ->
Trailer = emqx_connector_aggreg_csv:close(Container), Trailer = emqx_connector_aggreg_csv:close(Container),
{ok, Transfer} = Mod:process_append(Trailer, Transfer0), Transfer = Mod:process_append(Trailer, Transfer0),
{ok, Completed} = Mod:process_complete(Transfer), {ok, Completed} = Mod:process_complete(Transfer),
?tp(connector_aggreg_delivery_completed, #{action => Name, transfer => Completed}), ?tp(connector_aggreg_delivery_completed, #{action => Name, transfer => Completed}),
ok. ok.