Use extreme with backpressure

Author: burmajam

CHANGELOG.md

@@ -6,6 +6,10 @@ The format is based on [Keep a
 Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to
 [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## 1.1.0 - 2024-06-18
+
+- Use new `Extreme.ListenerWithBackPressure`
+
 ## 1.0.1 - 2024-01-31
 
 - Bump all dependencies including bumping `:extreme` to v1.0.5 which fixes a

config/test.exs

@@ -1,5 +1,12 @@
 import Config
 
+config :logger, :console,
+  format: "$time $metadata[$level] $message\n",
+  level: :debug,
+  metadata: [:pid, :module, :function]
+
+config :ex_unit, capture_log: true
+
 config :kelvin, ExtremeClient,
   db_type: :node,
   host: System.get_env("EVENTSTORE_HOST") || "localhost",

lib/kelvin/in_order_subscription.ex

@@ -35,7 +35,7 @@ defmodule Kelvin.InOrderSubscription do
 
   defstruct [
     :config,
-    :subscription,
+    :extreme_listener,
     :self,
     :max_buffer_size,
     demand: 0,
@@ -56,7 +56,31 @@ defmodule Kelvin.InOrderSubscription do
         Application.get_env(:kelvin, :catch_up_chunk_size, 256)
       )
 
+    connection = Keyword.fetch!(opts, :connection)
+    stream_name = Keyword.fetch!(opts, :stream_name)
+
+    listener_name =
+      opts
+      |> Keyword.get(:name, __MODULE__)
+      |> Module.concat(ExtremeListener)
+
+    {:ok, extreme_listener} =
+      Kelvin.Listener.start_link(connection, stream_name,
+        read_per_page: max_buffer_size,
+        auto_subscribe: false,
+        ack_timeout: :infinity,
+        name: listener_name,
+        producer: self(),
+        get_stream_position_fun: fn ->
+          opts
+          |> Keyword.fetch!(:restore_stream_position!)
+          |> _do_function()
+          |> Kernel.+(1)
+        end
+      )
+
     state = %__MODULE__{
+      extreme_listener: extreme_listener,
       config: Map.new(opts),
       self: Keyword.get(opts, :name, self()),
       max_buffer_size: max_buffer_size
@@ -92,23 +116,9 @@ defmodule Kelvin.InOrderSubscription do
   end
 
   def handle_info(:subscribe, state) do
-    if state.subscription do
-      # coveralls-ignore-start
-      Logger.warn("#{inspect(__MODULE__)} is already subscribed.")
-      # coveralls-ignore-stop
-    else
-      case _subscribe(state) do
-        {:ok, sub} ->
-          Process.link(sub)
-          {:noreply, [], put_in(state.subscription, sub)}
-
-        # coveralls-ignore-start
-        {:error, reason} ->
-          {:stop, reason, state}
+    Kelvin.Listener.subscribe(state.extreme_listener)
 
-          # coveralls-ignore-stop
-      end
-    end
+    {:noreply, [], state}
   end
 
   def handle_info(_info, state), do: {:noreply, [], state}
@@ -162,18 +172,6 @@ defmodule Kelvin.InOrderSubscription do
     end
   end
 
-  defp _subscribe(state) do
-    state.config.connection
-    |> Extreme.RequestManager._name()
-    |> GenServer.call(
-      {:read_and_stay_subscribed, self(),
-       {state.config.stream_name,
-        _do_function(state.config.restore_stream_position!) + 1,
-        state.max_buffer_size, true, false, :infinity}},
-      :infinity
-    )
-  end
-
   defp _do_function(func) when is_function(func, 0), do: func.()
 
   defp _do_function({m, f, a}) when is_atom(m) and is_atom(f) and is_list(a) do

lib/kelvin/listener.ex

@@ -0,0 +1,18 @@
+defmodule Kelvin.Listener do
+  use Extreme.ListenerWithBackPressure
+
+  def on_init(opts) do
+    state = %{
+      producer: Keyword.fetch!(opts, :producer),
+      get_stream_position_fun: Keyword.fetch!(opts, :get_stream_position_fun)
+    }
+
+    {:ok, state}
+  end
+
+  defp get_last_event(_stream_name, %{} = state),
+    do: state.get_stream_position_fun.()
+
+  defp process_push(push, _stream_name, %{} = state),
+    do: GenServer.call(state.producer, {:on_event, push})
+end

mix.exs

@@ -45,7 +45,8 @@ defmodule Kelvin.MixProject do
   defp deps do
     [
       {:gen_stage, "~> 1.0"},
-      {:extreme, "~> 1.1.0-rc1"},
+      # {:extreme, "~> 1.1.0-rc1"},
+      {:extreme, github: "exponentially/extreme", branch: "event-producer"},
       # docs
       {:ex_doc, ">= 0.0.0", only: :dev, runtime: false},
       # test

mix.lock

@@ -7,7 +7,7 @@
   "ex_doc": {:hex, :ex_doc, "0.31.1", "8a2355ac42b1cc7b2379da9e40243f2670143721dd50748bf6c3b1184dae2089", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.1", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "3178c3a407c557d8343479e1ff117a96fd31bafe52a039079593fb0524ef61b0"},
   "excoveralls": {:hex, :excoveralls, "0.18.0", "b92497e69465dc51bc37a6422226ee690ab437e4c06877e836f1c18daeb35da9", [:mix], [{:castore, "~> 1.0", [hex: :castore, repo: "hexpm", optional: true]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "1109bb911f3cb583401760be49c02cbbd16aed66ea9509fc5479335d284da60b"},
   "exprotobuf": {:hex, :exprotobuf, "1.2.17", "3003937da617f588a8fb63ebdd7b127a18d78d6502623c272076fd54c07c4de1", [:mix], [{:gpb, "~> 4.0", [hex: :gpb, repo: "hexpm", optional: false]}], "hexpm", "e07ec1e5ae6f8c1c8521450d5f6b658c8c700b1f34c70356e91ece0766f4361a"},
-  "extreme": {:hex, :extreme, "1.1.0-rc1", "43ee836564d7e647631e06a087a865a9096b21ec9ade0c3db86dde4dd83f32e4", [:mix], [{:elixir_uuid, "~> 1.2", [hex: :elixir_uuid, repo: "hexpm", optional: false]}, {:exprotobuf, "~> 1.2.9", [hex: :exprotobuf, repo: "hexpm", optional: false]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "fdd0528b488c77f337ffd1d5053cea0e9e1f9731e42d8ddcc659bb6b43ae2328"},
+  "extreme": {:git, "https://github.com/exponentially/extreme.git", "9cc7438a4f3659af92f081731f545b9a2e18638a", [branch: "event-producer"]},
   "file_system": {:hex, :file_system, "1.0.0", "b689cc7dcee665f774de94b5a832e578bd7963c8e637ef940cd44327db7de2cd", [:mix], [], "hexpm", "6752092d66aec5a10e662aefeed8ddb9531d79db0bc145bb8c40325ca1d8536d"},
   "gen_stage": {:hex, :gen_stage, "1.2.1", "19d8b5e9a5996d813b8245338a28246307fd8b9c99d1237de199d21efc4c76a1", [:mix], [], "hexpm", "83e8be657fa05b992ffa6ac1e3af6d57aa50aace8f691fcf696ff02f8335b001"},
   "gpb": {:hex, :gpb, "4.21.1", "72e229c242d252d690addcfd04a6416c26c4d4d2c3521e05570a7a78b48d3bd1", [:make, :rebar3], [], "hexpm", "c05c9aea9e25bd341367a43b3d3eb68e951563911072259c5ec4cb6642f4ef22"},

test/kelvin/in_order_subscription_test.exs

@@ -1,8 +1,6 @@
 defmodule Kelvin.InOrderSubscriptionTest do
   use ExUnit.Case, async: true
 
-  @moduletag :capture_log
-
   alias Extreme.Messages
 
   setup do
@@ -14,7 +12,7 @@ defmodule Kelvin.InOrderSubscriptionTest do
 
   describe "given events have been written to a stream" do
     setup c do
-      write_events(0..100, c.stream_name)
+      _write_events(0..100, c.stream_name)
       :ok
     end
 
@@ -33,7 +31,7 @@ defmodule Kelvin.InOrderSubscriptionTest do
         assert event.event.data == to_string(n)
       end
 
-      write_events(101..200, c.stream_name)
+      _write_events(101..200, c.stream_name)
 
       for n <- 101..200 do
         assert_receive {:events, [event]}, 1_000
@@ -65,15 +63,7 @@ defmodule Kelvin.InOrderSubscriptionTest do
 
       assert_receive {:DOWN, ^monitor_ref, _, _, _}
 
-      # we're hardcoding the restore_stream_position! function so this will
-      # restart from 0 instead of the current stream position as would be the
-      # case in a real-life system
-      for n <- 0..100 do
-        assert_receive {:events, [event]}, 10_000
-        assert event.event.data == to_string(n)
-      end
-
-      write_events(101..200, c.stream_name)
+      _write_events(101..200, c.stream_name)
 
       for n <- 101..200 do
         assert_receive {:events, [event]}, 1_000
@@ -84,7 +74,7 @@ defmodule Kelvin.InOrderSubscriptionTest do
 
   describe "given only a few events have been written to a stream" do
     setup c do
-      write_events(0..10, c.stream_name)
+      _write_events(0..10, c.stream_name)
       :ok
     end
 
@@ -113,7 +103,7 @@ defmodule Kelvin.InOrderSubscriptionTest do
 
   defp restore_stream_position!, do: -1
 
-  defp write_events(range, stream) do
+  defp _write_events(range, stream) do
     range
     |> Enum.map(fn n ->
       Messages.NewEvent.new(