AMQPDaemon

This module provides generic AMQP daemon and builder of common connection informations, which are defined as constants in edeposit.amqp.settings.

Daemon is used by edeposit_amqp_alephdaemon and edeposit_amqp_calibredaemon.

class edeposit.amqp.amqpdaemon.AMQPDaemon(con_param, queue, out_exch, out_key, react_fn, glob)

Bases: edeposit.amqp.pikadaemon.PikaDaemon

Parameters:

• con_param (ConnectionParameters) – see getConParams() for details
• queue (str) – name of the queue
• out_exch (str) – name of the exchange for outgoing messages
• out_key (str) – what key will be used to send messages back
• react_fn (fn) – function, which can react to messages, see Note for details
• glob (dict) – result of globals() call - used in deserializer to automatically build classes, which are not available in this namespace of this package

Note

react_fn parameter is expected to be function, which gets two parameters - message (some form of message, it can be also namedtuple), and UUID containing unique identificator of the message.

Example of function used as react_fn parameter:

def reactToAMQPMessage(message, UUID):
    response = None
    if message == 1:
        return 2
    elif message == "Hello":
        return "Hi"
    elif type(message) == dict:
        return {1: 2}

    raise UserWarning("Unrecognized message")

As you can see, protocol is pretty easy. You get message, to which you react somehow and return response. Thats all.

get_sendback(uuid, key)

Return function for sending progress messages back to original caller.

Parameters:

• uuid (str) – UUID of the received message.
• key (str) – Routing key.

Returns:

Reference to function which takes only one data argument.

Return type:

fn reference

onMessageReceived(method_frame, properties, body)

React to received message - deserialize it, add it to users reaction function stored in self.react_fn and send back result.

If Exception is thrown during process, it is sent back instead of message.

Note

In case of Exception, response message doesn’t have useful body, but in headers is stored following (string) parameters:

• exception, where the Exception’s message is stored
• exception_type, where e.__class__ is stored
• exception_name, where e.__class__.__name__ is stored
• traceback where the full traceback is stored (contains line number)

This allows you to react to unexpected cases at the other end of the AMQP communication.

parseKey(method_frame)

process_exception(e, uuid, routing_key, body, tb=None)

Callback called when exception was raised.

This method serializes the exception and sends it over AMQP back to caller.

Parameters:

• e (obj) – Instance of the exception.
• uuid (str) – UUID of the message that caused the exception to raise.
• routing_key (str) – Which routing key was used.
• body (str) – Body of the exception - the longer text.
• tb (str, default None) – Traceback (stacktrace)v of the exception.

edeposit.amqp.amqpdaemon.getConParams(virtualhost)

Connection object builder.

Parameters:virtualhost (str) – selected virtualhost in rabbitmq Returns:object filled by constants from edeposit.amqp.settings. Return type:pika.ConnectionParameters