Several small changes, resolves #20 and resolves #21

.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["mix.exs", "{config,lib}/**/*.{ex,exs}"]
+]

README.md

@@ -125,3 +125,9 @@ defmodule Consumer do
     connection_id: :connection_identifier,
     ...
 ```
+#### Specify generated exchanges type
+
+In your `config.exs` put:
+```
+config :tackle, exchange_type: :topic
+```

config/config.exs

@@ -21,10 +21,13 @@ use Mix.Config
 #     config :logger, level: :info
 #
 
+# :direct || :fanout || :topic || :headers || :match
+# config :tackle, exchange_type: :topic
+
 # It is also possible to import configuration files, relative to this
 # directory. For example, you can emulate configuration per environment
 # by uncommenting the line below and defining dev.exs, test.exs and such.
 # Configuration from the imported file will override the ones defined
 # here (which is why it is important to import them last).
 #
-#     import_config "#{Mix.env}.exs"
+import_config "#{Mix.env()}.exs"

config/dev.exs

@@ -0,0 +1 @@
+use Mix.Config

config/test.exs

@@ -0,0 +1,8 @@
+use Mix.Config
+
+config :logger,
+  backends: [{LoggerFileBackend, :error_log}]
+
+config :logger, :error_log,
+  path: "/dev/null",
+  level: :error

lib/tackle.ex

@@ -3,31 +3,31 @@ defmodule Tackle do
 
   require Logger
 
+  @impl Application
   def start(_type, _args) do
-    import Supervisor.Spec, warn: false
-
     children = [
-      worker(Tackle.Connection, []),
+      Tackle.Connection
     ]
 
     opts = [strategy: :one_for_one, name: Tackle.Supervisor]
     Supervisor.start_link(children, opts)
   end
 
-  def publish(message, options) do
+  def publish(message, options) when is_binary(message) do
     url = options[:url]
     exchange = options[:exchange]
     routing_key = options[:routing_key]
 
-    Logger.debug "Connecting to '#{url}'"
+    Logger.debug("Connecting to '#{Tackle.DebugHelper.safe_uri(url)}'")
     {:ok, connection} = AMQP.Connection.open(url)
     channel = Tackle.Channel.create(connection)
 
-    Logger.debug "Declaring an exchange '#{exchange}'"
-    AMQP.Exchange.direct(channel, exchange, durable: true)
+    Logger.debug("Declaring an exchange '#{exchange}'")
+    Tackle.Exchange.create_remote_exchange(channel, exchange)
 
     AMQP.Basic.publish(channel, exchange, routing_key, message, persistent: true)
 
+    AMQP.Channel.close(channel)
     AMQP.Connection.close(connection)
   end
 
@@ -40,5 +40,4 @@ defmodule Tackle do
 
     Tackle.Republisher.republish(url, queue, exchange, routing_key, count)
   end
-
 end

lib/tackle/channel.ex

@@ -4,8 +4,14 @@ defmodule Tackle.Channel do
 
   @prefetch_count 1
 
-  def create(connection) do create(connection, @prefetch_count) end
-  def create(connection, nil) do create(connection, @prefetch_count) end
+  def create(connection) do
+    create(connection, @prefetch_count)
+  end
+
+  def create(connection, nil) do
+    create(connection, @prefetch_count)
+  end
+
   def create(connection, prefetch_count) do
     {:ok, channel} = Channel.open(connection)
 
@@ -14,4 +20,7 @@ defmodule Tackle.Channel do
     channel
   end
 
+  def close(channel) do
+    Channel.close(channel)
+  end
 end

lib/tackle/connection.ex

@@ -1,17 +1,23 @@
 defmodule Tackle.Connection do
+  use Agent
   require Logger
 
   @moduledoc """
   Holds established connections.
   Each connection is identifed by name.
 
-  Connection name ':default' is speciall: it is NOT persisted ->
+  Connection name ':default' is special: it is NOT persisted ->
   each open() call with  :default connection name opens new connection
   (to preserve current behaviour).
   """
 
-  def start_link do
-    Agent.start_link(fn -> %{} end, name: __MODULE__)
+  def start_link, do: start_link([])
+
+  def start_link(opts) do
+    {cache, opts} = Keyword.pop(opts, :initial_value, %{})
+    opts = Keyword.merge([name: __MODULE__], opts)
+
+    Agent.start_link(fn -> cache end, opts)
   end
 
   @doc """
@@ -20,27 +26,46 @@ defmodule Tackle.Connection do
 
       open(:foo, [])
   """
-  def open(name, url) do open_(name, url) end
+  def open(name, url) do
+    do_open(name, url)
+  end
+
+  def close(conn) do
+    AMQP.Connection.close(conn)
+  end
+
+  def reset do
+    get_all()
+    |> Enum.each(fn {_name, conn} ->
+      Tackle.Connection.close(conn)
+      Agent.update(__MODULE__, fn _state -> %{} end)
+    end)
+
+    :ok
+  end
 
   @doc """
   Get a list of opened connections
   """
   def get_all do
-    Agent.get(__MODULE__, fn state -> state |> Map.to_list end)
+    Agent.get(__MODULE__, fn state -> state |> Map.to_list() end)
   end
 
-  defp open_(name=:default, url) do
+  defp do_open(name = :default, url) do
     connection = open(url)
-    Logger.info("Opening new connection #{inspect connection} for id: #{name}")
+    Logger.info("Opening new connection #{inspect(connection)} for id: #{name}")
     connection
   end
-  defp open_(name, url) do
+
+  defp do_open(name, url) do
     Agent.get(__MODULE__, fn state -> Map.get(state, name) end)
     |> case do
       nil ->
         open_and_persist(name, url)
+
       connection ->
-        Logger.info("Fetched existing connection #{inspect connection} for id: #{name}")
+        Logger.info("Fetched existing connection #{inspect(connection)} for id: #{name}")
+
         connection
         |> validate(name)
         |> reopen_on_validation_failure(name, url)
@@ -51,35 +76,42 @@ defmodule Tackle.Connection do
     case open(url) do
       response = {:ok, connection} ->
         Agent.update(__MODULE__, fn state -> Map.put(state, name, connection) end)
-        Logger.info("Opening new connection #{inspect connection} for id: #{name}")
+        Logger.debug("Opening new connection #{inspect(connection)} for id: #{name}")
         response
+
       error ->
         Logger.error("Failed to open new connection for id: #{name}: #{error}")
         error
     end
   end
 
   defp validate(connection, name) do
-    connection |> Map.get(:pid) |> validate_connection_process(connection, name)
+    connection.pid |> validate_connection_process(connection, name)
   end
 
   def reopen_on_validation_failure(state = {:error, _}, name, url) do
-    Logger.warn("Connection validation failed #{inspect state} for id: #{name}")
+    Logger.warn("Connection validation failed #{inspect(state)} for id: #{name}")
     Agent.update(__MODULE__, fn state -> Map.delete(state, name) end)
     open(name, url)
   end
+
   def reopen_on_validation_failure(connection, _name, _url) do
-    {:ok, connection} end
+    {:ok, connection}
+  end
 
   defp validate_connection_process(pid, connection, name) when is_pid(pid) do
-    pid |> Process.alive? |> validate_connection_process_rh(connection, name)
+    pid |> Process.alive?() |> validate_connection_process_rh(connection, name)
   end
+
   defp validate_connection_process(_pid, connection, name) do
     false |> validate_connection_process_rh(connection, name)
   end
 
-  defp validate_connection_process_rh(_alive? = true,  connection,  _name) do connection end
-  defp validate_connection_process_rh(_alive? = false, _connection, _name)  do
+  defp validate_connection_process_rh(_alive? = true, connection, _name) do
+    connection
+  end
+
+  defp validate_connection_process_rh(_alive? = false, _connection, _name) do
     {:error, :no_process}
   end