diff --git a/lib/event_store/notifications/listener.ex b/lib/event_store/notifications/listener.ex index cfaea7dd..77094990 100644 --- a/lib/event_store/notifications/listener.ex +++ b/lib/event_store/notifications/listener.ex @@ -13,7 +13,15 @@ defmodule EventStore.Notifications.Listener do alias EventStore.Notifications.{Listener, Notification} - defstruct [:listen_to, :query_timeout, :schema, :ref, demand: 0, queue: :queue.new()] + defstruct [ + :listen_to, + :monitor_ref, + :query_timeout, + :schema, + :ref, + demand: 0, + queue: :queue.new() + ] def start_link(opts) do {start_opts, listener_opts} = @@ -44,6 +52,14 @@ defmodule EventStore.Notifications.Listener do dispatch_events([], state) end + def handle_info({:DOWN, ref, :process, _object, reason}, %Listener{monitor_ref: ref} = state) do + {:stop, reason, state} + end + + def handle_info({:DOWN, _ref, :process, _object, _reason}, %Listener{} = state) do + {:noreply, [], state} + end + def handle_demand(incoming_demand, %Listener{} = state) do %Listener{demand: pending_demand} = state @@ -63,7 +79,9 @@ defmodule EventStore.Notifications.Listener do {:eventually, ref} -> ref end - %Listener{state | ref: ref} + monitor_ref = Process.monitor(listen_to) + + %Listener{state | ref: ref, monitor_ref: monitor_ref} end defp dispatch_events(events, %Listener{demand: 0} = state) do diff --git a/test/notifications/listener_test.exs b/test/notifications/listener_test.exs new file mode 100644 index 00000000..ecc2f3c1 --- /dev/null +++ b/test/notifications/listener_test.exs @@ -0,0 +1,83 @@ +defmodule EventStore.Notifications.ListenerTest do + use EventStore.StorageCase + + @moduletag :capture_log + + alias EventStore.{EventFactory, PubSub, Wait} + + @listener TestEventStore.EventStore.Notifications.Listener + @listen_to TestEventStore.Postgrex.Notifications + + describe "listen_to process down" do + test "keeps delivering events after the listen_to process dies" do + stream_uuid = "example-stream" + + :ok = PubSub.subscribe(TestEventStore, stream_uuid) + + listener = whereis(@listener) + + kill(@listen_to) + wait_until_listening(listener) + + :ok = append_events(stream_uuid, 3) + + assert_receive {:events, events}, 5_000 + assert length(events) == 3 + end + + test "stops with the reason reported for the listen_to process" do + listener = whereis(@listener) + ref = Process.monitor(listener) + + kill(@listen_to) + + assert_receive {:DOWN, ^ref, :process, ^listener, reason}, 5_000 + assert reason == :killed + end + + test "ignores a down message for an unrelated process" do + stream_uuid = "example-stream" + + :ok = PubSub.subscribe(TestEventStore, stream_uuid) + + listener = whereis(@listener) + + send(listener, {:DOWN, make_ref(), :process, spawn(fn -> :ok end), :normal}) + + :ok = append_events(stream_uuid, 3) + + assert_receive {:events, events}, 5_000 + assert length(events) == 3 + assert Process.alive?(listener) + end + end + + defp whereis(name) do + pid = Process.whereis(name) + assert is_pid(pid) + pid + end + + defp kill(name) do + pid = whereis(name) + Process.exit(pid, :kill) + end + + # The supervisor restarts the notifications connection and the listener. Wait + # for a replacement listener, and for a connection that is actually online: + # `listen/3` answers `{:eventually, ref}` while it is still reconnecting. + defp wait_until_listening(previous_listener) do + Wait.until(10_000, fn -> + listener = Process.whereis(@listener) + + assert is_pid(listener) + assert listener != previous_listener + assert {:ok, ref} = Postgrex.Notifications.listen(@listen_to, "listener_test_probe") + assert is_reference(ref) + end) + end + + defp append_events(stream_uuid, count) do + TestEventStore.append_to_stream(stream_uuid, :any_version, EventFactory.create_events(count)) + end +end