Packages

rambla

1.4.0
1.5.0 1.4.2 1.4.1 1.4.0 1.3.0 1.2.6 1.2.5 1.2.4 1.2.3 1.2.2 1.2.1 1.2.0 1.1.5 1.1.4 1.1.3 1.1.2 1.1.1 1.1.0 1.0.1 1.0.0 0.16.6 0.16.5 0.16.4 0.16.3 0.16.2 0.16.1 0.16.0 0.15.0 0.14.4 0.14.3 0.14.2 0.14.1 0.14.0 0.13.1 0.13.0 0.12.2 0.12.1 0.11.1 0.11.0 0.10.0 0.9.3 0.9.2 0.9.1 0.9.0 0.8.4 0.8.3 0.8.2 0.8.1 0.8.0 0.7.0 0.6.5 0.6.4 0.6.3 0.6.2 0.6.1 0.6.0 0.5.5 0.5.4 0.5.3 0.5.2 0.5.1 0.5.0 0.4.0 0.3.0 0.2.0 0.1.0

Easy publishing to many different targets. Supported back-ends: - Rabbit [Amqp](https://hexdocs.pm/amqp/) - Redis [Redix](https://hexdocs.pm/redix) - Http [:httpc](http://erlang.org/doc/man/httpc.html) - Smtp [:gen_smtp](https://hexdocs.pm/gen_smtp) - Slack [Envío](https://hexdocs.pm/envio)

HexDocs

Current section

66 Versions

Jump to
Readme
66 Versions
17 Dependencies
0 Dependants
Files
Activity
Hide whitespace

Compare versions

11 files changed
+94 additions
-14 deletions
  @@ -1,6 +1,6 @@
1 1 {<<"links">>,[{<<"GitHub">>,<<"https://github.com/am-kantox/rambla">>}]}.
2 2 {<<"name">>,<<"rambla">>}.
3 - {<<"version">>,<<"1.3.0">>}.
3 + {<<"version">>,<<"1.4.0">>}.
4 4 {<<"description">>,
5 5 <<69,97,115,121,32,112,117,98,108,105,115,104,105,110,103,32,116,111,32,109,
6 6 97,110,121,32,100,105,102,102,101,114,101,110,116,32,116,97,114,103,101,
  @@ -232,16 +232,47 @@ defmodule Rambla.Handler do
232 232 end
233 233
234 234 @impl Finitomata.Pool.Actor
235 - def actor(payload, %{} = state) when is_list(payload) do
236 - if Keyword.keyword?(payload),
237 - do: payload |> Map.new() |> actor(state),
238 - else: actor(payload, state)
239 - end
240 -
241 235 def actor(%RetryState{payload: payload, retries: retries}, state) do
242 236 actor(payload, %{state | retries: retries})
243 237 end
244 238
239 + def actor(payload, %{} = state) when is_list(payload) do
240 + payload
241 + |> Enum.group_by(&extract_options(&1, state))
242 + |> Enum.reduce(nil, fn {options, payloads}, acc ->
243 + case {do_one_actor(payloads, options, state, state.retries), acc} do
244 + {{any, {on, res}}, nil} ->
245 + {any, {on, [res]}}
246 +
247 + {{:ok, {on_result, result}}, {:ok, {on_result, results}}} ->
248 + {:ok, on_result, [result | results]}
249 +
250 + {{:ok, {on_result_new, result}}, {:ok, {on_result_old, results}}} ->
251 + Logger.warning(
252 + "[🖇️] #{__MODULE__}[⚡] → Bulk has divergent callbacks, using the last one"
253 + )
254 +
255 + {:ok, on_result_new, [result | results]}
256 +
257 + {{:error, {on_error, error}}, {:error, {on_error, errors}}} ->
258 + {:error, on_error, [error | errors]}
259 +
260 + {{:error, {on_error_new, error}}, {:error, {on_error_old, errors}}} ->
261 + Logger.warning(
262 + "[🖇️] #{__MODULE__}[⚡] → Bulk has divergent error callbacks, using the last one"
263 + )
264 +
265 + {:error, on_error_new, [error | errors]}
266 +
267 + {{:error, {on_error, error}}, _} ->
268 + {:error, on_error, [error]}
269 +
270 + {_, acc} ->
271 + acc
272 + end
273 + end)
274 + end
275 +
245 276 def actor(payload, state) do
246 277 options = extract_options(payload, state)
247 278
  @@ -250,6 +281,10 @@ defmodule Rambla.Handler do
250 281 do: Map.pop(payload, :retries, state.retries),
251 282 else: {state.retries, payload}
252 283
284 + do_one_actor(payload, options, state, retries)
285 + end
286 +
287 + defp do_one_actor(payload, options, state, retries) do
253 288 on_result = get_in(options, [:callbacks, :on_success])
254 289 on_error = get_in(options, [:callbacks, :on_failure])
255 290
  @@ -362,10 +397,6 @@ defmodule Rambla.Handler do
362 397 """
363 398 def publish(id, payload, pid \\ nil)
364 399
365 - def publish(id, payloads, pid) when is_list(payloads) do
366 - Enum.each(payloads, &publish(id, &1, pid))
367 - end
368 -
369 400 def publish(id, payload, pid) do
370 401 id |> fqn_id() |> Finitomata.Pool.run(payload, pid)
371 402 end
  @@ -33,6 +33,9 @@ if :amqp in Rambla.services() do
33 33
34 34 @impl Rambla.Handler
35 35 @doc false
36 + def handle_publish(messages, options, state) when is_list(messages),
37 + do: Enum.each(messages, &handle_publish(&1, options, state))
38 +
36 39 def handle_publish(%{message: message}, options, %{connection: %{channel: name}}) do
37 40 {exchange, options} = Map.pop(options, :exchange, "")
38 41 {declare?, options} = Map.pop(options, :declare?, false)
  @@ -66,6 +66,27 @@ if :clickhouse in Rambla.services() do
66 66
67 67 @impl Rambla.Handler
68 68 @doc false
69 + def handle_publish(
70 + messages,
71 + options,
72 + %{connection: %{params: conn_params, channel: name}} = state
73 + )
74 + when is_list(messages) do
75 + case Map.get(unquote(Macro.escape(connections)), Keyword.get(conn_params, :connection)) do
76 + nil ->
77 + Enum.each(messages, &handle_publish(&1, options, state))
78 +
79 + mod ->
80 + messages =
81 + Enum.map(messages, fn
82 + %{message: message} -> message
83 + message -> message
84 + end)
85 +
86 + do_handle_publish(mod, messages, options, name)
87 + end
88 + end
89 +
69 90 def handle_publish(%{message: message}, options, state) when is_binary(message) do
70 91 {serializer, options} = Map.pop(options, :serializer, Jason)
  @@ -35,6 +35,9 @@ if :httpc in Rambla.services() do
35 35
36 36 @impl Rambla.Handler
37 37 @doc false
38 + def handle_publish(messages, options, state) when is_list(messages),
39 + do: Enum.each(messages, &handle_publish(&1, options, state))
40 +
38 41 def handle_publish(%{message: message}, options, %{connection: %{channel: name}}) do
39 42 conn = config() |> get_in([:channels, name, :connection])
Loading more files…