Make PostHog.Sender a pool

Author: martosaur

lib/posthog/registry.ex

@@ -13,4 +13,7 @@ defmodule PostHog.Registry do
 
   def via(supervisor_name, server_name),
     do: {:via, Registry, {registry_name(supervisor_name), server_name}}
+
+  def via(supervisor_name, pool_name, index),
+    do: {:via, Registry, {registry_name(supervisor_name), {pool_name, index}}}
 end

lib/posthog/sender.ex

@@ -4,6 +4,7 @@ defmodule PostHog.Sender do
 
   defstruct [
     :registry,
+    :index,
     :api_client,
     :max_batch_time_ms,
     :max_batch_events,
@@ -15,7 +16,7 @@ defmodule PostHog.Sender do
     name =
       opts
       |> Keyword.fetch!(:supervisor_name)
-      |> PostHog.Registry.via(__MODULE__)
+      |> PostHog.Registry.via(__MODULE__, opts[:index])
 
     callers = Process.get(:"$callers", [])
     Process.flag(:trap_exit, true)
@@ -33,8 +34,16 @@ defmodule PostHog.Sender do
         PostHog.Test.remember_event(supervisor_name, event)
 
       _ ->
-        supervisor_name
-        |> PostHog.Registry.via(__MODULE__)
+        senders =
+          supervisor_name
+          |> PostHog.Registry.registry_name()
+          |> Registry.select([{{{__MODULE__, :_}, :"$1", :"$2"}, [], [{{:"$2", :"$1"}}]}])
+
+        # Pick the first available sender, otherwise random busy one.
+        senders
+        |> Keyword.get_lazy(:available, fn ->
+          senders |> Keyword.values() |> Enum.random()
+        end)
         |> GenServer.cast({:event, event})
     end
   end
@@ -45,6 +54,7 @@ defmodule PostHog.Sender do
   def init({opts, callers}) do
     state = %__MODULE__{
       registry: PostHog.Registry.registry_name(opts[:supervisor_name]),
+      index: Keyword.fetch!(opts, :index),
       api_client: Keyword.fetch!(opts, :api_client),
       max_batch_time_ms: Keyword.fetch!(opts, :max_batch_time_ms),
       max_batch_events: Keyword.fetch!(opts, :max_batch_events),
@@ -54,6 +64,9 @@ defmodule PostHog.Sender do
 
     Process.put(:"$callers", callers)
 
+    {:available, nil} =
+      Registry.update_value(state.registry, registry_key(state.index), fn _ -> :available end)
+
     {:ok, state}
   end
 
@@ -81,7 +94,9 @@ defmodule PostHog.Sender do
 
   @impl GenServer
   def handle_continue(:send_batch, state) do
+    Registry.update_value(state.registry, registry_key(state.index), fn _ -> :busy end)
     PostHog.API.post_batch(state.api_client, state.events)
+    Registry.update_value(state.registry, registry_key(state.index), fn _ -> :available end)
     {:noreply, %{state | events: [], num_events: 0}}
   end
 
@@ -91,4 +106,6 @@ defmodule PostHog.Sender do
   end
 
   def terminate(_reason, _state), do: :ok
+
+  defp registry_key(index), do: {__MODULE__, index}
 end

lib/posthog/supervisor.ex

@@ -29,19 +29,30 @@ defmodule PostHog.Supervisor do
         {Registry,
          keys: :unique,
          name: PostHog.Registry.registry_name(config.supervisor_name),
-         meta: [config: config]},
+         meta: [config: config]}
+      ] ++ senders(config)
+
+    Process.put(:"$callers", callers)
+
+    Supervisor.init(children, strategy: :one_for_one)
+  end
+
+  defp senders(config) do
+    pool_size = Map.get(config, :sender_pool_size, max(System.schedulers_online(), 2))
+
+    for index <- 1..pool_size do
+      Supervisor.child_spec(
         {PostHog.Sender,
          [
            api_client: config.api_client,
            supervisor_name: config.supervisor_name,
            max_batch_time_ms: Map.get(config, :max_batch_time_ms, to_timeout(second: 10)),
            max_batch_events: Map.get(config, :max_batch_events, 100),
-           test_mode: config.test_mode
-         ]}
-      ]
-
-    Process.put(:"$callers", callers)
-
-    Supervisor.init(children, strategy: :one_for_one)
+           test_mode: config.test_mode,
+           index: index
+         ]},
+        id: {PostHog.Sender, index}
+      )
+    end
   end
 end