Allow Listener.process_push/2 to return 'stop'

CHANGELOG.md

@@ -1,3 +1,8 @@
+# Changelog for extreme v1.0.4
+
+- Listener.process_push callback can return `:stop`, meaning subscription should be stopped
+  and pushes that are already in mailbox should be purged.
+
 # Changelog for extreme v1.0.3
 
 - Add subscribe/unsubscribe and auto_subscribe? option for starting `Extreme.Listener`

lib/listener.ex

@@ -72,17 +72,29 @@ defmodule Extreme.Listener do
       end
 
       @impl true
+      def handle_call({:on_event, _push}, _from, %{subscription: nil, mode: :live} = state),
+        do: {:reply, :ok, state}
+
       def handle_call(
             {:on_event, push},
             _from,
             %{subscription: subscription, mode: :live} = state
-          )
-          when not is_nil(subscription) do
-        {:ok, event_number} = process_push(push, state.stream_name)
-        {:reply, :ok, %{state | last_event: event_number}}
+          ) do
+        push
+        |> process_push(state.stream_name)
+        |> case do
+          {:ok, event_number} ->
+            {:reply, :ok, %{state | last_event: event_number}}
+
+          :stop ->
+            _unsubscribe(state)
+        end
       end
 
-      def handle_call(:unsubscribe, from, state) do
+      def handle_call(:unsubscribe, _from, state),
+        do: _unsubscribe(state)
+
+      defp _unsubscribe(state) do
         Logger.info(
           "#{__MODULE__} unsubscribed from #{state.stream_name}. Last processed event: #{state.last_event}"
         )

mix.exs

@@ -4,7 +4,7 @@ defmodule Extreme.Mixfile do
   def project do
     [
       app: :extreme,
-      version: "1.0.3",
+      version: "1.0.4",
       elixir: "~> 1.11",
       elixirc_paths: _elixirc_paths(Mix.env()),
       source_url: "https://github.com/exponentially/extreme",