Add direct consumer support

lib/amqp/basic.ex

@@ -346,9 +346,17 @@ defmodule AMQP.Basic do
 
     consumer_pid = consumer_pid || self()
 
-    receiver = ReceiverManager.register_handler(chan.pid, consumer_pid, :consume)
+    pid =
+      case chan.consumer_type do
+        :selective ->
+          %{pid: pid} = ReceiverManager.register_handler(chan.pid, consumer_pid, :consume)
+          pid
 
-    case :amqp_channel.subscribe(chan.pid, basic_consume, receiver.pid) do
+        :direct ->
+          consumer_pid
+      end
+
+    case :amqp_channel.subscribe(chan.pid, basic_consume, pid) do
       basic_consume_ok(consumer_tag: consumer_tag) -> {:ok, consumer_tag}
       error -> {:error, error}
     end

lib/amqp/channel.ex

@@ -5,18 +5,28 @@ defmodule AMQP.Channel do
 
   alias AMQP.{Connection, Channel}
 
-  defstruct [:conn, :pid]
-  @type t :: %Channel{conn: Connection.t(), pid: pid}
+  defstruct [:conn, :pid, :consumer_type]
+  @type t :: %Channel{conn: Connection.t(), pid: pid, consumer_type: consumer_type}
+  @type consumer_type :: :direct | :selective
+  @type opts :: [consumer_type: consumer_type, consumer_module: module, consumer_init_args: any]
 
   @doc """
   Opens a new Channel in a previously opened Connection.
+
+  ## Options
+    * `:consumer_type` - specifies the type of consumer used as callback module.
+      available options: `:selective` or `:direct`. Defaults to `:selective`;
+    * `:consumer_module` - specifies consumer callback module. Used only with `:consumer_type: :direct` \
+      ignored otherwise. Defaults to `AMQP.Channel.DirectReceiver`;
+    * `:consumer_init_args` - arguments that will be passed to `init/1` function of
+      consumer callback module. Used only with `:consumer_type: :direct` ignored otherwise. \
+      If used with default `:consumer_module` it expects `t:Process.dest/0` of process \
+      receiving AMQP messages. Defaults to caller `pid`.
   """
