This document describes the current stable version of Kombu (4.2). For development docs, go here.

librabbitmq AMQP transport - kombu.transport.librabbitmq

librabbitmq transport.

Transport

class kombu.transport.librabbitmq.Transport(client, **kwargs)[source]

AMQP Transport (librabbitmq).

class Connection(host='localhost', userid='guest', password='guest', virtual_host='/', port=5672, channel_max=65535, frame_max=131072, heartbeat=0, lazy=False, **kwargs)

AMQP Connection (librabbitmq).

class Channel(connection, channel_id)

AMQP Channel (librabbitmq).

class Message(channel, props, info, body)

AMQP Message (librabbitmq).

prepare_message(body, priority=None, content_type=None, content_encoding=None, headers=None, properties=None)

Encapsulate data into a AMQP message.

prepare_queue_arguments(arguments, **kwargs)
class Message(channel, props, info, body)

AMQP Message (librabbitmq).

channel_errors = (<class 'amqp.exceptions.ChannelError'>, <class '_librabbitmq.ChannelError'>)
close_connection(connection)[source]

Close the AMQP broker connection.

connection_errors = (<class 'amqp.exceptions.ConnectionError'>, <class '_librabbitmq.ConnectionError'>, <class 'socket.error'>, <type 'exceptions.IOError'>, <type 'exceptions.OSError'>)
create_channel(connection)[source]
default_connection_params
default_port = 5672
default_ssl_port = 5671
drain_events(connection, **kwargs)[source]
driver_name = u'librabbitmq'
driver_type = u'amqp'
driver_version()[source]
establish_connection()[source]

Establish connection to the AMQP broker.

get_manager(*args, **kwargs)[source]
implements = {'heartbeats': False, 'asynchronous': True, 'exchange_type': frozenset([u'topic', u'headers', u'fanout', u'direct'])}
qos_semantics_matches_spec(connection)[source]
register_with_event_loop(connection, loop)[source]
verify_connection(connection)[source]

Connection

class kombu.transport.librabbitmq.Connection(host='localhost', userid='guest', password='guest', virtual_host='/', port=5672, channel_max=65535, frame_max=131072, heartbeat=0, lazy=False, **kwargs)[source]

AMQP Connection (librabbitmq).

class Channel(connection, channel_id)

AMQP Channel (librabbitmq).

Consumer(*args, **kwargs)
class Message(channel, props, info, body)

AMQP Message (librabbitmq).

exception MessageStateError

The message has already been acknowledged.

args
message
accept
ack(multiple=False)

Acknowledge this message as being processed.

This will remove the message from the queue.

Raises:MessageStateError – If the message has already been acknowledged/requeued/rejected.
ack_log_error(logger, errors, multiple=False)
acknowledged

Set to true if the message has been acknowledged.

body
channel
content_encoding
content_type
decode()

Deserialize the message body.

Returning the original python structure sent by the publisher.

Note

The return value is memoized, use _decode to force re-evaluation.

delivery_info
delivery_tag
errors = None
headers
payload

The decoded message body.

properties
reject(requeue=False)

Reject this message.

The message will be discarded by the server.

Raises:MessageStateError – If the message has already been acknowledged/requeued/rejected.
reject_log_error(logger, errors, requeue=False)
requeue()

Reject this message and put it back on the queue.

Warning

You must not use this method as a means of selecting messages to process.

Raises:MessageStateError – If the message has already been acknowledged/requeued/rejected.
Producer(*args, **kwargs)
after_reply_message_received(queue)

Callback called after RPC reply received.

Notes

Reply queue semantics: can be used to delete the queue after transient reply message received.

basic_ack(delivery_tag, multiple=False)
basic_cancel(consumer_tag, **kwargs)
basic_consume(queue='', consumer_tag=None, no_local=False, no_ack=False, exclusive=False, callback=None, arguments=None, nowait=False)
basic_get(queue='', no_ack=False)
basic_publish(body, exchange='', routing_key='', mandatory=False, immediate=False, **properties)
basic_qos(prefetch_size=0, prefetch_count=0, _global=False)
basic_recover(requeue=True)
basic_reject(delivery_tag, requeue=True)
close()
exchange_declare(exchange='', type='direct', passive=False, durable=False, auto_delete=False, arguments=None, nowait=False)

Declare exchange.

Keyword Arguments:
 auto_delete – Not recommended and so it is ignored.
exchange_delete(exchange='', if_unused=False, nowait=False)
flow(active)
get_bindings()
is_open = False
no_ack_consumers = None
prepare_message(body, priority=None, content_type=None, content_encoding=None, headers=None, properties=None)

Encapsulate data into a AMQP message.

prepare_queue_arguments(arguments, **kwargs)
queue_bind(queue='', exchange='', routing_key='', arguments=None, nowait=False)
queue_declare(queue='', passive=False, durable=False, exclusive=False, auto_delete=False, arguments=None, nowait=False)
queue_delete(queue='', if_unused=False, if_empty=False, nowait=False)

nowait argument is not supported.

queue_purge(queue, nowait=False)
queue_unbind(queue='', exchange='', routing_key='', arguments=None, nowait=False)
class Message(channel, props, info, body)

AMQP Message (librabbitmq).

exception MessageStateError

The message has already been acknowledged.

args
message
accept
ack(multiple=False)

Acknowledge this message as being processed.

This will remove the message from the queue.

Raises:MessageStateError – If the message has already been acknowledged/requeued/rejected.
ack_log_error(logger, errors, multiple=False)
acknowledged

Set to true if the message has been acknowledged.

body
channel
content_encoding
content_type
decode()

Deserialize the message body.

Returning the original python structure sent by the publisher.

Note

The return value is memoized, use _decode to force re-evaluation.

delivery_info
delivery_tag
errors = None
headers
payload

The decoded message body.

properties
reject(requeue=False)

Reject this message.

The message will be discarded by the server.

Raises:MessageStateError – If the message has already been acknowledged/requeued/rejected.
reject_log_error(logger, errors, requeue=False)
requeue()

Reject this message and put it back on the queue.

Warning

You must not use this method as a means of selecting messages to process.

Raises:MessageStateError – If the message has already been acknowledged/requeued/rejected.
callbacks
channel(channel_id=None)[source]
channel_max
close()[source]
connect()

Establish connection to the broker.

connected
drain_events(timeout=None)[source]
fileno()

File descriptor number.

frame_max
heartbeat
hostname
password
port
reconnect()[source]
server_properties
userid
virtual_host

Channel

class kombu.transport.librabbitmq.Channel(connection, channel_id)[source]

AMQP Channel (librabbitmq).

class Message(channel, props, info, body)

AMQP Message (librabbitmq).

prepare_message(body, priority=None, content_type=None, content_encoding=None, headers=None, properties=None)[source]

Encapsulate data into a AMQP message.

prepare_queue_arguments(arguments, **kwargs)[source]

Message

class kombu.transport.librabbitmq.Message(channel, props, info, body)[source]

AMQP Message (librabbitmq).