Project description

Linger

-- Message queue and pub-sub service with HTTP API

About

The linger package provides the Linger server, which is a message queue and pub-sub service with a REST HTTP API.

The message queue distributes messages in channels from publishers to consumers. The message queue can be used in situations where a work queue is needed, e.g., for running time-consuming tasks asynchronously.

The pub-sub (publish-subscribe) service distribute messages from publishers to subscribers (both fan-in and fan-out). It can be used in situations where message notification is needed, e.g., for delivering status updates, or change notifications.

The HTTP API server is implemented in Python 3 using a non-blocking, single-threaded Tornado web server.

Message queue

Linger provides a message queue where messages are added by producers, and made available to consumers.

Producers may add messages to any named channel, which are created on-demand.

Consumers may subscribe and consume messages from any named channel, but is limited to retrieving a message from one channel per request.

Messages are delivered to consumers at least once, and they may be re-delivered, if the consumer doesn't delete the delivered message.

Messages can be in one of two states: ready or hidden. Where 'ready' messages are those available for delivery to consumers, and hidden messages are not made available to consumers. A message is hidden during a visibility timeout, after being delivered to a consumer.

The life of a message in the queue

A producer adds a new message to a named channel (the queue).

A consumer requests a message from the channel, the message is delivered to the consumer.

Once the message has been delivered, it will be hidden and not delivered again until the visibility timeout has passed. (This keeps multiple consumers from requesting the same message.)

When the consumer has successfully processed the message, they delete the message from the channel.

A message can be assigned a priority which is different from the default, so messages are delivered in a non-sequential order (an order not based on the time messages are added to the queue).

A retention (linger) period may be set on a message, to limit the lifetime of the message in the channel. This can be useful when the channel (for some reason) has no consumers, and you want to limit the age of messages in the channel.

Pub-sub service

Linger provides pub-sub functionality where messages are posted by publishers, and made available to subscribers.

A message is published on a named topic, and is distributed by Linger to one or more subscribed channels. Each channel may have one or more subscribers that consume the messages of that channel.

The life of a published message

A publisher post a new message to a named topic.

The message is distributed to channels subscribed to the topic.

A subscriber requests a message from a channel, the message is returned.

Once the message has been delivered, it will not be delivered again until the visibility timeout has passed.

When the publisher has successfully received a message, they delete the message from the channel.

Notice that step 3 to 5 here is the same as step 2 to 4 in "the life of a message in the queue". In a standard pub-sub model there is only one subscriber per channel, but Linger allows for multiple subscribers, which consume the messages delivered to the channel. Hence, multiple subscribers to same channel will not see the same messages.

When subscribing a channel to a topic, you can set the message priority, retention (linger) period, etc., which will be applied to messages on that topic delivered to the channel.

Getting started

Install

Install the linger script in /usr/local/bin (on Linux and OS X) with the following pip command:

pip install https://github.com/nephics/linger/archive/master.zip

This will also ensure that the dependency package tornado is installed.

Run

Start a Linger server by running the script:

linger

from the command line, or use systemd or a process-daemon tool like Supervisor.

The default port is 8989. This and other options can be set from the command line, or using a config file (with path specified as a command line option). Command line options can be listed using argument --help.

By default the database is kept in memory, but the server can store the database on disk, both to reduce memory footprint and to allow for full restore in the event of a server/process restart. Use the --dbfile command line option to store the database on disk.

It is also possible to define a global a high-level mark limiting the number of messages in any channel. The default is no high-level mark, this setting may be changed using the --hlm command line option.

Security

The HTTP API is unauthenticated, all clients have unrestricted access to the full API, but you may use a reverse proxy like nginx to require authentication for network access, see HTTP Basic Auth.

Limits and performance

Long-polling duration is limited to about 2 mins.

Messages are by default limited to 256 KB in size, and may contain any sequences of bytes.

Linger is not currently optimised with regard to memory usage, and it has not been tested for high-performance usage scenarios, such as delivering billions of messages a day. But, for most real world situations, Linger will serve you reliably.

HTTP API overview

The Linger HTTP API consists of these methods:

GET /channels- list channels

GET /channels/<channel>- get message from channel

POST /channels/<channel>- add message to channel

DELETE /channels/<channel>- drain the channel

GET /channels/<channel>/stats- get channel stats

GET /channels/<channel>/topics- list topics a channel is subscribed to

Where <channel> is the channel name, where messages can be added and removed. Channel names may contain characters from a-zA-Z0-9_%-. Remember to URL-encode the name when using it in requests, particularly important if the name includes space or slash.

Channels are created on-demand, and are automatically destroyed when they are empty and not in use.

The <msg-id> is an integer that identifies a message (uniquely across all channels).

The API methods are described in more detailed in the following sections, with examples using cURL.

List channels

The list of current channels can be retrieved using a HTTP GET request to /channels. Example request:

curl -X GET http://127.0.0.1:8989/channels

The server responds with HTTP status code 200, and the channel list is included in the response body, which is text encoded as JSON.
Example response:

{"channels": ['test']}

Add message to a channel

Add a message to a named channel using a HTTP POST request to /channels/<channel>. Example request:

The server responds with HTTP status code 202, and the message id (an integer) encoded as JSON in the response body. Example response:

{"id": 1}

You can set the message priority, visibility timeout, max delivery attempts, and message retention limit using query parameters:

priority=10 message priority, a lower number means a
higher priority in the channel,
default is zero
(any integer is accepted)
timeout=60 wait for 60 seconds before delivering the
message to another client, default is
30 seconds
(accepts an integer greater than zero)
deliver=5 deliver the message at most 5 times before
discarding it, default is zero, which means
never discard the message
(any positive integer is accepted)
linger=60 keep the message for 60 seconds, before
discarding it, default is zero, which means
never discard the message
(any positive integer is accepted)

The parameters can be combined in a request. Example request with JSON encoded message:

The response content-type will be the same as specified when adding the message to the channel (default is text/plain).

By adding the nowait query parameter, you may prevent long-polling, and have the server send a response immediately. This implies that the server will return an empty reply, if there is no message waiting in the channel. Example request:

curl http://127.0.0.1:8989/channels/test?nowait

Drain the channel

Drain (discard) all messages from channel using a HTTP DELETE request to /channels/<channel>. Example request:

curl -X DELETE http://127.0.0.1:8989/channels/test

The server responds with HTTP status code 204.

Get channel stats

Retrieve statistics about the messages in a channel using a HTTP GET request to /channels/<channel>/stats. Example request:

curl -X GET http://127.0.0.1:8989/channels/test/stats

The server responds with HTTP status code 200, and the response body contains a JSON encoded mapping of stats about messages in the channel.

Example response:

{"ready": 2, "hidden": 0}

List topics a channel is subscribed to

The list of topics a channel is subscribed to can be retrieved using a HTTP GET request to /channels/<channel>/topics. Example request:

curl http://127.0.0.1:8989/channels/test/topics

Example response:

{"topics": ["some-topic"]}

Subscribe channel to a topic

Subscribe a channel to a named topic using a HTTP PUT request to /channels/<channel>/topics/<topic>. Example request:

curl -X PUT http://127.0.0.1:8989/channels/test/topics/some-topic

The server responds with HTTP status code 204.

If you add query parameters, the subscription applies these parameters to all messages published to the channel on the specific topic. The possible query parameters are the same as available for when adding a message to a channel (see above). This includes message priority, visibility timeout, max delivery attempts, and message retention limit.

Example request limiting the message retention to 60 seconds for all messages published to the channel on the specific topic: