Repository: plataformatec/broadway
Pull request #45 (2019-02-15 12:55:20)

Update docs


Patch:
diff --git a/guides/examples/Custom Producers.md b/guides/examples/Custom Producers.md
index d787dc0..6e9ae6f 100644
--- a/guides/examples/Custom Producers.md
+++ b/guides/examples/Custom Producers.md
@@ -1,7 +1,7 @@
 # Custom Producers
 
 If you want to use Broadway but there is no existing Broadway producer
-for your technology of your choice, you can integrate any existing GenStage
+for the technology of your choice, you can integrate any existing GenStage
 producer into the pipeline with relative ease.
 
 ## Example
@@ -73,9 +73,9 @@ generated by the producer into proper Broadway messages:
       end
     end
 
-Notice that you need pass three options to the producer:
+Notice that you need to pass three options to the producer:
 
-  * `:module` - the module of that implements the GenStage behaviour.
+  * `:module` - the module that implements the GenStage behaviour.
 
   * `:arg` - the argument that will be given to the `init` callback
     of the GenStage. It is very important to note that Broadway
diff --git a/guides/internals/Architecture.md b/guides/internals/Architecture.md
index 0ce84b3..6522989 100644
--- a/guides/internals/Architecture.md
+++ b/guides/internals/Architecture.md
@@ -8,7 +8,7 @@ own our pipeline.
 
 ## The pipeline model
 
-```
+```asciidoc
                                    [producers]   <- pulls data from SQS, RabbitMQ, etc.
                                         |
                                         |   (demand dispatcher)
@@ -41,7 +41,7 @@ Broadway was designed to always go back to a working state in case
 of failures thanks to the use of supervisors. Our supervision tree
 is designed as follows:
 
-```
+```asciidoc
                         [Broadway GenServer]
                                  |
                                  |
@@ -91,7 +91,7 @@ shutdown. We know that either all processes are running OR they are all
 being shutdown. Therefore, to gracefully shutdown the supervision tree,
 a terminator process is activated, which starts the following steps:
 
-  1. It notifies the first layer of processors that they should no
+  1. It notifies the first layer of processors that they should not
      resubscribe to producers once they exit
 
   2. It tells all producers to no longer accept demand, flush all
diff --git a/lib/broadway.ex b/lib/broadway.ex
index 9cdf2f3..5c36d08 100644
--- a/lib/broadway.ex
+++ b/lib/broadway.ex
@@ -84,7 +84,7 @@ defmodule Broadway do
 
   Here is how this pipeline would be represented:
 
-  ```
+  ```asciidoc
                        [producer_1]
                            / \
                           /   \
@@ -133,7 +133,7 @@ defmodule Broadway do
           |> Message.put_batcher(:sqs)
         end
 
-        def handle_message(_, %Message{data: data} = message, _context) is_even(data) do
+        def handle_message(_, %Message{data: data} = message, _) when is_even(data) do
           message
           |> Message.update_data(&process_data/1)
           |> Message.put_batcher(:s3)
@@ -174,13 +174,42 @@ defmodule Broadway do
 
   Note however, that `Broadway` does not provide any sort of retries
   out of the box. This is left completely as a responsibility of the
-  producer. For instance, if you are using Amazon SQS and you would
-  like unacknowledged messages to be retried, it is your responsibility
-  to configure a dead-letter queue and set the relevant timeouts.
+  producer. For instance, if you are using Amazon SQS, the default
+  behaviour is to retry unacknowledged messages after a user-defined
+  timeout. If you don't want unacknowledged messages to be retried,
+  is your responsibility to configure a dead-letter queue as target
+  for those messages.
 
   ## Testing
 
-  TODO.
+  Testing Broadway pipelines can be done with `test_messages/2`.
+  With `test_messages/2`, you can push some sample data into the
+  pipeline and receive a process message when the pipeline
+  acknowledges the data you have pushed has been processed.
+  This is very useful as a synchronization mechanism. Because
+  many pipelines end-up working with side-effects, you can use
+  the test message acknowledgment to guarantee the message has
+  been processed and therefore side-effects should be visible.
+
+  For example, if you have a pipeline named `MyApp.Broadway` that
+  writes to the database on every message, you could test it as:
+
+      # Push 3 messages with the data field set to 1, 2, and 3 respectively
+      ref = Broadway.test_messages(MyApp.Broadway, [1, 2, 3])
+
+      # Assert that the messages have been consumed
+      assert_receive {:ack, ^ref, [_, _, _] = _successful, failed}
+
+      # Now assert the database side-effects
+      ...
+
+  Keep in mind that multiple acknowledgement messages may be sent.
+  For example, if the batcher in the example above has size of 2,
+  then two batches would be created and therefore two ack messages
+  would be sent. Similarly, if any of the messages fail when
+  processed, an acknowledgement of their failure may be sent early
+  on. On the positive side, if you always push just a single test
+  message, then there is always one acknowledgment.
   """
 
   alias Broadway.{BatchInfo, Message, Options, Server, Producer}
@@ -380,17 +409,49 @@ defmodule Broadway do
   end
 
   @doc """
