Repository: plataformatec/broadway (elixir)

Issue:
Add Broadway.push_messages/2
It is a wrapper around Broadway.Producer.push_messages. We first need to task the Broadway.Server for a name of a producer (it can be randomic) which we return to the client and then the client invokes it.

We also need to validate that the messages are properly formatted (i.e. it has the acknowledger field and what not).

Pull request:
Add Broadway.push_messages/2
Closes #10.

Patch:
diff --git a/lib/broadway.ex b/lib/broadway.ex
index 6cb1268..d570868 100644
--- a/lib/broadway.ex
+++ b/lib/broadway.ex
@@ -297,7 +297,7 @@ defmodule Broadway do
   currently fetched messages are processed and published (consumer).
   """
 
-  alias Broadway.{BatchInfo, Message, Options, Server}
+  alias Broadway.{BatchInfo, Message, Options, Server, Producer}
 
   @doc """
   Invoked to handle/process indiviual messages sent from a producer.
@@ -473,6 +473,18 @@ defmodule Broadway do
     end
   end
 
+  @doc """
+  Sends a list of messages using one of the configured producers.
+
+  The producer is randomly chosen among all sets of producers/stages.
+  """
+  @spec push_messages(GenServer.server(), messages :: [Message.t()]) :: :ok
+  def push_messages(server, messages) do
+    server
+    |> Server.get_random_producer()
+    |> Producer.push_messages(messages)
+  end
+
   defp configuration_spec() do
     [
       name: [required: true, type: :atom],
diff --git a/lib/broadway/consumer.ex b/lib/broadway/consumer.ex
index 08344e7..76e2fa6 100644
--- a/lib/broadway/consumer.ex
+++ b/lib/broadway/consumer.ex
@@ -3,6 +3,8 @@ defmodule Broadway.Consumer do
   use GenStage
   use Broadway.Subscriber
 
+  require Logger
+
   alias Broadway.{Acknowledger, Message}
   @subscription_options [max_demand: 1, min_demand: 0]
 
@@ -32,9 +34,14 @@ defmodule Broadway.Consumer do
 
     {successful_messages, failed_messages} =
       handle_batch(batcher_key, messages, batch_info, state)
-      |> Enum.split_with(&(&1.status == :ok))
 
-    Acknowledger.ack_messages(successful_messages, failed_messages)
+    try do
+      Acknowledger.ack_messages(successful_messages, failed_messages)
+    catch
+      kind, reason ->
+        Logger.error(Exception.format(kind, reason, System.stacktrace()))
+    end
+
     {:noreply, [], state}
   end
 
@@ -43,10 +50,12 @@ defmodule Broadway.Consumer do
 
     try do
       module.handle_batch(batcher_key, messages, batch_info, context)
-    rescue
-      e ->
-        error_message = Exception.message(e)
-        Enum.map(messages, &Message.failed(&1, error_message))
+      |> Enum.split_with(fn %Message{status: status} -> status == :ok end)
+    catch
+      kind, reason ->
+        Logger.error(Exception.format(kind, reason, System.stacktrace()))
+        failed = "due to an unhandled #{kind}"
+        {[], Enum.map(messages, &Message.failed(&1, failed))}
     end
   end
 end
diff --git a/lib/broadway/server.ex b/lib/broadway/server.ex
index d8db94d..9573335 100644
--- a/lib/broadway/server.ex
+++ b/lib/broadway/server.ex
@@ -13,7 +13,14 @@ defmodule Broadway.Server do
     Process.flag(:trap_exit, true)
     config = init_config(module, opts)
     {:ok, supervisor_pid} = start_supervisor(config)
-    {:ok, %{supervisor_pid: supervisor_pid, terminator: config.terminator}}
+
+    {:ok,
+     %{
+       supervisor_pid: supervisor_pid,
+       terminator: config.terminator,
+       name: opts[:name],
+       producers_names: producers_names(opts[:name], config.producers_config)
+     }}
   end
 
   @impl true
@@ -25,6 +32,12 @@ defmodule Broadway.Server do
     {:noreply, state}
   end
 
+  @impl true
+  def handle_call(:get_random_producer, _from, state) do
+    producer = Enum.random(state.producers_names)
+    {:reply, producer, state}
+  end
+
   @impl true
   def terminate(reason, %{supervisor_pid: supervisor_pid, terminator: terminator}) do
     Broadway.Terminator.trap_exit(terminator)
@@ -36,6 +49,10 @@ defmodule Broadway.Server do
     end
   end
 
+  def get_random_producer(server) do
+    GenServer.call(server, :get_random_producer)
+  end
+
   defp reason_to_signal(:killed), do: :kill
   defp reason_to_signal(other), do: other
 
@@ -95,7 +112,7 @@ defmodule Broadway.Server do
 
     names =
       for index <- 1..n_producers do
-        process_name(broadway_name, "Producer_#{key}", index)
+        producer_name(broadway_name, key, index)
       end
 
     specs =
@@ -271,6 +288,16 @@ defmodule Broadway.Server do
     :"#{prefix}.#{type}_#{key}"
   end
 
+  defp producer_name(broadway_name, key, index) do
+    process_name(broadway_name, "Producer_#{key}", index)
+  end
+
+  defp producers_names(broadway_name, producers_config) do
+    for {key, config} <- producers_config, index <- 1..config[:stages] do
+      producer_name(broadway_name, key, index)
+    end
+  end
+
   defp build_producer_supervisor_spec(config, children) do
     name = :"#{config.name}.ProducerSupervisor"
     children_count = length(children)


Test patch:
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index f69c4d2..484c302 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -269,12 +269,10 @@ defmodule BroadwayTest do
     end
   end
 
-  describe "producer" do
-    test "push_messages/2" do
-      broadway = new_unique_name()
-
+  test "push_messages/2" do
+    {:ok, pid} =
       Broadway.start_link(Forwarder,
-        name: broadway,
+        name: new_unique_name(),
         context: %{test_pid: self()},
         producers: [
           default: [module: ManualProducer, arg: []]
@@ -286,16 +284,13 @@ defmodule BroadwayTest do
         ]
       )
 
-      producer = get_producer(broadway)
-
-      Producer.push_messages(producer, [
-        %Message{data: 1, acknowledger: {__MODULE__, 1}},
-        %Message{data: 3, acknowledger: {__MODULE__, 3}}
-      ])
+    Broadway.push_messages(pid, [
+      %Message{data: 1, acknowledger: {__MODULE__, 1}},
+      %Message{data: 3, acknowledger: {__MODULE__, 3}}
+    ])
 
-      assert_receive {:message_handled, 1}
-      assert_receive {:message_handled, 3}
-    end
+    assert_receive {:message_handled, 1}
+    assert_receive {:message_handled, 3}
   end
 
   describe "processor" do
@@ -381,6 +376,7 @@ defmodule BroadwayTest do
 
       producer = get_producer(broadway_name, :default)
       processor = get_processor(broadway_name, :default)
+      Process.link(Process.whereis(processor))
 
       %{producer: producer, processor: processor}
     end
@@ -452,8 +448,7 @@ defmodule BroadwayTest do
       assert capture_log(fn ->
                Producer.push_messages(producer, [one, raise, four])
                assert_receive {:ack, [%{data: 1}, %{data: 4}], []}
-             end) =~
-               "[error] ** (UndefinedFunctionError) function Unknown.ack/2 is undefined"
+             end) =~ "[error] ** (UndefinedFunctionError) function Unknown.ack/2 is undefined"
 
       refute_received {:EXIT, _, ^processor}
     end
