# 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.4.1"}]

add the `:spotter` application to the extra_applications:

  def application do
    [extra_applications: [:spotter]]

# 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.

# Examplee

  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.

  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:

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

    # Specify here a router that will be using during processing a message
      {"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 =, @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]}

    # 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}

    # 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)
  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:
    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)

# 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](