Repository: plataformatec/broadway
Pull request #109 (2019-09-11 08:23:09)

Add Broadway.Message.configure_ack/3
Some notes:

  * I didn't add the `configure/3` callback to `NoopAcknowledger` because I think it's good if it raises if you call `configure_ack/3` after calling `ack_immediately`. Do we want to implement it and raise an even better message?

  * Do we need to support a list of messages in `configure_ack/3`?


Patch:
diff --git a/lib/broadway/acknowledger.ex b/lib/broadway/acknowledger.ex
index 507f140..7c3c788 100644
--- a/lib/broadway/acknowledger.ex
+++ b/lib/broadway/acknowledger.ex
@@ -35,9 +35,26 @@ defmodule Broadway.Acknowledger do
       could not be processed or published.
 
   """
-  @callback ack(ack_ref :: {pid, reference}, successful :: [Message.t()], failed :: [Message.t()]) ::
+  @callback ack(ack_ref :: term, successful :: [Message.t()], failed :: [Message.t()]) ::
               :ok
 
+  @doc """
+  Configures the acknowledger with new `options`.
+
+  Every acknowledger can decide how to incorporate the given `options` into its
+  `ack_data`. The `ack_data` is the current acknowledger's data. The return value
+  of this function is `{:ok, new_ack_data}` where `new_ack_data` is the updated
+  data for the acknowledger.
+
+  Note that `options` are different for every acknowledger, as the acknowledger
+  is what specifies what are the supported options. Check the documentation for the
+  acknowledger you're using to see the supported options.
+  """
+  @callback configure(ack_ref :: term, ack_data :: term, options :: keyword) ::
+              {:ok, new_ack_data :: term}
+
+  @optional_callbacks [configure: 3]
+
   @doc """
   Acknowledges successful and failed messages grouped by `{acknowledger, ack_ref}`.
   """
diff --git a/lib/broadway/caller_acknowledger.ex b/lib/broadway/caller_acknowledger.ex
index e0c317c..9726d21 100644
--- a/lib/broadway/caller_acknowledger.ex
+++ b/lib/broadway/caller_acknowledger.ex
@@ -13,6 +13,12 @@ defmodule Broadway.CallerAcknowledger do
   It sends a message in the format:
 
       {:ack, ref, successful_messages, failed_messages}
+
+  If `Broadway.Message.configure_ack/3` is called on a message that
+  uses this acknowledger, then the following message is sent:
+
+      {:configure, ref, options}
+
   """
 
   @behaviour Broadway.Acknowledger
@@ -21,4 +27,10 @@ defmodule Broadway.CallerAcknowledger do
   def ack({pid, ref}, successful, failed) do
     send(pid, {:ack, ref, successful, failed})
   end
+
+  @impl true
+  def configure({pid, ref}, ack_data, options) do
+    send(pid, {:configure, ref, options})
+    {:ok, ack_data}
+  end
 end
diff --git a/lib/broadway/message.ex b/lib/broadway/message.ex
index b4b6dc0..cbc6003 100644
--- a/lib/broadway/message.ex
+++ b/lib/broadway/message.ex
@@ -81,6 +81,27 @@ defmodule Broadway.Message do
     %Message{message | batch_mode: mode}
   end
 
+  @doc """
+  Configures the acknowledger of this message.
+
+  This function calls the `c:Broadway.Acknowledger.configure/3` callback to
+  change the configuration of the acknowledger for the given `message`.
+
+  This function can only be called if the acknowledger implements the `configure/3`
+  callback. If it doesn't, an error is raised.
+  """
+  @spec configure_ack(message :: Message.t(), options :: keyword) :: Message.t()
+  def configure_ack(%Message{} = message, options) when is_list(options) do
+    %{acknowledger: {module, ack_ref, ack_data}} = message
+
+    if function_exported?(module, :configure, 3) do
+      {:ok, ack_data} = module.configure(ack_ref, ack_data, options)
+      %{message | acknowledger: {module, ack_ref, ack_data}}
+    else
+      raise "the configure/3 callback is not defined by acknowledger #{inspect(module)}"
+    end
+  end
+
   @doc """
   Mark a message as failed.
 


Test patch:
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index 6565ac9..e705618 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -749,6 +749,63 @@ defmodule BroadwayTest do
     end
   end
 
+  describe "configure_ack" do
+    test "raises if the acknowledger doesn't implement the configure/3 callback" do
+      broadway_name = new_unique_name()
+
+      handle_message = fn message, _ ->
+        Message.configure_ack(message, test_pid: self())
+      end
+
+      {:ok, broadway} =
+        Broadway.start_link(CustomHandlers,
+          name: broadway_name,
+          context: %{handle_message: handle_message},
+          producers: [
+            default: [module: {ManualProducer, []}]
+          ],
+          processors: [default: []]
+        )
+
+      log =
+        capture_log(fn ->
+          Broadway.push_messages(broadway, [
+            %Message{data: 1, acknowledger: {Acker, nil, %{test_pid: self()}}}
+          ])
+
+          assert_receive {:ack, _successful = [], [failed]}
+          assert failed.status == {:failed, "due to an unhandled error"}
+        end)
+
+      assert log =~ "the configure/3 callback is not defined by acknowledger BroadwayTest.Acker"
+    end
+
+    test "configures the acknowledger" do
+      broadway_name = new_unique_name()
+      configure_options = [some_unique_term: make_ref()]
+
+      handle_message = fn message, _ ->
+        Message.configure_ack(message, configure_options)
+      end
+
+      {:ok, broadway} =
+        Broadway.start_link(CustomHandlers,
+          name: broadway_name,
+          context: %{handle_message: handle_message},
+          producers: [
+            default: [module: {ManualProducer, []}]
+          ],
+          processors: [default: []]
+        )
+
+      ref = Broadway.test_messages(broadway, [1])
+
+      assert_receive {:configure, ^ref, ^configure_options}
+      assert_receive {:ack, ^ref, [success], _failed = []}
+      assert success.data == 1
+    end
+  end
+
   describe "transformer" do
     setup tags do
       broadway_name = new_unique_name()