@@ -854,6 +849,7 @@ defmodule BroadwayTest do
           case message.data do
             :fail -> Message.failed(message, "Failed message")
             :raise -> raise "Error raised"
+            :bad_return -> :oops
             _ -> message
           end
         end)
@@ -878,21 +874,10 @@ defmodule BroadwayTest do
         )
 
       producer = get_producer(broadway_name, :default)
+      consumer = get_consumer(broadway_name, :default)
+      Process.link(Process.whereis(consumer))
 
-      %{producer: producer}
-    end
-
-    test "all messages in the batch are marked as {:failed, reason} when an error is raised",
-         %{producer: producer} do
-      push_messages(producer, [1, 2, 3, :raise])
-
-      assert_receive {:ack, _,
-                      [
-                        %{data: 1, status: {:failed, "Error raised"}},
-                        %{data: 2, status: {:failed, "Error raised"}},
-                        %{data: 3, status: {:failed, "Error raised"}},
-                        %{data: :raise, status: {:failed, "Error raised"}}
-                      ]}
+      %{producer: producer, consumer: consumer}
     end
 
     test "messages are grouped as successful and failed before sent for acknowledgement",
@@ -901,6 +886,58 @@ defmodule BroadwayTest do
 
       assert_receive {:ack, [%{data: 1}, %{data: 4}], [%{data: :fail}, %{data: :fail}]}
     end
+
+    test "all messages in the batch are marked as {:failed, reason} when an error is raised",
+         %{producer: producer, consumer: consumer} do
+      assert capture_log(fn ->
+               push_messages(producer, [1, 2, :raise, 3])
+
+               assert_receive {:ack, _,
+                               [
+                                 %{data: 1, status: {:failed, "due to an unhandled error"}},
+                                 %{data: 2, status: {:failed, "due to an unhandled error"}},
+                                 %{data: :raise, status: {:failed, "due to an unhandled error"}},
+                                 %{data: 3, status: {:failed, "due to an unhandled error"}}
+                               ]}
+             end) =~ "[error] ** (RuntimeError) Error raised"
+
+      refute_received {:EXIT, _, ^consumer}
+    end
+
+    test "all messages in the batch are marked as {:failed, reason} on bad return",
+         %{producer: producer, consumer: consumer} do
+      assert capture_log(fn ->
+               push_messages(producer, [1, 2, :bad_return, 3])
+
+               assert_receive {:ack, _,
+                               [
+                                 %{data: 1, status: {:failed, "due to an unhandled error"}},
+                                 %{data: 2, status: {:failed, "due to an unhandled error"}},
+                                 %{
+                                   data: :bad_return,
+                                   status: {:failed, "due to an unhandled error"}
+                                 },
+                                 %{data: 3, status: {:failed, "due to an unhandled error"}}
+                               ]}
+             end) =~ "[error]"
+
+      refute_received {:EXIT, _, ^consumer}
+    end
+
+    test "consumers do not crash on bad acknowledger",
+         %{producer: producer, consumer: consumer} do
+      [one, two, raise, three] = wrap_messages([1, 2, :raise, 3])
+      raise = %{raise | acknowledger: {Unknown, :ok}}
+
+      assert capture_log(fn ->
+               Producer.push_messages(producer, [one, two, raise, three])
+               push_messages(producer, [1, 2, 3, 4])
+               assert_receive {:ack, [%{data: 1}, %{data: 2}, %{data: 3}, %{data: 4}], []}
+             end) =~
+               "[error] ** (UndefinedFunctionError) function Unknown.ack/2 is undefined"
+
+      refute_received {:EXIT, _, ^consumer}
+    end
   end
 
   describe "shutdown" do