nebo15/rbmq

Simple API for spawning RabbitMQ Producers and Consumers.

RBMQ

Simple and easy creation of producers and consumers for RabbitMQ. Written over AMQP

Installation

The package can be installed as:

1. Add rbmq to your list of dependencies in mix.exs:

def deps do
  [{:rbmq, "~> 0.2.2"}]
end

2. Ensure rbmq is started before your application:

def application do
  [applications: [:rbmq]]
end

Configuration

You can define connection configuration in your config.exs:

config :my_app, MyAMQPConnection,
  host: {:system, "AMQP_HOST", "localhost"},
  port: {:system, "AMQP_PORT", 5672},
  username: {:system, "AMQP_USER", "guest"},
  password: {:system, "AMQP_PASSWORD", "guest"},
  virtual_host: {:system, "AMQP_VHOST", "/"},
  connection_timeout: {:system, "AMQP_TIMEOUT", 15_000},

RBMQ support linking to runtime environment conflagration via {:system, "ENV_VAR_NAME", "default_value"} and {:system, "ENV_VAR_NAME"} tuples. But are free to set raw values whenever you need.

By default RBMQ read environment configuration to establish AMQP connection:

• AMQP_HOST - host, default: localhost
• AMQP_PORT - port, default: 5672
• AMQP_USER - username, default: guest
• AMQP_PASSWORD - password, default: guest
• AMQP_VHOST - default vhost, default: /
• AMQP_TIMEOUT - timeout, default: 15 sec.

Other connections settings can be found in AMQP client docs.

Usage

1. Define your connection

defmodule MyAMQPConnection do
  use RBMQ.Connection,
    otp_app: :my_app
    # Optionally you can define queue params right here,
    # but it's better to do so in producer and consumer separately
end

2. Define your Producer and/or Consumer

defmodule MyProducer do
  use RBMQ.Producer,
    connection: MyAMQPConnection,

    # Queue params
    queue: [
      name: "prodcer_queue",
      error_name: "prodcer_queue_errors",
      routing_key: "prodcer_queue",
      durable: false
    ],
    exchange: [
      name: "prodcer_queue_exchange",
      type: :direct,
      durable: false
    ]
end

defmodule MyConsumer do
  use RBMQ.Consumer,
    connection: MyAMQPConnection,

    # Queue params
    queue: [
      name: "consomer_queue",
      durable: false
    ],
    qos: [
      prefetch_count: 10
    ]

  def consume(_payload, [tag: tag, redelivered?: _redelivered]) do
    ack(tag)
  end
end

Pay attention to consume/2 method. Write your consuming logic there. We recommend to send async messages to GenServer that will consume them, so queue read wouldn't be blocked by a single thread.

If your queue required acknowledgements, use ack\1 and nack\1 methods.

3. Add everything to your application supervisor:

defmodule MyApp do
  use Application

  # See http://elixir-lang.org/docs/stable/elixir/Application.html
  # for more information on OTP Applications
  def start(_type, _args) do
    import Supervisor.Spec, warn: false

    # Define workers and child supervisors to be supervised
    children = [
      # Start the AMQP connection
      supervisor(MyAMQPConnection, []),
      # Start producer and consumer
      worker(MyProducer, []),
      worker(MyConsumer, []),
    ]

    opts = [strategy: :one_for_one, name: AssetProcessor.API.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

rbmq's Issues

Change config usage

It's better to move config definitions to target application, it's very useful when we use umbrella apps.

Bad:

config :rbmq, params

Good:

config :ap_api, :amqp, params

Also we should consider config setup totally to application, moving it out of this module.

Refactoring for configurations and stable tests

Code not matching published hex library

I noticed the code in the hex repo is version 0.30 but it does not seem to match the code here.

Handle shutdown correctly

=SUPERVISOR REPORT==== 26-Aug-2016::21:24:45 ===
     Supervisor: {local,producer_supervisor}
     Context:    child_terminated
     Reason:     shutdown
     Offender:   [{pid,<0.1685.0>},
                  {id,'Elixir.Rbmq.Producer'},
                  {mfargs,
                      {'Elixir.Rbmq.Producer',start_link,
                          [<<"p1">>,<<"os_decision_queue">>,nil,nil]}},
                  {restart_type,permanent},
                  {shutdown,5000},
                  {child_type,worker}]

Questions about the current state of the package

Hello,

I'm curious about using this package in my own project, but noticed that the package isn't updated a long time ago and doesn't developing further. So, I have a couple of questions related to the package:

• Do you have any plans to support this project? It not also related to the adding functionality, but updating the requirements as well (for example, the amqp package which is actively it using).
• If the project is being no longer supported, then can I reuse some pieces of this codebase in a custom package? But of course, in this case I should to mention about re-using the code from this library. Perhaps in AUTHORS file in the custom package? Or maybe somewhere else?

Follow naming conventions

Rbmq -> RBMQ since common names can be uppercases by guidelines.

Hard-coded consumer prefetch-count