-  @spec open(Connection.t()) :: {:ok, Channel.t()} | {:error, any}
-  def open(%Connection{pid: pid} = conn) do
-    case :amqp_connection.open_channel(pid) do
-      {:ok, chan_pid} -> {:ok, %Channel{conn: conn, pid: chan_pid}}
-      error -> error
-    end
+  @spec open(Connection.t(), opts) :: {:ok, Channel.t()} | {:error, any}
+  def open(%Connection{} = conn, opts \\ []) do
+    consumer_type = Keyword.get(opts, :consumer_type, :selective)
+    do_open_channel(conn, consumer_type, opts)
   end
 
   @doc """
@@ -29,4 +39,21 @@ defmodule AMQP.Channel do
       error -> {:error, error}
     end
   end
+
+  defp do_open_channel(conn, type = :selective, _opts) do
+    case :amqp_connection.open_channel(conn.pid) do
+      {:ok, chan_pid} -> {:ok, %Channel{consumer_type: type, conn: conn, pid: chan_pid}}
+      error -> error
+    end
+  end
+
+  defp do_open_channel(conn, type = :direct, opts) do
+    consumer_module = Keyword.get(opts, :consumer_module, AMQP.Channel.DirectReceiver)
+    consumer_init_args = Keyword.get(opts, :consumer_init_args, self())
+
+    case :amqp_connection.open_channel(conn.pid, {consumer_module, consumer_init_args}) do
+      {:ok, chan_pid} -> {:ok, %Channel{consumer_type: type, conn: conn, pid: chan_pid}}
+      error -> error
+    end
+  end
 end

lib/amqp/channel/direct_receiver.ex

@@ -0,0 +1,139 @@
+defmodule AMQP.Channel.DirectReceiver do
+  @moduledoc false
+
+  #  Direct consumer callback module that implements `:amqp_gen_consumer` behavior.
+  #  Based on `amqp_direct_consumer.erl`.
+  #  For more information see: https://github.com/rabbitmq/rabbitmq-erlang-client/blob/master/src/amqp_direct_consumer.erl .
+
+  import AMQP.Core
+  @behaviour :amqp_gen_consumer
+
+  #########################################################
+  ### amqp_gen_consumer callbacks
+  #########################################################
+
+  @doc false
+  def init(consumer) do
+    _ref = Process.monitor(consumer)
+    {:ok, consumer}
+  end
+
+  @doc false
+  def handle_consume(basic_consume(), _pid, consumer) do
+    # silently discard
+    {:ok, consumer}
+  end
+
+  @doc false
+  def handle_consume_ok(basic_consume_ok(consumer_tag: consumer_tag), _args, consumer) do
+    _ = send(consumer, {:basic_consume_ok, %{consumer_tag: consumer_tag}})
+    {:ok, consumer}
+  end
+
+  @doc false
+  def handle_cancel(basic_cancel(consumer_tag: consumer_tag, nowait: no_wait), consumer) do
+    _ = send(consumer, {:basic_cancel, %{consumer_tag: consumer_tag, no_wait: no_wait}})
+    {:ok, consumer}
+  end
+
+  @doc false
+  def handle_cancel_ok(basic_cancel_ok(consumer_tag: consumer_tag), _args, consumer) do
+    _ = send(consumer, {:basic_cancel_ok, %{consumer_tag: consumer_tag}})
+    {:ok, consumer}
+  end
+
+  @doc false
+  def handle_server_cancel(basic_cancel(consumer_tag: consumer_tag, nowait: no_wait), consumer) do
+    _ = send(consumer, {:basic_cancel, %{consumer_tag: consumer_tag, no_wait: no_wait}})
+    {:ok, consumer}
+  end
+
+  @doc false
+  def handle_deliver(
+        basic_deliver(
+          consumer_tag: consumer_tag,
+          delivery_tag: delivery_tag,
+          redelivered: redelivered,
+          exchange: exchange,
+          routing_key: routing_key
+        ),
+        amqp_msg(
+          props:
+            p_basic(
+              content_type: content_type,
+              content_encoding: content_encoding,
+              headers: headers,
+              delivery_mode: delivery_mode,
+              priority: priority,
+              correlation_id: correlation_id,
+              reply_to: reply_to,
+              expiration: expiration,
+              message_id: message_id,
+              timestamp: timestamp,
+              type: type,
+              user_id: user_id,
+              app_id: app_id,
+              cluster_id: cluster_id
+            ),
+          payload: payload
+        ),
+        consumer
+      ) do
+    send(
+      consumer,
+      {:basic_deliver, payload,
+       %{
+         consumer_tag: consumer_tag,
+         delivery_tag: delivery_tag,
+         redelivered: redelivered,
+         exchange: exchange,
+         routing_key: routing_key,
+         content_type: content_type,
+         content_encoding: content_encoding,
+         headers: headers,
+         persistent: delivery_mode == 2,
+         priority: priority,
+         correlation_id: correlation_id,
+         reply_to: reply_to,
+         expiration: expiration,
+         message_id: message_id,
+         timestamp: timestamp,
+         type: type,
+         user_id: user_id,
+         app_id: app_id,
+         cluster_id: cluster_id
+       }}
+    )
+
+    {:ok, consumer}
+  end
+
+  @doc false
+  def handle_deliver(basic_deliver(), _args, _ctx, _consumer) do
+    # there's no support for direct connection
+    # this callback implementation should be added with library support
+    {:error, :undefined}
+  end
+
+  @doc false
+  def handle_info({:DOWN, _mref, :process, consumer, :normal}, consumer) do
+    {:ok, consumer}
+  end
+
+  def handle_info({:DOWN, _mref, :process, consumer, info}, consumer) do
+    {:error, {:consumer_died, info}, consumer}
+  end
+
+  def handle_info(down = {:DOWN, _, _, _, _}, consumer) do
+    _ = send(consumer, down)
+    {:ok, consumer}
+  end
+
+  @doc false
+  def handle_call(_req, _from, consumer) do
+    {:reply, {:error, :undefined}, consumer}
+  end
+
+  @doc false
+  def terminate(_reason, consumer), do: consumer
+end