feat: allow using custom delimiters for pubsub topics

.formatter.exs

@@ -74,6 +74,7 @@ spark_locals_without_parens = [
   define_calculation: 2,
   define_for: 1,
   delay_global_validations?: 1,
+  delimiter: 1,
   description: 1,
   destination_attribute: 1,
   destination_attribute_on_join_resource: 1,

documentation/topics/pub_sub.md

@@ -68,6 +68,18 @@ Would produce the following messages, given a `team_id` of 1, a `tenant` of `org
 "org_1:updated"
 ```
 
+## Custom Delimiters
+
+It's possible to change the default delimiter used when generating topics. This is useful when working with message brokers
+like RabbitMQ, which rely on a different set of delimiters for routing.
+
+
+```elixir
+pub_sub do
+  delimiter "."
+end
+```
+
 ## Named Pubsub modules
 
 If you are using a phoenix `Endpoint` module for pubsub then this is unnecessary. If you want to use a custom pub sub started with something like `{Phoenix.PubSub, name: MyName}`, then you can provide `MyName` to

lib/ash/notifier/pub_sub/info.ex

@@ -16,12 +16,17 @@ defmodule Ash.Notifier.PubSub.Info do
     Spark.Dsl.Extension.get_opt(resource, [:pub_sub], :prefix, nil)
   end
 
+  @doc "The delimiter to use when generating message topics"
+  def delimiter(resource) do
+    Spark.Dsl.Extension.get_opt(resource, [:pub_sub], :delimiter, ":")
+  end
+
   @doc "The pubsub name for a resource"
   def name(resource) do
     Spark.Dsl.Extension.get_opt(resource, [:pub_sub], :name, nil)
   end
 
-  @doc "The broadcast type for aresource"
+  @doc "The broadcast type for a resource"
   def broadcast_type(resource) do
     Spark.Dsl.Extension.get_opt(resource, [:pub_sub], :broadcast_type, nil)
   end

lib/ash/notifier/pub_sub/pub_sub.ex

@@ -70,6 +70,10 @@ defmodule Ash.Notifier.PubSub do
         doc:
           "A prefix for all pubsub messages, e.g `users`. A message with `created` would be published as `users:created`"
       ],
+      delimiter: [
+        type: :string,
+        doc: "A delimiter for building topics. Default is a colon (:)"
+      ],
       broadcast_type: [
         type: {:one_of, [:notification, :phoenix_broadcast, :broadcast]},
         default: :notification,
@@ -106,6 +110,8 @@ defmodule Ash.Notifier.PubSub do
 
   use Ash.Notifier
 
+  alias Ash.Notifier.PubSub.Info
+
   @doc false
   def notify(%Ash.Notifier.Notification{resource: resource} = notification) do
     resource
@@ -118,16 +124,17 @@ defmodule Ash.Notifier.PubSub do
     debug? = Application.get_env(:ash, :pub_sub)[:debug?] || false
     event = publish.event || to_string(notification.action.name)
     prefix = prefix(notification.resource) || ""
+    delimiter = Info.delimiter(notification.resource)
 
     topics =
       publish.topic
-      |> fill_template(notification)
+      |> fill_template(notification, delimiter)
       |> Enum.map(fn topic ->
         case {prefix, topic} do
           {"", ""} -> ""
           {prefix, ""} -> prefix
           {"", topic} -> topic
-          {prefix, topic} -> "#{prefix}:#{topic}"
+          {prefix, topic} -> "#{prefix}#{delimiter}#{topic}"
         end
       end)
 
@@ -188,13 +195,13 @@ defmodule Ash.Notifier.PubSub do
     end
   end
 
-  defp fill_template(topic, _) when is_binary(topic), do: [topic]
+  defp fill_template(topic, _notification, _delimiter) when is_binary(topic), do: [topic]
 
-  defp fill_template(topic, notification) do
+  defp fill_template(topic, notification, delimiter) do
     topic
     |> all_combinations_of_values(notification, notification.action.type)
     |> Enum.map(&List.flatten/1)
-    |> Enum.map(&Enum.join(&1, ":"))
+    |> Enum.map(&Enum.join(&1, delimiter))
     |> Enum.uniq()
   end
 

test/notifier/pubsub_test.exs

@@ -45,12 +45,44 @@ defmodule Ash.Test.Notifier.PubSubTest do
     end
   end
 
+  defmodule User do
+    @moduledoc false
+    use Ash.Resource,
+      data_layer: Ash.DataLayer.Ets,
+      notifiers: [
+        Ash.Notifier.PubSub
+      ]
+
+    pub_sub do
+      module PubSub
+      prefix "users"
+      delimiter "."
+
+      publish :create, [:id, "created"]
+    end
+
+    ets do
+      private?(true)
+    end
+
+    actions do
+      defaults [:create]
+    end
+
+    attributes do
+      uuid_primary_key :id
+
+      attribute :name, :string
+    end
+  end
+
   defmodule Registry do
     @moduledoc false
     use Ash.Registry
 
     entries do
       entry Post
+      entry User
     end
   end
 
@@ -144,4 +176,14 @@ defmodule Ash.Test.Notifier.PubSubTest do
     message = "post:foo:#{new_id}"
     assert_receive {:broadcast, ^message, "update_pkey", %Ash.Notifier.Notification{}}
   end
+
+  test "publishing a message with a different delimiter" do
+    user =
+      User
+      |> Ash.Changeset.new(%{name: "Dave"})
+      |> Api.create!()
+
+    message = "users.#{user.id}.created"
+    assert_receive {:broadcast, ^message, "create", %Ash.Notifier.Notification{}}
+  end
 end