CL-DISQUE

A Disque client for Common Lisp.

Disque is an in-memory job queue and message broker. It's distributed and fault-tolerant so it works as a middle layer among processes that want to exchange messages.

Learn more about message queues here, and more about the Disque implementation here.

cl-disque provides a client for working with Disque through sending and receiving jobs and commands.

1Usage

1.1Quickstart

First, make sure a Disque server is running.

Load cl-disque and connect to the Disque server on the given host and port:

     (ql:quickload 'cl-disque)

     ; host defaults to 127.0.0.1, port defaults to 7711
     (cl-disque:connect :host <host> :port <port>)

You can interact with the server using commands from the disque package.

   (disque:hello)
   ;=> (1 "node-id"
   ;    ("node-ids" "host" "7711" "1"))

Disconnect from the Disque server with:

   (cl-disque:disconnect)

Alternatively, you can wrap all interactions in the with-connection macro, which creates a new connection to execute the given body, and assures a disconnect afterwards:

   (cl-disque:with-connection ()
     (disque:addjob "queue" "job" 0)
     (disque:getjob "queue" 1))
   ;=> (("queue" "job-hash" "job"))

The Makefile offers a couple of commands for running the test-suite and loading cl-disque into an SBCL repl:

   # To run the test suite
   $ make test

   # To load an SBCL repl
   $ make sbcl-repl

1.2Available commands

Cl-Disque supports all of the Disque client commands and theirarguments. See The Disque Documentation for more specifics on eachcommand

1.2.1INFO

• Args: ()
• Response-type: :bulk

1.2.2HELLO

• Args: ()
• Response-type: :multi

1.2.3QLEN

• Args: (queue)
• Response-type: :integer

1.2.4QPEEK

• Args: (queue count)
• Response-type: :multi

1.2.5QSCAN

• Args: (&rest args &key count busyloop minlen maxlen importrate)
• Response-type: :multi

1.2.6GETJOB

• Args: (queues &rest args &key nohang timeout count withcounters)
• Reponse-type: :multi

Note: queues can either be a single queue or a list of queues:

    (disque:getjob "queue1")
    ;; or
    (disque:getjob '("queue1" "queue2" "queue3")

1.2.7ADDJOB

• Args: (queue job timeout &rest args &key replicate delay retry ttl

maxlen async)

• Response-type: :status

1.2.8ACKJOB

• Args: (job &rest jobs)
• Response-type: :integer

1.2.9FASTACK

• Args: (job &rest jobs)
• Response-type: :integer

1.2.10WORKING

• Args: (job)
• Response-type: :integer

1.2.11NACK

• Args: (job &rest jobs)
• Response-type: :integer

1.2.12ENQUEUE

• Args: (job &rest jobs)
• Reponse-type: :integer

1.2.13DEQUEUE

• Args: (job &rest jobs)
• Response-type :integer

1.2.14DELJOB

• Args: (job &rest jobs)
• Response-type: :integer

1.2.15SHOW

• Args: (job)
• Response-type: :multi

1.2.16JSCAN

• Args: (cursor &rest args &key count blocking queue state reply)
• Response-type: :multi

1.3Code organization

The system provides two packages: CL-DISQUE and DISQUE.

Everything is available in the CL-DISQUE package.

The DISQUE package contains all of the commands for interacting with a Disque server. This is simply syntactic sugar, as all of the commands are also available in the CL-DISQUE package with a command prefix. For Example:

   (disque:info)
   ; is the same as
   (cl-disque:disque-info)

1.4Installation

Git clone this repo into your ~/quicklisp/local-projects/ directory, and (ql:quickload :cl-disque).

1.5Dependencies

• usocket
• flexi-streams
• rutils
• prove (only for tests)

1.6Debugging and error recovery

If *echo-p* is T, all client-server communications will be echoed to the stream *echo-stream*, which defaults to *standard-output*.

Error handling is mimicked after Postmodern. In particular, whenever an error occurs that breaks the communication stream, a condition of type disque-connection-error is signalled offering a :reconnect restart. If it is selected the whole Disque command will be resent, if the reconnection attempt succeeds. Furthermore, connect checks if a connection to Disque is already established, and offers two restarts (:leave and :replace) if this is the case.

When the server respondes with an error reply a condition of type disque-error-reply is signalled.

There's also a high-level with-persistent-connection macro, that tries to do the right thing™ (i.e. automatically reopen the connection once, if it is broken).

1.7Advanced usage

1.7.1Pipelining

For better performance Disque allows to pipeline commands and delay receiving results until the end, and process them all in oine batch afterwards. To support that there's with-pipelining macro.

Note, that with-pipelining calls theoretically may nest, but the results will only be available to the highest-level pipeline, all the nested pipelines will return :PIPELINED. So a warining is signalled in this situation.

Note: Pipelining has not been tested since being ported form cl-redis.

1.8Credits

Cody Reichert <codyreichert@gmail.com> is the maintainer of CL-DISQUE.

CL-DISQUE is a ported of the CL-REDIS client, which is developed and maintained by Vsevolod Dyomkin <vseloved@gmail.com>. Many thanks to him for implementing the protocol and providing most of the internals.

Alexandr Manzyuk <manzyuk@googlemail.com> also contributed to CL-REDIS client and developed the connection handling code following the implementation in Postmodern. It was since partially rewritten to accommodate more advanced connection handling strategies, like persistent connection.

1.9License

MIT (See LICENSE file for details).