kombu server.py

gistfile1.txt

from kombu import Connection, Exchange, Queue
import logging

LOG = logging.getLogger(__name__)

cvworker_exchange = Exchange('cvworker', 'direct', durable=True)
queue = Queue('video', exchange=cvworker_exchange, routing_key='model1')
return_queue = Queue('return_queue')


def process_request(request, message):
    """Process incoming request."""
    LOG.debug("Start processing request {0}.".format(request))

    correlation_id = message.properties.get('correlation_id', None)
    if not correlation_id:
        LOG.error("The 'correlation_id' message property is missing.")
        return
    else:
        LOG.debug("Correlation id: {0}".format(correlation_id))

    reply_to = message.properties.get('reply_to')
    if not reply_to:
        LOG.error("The 'reply_to' message property is missing.")
        return
    else:
        LOG.debug("Reply to: {0}".format(reply_to))

    # execute function
    #
    #
    #
    response = 'response'
        with Connection('amqp://guest:guest@localhost//') as conn:
        producer = conn.Producer(serializer='json')
        producer.publish(
            body=response,
            exchange='',
            correlation_id=correlation_id,
            declare=[return_queue]
        )


def on_request(request, message):
    """This method is automatically called when a request is incoming.

    :param request: the body of the amqp message already deserialized by kombu

    :param message: the plain amqp kombu.message with additional information
    """
    LOG.debug("Got request: {0}".format(request))
    try:
        message.ack()
    except Exception:
        LOG.exception("Failed to acknowledge AMQP message.")
    else:
        LOG.debug("AMQP message acknowledged.")

        # process request
        process_request(request, message)

with Connection('amqp://guest:guest@localhost//') as conn:
    with conn.Consumer(queues=queue, callback=[on_request]):
        while True:
            conn.drain_events()