Repository: plataformatec/broadway (elixir)

Issue:
Add API to ack at will
@josevalim and I discussed a new API to manually ack a message instead of having to wait for the Broadway pipeline to be done so that the message is acked automatically.

This API would call the acknowledger in the message directly and then set it to a no-op acknowledger.

```elixir
Broadway.Message.ack(message)
```

We are not passing options to this functions because the acknowledger can be configured individually for each message through the API proposed in #104.

Pull request:
Add Broadway.Message.ack_immediately/1
Took a stab at this. I have very little idea of what I'm doing 🎊 

Closes #105.

\cc @josevalim @msaraiva 

Patch:
diff --git a/lib/broadway/message.ex b/lib/broadway/message.ex
index c60b194..b4b6dc0 100644
--- a/lib/broadway/message.ex
+++ b/lib/broadway/message.ex
@@ -13,6 +13,7 @@ defmodule Broadway.Message do
   """
 
   alias __MODULE__, as: Message
+  alias Broadway.{Acknowledger, NoopAcknowledger}
 
   @type t :: %Message{
           data: term,
@@ -90,4 +91,34 @@ defmodule Broadway.Message do
   def failed(%Message{} = message, reason) do
     %Message{message | status: {:failed, reason}}
   end
+
+  @doc """
+  Immediately acknowledges the given message or list of messages.
+
+  This function can be used to acknowledge a message (or list of messages)
+  immediately withouth waiting for the rest of the pipeline.
+
+  Acknowledging a message sets that message's acknowledger to a no-op
+  acknowledger so that it's safe to ack at the end of the pipeline.
+
+  Returns the updated acked message if a message is passed in,
+  or the updated list of acked messages if a list of messages is passed in.
+  """
+  @spec ack_immediately(message :: Message.t()) :: Message.t()
+  @spec ack_immediately(messages :: [Message.t(), ...]) :: [Message.t(), ...]
+  def ack_immediately(message_or_messages)
+
+  def ack_immediately(%Message{} = message) do
+    [message] = ack_immediately([message])
+    message
+  end
+
+  def ack_immediately(messages) when is_list(messages) and messages != [] do
+    {successful, failed} = Enum.split_with(messages, &(&1.status == :ok))
+    _ = Acknowledger.ack_messages(successful, failed)
+
+    for message <- messages do
+      %{message | acknowledger: {NoopAcknowledger, _ack_ref = nil, _data = nil}}
+    end
+  end
 end
diff --git a/lib/broadway/noop_acknowledger.ex b/lib/broadway/noop_acknowledger.ex
new file mode 100644
index 0000000..293ce4c
--- /dev/null
+++ b/lib/broadway/noop_acknowledger.ex
@@ -0,0 +1,15 @@
+defmodule Broadway.NoopAcknowledger do
+  @moduledoc """
+  An acknowledger that does nothing.
+
+  Set automatically on messages that have been acked immediately
+  via `Broadway.Message.ack_immediately/1`.
+  """
+
+  @behaviour Broadway.Acknowledger
+
+  @impl true
+  def ack(_ack_ref = nil, _successful, _failed) do
+    :ok
+  end
+end


Test patch:
diff --git a/test/broadway_test.exs b/test/broadway_test.exs
index 87c19a9..6565ac9 100644
--- a/test/broadway_test.exs
+++ b/test/broadway_test.exs
@@ -682,6 +682,73 @@ defmodule BroadwayTest do
     end
   end
 
+  describe "ack_immediately" do
+    test "acks a single message" do
+      broadway_name = new_unique_name()
+      test_pid = self()
+
+      handle_message = fn message, _ ->
+        message = Message.ack_immediately(message)
+        send(test_pid, :manually_acked)
+        message
+      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 {:ack, ^ref, [%Message{data: 1}], []}
+      assert_receive :manually_acked
+      refute_receive {:ack, ^ref, _successful, _failed}
+    end
+
+    test "acks multiple messages" do
+      broadway_name = new_unique_name()
+      test_pid = self()
+
+      handle_message = fn message, _ ->
+        Message.put_batcher(message, :default)
+      end
+
+      handle_batch = fn :default, batch, _batch_info, _ ->
+        batch =
+          Enum.map(batch, fn
+            %Message{data: :ok} = message -> message
+            %Message{data: :fail} = message -> Message.failed(message, :manually_failed)
+          end)
+
+        batch = Message.ack_immediately(batch)
+        send(test_pid, :manually_acked)
+        batch
+      end
+
+      {:ok, broadway} =
+        Broadway.start_link(CustomHandlers,
+          name: broadway_name,
+          context: %{handle_message: handle_message, handle_batch: handle_batch},
+          producers: [
+            default: [module: {ManualProducer, []}]
+          ],
+          processors: [default: []],
+          batchers: [default: [batch_size: 4, batch_timeout: 1000]]
+        )
+
+      ref = Broadway.test_messages(broadway, [:ok, :ok, :fail, :ok], batch_mode: :bulk)
+
+      assert_receive {:ack, ^ref, [_, _, _], [_]}
+      assert_receive :manually_acked
+      refute_receive {:ack, ^ref, _successful, _failed}
+    end
+  end
+
   describe "transformer" do
     setup tags do
       broadway_name = new_unique_name()