-  Sends a list of messages using one of the configured producers.
+  Sends a list of `Broadway.Message`s to the Broadway pipeline.
 
   The producer is randomly chosen among all sets of producers/stages.
+  This is used to send out of band data to a Broadway pipeline.
   """
   @spec push_messages(GenServer.server(), messages :: [Message.t()]) :: :ok
-  def push_messages(server, messages) do
-    server
+  def push_messages(broadway, messages) when is_list(messages) do
+    broadway
     |> Server.get_random_producer()
     |> Producer.push_messages(messages)
   end
 
+  @doc """
+  Sends a list of data as messsages to the Broadway pipeline.
+
+  This is a convenience used mostly for testing. The given data
+  is automaticaly wrapped in a `Broadway.Message` with
+  `Broadway.CallerAcknowledger` configured to send a message
+  back to the caller once the message has been fully processed.
+  It uses `push_messages/2` for dispatching.
+
+  It returns a reference that can be used to identify the ack
+  messages.
+
+  ## Examples
+
+  For example, in your tests, you may do:
+
+      ref = Broadway.test_messages(broadway, [1, 2, 3])
+      assert_receive {:ack, ^ref, successful, failed}
+      assert length(successful) == 3
+      assert length(failed) == 0
+
+  """
+  @spec test_messages(GenServer.server(), data :: [term]) :: reference()
+  def test_messages(broadway, data) when is_list(data) do
+    ref = make_ref()
+    ack = {Broadway.CallerAcknowledger, {self(), ref}, :ok}
+    messages = Enum.map(data, &%Message{data: &1, acknowledger: ack})
+    :ok = push_messages(broadway, messages)
+    ref
+  end
+
   defp configuration_spec() do
     [
       name: [required: true, type: :atom],
diff --git a/lib/broadway/acknowledger.ex b/lib/broadway/acknowledger.ex
index 98d4356..c2ac764 100644
--- a/lib/broadway/acknowledger.ex
+++ b/lib/broadway/acknowledger.ex
@@ -25,7 +25,7 @@ defmodule Broadway.Acknowledger do
       you have a scenario where messages are coming from
       different producers. Broadway will use this information
       to correctly identify the acknowledger and pass it among
-      with the messages so you can property communicate with
+      with the messages so you can properly communicate with
       the source of the data for acknowledgement.
 
     * `successful` is the list of messages that were
@@ -38,7 +38,7 @@ defmodule Broadway.Acknowledger do
   @callback ack(ack_ref :: reference, successful :: [Message.t()], failed :: [Message.t()]) :: :ok
 
   @doc """
