Packages

yggdrasil

1.1.0
6.0.3 6.0.2 6.0.1 6.0.0 5.0.2 5.0.1 5.0.0 4.1.6 4.1.5 4.1.4 4.1.3 4.1.2 4.1.1 4.1.0 4.0.0 3.3.4 3.3.3 3.3.2 3.3.1 3.3.0 3.2.3 3.2.2 3.2.1 3.2.0 3.1.0 3.0.3 3.0.2 3.0.1 3.0.0 2.1.0 2.0.8 2.0.7 2.0.6 2.0.5 2.0.4 2.0.3 2.0.2 2.0.1 2.0.0 1.2.3 1.2.2 1.2.1 1.1.1 1.1.0 1.0.0

Agnostic pub/sub with Redis, RabbitMQ and Postgres support

HexDocs

Current section

Files

Jump to
Readme
45 Versions
2 Dependencies
Dependants
Files
Activity
yggdrasil lib yggdrasil broker rabbitmq.ex
Raw

lib/yggdrasil/broker/rabbitmq.ex 

defmodule Yggdrasil.Broker.RabbitMQ do

  use Yggdrasil.Broker



  def subscribe({exchange, routing_key}, _callback) do

    config = Yggdrasil.Broker.RabbitMQConfig.fetch_env



    {:ok, conn} = AMQP.Connection.open(username: config.username,

                                       password: config.password,

                                       host: config.host, port: config.port,

                                       virtual_host: config.virtual_host) 

    {:ok, chan} = AMQP.Channel.open conn

    {:ok, %{queue: queue}} = AMQP.Queue.declare(chan, "", exclusive: true)

    AMQP.Queue.bind(chan, queue, exchange, routing_key: routing_key)

    {:ok, _consumer_tag} = AMQP.Basic.consume chan, queue

    {:ok, %{connection: conn, channel: chan}}

  end



  def unsubscribe(%{connection: conn, channel: chan}) do

    AMQP.Channel.close chan

    AMQP.Connection.close conn

  end



  def handle_message(_conn, {:basic_consume_ok, %{consumer_tag: _tag}}), do:

    :subscribed

  def handle_message(_conn, {:basic_cancel, _}), do:

    {:stop, :shutdown}

  def handle_message(%{channel: chan},

                     {:basic_deliver, message,

                      %{delivery_tag: tag, redelivered: redelivered}}) do

    try do

      spawn fn -> AMQP.Basic.ack(chan, tag) end

      {:message, message}

    rescue

      _exception ->

        spawn fn -> AMQP.Basic.reject(chan, tag, requeue: not redelivered) end

        {:stop, :shutdown}

    end

  end

  def handle_message(_conn, _ignored), do:

    :whatever

end