Use one service per exchange

renderedtext · Nov 23, 2023 · 85a7081 · 85a7081
1 parent e291fa1
commit 85a7081
Show file tree

Hide file tree

Showing 2 changed files with 67 additions and 16 deletions.
diff --git a/lib/tackle/multiconsumer.ex b/lib/tackle/multiconsumer.ex
@@ -51,7 +51,7 @@ defmodule Tackle.Multiconsumer do
           defmodule unquote(module_name) do
             use Tackle.Consumer,
               url: unquote(opts[:url]),
-              service: unquote(service),
+              service: "#{unquote(service)}.#{unquote(exchange)}",
               exchange: unquote(exchange),
               routing_key: unquote(routing_key),
               queue: unquote(queue),

diff --git a/test/tackle/multiconsumer_test.exs b/test/tackle/multiconsumer_test.exs
@@ -105,35 +105,31 @@ defmodule Tackle.MulticonsumerTest do
       {:ok, _pid} =
         MulticonsumerWithDynamicQueueName.start_link(process_name: {:global, make_ref()})
 
-      Tackle.publish("HELLO!",
-        url: "amqp://rabbitmq:5672",
-        exchange: "MulticonsumerWithMultipleHandlersExchange",
-        routing_key: "routing.key1"
-      )
+      {:ok, c} = Tackle.Connection.open(:publisher, "amqp://rabbitmq:5672")
+      channel = Tackle.Channel.create(c)
+      exchange1 = Tackle.Exchange.create(channel, "MulticonsumerWithMultipleHandlersExchange")
+      exchange2 = Tackle.Exchange.create(channel, "Exchange2")
+
+      Tackle.Exchange.publish(channel, exchange1, "HELLO!", "routing.key1")
+      Tackle.Exchange.publish(channel, exchange2, "HELLO!", "routing.key1")
 
       assert_receive "first handler fired", 1000
       assert_receive "first handler fired", 1000
       refute_receive "first handler fired", 1000
       refute_receive "second handler fired", 1000
       refute_receive "third handler fired", 1000
 
-      Tackle.publish("HELLO!",
-        url: "amqp://rabbitmq:5672",
-        exchange: "MulticonsumerWithMultipleHandlersExchange",
-        routing_key: "routing.key2"
-      )
+      Tackle.Exchange.publish(channel, exchange1, "HELLO!", "routing.key2")
+      Tackle.Exchange.publish(channel, exchange2, "HELLO!", "routing.key2")
 
       assert_receive "second handler fired", 1000
       assert_receive "second handler fired", 1000
       refute_receive "first handler fired", 1000
       refute_receive "second handler fired", 1000
       refute_receive "third handler fired", 1000
 
-      Tackle.publish("HELLO!",
-        url: "amqp://rabbitmq:5672",
-        exchange: "MulticonsumerWithMultipleHandlersExchange",
-        routing_key: "routing.key3"
-      )
+      Tackle.Exchange.publish(channel, exchange1, "HELLO!", "routing.key3")
+      Tackle.Exchange.publish(channel, exchange2, "HELLO!", "routing.key3")
 
       assert_receive "third handler fired", 1000
       assert_receive "third handler fired", 1000
@@ -142,4 +138,59 @@ defmodule Tackle.MulticonsumerTest do
       refute_receive "third handler fired", 1000
     end
   end
+
+  describe "Multiconsumer with service per exchange" do
+    defmodule MulticonsumerWithServicePerExchange do
+      use Tackle.Multiconsumer,
+        url: "amqp://rabbitmq:5672",
+        service: "MulticonsumerWithSerivcePerExchangeService",
+        routes: [
+          {"Exchange1", "routing.key4", :first_handler},
+          {"Exchange2", "routing.key4", :second_handler},
+        ]
+
+      def first_handler(_message) do
+        send(:checker, "first handler fired")
+      end
+
+      def second_handler(_message) do
+        send(:checker, "second handler fired")
+      end
+    end
+
+    test "one message should be process by one handler" do
+      Process.register(self(), :checker)
+
+      {:ok, _pid} =
+        MulticonsumerWithServicePerExchange.start_link(process_name: {:global, make_ref()})
+
+      {:ok, _pid} =
+        MulticonsumerWithServicePerExchange.start_link(process_name: {:global, make_ref()})
+
+      {:ok, c} = Tackle.Connection.open(:publisher, "amqp://rabbitmq:5672")
+      channel = Tackle.Channel.create(c)
+      exchange1 = Tackle.Exchange.create(channel, "Exchange1")
+      exchange2 = Tackle.Exchange.create(channel, "Exchange2")
+
+      Tackle.Exchange.publish(channel, exchange1, "HELLO!", "routing.key4")
+      Tackle.Exchange.publish(channel, exchange1, "HELLO!", "routing.key4")
+      Tackle.Exchange.publish(channel, exchange1, "HELLO!", "routing.key4")
+
+      assert_receive "first handler fired", 1000
+      assert_receive "first handler fired", 1000
+      assert_receive "first handler fired", 1000
+      refute_receive "second handler fired", 1000
+      refute_receive "first handler fired", 1000
+
+      Tackle.Exchange.publish(channel, exchange2, "HELLO!", "routing.key4")
+      Tackle.Exchange.publish(channel, exchange2, "HELLO!", "routing.key4")
+      Tackle.Exchange.publish(channel, exchange2, "HELLO!", "routing.key4")
+
+      assert_receive "second handler fired", 1000
+      assert_receive "second handler fired", 1000
+      assert_receive "second handler fired", 1000
+      refute_receive "first handler fired", 1000
+      refute_receive "second handler fired", 1000
+    end
+  end
 end