Zurga · Zurga · Mar 24, 2025 · Mar 24, 2025 · Mar 24, 2025 · Mar 24, 2025
diff --git a/lib/ecto_watch/options.ex b/lib/ecto_watch/options.ex
@@ -3,12 +3,13 @@ defmodule EctoWatch.Options do
 
   alias EctoWatch.Options.WatcherOptions
 
-  defstruct [:repo_mod, :pub_sub_mod, :watchers, :debug?]
+  defstruct [:repo_mod, :pub_sub_mod, :dispatcher, :watchers, :debug?]
 
   def new(opts) do
     %__MODULE__{
       repo_mod: opts[:repo],
       pub_sub_mod: opts[:pub_sub],
+      dispatcher: opts[:dispatcher],
       watchers:
         Enum.map(opts[:watchers], fn watcher_opts ->
           WatcherOptions.new(watcher_opts, opts[:debug?])
@@ -26,6 +27,7 @@ defmodule EctoWatch.Options do
         type: {:custom, __MODULE__, :check_valid_pubsub_module, []},
         required: true
       ],
+      dispatcher: [type: :atom, required: false, default: Phoenix.PubSub],
       watchers: [
         type: {:custom, WatcherOptions, :validate_list, []},
         required: true

diff --git a/lib/ecto_watch/watcher_server.ex b/lib/ecto_watch/watcher_server.ex
@@ -35,16 +35,16 @@ defmodule EctoWatch.WatcherServer do
     end
   end
 
-  def start_link({repo_mod, pub_sub_mod, watcher_options}) do
+  def start_link({repo_mod, pub_sub_mod, dispatcher, watcher_options}) do
     GenServer.start_link(
       __MODULE__,
-      {repo_mod, pub_sub_mod, watcher_options},
+      {repo_mod, pub_sub_mod, dispatcher, watcher_options},
       name: unique_label(watcher_options)
     )
   end
 
   @impl true
-  def init({repo_mod, pub_sub_mod, options}) do
+  def init({repo_mod, pub_sub_mod, dispatcher, options}) do
     debug_log(options, "Starting server")
 
     unique_label = "#{unique_label(options)}"
@@ -121,6 +121,7 @@ defmodule EctoWatch.WatcherServer do
      %{
        repo_mod: repo_mod,
        pub_sub_mod: pub_sub_mod,
+       dispatcher: dispatcher,
        unique_label: unique_label,
        identifier_columns:
          MapSet.put(
@@ -226,7 +227,7 @@ defmodule EctoWatch.WatcherServer do
         "Broadcasting to Phoenix PubSub topic `#{topic}`: #{inspect(message)}"
       )
 
-      Phoenix.PubSub.broadcast(state.pub_sub_mod, topic, message)
+      Phoenix.PubSub.broadcast(state.pub_sub_mod, topic, message, state.dispatcher)
     end
 
     {:noreply, state}

diff --git a/lib/ecto_watch/watcher_supervisor.ex b/lib/ecto_watch/watcher_supervisor.ex
@@ -20,7 +20,7 @@ defmodule EctoWatch.WatcherSupervisor do
           id: WatcherServer.name(watcher_options),
           start:
             {WatcherServer, :start_link,
-             [{options.repo_mod, options.pub_sub_mod, watcher_options}]}
+             [{options.repo_mod, options.pub_sub_mod, options.dispatcher, watcher_options}]}
         }
       end)