Allow config to be passed to topology functions

lib/conduit/broker.ex

@@ -72,7 +72,7 @@ defmodule Conduit.Broker do
           |> Application.get_env(__MODULE__, [])
           |> Keyword.merge(opts)
 
-        Conduit.Broker.init(__MODULE__, topology(), subscribe_routes(), config)
+        Conduit.Broker.init(__MODULE__, topology(config), subscribe_routes(), config)
       end
     end
   end

lib/conduit/broker/dsl.ex

@@ -220,7 +220,9 @@ defmodule Conduit.Broker.DSL do
   defmacro __before_compile__(env) do
     quote do
       import Conduit.Plug.MessageActions
-      unquote(Topology.methods())
+      require Conduit.Broker.Topology
+
+      Conduit.Broker.Topology.deftopology()
       unquote(Pipeline.methods())
       unquote(IncomingScope.methods(env.module))
       unquote(OutgoingScope.methods(env.module))

lib/conduit/broker/topology.ex

@@ -47,16 +47,18 @@ defmodule Conduit.Broker.Topology do
     end
   end
 
-  def methods do
-    quote unquote: false do
-      topology =
-        @topology
-        |> Enum.map(fn %{__struct__: module} = data ->
-          module.escape(data)
-        end)
-        |> Enum.reverse()
+  defmacro deftopology() do
+    topology =
+      __CALLER__.module
+      |> Module.get_attribute(:topology)
+      |> Enum.map(fn %{__struct__: module} = data ->
+        module.escape(data)
+      end)
+      |> Enum.reverse()
 
-      def topology do
+    # Generate topology function and replace body with constructed one
+    quote do
+      def topology(config) do
         unquote(topology)
       end
     end

lib/conduit/broker/topology/exchange.ex

@@ -24,8 +24,9 @@ defmodule Conduit.Broker.Topology.Exchange do
   end
 
   @doc false
+  # Conduit.Topology.Exchange.new(name, opts, config)
   def escape(%__MODULE__{} = exchange) do
     quote(do: Conduit.Topology.Exchange.new())
-    |> put_elem(2, [exchange.name, exchange.opts])
+    |> put_elem(2, [exchange.name, exchange.opts, Macro.var(:config, Conduit.Broker.Topology)])
   end
 end

lib/conduit/broker/topology/queue.ex

@@ -18,9 +18,9 @@ defmodule Conduit.Broker.Topology.Queue do
   end
 
   @doc false
-  # Conduit.Topology.Queue.new(name, opts)
+  # Conduit.Topology.Queue.new(name, opts, config)
   def escape(%__MODULE__{} = queue) do
     quote(do: Conduit.Topology.Queue.new())
-    |> put_elem(2, [queue.name, queue.opts])
+    |> put_elem(2, [queue.name, queue.opts, Macro.var(:config, Conduit.Broker.Topology)])
   end
 end

lib/conduit/topology/exchange.ex

@@ -8,6 +8,7 @@ defmodule Conduit.Topology.Exchange do
         }
   @type name :: String.t() | (() -> String.t())
   @type opts :: Keyword.t() | (() -> Keyword.t())
+  @type config :: Keyword.t()
 
   defstruct name: nil, opts: []
 
@@ -18,19 +19,27 @@ defmodule Conduit.Topology.Exchange do
 
   ## Examples
 
-      iex> Conduit.Topology.Exchange.new("topic")
-      %Conduit.Topology.Exchange{name: "topic", opts: []}
-      iex> Conduit.Topology.Exchange.new("topic", type: :fanout)
+      iex> Conduit.Topology.Exchange.new("topic", [type: :fanout], [])
       %Conduit.Topology.Exchange{name: "topic", opts: [type: :fanout]}
