spotter

This project is focused on providing opportunities for implementing workers and middleware layers for AMQP queues which are actively used in you system. Workers as middlewares will help you to restrict an access to the certain message queues and doing pre-processing or validating data before passing it later to the next queue in the processing chain.

For example, it's very important for a cases when you should guaranteed that the data on each stage of pipeline will be correct and valid so that the last stage will send a response to the client as expected, instead of let it crash at some stage without sending a detailed error.

Features

• Restricting an access to the certain message queues (or resources) via checking permissions
• Pre-processing an input data before passing it to the next stage
• Monitoring and re-establishing failed AMQP connections

Installing

The package can be installed via adding the spotter dependency to your list of dependencies in mix.exs:

def deps do
  [{:spotter, "~> 0.6.1"}]
end

add the :spotter application to the extra_applications:

def application do
  [extra_applications: [:spotter]]
end

Configuration

By default Spotter reads environment configuration and trying to establish a AMQP connection with the following parameters:

• SPOTTER_AMQP_USERNAME - username. Default: guest
• SPOTTER_AMQP_PASSWORD - password. Default: guest
• SPOTTER_AMQP_HOST - host. Default: localhost
• SPOTTER_AMQP_PORT - port. Default: 5672
• SPOTTER_AMQP_VHOST - default virtual host. Default: /
• SPOTTER_AMQP_TIMEOUT - timeout, Default: 60000 milliseconds.

Also it is possible to specify other connections that can be found in AMQP client docs. Any of those arguments (that were mentioned in the documentation) can be specified in GenServer.start_link/3 function.

Example

1. Define a connection module which is going be used later

defmodule AppAMQPConnection do
  use Spotter.AMQP.Connection,
  otp_app: :custom_app
  # You can specify here queue, exchange and QoS parameters when it necessary.
  # However, will be better to store a configuration per each worker separately.
end

2. Implement your own worker, like here

defmodule CustomWorker do
  use Spotter.Worker,
  otp_app: :custom_app,
  connection: AppAMQPConnection

  # Also you could specify here the `queue` \ `exchange` \ `qos` options
  # instead of instantiating and binding the @queue_validate queue manually.
  # For more details see: https://github.com/Nebo15/rbmq

  @exchange "amqp.direct"
  @queue_validate "validate.queue"
  @queue_genstage "genstage.queue"

  # Specify here a router that will be using during processing a message
  @router Spotter.Router.new([
    {"my.test.endpoint", ["get", "post"]},
  ])

  # Specify here the queue that you want to use
  # `opts` will contain options (as a Map) that were specified in child_spec for supervisor 
  def configure(channel_name, _opts) do
    # For getting an access to the channel by its name use `get_channel(channel_name)` function
    channel = get_channel(channel_name)
  
    :ok = AMQP.Exchange.direct(channel, @exchange, durable: true)

    # An initial point where the worker do required stuff
    {:ok, queue_request} = AMQP.Queue.declare(channel, @queue_validate, durable: true, auto_delete: true)
    :ok = AMQP.Queue.bind(channel, @queue_validate, @exchange, routing_key: @queue_validate)

    # Queue for a valid messages
    {:ok, queue_forward} = AMQP.Queue.declare(channel, @queue_genstage, durable: true, auto_delete: true)
    :ok = AMQP.Queue.bind(channel, @queue_genstage, @exchange, routing_key: @queue_genstage)

    # Specify a consumer here
    {:ok, _} = AMQP.Basic.consume(channel, @queue_validate)

    # You must return here the tuple, where the first element is `:ok` atom, and the
    # second element whill be any type what you would like. The second element will
    # be set for the GenServer state, so that you can get an access to it via the
    # `state[:meta]` expression
    {:ok, [queue_request: queue_request, queue_forward: queue_forward]}
  end

  # Invoked when a message successfully consumed
  def handle_info({:basic_deliver, payload, %{delivery_tag: tag, reply_to: reply_to, headers: headers}}, state) do
    channel_name = state[:channel_name]
    spawn fn -> consume(channel_name, tag, reply_to, headers, payload) end
    {:noreply, state}
  end

  # Processing a consumed message
  defp consume(channel_name, tag, reply_to, headers, payload) do
    # Do some usefull stuff here ...

    # And don't forget to ack a processed message. Or perhaps even use nack 
    # when it will be neceessary.
    # We will wrap the call into `safe_run(channel_name, func)` call, so that it will retry 
    # the executed code when the used channel is failed
    safe_run(channel_name, fn(channel) -> AMQP.Basic.ack(channel, tag) end)
  end
end

Pay attention to this consume/5 method. I recommend to send async messages to GenServer that will be consumed later, so that when the message is processing a single thread wouldn't be blocked.
After that just specify this CustomWorker in your OTP application with supervisor and invoke GenServer.start_link/3.

3. Add the connection with the worker to your application

defmodule CustomApp do
  use Application

  # For more detail see: http://elixir-lang.org/docs/stable/elixir/Application.html
  def start(_type, _args) do
    # Define workers and child supervisors to be supervised
    children = [
      %{
        id: AppAMQPConnection, 
        start: {AppAMQPConnection, :start_link, []},
        restart: :transient
      },
      %{
        id: CustomWorker, 
        start: {CustomWorker, :start_link, []},
        restart: :transient
      }
    ]

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

Thanks

This package is heavily inspired and built on the top of the RBMQ package. The orignal project is published under the MIT license and you can find the source code here.