bucker

  1. bucker
    1. What is it?
    2. What is it for?
    3. Requirements
    4. License
    5. How does it work?
    6. Run it
    7. Messages allowed
      1. List all the queues
      2. Create a new queue
      3. Delete queue
      4. List new messages
      5. Get a message
      6. Push a message
      7. Delete a message

This is a work in progress so expect a lot of changes. bucker is not yet stable but it works.

What is it?

bucker is a queue system that supports multiple storage for the queue (memcached, Amazon SQS for now) and is driven by XML messages sent over a TCP connections between a client and the queue server.

What is it for?

bucker aims at providing a simple, yet powerful queue system for your applications. This queue system can be seen as a way to perform inter-process communication between different components of your application allowing to scale horizontally.

Requirements

  • python-memcached if you plan on using the memcached storage
  • CherryPy svn trunk (for the scripts, it's not technically used by bucker internally but provides some tools to run the servers until Kamaelia offers a nice way to stop components).

License

BSD license

How does it work?

A server runs and listens for incoming TCP connections from client. That server understands the limited set of commands that are explained below and then acts accordingly on the given storage (memcached, etc.). A client sends commands to the server as XML messages. In returns, it gets XML messages informing of the success or failure of the command.

Internally bucker is completely based on Kamaelia, a nice Python library that allows for heterogeneous components to be plugged to each other by linking their output to their input by using a concept of boxes. bucker therefore creates a graph of components which input boxes are connected to output boxes of other components. Internally each component can then process an input message and sends another message to one of its outbox, Kamaelia handles the message transit. See its documentation for more information.

In addition to the main message server, bucker provides what it refers to a bus. This component connects to a bucker server and monitors a set of specified queues for new messages. When a message is pushed to the server by a client, the bus gets it and deletes it from the queue server. Clients (different from the ones producing messages) connects to that bus and simply waits for the bus to pass new messages automatically. Therefore those clients do not have to poll the queue server themselves, this operation is done by the bus, they simply react when the bus forwards them a message.

Run it

bucker comes with two scripts, one of the queue message server and one for the bus (if you need to use it).

To get some help about the queue server: 

$> queue-server.py --help

To start the queue server with verbose output: 

$> queue-server.py -v

This supposes that you have a memcached server running on its default port.

To get the help of the bus server: 

$> queue-bus.py --help

To start the bus server: 

$> queue-bus.py -v -q test

This starts the bus server and connects to a default queue server. It also monitors the test queue (if it doesn't exist yet it will be created).

To test it: 

from bucker.api.message import ListQueues
from bucker.api.client import QueueClient
q = QueueClient('127.0.0.1', 9876)
q.connect()
q.send(ListQueues())
q.recv()
q.disconnect()

Messages allowed

Notes:

  • In all the following messages the op element is the only compulsory for any message. In some cases other elements can be compulsory.
  • The compulsory namespace is: http://purl.oclc.org/DEFUZE/qs and must be present on the qs element. For clarity it won't be shown below.

List all the queues

<qs type="request"><op><list-queues /></op></qs>

Create a new queue

<qs type="request"><op><new-queue /></op><qid>QUEUE_ID</qid></qs>

  • new-queue is compulsory
  • qid is compulsory
  • QUEUE_ID is a string identifying a queue

If a queue already exists, an error is returned as follow: 

<qs resp="3c57bcc8414b5c3540e650d2c4dc38848c01834ab5d5b5032da57e00daebb74d" xml:lang="en-US" type="error"><error code="409" type="conflict">'QUEUE_ID' is already an existing queue</error></qs>

Delete queue

<qs type="request"><op force="yes"><delete-queue /></op><qid>QUEUE_ID</qid></qs>
  • delete-queue is compulsory
  • qid is compulsory
  • QUEUE_ID is a string identifying a queue
  • If the force attribute is set to "yes", it will result in deleting the queue even if it isn't empty. Otherwise an error will be returned. If not present or if its value is different from "yes", it considers that the force should not be applied.

List new messages

You may request the queue for new pushed messages. The returned list will be the message ids of messages in the queue that have not yet been retrieved once. 

<qs type="request"><op><list-messages /></op><qid>QUEUE_ID</qid></qs>
  • list-messages is compulsory
  • qid is compulsory
  • QUEUE_ID is a string identifying a queue

The returned XMl document will contain a messages element which has zero or many mid elements. If present a mid element's content is the id of the message.

Get a message

<qs type="request"><op peek="yes|no"><get-message /></op><qid>QUEUE_ID</qid><mid>MESSAGE_ID</mid></qs>
  • get-message is compulsory
  • qid is compulsory
  • mid is compulsory
  • peek is not mandatory. It is only if you want the message to stay in the list of recent messages (via list-messages), in that case set to 'yes'

Notes:

  • If the message is retrieved for the first time, it will be removed from the recent messages list (unless peek equals 'yes')
  • Each message has a visibility window during which it cannot be seen from any other client. This means that when the window is free, the message is returned and the window set for a certain period of time (default to 30s). Once that period has passed, the message becomes visible and accessible to any client again. In any case the response informs the client of the status of the message.

Push a message

<qs type="request"><op><push-message /></op><qid>QUEUE_ID</qid><payload>PAYLOAD</payload></qs>
  • push-message is compulsory
  • qid is compulsory
  • It is recommended to encode (like using base64 encoding) the payload so that no encoding issue will come from the storage.

The payload can be any content you wish to store into the queue. The returned response contains the id of the message.

Delete a message

<qs type="request"><op><delete-message /></op><qid>QUEUE_ID</qid><mid>MESSAGE_ID</mid></qs>
  • delete-message is compulsory
  • qid is compulsory
  • mid is compulsory