aioamqp_consumer 0.3.4
consumer/producer/rpc library built over aioamqp
Keywords: amqp

aioamqp_consumer
================

:info: consumer/producer/rpc library built over aioamqp

.. image:: https://img.shields.io/travis/aio-libs/aioamqp_consumer.svg
    :target: https://travis-ci.org/aio-libs/aioamqp_consumer

.. image:: https://img.shields.io/pypi/v/aioamqp_consumer.svg
    :target: https://pypi.python.org/pypi/aioamqp_consumer

Installation
------------

.. code-block:: shell

    pip install aioamqp_consumer

Consumer/Producer usage
-----------------------

.. code-block:: python

    import asyncio

    from aioamqp_consumer import Consumer, Producer


    async def task(payload, properties):
        await asyncio.sleep(1)
        print(payload)


    async def main():
        amqp_url = 'amqp://guest:guest@127.0.0.1:5672//'
        amqp_queue = 'your-queue-here'
        queue_kwargs = {
            'durable': True,
        }
        # https://aioamqp.readthedocs.io/en/latest/api.html#aioamqp.connect
        amqp_kwargs = {}

        async with Producer(amqp_url, amqp_kwargs=amqp_kwargs) as producer:
            for _ in range(5):
                await producer.publish(
                    b'hello',
                    amqp_queue,
                    queue_kwargs=queue_kwargs,
                )

        consumer = Consumer(
            amqp_url,
            task,
            amqp_queue,
            queue_kwargs=queue_kwargs,
            amqp_kwargs=amqp_kwargs,
        )
        await consumer.scale(20)  # scale up to 20 background coroutines
        await consumer.scale(5)  # downscale to 5 background coroutines
        # wait for rabbitmq queue is empty and all local messages are processed
        await consumer.join()
        consumer.close()
        await consumer.wait_closed()


    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

RPC usage
---------

.. code-block:: python

    import asyncio

    from aioamqp_consumer import RpcClient, RpcServer, rpc

    payload = b'test'


    @rpc(queue_name='random_queue')
    async def method(payload):
        print(payload)
        return payload


    async def main():
        amqp_url = 'amqp://guest:guest@127.0.0.1:5672//'

        server = RpcServer(amqp_url, method=method)

        client = RpcClient(amqp_url)

        ret = await client.wait(method(payload))

        assert ret == payload

        await client.close()

        await server.stop()


    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())
    loop.close()

For built-in json encoding/decoding, take a look on `aioamqp_consumer.json_rpc`

For production deploying `aioamqp_consumer.Consumer`/`aioamqp_consumer.RpcServer` there is built-in simpler runner:

.. code-block:: python

    from aioamqp_consumer import RpcServer, json_rpc

    amqp_url = 'amqp://guest:guest@127.0.0.1:5672//'


    @json_rpc(queue_name='random_queue')
    async def square(*, x):
        ret = x ** 2

        print(x, ret)

        return ret

    if __name__ == '__main__':
        RpcServer(amqp_url, method=square).run()

Thanks
------

The library was donated by `Ocean S.A. <https://ocean.io/>`_

Thanks to the company for contribution.