-  Acknowledges successful and failed messages grouped by acknowledger.
+  Acknowledges successful and failed messages grouped by `{acknowledger, ack_ref}`.
   """
   @spec ack_messages([Message.t()], [Message.t()]) :: no_return
   def ack_messages(successful, failed) do
diff --git a/lib/broadway/caller_acknowledger.ex b/lib/broadway/caller_acknowledger.ex
new file mode 100644
index 0000000..e0c317c
--- /dev/null
+++ b/lib/broadway/caller_acknowledger.ex
@@ -0,0 +1,24 @@
+defmodule Broadway.CallerAcknowledger do
+  @moduledoc """
+  A simple acknowledger that sends a message back to a caller.
+
+  It must be stored as:
+
+      acknowledger: {Broadway.CallerAcknowledger, {pid, ref}, term}
+
+  The second element is a tuple with the pid to receive the messages
+  and a unique identifier (usually a reference). The third element,
+  which is per message, is ignored.
+
+  It sends a message in the format:
+
+      {:ack, ref, successful_messages, failed_messages}
+  """
+
+  @behaviour Broadway.Acknowledger
+
+  @impl true
+  def ack({pid, ref}, successful, failed) do
+    send(pid, {:ack, ref, successful, failed})
+  end
+end


Test patch:
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index f15a088..2305b0f 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -4,7 +4,7 @@ defmodule BroadwayTest do
   import Integer
   import ExUnit.CaptureLog
 
-  alias Broadway.{Producer, Message, BatchInfo}
+  alias Broadway.{Message, BatchInfo, CallerAcknowledger}
 
   defmodule Acker do
     @behaviour Broadway.Acknowledger
@@ -298,7 +298,7 @@ defmodule BroadwayTest do
 
       broadway_name = new_unique_name()
 
-      {:ok, _pid} =
+      {:ok, broadway} =
         Broadway.start_link(CustomHandlers,
           name: broadway_name,
           context: context,
@@ -309,55 +309,54 @@ defmodule BroadwayTest do
           batchers: [default: [batch_size: 2]]
         )
 
-      producer = get_producer(broadway_name, :default)
       processor = get_processor(broadway_name, :default)
       Process.link(Process.whereis(processor))
-
-      %{producer: producer, processor: processor}
+      %{broadway: broadway, processor: processor}
     end
 
-    test "successful messages are marked as :ok", %{producer: producer} do
-      push_messages(producer, [1, 2])
+    test "successful messages are marked as :ok", %{broadway: broadway} do
+      ref = Broadway.test_messages(broadway, [1, 2])
       assert_receive {:batch_handled, [%{status: :ok}, %{status: :ok}]}
+      assert_receive {:ack, ^ref, [%{status: :ok}, %{status: :ok}], []}
     end
 
-    test "failed messages are marked as {:failed, reason}", %{producer: producer} do
-      push_messages(producer, [:fail])
-      assert_receive {:ack, _, [%{status: {:failed, "Failed message"}}]}
-    end
-
-    test "failed messages are not forwarded to the batcher", %{producer: producer} do
-      push_messages(producer, [1, :fail, :fail, 4])
-
-      assert_receive {:batch_handled, batch}
-      assert Enum.map(batch, & &1.data) == [1, 4]
+    test "failed messages are marked as {:failed, reason}", %{broadway: broadway} do
+      ref = Broadway.test_messages(broadway, [:fail])
+      assert_receive {:ack, ^ref, _, [%{status: {:failed, "Failed message"}}]}
     end
 
-    test "messages are grouped as successful and failed before sent for acknowledgement",
-         %{producer: producer} do
-      push_messages(producer, [1, :fail, 4])
+    test "failed messages are not forwarded to the batcher", %{broadway: broadway} do
+      ref = Broadway.test_messages(broadway, [1, :fail, :fail, 4])
 
-      assert_receive {:ack, [], [%{data: :fail}]}
-      assert_receive {:ack, [%{data: 1}, %{data: 4}], []}
+      assert_receive {:batch_handled, [%{data: 1}, %{data: 4}]}
+      assert_receive {:ack, ^ref, [%{status: :ok}, %{status: :ok}], []}
+      assert_receive {:ack, ^ref, _, [%{status: {:failed, "Failed message"}}]}
+      assert_receive {:ack, ^ref, _, [%{status: {:failed, "Failed message"}}]}
     end
 
     test "messages are marked as {:failed, reason} when an error is raised while processing",
-         %{producer: producer, processor: processor} do
+         %{broadway: broadway, processor: processor} do
       assert capture_log(fn ->
-               push_messages(producer, [1, :raise, 4])
-               assert_receive {:ack, _, [%{status: {:failed, "due to an unhandled error"}}]}
-               assert_receive {:ack, [%{data: 1}, %{data: 4}], []}
+               ref = Broadway.test_messages(broadway, [1, :raise, 4])
+
+               assert_receive {:ack, ^ref, [],
+                               [%{status: {:failed, "due to an unhandled error"}}]}
+
+               assert_receive {:ack, ^ref, [%{data: 1}, %{data: 4}], []}
              end) =~ "[error] ** (RuntimeError) Error raised"
 
       refute_received {:EXIT, _, ^processor}
     end
 
     test "messages are marked as {:failed, reason} on bad return",
-         %{producer: producer, processor: processor} do
+         %{broadway: broadway, processor: processor} do
       assert capture_log(fn ->
-               push_messages(producer, [1, :bad_return, 4])
-               assert_receive {:ack, _, [%{status: {:failed, "due to an unhandled error"}}]}
-               assert_receive {:ack, [%{data: 1}, %{data: 4}], []}
+               ref = Broadway.test_messages(broadway, [1, :bad_return, 4])
+
+               assert_receive {:ack, ^ref, [],
+                               [%{status: {:failed, "due to an unhandled error"}}]}
+
+               assert_receive {:ack, ^ref, [%{data: 1}, %{data: 4}], []}
              end) =~
                "[error] ** (RuntimeError) expected a Broadway.Message from handle_message/3, got :oops"
 
@@ -365,10 +364,14 @@ defmodule BroadwayTest do
     end
 
     test "messages are marked as {:failed, reason} on bad batcher",
-         %{producer: producer, processor: processor} do
+         %{broadway: broadway, processor: processor} do
       assert capture_log(fn ->
-               push_messages(producer, [1, :bad_batcher, 4])
-               assert_receive {:ack, _, [%{status: {:failed, "due to an unhandled error"}}]}
+               ref = Broadway.test_messages(broadway, [1, :bad_batcher, 4])
+
+               assert_receive {:ack, ^ref, [],
+                               [%{status: {:failed, "due to an unhandled error"}}]}
+
+               assert_receive {:ack, ^ref, [%{data: 1}, %{data: 4}], []}
              end) =~
                "[error] ** (RuntimeError) message was set to unknown batcher :unknown. The known batchers are [:default]"
 
@@ -376,13 +379,13 @@ defmodule BroadwayTest do
     end
 
     test "processors do not crash on bad acknowledger",
-         %{producer: producer, processor: processor} do
+         %{broadway: broadway, processor: processor} do
       [one, raise, four] = wrap_messages([1, :raise, 4])
       raise = %{raise | acknowledger: {Unknown, :ack_ref, :ok}}
 
       assert capture_log(fn ->
-               Producer.push_messages(producer, [one, raise, four])
-               assert_receive {:ack, [%{data: 1}, %{data: 4}], []}
+               Broadway.push_messages(broadway, [one, raise, four])
+               assert_receive {:ack, _, [%{data: 1}, %{data: 4}], []}
              end) =~ "[error] ** (UndefinedFunctionError) function Unknown.ack/3 is undefined"
 
       refute_received {:EXIT, _, ^processor}
@@ -391,7 +394,7 @@ defmodule BroadwayTest do
 
   describe "put_batcher" do
     setup do
-      broadway = new_unique_name()
+      broadway_name = new_unique_name()
       test_pid = self()
 
       handle_message = fn message, _ ->
@@ -410,9 +413,9 @@ defmodule BroadwayTest do
         end
       }
 
-      {:ok, _pid} =
+      {:ok, broadway} =
         Broadway.start_link(CustomHandlers,
-          name: broadway,
+          name: broadway_name,
           context: context,
           producers: [
             default: [module: ManualProducer, arg: []]
@@ -424,11 +427,11 @@ defmodule BroadwayTest do
           ]
         )
 
-      %{producer: get_producer(broadway)}
+      %{broadway: broadway}
     end
 
-    test "generate batches based on :batch_size", %{producer: producer} do
-      push_messages(producer, 1..40)
+    test "generate batches based on :batch_size", %{broadway: broadway} do
+      Broadway.test_messages(broadway, Enum.to_list(1..40))
 
       assert_receive {:batch_handled, :odd, messages} when length(messages) == 10
       assert_receive {:batch_handled, :odd, messages} when length(messages) == 10
@@ -440,8 +443,8 @@ defmodule BroadwayTest do
     end
 
     test "generate batches with the remaining messages after :batch_timeout is reached",
-         %{producer: producer} do
-      push_messages(producer, 1..5)
+         %{broadway: broadway} do
+      Broadway.test_messages(broadway, [1, 2, 3, 4, 5])
 
       assert_receive {:batch_handled, :odd, messages} when length(messages) == 3
       assert_receive {:batch_handled, :even, messages} when length(messages) == 2
@@ -451,7 +454,7 @@ defmodule BroadwayTest do
 
   describe "put_partition" do
     setup do
-      broadway = new_unique_name()
+      broadway_name = new_unique_name()
 
       handle_message = fn message, _ ->
         if is_odd(message.data) do
@@ -466,31 +469,33 @@ defmodule BroadwayTest do
         handle_batch: fn _, batch, _, _ -> batch end
       }
 
-      {:ok, _pid} =
+      {:ok, broadway} =
         Broadway.start_link(CustomHandlers,
-          name: broadway,
+          name: broadway_name,
           context: context,
           producers: [default: [module: ManualProducer, arg: []]],
           processors: [default: []],
           batchers: [default: [batch_size: 2, batch_timeout: 20]]
         )
 
-      %{producer: get_producer(broadway)}
+      %{broadway: broadway}
     end
 
-    test "generate batches based on :batch_size", %{producer: producer} do
-      push_messages(producer, 1..10)
+    test "generate batches based on :batch_size", %{broadway: broadway} do
+      ref = Broadway.test_messages(broadway, Enum.to_list(1..10))
 
-      assert_receive {:ack, [%{data: 1, partition: :odd}, %{data: 3, partition: :odd}], []}
-      assert_receive {:ack, [%{data: 2, partition: :even}, %{data: 4, partition: :even}], []}
+      assert_receive {:ack, ^ref, [%{data: 1, partition: :odd}, %{data: 3, partition: :odd}], []}
+
+      assert_receive {:ack, ^ref, [%{data: 2, partition: :even}, %{data: 4, partition: :even}],
+                      []}
     end
 
     test "generate batches with the remaining messages after :batch_timeout is reached",
-         %{producer: producer} do
-      push_messages(producer, 1..2)
+         %{broadway: broadway} do
+      ref = Broadway.test_messages(broadway, [1, 2])
 
-      assert_receive {:ack, [%{data: 1, partition: :odd}], []}
-      assert_receive {:ack, [%{data: 2, partition: :even}], []}
+      assert_receive {:ack, ^ref, [%{data: 1, partition: :odd}], []}
+      assert_receive {:ack, ^ref, [%{data: 2, partition: :even}], []}
     end
   end
 
@@ -554,7 +559,7 @@ defmodule BroadwayTest do
 
       broadway_name = new_unique_name()
 
-      {:ok, _pid} =
+      {:ok, broadway} =
         Broadway.start_link(CustomHandlers,
           name: broadway_name,
           resubscribe_interval: 0,
@@ -574,37 +579,41 @@ defmodule BroadwayTest do
       batcher = get_batcher(broadway_name, :default)
       consumer = get_consumer(broadway_name, :default)
 
-      %{producer: producer, processor: processor, batcher: batcher, consumer: consumer}
+      %{
+        broadway: broadway,
+        producer: producer,
+        processor: processor,
+        batcher: batcher,
+        consumer: consumer
+      }
     end
 
-    test "only the producer will be restarted", context do
-      %{producer: producer, processor: processor, batcher: batcher, consumer: consumer} = context
-
+    test "only the producer will be restarted",
+         %{producer: producer, processor: processor, batcher: batcher, consumer: consumer} do
       ref_producer = Process.monitor(producer)
       ref_batcher = Process.monitor(batcher)
       ref_processor = Process.monitor(processor)
       ref_consumer = Process.monitor(consumer)
-
       GenStage.stop(producer)
 
       assert_receive {:DOWN, ^ref_producer, _, _, _}
       refute_receive {:DOWN, ^ref_processor, _, _, _}
       refute_receive {:DOWN, ^ref_batcher, _, _, _}
       refute_receive {:DOWN, ^ref_consumer, _, _, _}
-
       assert_receive {:producer_initialized, ^producer}
     end
 
-    test "processors resubscribe to the restarted producers and keep processing messages" do
-      assert_receive {:producer_initialized, producer}
+    test "processors resubscribe to the restarted producers and keep processing messages",
+         %{broadway: broadway, producer: producer} do
+      assert_receive {:producer_initialized, ^producer}
 
-      push_messages(producer, [1])
+      Broadway.test_messages(broadway, [1])
       assert_receive {:message_handled, %{data: 1}}
 
       GenStage.stop(producer)
       assert_receive {:producer_initialized, ^producer}
 
-      push_messages(producer, [2])
+      Broadway.test_messages(broadway, [2])
       assert_receive {:message_handled, %{data: 2}}
     end
   end
@@ -634,7 +643,7 @@ defmodule BroadwayTest do
 
       broadway_name = new_unique_name()
 
-      {:ok, _pid} =
+      {:ok, broadway} =
         Broadway.start_link(CustomHandlers,
           name: broadway_name,
           context: context,
@@ -650,42 +659,53 @@ defmodule BroadwayTest do
       batcher = get_batcher(broadway_name, :default)
       consumer = get_consumer(broadway_name, :default)
 
-      %{producer: producer, processor: processor, batcher: batcher, consumer: consumer}
+      %{
+        broadway: broadway,
+        producer: producer,
+        processor: processor,
+        batcher: batcher,
+        consumer: consumer
+      }
     end
 
     test "processor will be restarted in order to handle other messages",
-         %{producer: producer, processor: processor} do
-      push_messages(producer, [1])
+         %{broadway: broadway, processor: processor} do
+      Broadway.test_messages(broadway, [1])
       assert_receive {:message_handled, %{data: 1}}
 
       pid = Process.whereis(processor)
       ref = Process.monitor(processor)
-      push_messages(producer, [:kill_processor])
+      Broadway.test_messages(broadway, [:kill_processor])
       assert_receive {:DOWN, ^ref, _, _, _}
 
-      push_messages(producer, [2])
+      Broadway.test_messages(broadway, [2])
       assert_receive {:message_handled, %{data: 2}}
       assert Process.whereis(processor) != pid
     end
 
     test "processor crashes cascade down (but not up)", context do
-      %{producer: producer, processor: processor, batcher: batcher, consumer: consumer} = context
+      %{
+        broadway: broadway,
+        producer: producer,
+        processor: processor,
+        batcher: batcher,
+        consumer: consumer
+      } = context
 
       ref_batcher = Process.monitor(batcher)
       ref_processor = Process.monitor(processor)
       ref_producer = Process.monitor(producer)
       ref_consumer = Process.monitor(consumer)
 
-      push_messages(producer, [:kill_processor])
-
+      Broadway.test_messages(broadway, [:kill_processor])
       assert_receive {:DOWN, ^ref_processor, _, _, _}
       assert_receive {:DOWN, ^ref_batcher, _, _, _}
       assert_receive {:DOWN, ^ref_consumer, _, _, _}
       refute_received {:DOWN, ^ref_producer, _, _, _}
     end
 
-    test "only the messages in the crashing processor are lost", %{producer: producer} do
-      push_messages(producer, [1, 2, :kill_processor, 3, 4, 5])
+    test "only the messages in the crashing processor are lost", %{broadway: broadway} do
+      Broadway.test_messages(broadway, [1, 2, :kill_processor, 3, 4, 5])
 
       assert_receive {:message_handled, %{data: 1}}
       assert_receive {:message_handled, %{data: 2}}
@@ -697,8 +717,8 @@ defmodule BroadwayTest do
       assert_receive {:message_handled, %{data: 5}}
     end
 
-    test "batches are created normally (without the lost messages)", %{producer: producer} do
-      push_messages(producer, [1, 2, :kill_processor, 3, 4, 5])
+    test "batches are created normally (without the lost messages)", %{broadway: broadway} do
+      Broadway.test_messages(broadway, [1, 2, :kill_processor, 3, 4, 5])
 
       assert_receive {:batch_handled, messages}
       values = messages |> Enum.map(& &1.data)
@@ -730,7 +750,7 @@ defmodule BroadwayTest do
 
       broadway_name = new_unique_name()
 
-      {:ok, _pid} =
+      {:ok, broadway} =
         Broadway.start_link(CustomHandlers,
           name: broadway_name,
           context: context,
@@ -746,58 +766,66 @@ defmodule BroadwayTest do
       batcher = get_batcher(broadway_name, :default)
       consumer = get_consumer(broadway_name, :default)
 
-      %{producer: producer, processor: processor, batcher: batcher, consumer: consumer}
+      %{
+        broadway: broadway,
+        producer: producer,
+        processor: processor,
+        batcher: batcher,
+        consumer: consumer
+      }
     end
 
-    test "batcher will be restarted in order to handle other messages", %{producer: producer} do
-      push_messages(producer, [1, 2])
+    test "batcher will be restarted in order to handle other messages", %{broadway: broadway} do
+      Broadway.test_messages(broadway, [1, 2])
       assert_receive {:batch_handled, _, %BatchInfo{batcher_pid: batcher1}}
-
       ref = Process.monitor(batcher1)
 
-      push_messages(producer, [:kill_batcher, 3])
+      Broadway.test_messages(broadway, [:kill_batcher, 3])
       assert_receive {:batch_handled, _, %BatchInfo{batcher_pid: ^batcher1}}
-
       assert_receive {:DOWN, ^ref, _, obj, _}
 
-      push_messages(producer, [4, 5])
+      Broadway.test_messages(broadway, [4, 5])
       assert_receive {:batch_handled, _, %BatchInfo{batcher_pid: batcher2}}
-
       assert batcher1 != batcher2
     end
 
     test "batcher crashes cascade down (but not up)", context do
-      %{producer: producer, processor: processor, batcher: batcher, consumer: consumer} = context
+      %{
+        broadway: broadway,
+        producer: producer,
+        processor: processor,
+        batcher: batcher,
+        consumer: consumer
+      } = context
 
       ref_batcher = Process.monitor(batcher)
       ref_processor = Process.monitor(processor)
       ref_producer = Process.monitor(producer)
       ref_consumer = Process.monitor(consumer)
 
-      push_messages(producer, [:kill_batcher, 3])
-
+      Broadway.test_messages(broadway, [:kill_batcher, 3])
       assert_receive {:DOWN, ^ref_batcher, _, _, _}
       assert_receive {:DOWN, ^ref_consumer, _, _, _}
       refute_received {:DOWN, ^ref_producer, _, _, _}
       refute_received {:DOWN, ^ref_processor, _, _, _}
     end
 
-    test "only the messages in the crashing batcher are lost", %{producer: producer} do
-      push_messages(producer, [1, 2, :kill_batcher, 3, 4, 5, 6, 7])
+    test "only the messages in the crashing batcher are lost", %{broadway: broadway} do
+      ref = Broadway.test_messages(broadway, [1, 2, :kill_batcher, 3, 4, 5, 6, 7])
 
-      assert_receive {:ack, successful, _failed}
+      assert_receive {:ack, ^ref, successful, _failed}
       values = Enum.map(successful, & &1.data)
       assert values == [1, 2]
 
-      assert_receive {:ack, successful, _failed}
+      assert_receive {:ack, ^ref, successful, _failed}
       values = Enum.map(successful, & &1.data)
       assert values == [:kill_batcher, 3]
 
-      assert_receive {:ack, successful, _failed}
+      assert_receive {:ack, ^ref, successful, _failed}
       values = Enum.map(successful, & &1.data)
       assert values == [6, 7]
 
-      refute_receive {:ack, _successful, _failed}
+      refute_receive {:ack, _, _successful, _failed}
     end
   end
 
@@ -836,32 +864,30 @@ defmodule BroadwayTest do
           batchers: [default: [batch_size: 4]]
         )
 
-      producer = get_producer(broadway_name, :default)
       consumer = get_consumer(broadway_name, :default)
       Process.link(Process.whereis(consumer))
-
-      %{broadway: broadway, producer: producer, consumer: consumer}
+      %{broadway: broadway, consumer: consumer}
     end
 
     test "messages are grouped by ack_ref + status (successful or failed) before sent for acknowledgement",
          %{broadway: broadway} do
       Broadway.push_messages(broadway, [
-        %Message{data: 1, acknowledger: {Acker, :ack_ref_1, %{test_pid: self(), data_id: 1}}},
-        %Message{data: :fail, acknowledger: {Acker, :ack_ref_2, %{test_pid: self(), data_id: 2}}},
-        %Message{data: :fail, acknowledger: {Acker, :ack_ref_1, %{test_pid: self(), data_id: 3}}},
-        %Message{data: 4, acknowledger: {Acker, :ack_ref_2, %{test_pid: self(), data_id: 4}}}
+        %Message{data: 1, acknowledger: {CallerAcknowledger, {self(), :ack_ref_1}, :ok}},
+        %Message{data: :fail, acknowledger: {CallerAcknowledger, {self(), :ack_ref_2}, :ok}},
+        %Message{data: :fail, acknowledger: {CallerAcknowledger, {self(), :ack_ref_1}, :ok}},
+        %Message{data: 4, acknowledger: {CallerAcknowledger, {self(), :ack_ref_2}, :ok}}
       ])
 
-      assert_receive {:ack, [%{data: 1}], [%{data: :fail, acknowledger: {_, _, %{data_id: 3}}}]}
-      assert_receive {:ack, [%{data: 4}], [%{data: :fail, acknowledger: {_, _, %{data_id: 2}}}]}
+      assert_receive {:ack, :ack_ref_1, [%{data: 1}], [%{data: :fail}]}
+      assert_receive {:ack, :ack_ref_2, [%{data: 4}], [%{data: :fail}]}
     end
 
     test "all messages in the batch are marked as {:failed, reason} when an error is raised",
-         %{producer: producer, consumer: consumer} do
+         %{broadway: broadway, consumer: consumer} do
       assert capture_log(fn ->
-               push_messages(producer, [1, 2, :raise, 3])
+               ref = Broadway.test_messages(broadway, [1, 2, :raise, 3])
 
-               assert_receive {:ack, _,
+               assert_receive {:ack, ^ref, [],
                                [
                                  %{data: 1, status: {:failed, "due to an unhandled error"}},
                                  %{data: 2, status: {:failed, "due to an unhandled error"}},
@@ -874,11 +900,11 @@ defmodule BroadwayTest do
     end
 
     test "all messages in the batch are marked as {:failed, reason} on bad return",
-         %{producer: producer, consumer: consumer} do
+         %{broadway: broadway, consumer: consumer} do
       assert capture_log(fn ->
-               push_messages(producer, [1, 2, :bad_return, 3])
+               ref = Broadway.test_messages(broadway, [1, 2, :bad_return, 3])
 
-               assert_receive {:ack, _,
+               assert_receive {:ack, ^ref, [],
                                [
                                  %{data: 1, status: {:failed, "due to an unhandled error"}},
                                  %{data: 2, status: {:failed, "due to an unhandled error"}},
@@ -894,14 +920,14 @@ defmodule BroadwayTest do
     end
 
     test "consumers do not crash on bad acknowledger",
-         %{producer: producer, consumer: consumer} do
+         %{broadway: broadway, consumer: consumer} do
       [one, two, raise, three] = wrap_messages([1, 2, :raise, 3])
       raise = %{raise | acknowledger: {Unknown, :ack_ref, :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}], []}
+               Broadway.push_messages(broadway, [one, two, raise, three])
+               ref = Broadway.test_messages(broadway, [1, 2, 3, 4])
+               assert_receive {:ack, ^ref, [%{data: 1}, %{data: 2}, %{data: 3}, %{data: 4}], []}
              end) =~ "[error] ** (UndefinedFunctionError) function Unknown.ack/3 is undefined"
 
       refute_received {:EXIT, _, ^consumer}
@@ -923,7 +949,7 @@ defmodule BroadwayTest do
         handle_batch: fn _, batch, _, _ -> batch end
       }
 
-      {:ok, pid} =
+      {:ok, broadway} =
         Broadway.start_link(CustomHandlers,
           name: broadway_name,
           producers: [
@@ -936,7 +962,7 @@ defmodule BroadwayTest do
         )
 
       producer = get_producer(broadway_name, :default)
-      %{broadway: pid, producer: producer}
+      %{broadway: broadway, producer: producer}
     end
 
     test "killing the supervisor brings down the broadway GenServer",
@@ -962,20 +988,20 @@ defmodule BroadwayTest do
       async_push_messages(producer, [1, 2, 3, 4])
       Process.exit(broadway, :shutdown)
       :sys.resume(producer)
-      assert_receive {:ack, [%{data: 1}, %{data: 2}, %{data: 3}, %{data: 4}], []}
+      assert_receive {:ack, _, [%{data: 1}, %{data: 2}, %{data: 3}, %{data: 4}], []}
     end
 
     test "shutting down broadway waits until all events are processed even on incomplete batches",
-         %{broadway: broadway, producer: producer} do
-      push_messages(producer, [1, 2])
+         %{broadway: broadway} do
+      ref = Broadway.test_messages(broadway, [1, 2])
       Process.exit(broadway, :shutdown)
-      assert_receive {:ack, [%{data: 1}, %{data: 2}], []}
+      assert_receive {:ack, ^ref, [%{data: 1}, %{data: 2}], []}
     end
 
     @tag shutdown: 1
     test "shutting down broadway respects shutdown value",
-         %{broadway: broadway, producer: producer} do
-      push_messages(producer, [:sleep, 1, 2, 3])
+         %{broadway: broadway} do
+      Broadway.test_messages(broadway, [:sleep, 1, 2, 3])
       Process.exit(broadway, :shutdown)
       assert_receive {:EXIT, ^broadway, :shutdown}
     end
@@ -985,7 +1011,7 @@ defmodule BroadwayTest do
     :"Elixir.Broadway#{System.unique_integer([:positive, :monotonic])}"
   end
 
-  defp get_producer(broadway_name, key \\ :default, index \\ 1) do
+  defp get_producer(broadway_name, key, index \\ 1) do
     :"#{broadway_name}.Producer_#{key}_#{index}"
   end
 
@@ -1017,13 +1043,8 @@ defmodule BroadwayTest do
     send(producer, {:"$gen_call", {self(), make_ref()}, {:push_messages, wrap_messages(list)}})
   end
 
-  defp push_messages(producer, list) do
-    Producer.push_messages(producer, wrap_messages(list))
-  end
-
   defp wrap_messages(list) do
-    Enum.map(list, fn data ->
-      %Message{data: data, acknowledger: {Acker, :ack_ref, %{id: data, test_pid: self()}}}
-    end)
+    ack = {CallerAcknowledger, {self(), make_ref()}, :ok}
+    Enum.map(list, &%Message{data: &1, acknowledger: ack})
   end
 end