kombu/docs/userguide/simple.rst

.. _guide-simple:

==================
 Simple Interface
==================

.. contents::
    :local:


:mod:`kombu.simple` is a simple interface to AMQP queueing.
It is only slightly different from the :class:`~Queue.Queue` class in the
Python Standard Library, which makes it excellent for users with basic
messaging needs.

Instead of defining exchanges and queues, the simple classes only requires
two arguments, a connection channel and a name. The name is used as the
queue, exchange and routing key. If the need arises, you can specify
a :class:`~kombu.Queue` as the name argument instead.

In addition, the :class:`~kombu.Connection` comes with
shortcuts to create simple queues using the current connection:

.. code-block:: pycon

    >>> queue = connection.SimpleQueue('myqueue')
    >>> # ... do something with queue
    >>> queue.close()


This is equivalent to:

.. code-block:: pycon

    >>> from kombu.simple import SimpleBuffer

    >>> channel = connection.channel()
    >>> queue = SimpleBuffer(channel, 'mybuffer')
    >>> # ... do something with queue
    >>> channel.close()
    >>> queue.close()

.. _simple-send-receive:

Sending and receiving messages
==============================

The simple interface defines two classes; :class:`~kombu.simple.SimpleQueue`,
and :class:`~kombu.simple.SimpleBuffer`. The former is used for persistent
messages, and the latter is used for transient, buffer-like queues.
They both have the same interface, so you can use them interchangeably.

Here is an example using the :class:`~kombu.simple.SimpleQueue` class
to produce and consume logging messages:

.. code-block:: python

    import socket
    import datetime
    from time import time
    from kombu import Connection


    class Logger:

        def __init__(self, connection, queue_name='log_queue',
                serializer='json', compression=None):
            self.queue = connection.SimpleQueue(queue_name)
            self.serializer = serializer
            self.compression = compression

        def log(self, message, level='INFO', context={}):
            self.queue.put({'message': message,
                            'level': level,
                            'context': context,
                            'hostname': socket.gethostname(),
                            'timestamp': time()},
                            serializer=self.serializer,
                            compression=self.compression)

        def process(self, callback, n=1, timeout=1):
            for i in xrange(n):
                log_message = self.queue.get(block=True, timeout=1)
                entry = log_message.payload # deserialized data.
                callback(entry)
                log_message.ack() # remove message from queue

        def close(self):
            self.queue.close()


    if __name__ == '__main__':
        from contextlib import closing

        with Connection('amqp://guest:guest@localhost:5672//') as conn:
            with closing(Logger(conn)) as logger:

                # Send message
                logger.log('Error happened while encoding video',
                            level='ERROR',
                            context={'filename': 'cutekitten.mpg'})

                # Consume and process message

                # This is the callback called when a log message is
                # received.
                def dump_entry(entry):
                    date = datetime.datetime.fromtimestamp(entry['timestamp'])
                    print('[%s %s %s] %s %r' % (date,
                                                entry['hostname'],
                                                entry['level'],
                                                entry['message'],
                                                entry['context']))

                # Process a single message using the callback above.
                logger.process(dump_entry, n=1)