-      iex> Conduit.Topology.Exchange.new(fn -> "dynamic.name" end, fn -> [type: :fanout] end)
+      iex> Conduit.Topology.Exchange.new(fn -> "dynamic.name" end, fn -> [type: :fanout] end, [])
+      %Conduit.Topology.Exchange{name: "dynamic.name", opts: [type: :fanout]}
+      iex> config = [name: "dynamic.name", opts: [type: :fanout]]
+      iex> Conduit.Topology.Exchange.new(fn config -> config[:name] end, fn config -> config[:opts] end, config)
       %Conduit.Topology.Exchange{name: "dynamic.name", opts: [type: :fanout]}
   """
-  @spec new(name, opts) :: t()
-  def new(name, opts \\ [])
-  def new(name, opts) when is_function(name), do: new(name.(), opts)
-  def new(name, opts) when is_function(opts), do: new(name, opts.())
+  @spec new(name, opts, config) :: t()
+  def new(name, opts, config) when is_function(name), do: new(eval(name, config), opts, config)
+  def new(name, opts, config) when is_function(opts), do: new(name, eval(opts, config), config)
 
-  def new(name, opts) when is_binary(name) and is_list(opts) do
+  def new(name, opts, _config) when is_binary(name) and is_list(opts) do
     %__MODULE__{name: name, opts: opts}
   end
+
+  defp eval(fun, config), do: eval(:erlang.fun_info(fun, :arity), fun, config)
+
+  defp eval({:arity, 0}, fun, _config), do: fun.()
+  defp eval({:arity, 1}, fun, config), do: fun.(config)
+  defp eval({:arity, arity}, _fun, _config) do
+    raise Conduit.BadArityError, "Exchange declared with function that has invalidy arity. Expected 0 or 1, got #{arity}"
+  end
 end

lib/conduit/topology/queue.ex

@@ -8,6 +8,7 @@ defmodule Conduit.Topology.Queue do
         }
   @type name :: String.t() | (() -> String.t())
   @type opts :: Keyword.t() | (() -> Keyword.t())
+  @type config :: Keyword.t()
 
   defstruct name: nil, opts: []
 
@@ -18,19 +19,27 @@ defmodule Conduit.Topology.Queue do
 
   ## Examples
 
-    iex> Conduit.Topology.Queue.new("my_app.event")
-    %Conduit.Topology.Queue{name: "my_app.event", opts: []}
-    iex> Conduit.Topology.Queue.new("my_app.event", from: ["my_app.event_queue"])
+    iex> Conduit.Topology.Queue.new("my_app.event", [from: ["my_app.event_queue"]], [])
     %Conduit.Topology.Queue{name: "my_app.event", opts: [from: ["my_app.event_queue"]]}
-    iex> Conduit.Topology.Queue.new(fn -> "dynamic.name" end, fn -> [from: ["my_app.event_queue"]] end)
+    iex> Conduit.Topology.Queue.new(fn -> "dynamic.name" end, fn -> [from: ["my_app.event_queue"]] end, [])
+    %Conduit.Topology.Queue{name: "dynamic.name", opts: [from: ["my_app.event_queue"]]}
+    iex> config = [name: "dynamic.name", opts: [from: ["my_app.event_queue"]]]
+    iex> Conduit.Topology.Queue.new(fn config -> config[:name] end, fn -> config[:opts] end, config)
     %Conduit.Topology.Queue{name: "dynamic.name", opts: [from: ["my_app.event_queue"]]}
   """
-  @spec new(name, opts) :: t()
-  def new(name, opts \\ [])
-  def new(name, opts) when is_function(name), do: new(name.(), opts)
-  def new(name, opts) when is_function(opts), do: new(name, opts.())
+  @spec new(name, opts, config) :: t()
+  def new(name, opts, config) when is_function(name), do: new(eval(name, config), opts, config)
+  def new(name, opts, config) when is_function(opts), do: new(name, eval(opts, config), config)
 
-  def new(name, opts) when is_binary(name) and is_list(opts) do
+  def new(name, opts, _config) when is_binary(name) and is_list(opts) do
     %__MODULE__{name: name, opts: opts}
   end
+
+  defp eval(fun, config), do: eval(:erlang.fun_info(fun, :arity), fun, config)
+
+  defp eval({:arity, 0}, fun, _config), do: fun.()
+  defp eval({:arity, 1}, fun, config), do: fun.(config)
+  defp eval({:arity, arity}, _fun, _config) do
+    raise Conduit.BadArityError, "Queue declared with function that has invalidy arity. Expected 0 or 1, got #{arity}"
+  end
 end

test/conduit/broker/dsl_test.exs

@@ -18,6 +18,7 @@ defmodule Conduit.Broker.DSLTest do
       exchange "amq.topic"
 
       queue "my_app.created.stuff", from: ["#.created.stuff"]
+      queue fn config -> config[:node] <> ".dynamic" end, fn config -> [from: ["#.#{config[:node]}.stuff"]] end
     end
 
     pipeline :incoming do
@@ -50,11 +51,12 @@ defmodule Conduit.Broker.DSLTest do
     end
   end
 
-  describe ".topology" do
+  describe "topology/1" do
     test "returns a list of everything to setup" do
-      assert Broker.topology() == [
+      assert Broker.topology([node: "node1"]) == [
                %Conduit.Topology.Exchange{name: "amq.topic", opts: []},
-               %Conduit.Topology.Queue{name: "my_app.created.stuff", opts: [from: ["#.created.stuff"]]}
+               %Conduit.Topology.Queue{name: "my_app.created.stuff", opts: [from: ["#.created.stuff"]]},
+               %Conduit.Topology.Queue{name: "node1.dynamic", opts: [from: ["#.node1.stuff"]]}
              ]
     end
   end