===============
AMQP Dispatcher
===============

.. image:: https://travis-ci.org/opschops/amqp-dispatcher.svg?branch=master
    :target: https://travis-ci.org/opschops/amqp-dispatcher

A daemon to run AMQP consumers

Requirements
============

* Python 2.6+

Installation
============

Using PIP:

From Github::

    pip install git+git://github.com/opschops/amqp-dispatcher.git@0.11.0#egg=amqp-dispatcher

From PyPI::

    pip install amqp-dispatcher==0.11.0


Running
=======

.. code:: bash

    amqp-dispatcher --config amqp-dispatcher-config.yml

The environment variable ``RABBITMQ_URL`` can also be used which will cause
attempt to connect to the defined data source name. Hosts are separated
via commas, and they are connected to in random order.

Note that you can specify a heartbeat via the ``heartbeat`` querystring value
on the ``RABBITMQ_URL``. By default, it is set to ``None``, which ensures that
the client respects the broker specified heartbeat settings. You may wish to
override this for a particular environment.

Additionally, you can specifically set a heartbeat override via the ``RABBITMQ_HEARTBEAT`` environment variable.
This will take precedence over the heartbeat set in ``RABBITMQ_URL``.

To see an example run:

.. code:: bash

    make example

Consumers
---------

Consumers are a class with 2 required methods: ``consume`` and ``shutdown``.
amqp-dispatcher will not monkey patch the environment, you will have to do
that yourself.

- ``consume``: ``consume`` is called once for each message being handled. It should take 2 parameters, a proxy for AMQP operations (``amqp``) and the message (``msg``).
- ``shutdown`` - ``shutdown`` is called before the instance of the consumer is removed. It takes a single argument ``exception`` which may be ``None``. If your consumer raises an exception while consuming the ``shutdown`` method will be called. Once ``shutdown`` is finished a new instance of your consumer will be created to replace the one that raised the exception. If you would like to rate limit instance replacement you can call ``gevent.sleep(X)`` to sleep for ``X`` seconds after a failure.


Example consumer:

.. code:: python

    class Consumer(object):

        def __init__(self):
            self.init_msg = "I've been initiliazed"

        def consume(self, amqp, msg):
            print 'Consuming message', msg.body
            gevent.sleep(1)
            val = random.random()
            if val > .8:
                print 'publishing'
                amqp.publish('test_exchange', 'test_routing_key', {}, 'New body!')
            if val < .5:
                raise ValueError()
            print 'Done sleeping'
            amqp.ack()

        def shutdown(self, exception=None):
            print 'Shut down'


Configuration
-------------

amqp-dispatcher will read environment variable for connection information and a
YAML file for worker configuration.

You can validate the yaml file configuration with the following command:

.. code:: bash

    amqp-dispatcher --validate --config amqp-dispatcher-config.yml

This will validate that the ``startup_handler`` and ``consumers`` exist and can be
imported. Note that if there is any logic contained outside of those functions, said
logic will be executed.

Environment Variables
---------------------

- ``RABBITMQ_URL``: Connection string of the form ``amqp://USER:PASS@HOST:PORT/VHOST``

Startup Configuration
---------------------

If you need to perform custom actions (configure your logging, create initial objects) you can add a startup handler.

This is configured in the config yml with the ``startup_handler`` option.

.. code:: yaml

    startup_handler: amqpdispatcher.example_startup:startup

Queue configuration
-------------------

Queues can be created on the fly by amqp dispatcher, and may bind existing exchanges on the fly as well.

There are a few obvious constraints:

* To create a non-passive queue (typical behavior) the current user must have ``configure=queue`` permission
* To bind to an exchange, the current user must have ``read`` permission on the binding exchange

Queue configuration is as follows:

- ``queue``: (required) name of the queue
- ``durable``: (optional) queue created in "durable" mode (default = True)
- ``auto_delete``: (optional) queue created in "auto_delete" mode (default = False), meaning it will be deleted automatically once all consumers disconnect from it (e.g. on restart)
- ``exclusive``: (optional) queue created in "exclusive" mode (default = False) meaning it will only be accessible by this process
- ``x_dead_letter_exchange``: (optional) name of dead letter exchange
- ``x_dead_letter_routing_key``: (optional) dead letter routing key
- ``x_max_length``: (optional) maximum length of ready messages. (default = INFINITE)
- ``x_expires``: (optional) How long a queue can be unused for before it is automatically deleted (milliseconds) (default=INFINITE)
- ``x_message_ttl``: (optional) How long a message published to a queue can live before it is discarded (milliseconds) (default=INFINITE)

Bindings
--------

``bindings``  should contain a list of ``exchange``/``routing_key`` pairs and defines the binding for the queue (there can be multiple)

A complete configuration example would look like:

.. code:: yaml

    queues:
      - queue: notify_mat_job
        durable: true
        auto_delete: false
        passive: true
        exclusive: false
        x_dead_letter_exchange: null
        x_dead_letter_routing_key: null
        x_max_length: null
        x_expires: null
        x_message_ttl: null
        bindings:
          - exchange: notify
            routing_key: transaction.*
          - exchange: notify
            routing_key: click.*

      - queue: notify_apsalar_job
        bindings:
          - exchange: notify
            routing_key: transaction.*
          - exchange: notify
            routing_key: click.*

Worker configuration
--------------------

Workers are autoloaded when AMQP Dispatcher starts. This means your worker must
be importable from the environment.

A complete configuration example would look like:

.. code:: yaml

    consumers:
      - consumer: workers.module:Consumer
        consumer_count: 1
        queue: test_queue
        prefetch_count: 2
      - consumer: workers.module_2:Consumer
        consumer_count: 2
        queue: test_queue_2
        prefetch_count: 10

``prefetch_count`` is the AMQP ``prefetch_count`` when consuming. The
``consumer_count`` is the number of instances of your consumer to handle messages
from that queue.  Connection pools are highly recommended. MySQL will require the
`MySQL Connector <http://pypi.python.org/pypi/mysql-connector-python>`_  instead of
``mysqldb`` in order for gevent to switch properly.

Pools can be created and attached to the consumer class during the ``__init__``. Example with SQLAlchemy

.. code:: python

    class Consumer(object):

        session_maker = None

        def __init__(self):
            self.session = None

            if Consumer._engine is None:
                print 'Creating session maker'
                Consumer._engine = create_engine(...)
                Consumer.sessionmaker = sessionmaker(bind=Consumer._engine)

And then a session created during the consume method.

.. code:: python

        def consume(self, proxy, msg):
            session = self.sessionmaker()
            # Do something with the session
            session.close()

Logging
-------

Logging is performed on the logger ``amqp-dispatcher``. The RabbitMQ connection
provided by Haigha will log on ``amqp-dispatcher.haigha``.

You can also configure the logger by using the ``LOGGING_FILE_CONFIG``
environment variable to specify a file config path. This will be used by
``logging.config.fileConfig`` before creating the initial logger.
