Merge pull request #97 from nsweeting/v0.19.0

Ensure :EXIT signals go through handle_error/1

Author: nsweeting

.github/workflows/ci.yml

@@ -10,7 +10,7 @@ on:
 
 jobs:
   test:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-20.04
     name: Elixir ${{ matrix.elixir }} / OTP ${{ matrix.otp }}
     strategy:
       matrix:

lib/rabbit/consumer/executer.ex

@@ -36,6 +36,8 @@ defmodule Rabbit.Consumer.Executer do
   @doc false
   @impl GenServer
   def init({message, opts}) do
+    Process.flag(:trap_exit, true)
+
     with {:ok, opts} <- validate_opts(opts, @opts_schema) do
       state = init_state(message, opts)
       set_timeout(state.timeout)
@@ -58,6 +60,17 @@ defmodule Rabbit.Consumer.Executer do
     {:stop, :timeout, state}
   end
 
+  def handle_info({:EXIT, pid1, reason}, %{executer: pid2} = state) when pid1 == pid2 do
+    {reason, stack} =
+      case reason do
+        {%_{} = reason, stack} -> {reason, stack}
+        reason -> {reason, []}
+      end
+
+    handle_error(state, reason, stack)
+    {:stop, reason, state}
+  end
+
   @doc false
   @impl GenServer
   def handle_cast({:complete, ref1}, %{executer_ref: ref2} = state) when ref1 == ref2 do

test/consumer_test.exs

@@ -101,6 +101,36 @@ defmodule Rabbit.ConsumerTest do
     end
   end
 
+  defmodule ErrorTestConsumer do
+    use Rabbit.Consumer
+
+    @impl Rabbit.Consumer
+    def init(:consumer, opts) do
+      {:ok, opts}
+    end
+
+    @impl Rabbit.Consumer
+    def handle_setup(state) do
+      AMQP.Queue.declare(state.channel, state.queue, auto_delete: true)
+      AMQP.Queue.purge(state.channel, state.queue)
+      :ok
+    end
+
+    @impl Rabbit.Consumer
+    def handle_message(msg) do
+      decoded_payload = Base.decode64!(msg.payload)
+      {_, _, fun} = :erlang.binary_to_term(decoded_payload)
+      fun.()
+    end
+
+    @impl Rabbit.Consumer
+    def handle_error(msg) do
+      decoded_payload = Base.decode64!(msg.payload)
+      {pid, ref, _} = :erlang.binary_to_term(decoded_payload)
+      send(pid, {:handle_error, ref, msg})
+    end
+  end
+
   setup do
     {:ok, connection} = Connection.start_link(TestConnection)
     {:ok, producer} = Producer.start_link(TestProducer, connection: connection)
@@ -263,6 +293,43 @@ defmodule Rabbit.ConsumerTest do
     assert msg.custom_meta == %{foo: "bar"}
   end
 
+  test "will run the handle_error/1 callback on exceptions", meta do
+    {:ok, _consumer, queue} = start_consumer(ErrorTestConsumer, meta, [])
+    error = fn -> raise "boom" end
+    ref = publish_message(meta, queue, msg: error)
+
+    assert_receive {:handle_error, ^ref, msg}
+    assert msg.error_reason == %RuntimeError{message: "boom"}
+  end
+
+  test "will run the handle_error/1 callback on linked processes crashing", meta do
+    {:ok, _consumer, queue} = start_consumer(ErrorTestConsumer, meta, [])
+
+    error = fn ->
+      spawn_link(fn -> raise "boom" end)
+      :timer.sleep(5000)
+    end
+
+    ref = publish_message(meta, queue, msg: error)
+
+    assert_receive {:handle_error, ^ref, msg}
+    assert msg.error_reason == %RuntimeError{message: "boom"}
+  end
+
+  test "will run the handle_error/1 callback on process exits", meta do
+    {:ok, _consumer, queue} = start_consumer(ErrorTestConsumer, meta, [])
+
+    error = fn ->
+      Process.exit(self(), :boom)
+      :timer.sleep(5000)
+    end
+
+    ref = publish_message(meta, queue, msg: error)
+
+    assert_receive {:handle_error, ^ref, msg}
+    assert msg.error_reason == :boom
+  end
+
   defp start_consumer(meta, opts \\ []), do: start_consumer(TestConsumer, meta, opts)
 
   defp start_consumer(module, meta, opts) do