Package coilmq :: Module queue :: Class QueueManager

Class QueueManager

object --+
         |
        QueueManager

Class that manages distribution of messages to queue subscribers.

This class uses threading.RLock to guard the public methods. This is probably a bit excessive, given 1) the actomic nature of basic dict read/write operations and 2) the fact that most of the internal data structures are keying off of the STOMP connection, which is going to be thread-isolated. That said, this seems like the technically correct approach and should increase the chance of this code being portable to non-GIL systems.

Instance Methods

__init__(self, store, subscriber_scheduler=None, queue_scheduler=None)
x.__init__(...) initializes x; see help(type(x)) for signature

close(self)
Closes all resources/backends associated with this queue manager.

subscriber_count(self, destination=None)
Returns a count of the number of subscribers.

subscribe(self, connection, destination)
Subscribes a connection to the specified destination (topic or queue).

unsubscribe(self, connection, destination)
Unsubscribes a connection from a destination (topic or queue).

disconnect(self, connection)
Removes a subscriber connection, ensuring that any pending commands get requeued.

send(self, message)
Sends a MESSAGE frame to an eligible subscriber connection.

ack(self, connection, frame, transaction=None)
Acknowledge receipt of a message.

resend_transaction_frames(self, connection, transaction)
Resend the messages that were ACK'd in specified transaction.

clear_transaction_frames(self, connection, transaction)
Clears out the queued ACK frames for specified transaction.

Inherited from object: __delattr__, __format__, __getattribute__, __hash__, __new__, __reduce__, __reduce_ex__, __repr__, __setattr__, __sizeof__, __str__, __subclasshook__

Instance Variables
coilmq.scheduler.QueuePriorityScheduler	queue_scheduler The scheduler that chooses which queue to select for sending backlogs for a single connection.
coilmq.store.QueueStore	store The queue storage backend to use.
coilmq.scheduler.SubscriberPriorityScheduler	subscriber_scheduler The scheduler that chooses which subscriber to send messages to.

Properties
Inherited from `object`: `__class__`

Method Details

init(self, store, subscriber_scheduler=None, queue_scheduler=None)
(Constructor)

x.__init__(...) initializes x; see help(type(x)) for signature

Parameters:

store (coilmq.store.QueueStore) - The queue storage backend.
subscriber_scheduler (coilmq.scheduler.SubscriberPriorityScheduler) - The scheduler that chooses which subscriber to send messages to.
queue_scheduler (coilmq.scheduler.QueuePriorityScheduler) - The scheduler that chooses which queue to select for sending backlogs for a single connection.

Overrides: object.__init__

close(self)

Closes all resources/backends associated with this queue manager.

Decorators:

@synchronized

subscriber_count(self, destination=None)

Returns a count of the number of subscribers.

If destination is specified then it only returns count of subscribers for that specific destination.

Parameters:

destination (str) - The optional topic/queue destination (e.g. '/queue/foo')

Decorators:

@synchronized

subscribe(self, connection, destination)

Subscribes a connection to the specified destination (topic or queue).

Parameters:

connection (coilmq.server.StompConnection) - The connection to subscribe.
destination (str) - The topic/queue destination (e.g. '/queue/foo')

Decorators:

@synchronized

unsubscribe(self, connection, destination)

Unsubscribes a connection from a destination (topic or queue).

Parameters:

connection (coilmq.server.StompConnection) - The client connection to unsubscribe.
destination (str) - The topic/queue destination (e.g. '/queue/foo')

Decorators:

@synchronized

disconnect(self, connection)

Removes a subscriber connection, ensuring that any pending commands get requeued.

Parameters:

connection (coilmq.server.StompConnection) - The client connection to unsubscribe.

Decorators:

@synchronized

send(self, message)

Sends a MESSAGE frame to an eligible subscriber connection.

Note that this method will modify the incoming message object to add a message-id header (if not present) and to change the command to 'MESSAGE' (if it is not).

Parameters:

message (stompclient.frame.Frame) - The message frame.

Decorators:

@synchronized

ack(self, connection, frame, transaction=None)

Acknowledge receipt of a message.

If the `transaction` parameter is non-null, the frame being ack'd will be queued so that it can be requeued if the transaction is rolled back.

Parameters:

connection (coilmq.server.StompConnection) - The connection that is acknowledging the frame.
frame - The frame being acknowledged.

Decorators:

@synchronized

resend_transaction_frames(self, connection, transaction)

Resend the messages that were ACK'd in specified transaction.

This is called by the engine when there is an abort command.

Parameters:

connection (coilmq.server.StompConnection) - The client connection that aborted the transaction.
transaction (str) - The transaction id (which was aborted).

Decorators:

@synchronized

clear_transaction_frames(self, connection, transaction)

Clears out the queued ACK frames for specified transaction.

This is called by the engine when there is a commit command.

Parameters:

connection (coilmq.server.StompConnection) - The client connection that committed the transaction.
transaction (str) - The transaction id (which was committed).

Decorators:

@synchronized

Class QueueManager

__init__(self, store, subscriber_scheduler=None, queue_scheduler=None) (Constructor)

close(self)

subscriber_count(self, destination=None)

subscribe(self, connection, destination)

unsubscribe(self, connection, destination)

disconnect(self, connection)

send(self, message)

ack(self, connection, frame, transaction=None)

resend_transaction_frames(self, connection, transaction)

clear_transaction_frames(self, connection, transaction)

init(self, store, subscriber_scheduler=None, queue_scheduler=None)
(Constructor)