path: root/posts/2010-04-14-presque-a-redis-tatsumaki-based-message-queue.org

[[http://github.com/franckcuny/presque/tree/][presque]] is a small
message queue service build on top of
[[http://code.google.com/p/redis/][redis]] and
[[http://search.cpan.org/perldoc?Tatsumaki][Tatsumaki]]. It's heavily
inspired by [[http://github.com/gleicon/restmq][RestMQ]] and
[[http://github.com/defunkt/resque][resque]].

-  Communications are done in JSON over HTTP
-  Queues and messages are organized as REST resources
-  A worker can be writen in any language that make a HTTP request and
   read JSON
-  Thanks to redis, the queues are persistent

** Overview

resque need a configuration file, writen in YAML that contains the host
and port for the Redis server.

#+BEGIN_EXAMPLE
    redis:
        host: 127.0.0.1
        port: 6379
#+END_EXAMPLE

Let's start the server:

#+BEGIN_EXAMPLE
    % plackup app.psgi --port 5000
#+END_EXAMPLE

The applications provides some HTTP routes:

-  */*: a basic HTML page with some information about the queues
-  */q/*: REST API to get and post job to a queue
-  */j/*: REST API to get some information about a queue
-  */control/*: REST API to control a queue (start or stop consumers)
-  */stats/*: REST API to fetch some stats (displayed on the index page)

Queues are created on the fly, when a job for an unknown queue is
inserted. When a new job is created, the JSON send in the POST will be
stored "as is". There is no restriction on the schema or the content of
the JSON.

Creating a new job simply consist to :

#+BEGIN_EXAMPLE
    % curl -X POST "http://localhost:5000/q/foo" -d '{"foo":"bar", "foo2":"bar" }'
#+END_EXAMPLE

and fetching the job:

#+BEGIN_EXAMPLE
    % curl "http://localhost:5000/q/foo"
#+END_EXAMPLE

When a job is fetched, it's removed from the queue.

** A basic worker

I've also pushed
[[http://git.lumberjaph.net/p5-presque-worker.git/][presque::worker]].
It's based on
[[http://search.cpan.org/perldoc?AnyEvent::HTTP][AnyEvent::HTTP]] and
[[http://search.cpan.org/perldoc?Moose][Moose]]. Let's write a basic
worker using this class:

#+BEGIN_SRC perl
    use strict;
    use warnings;
    use 5.012;    # w00t

    package simple::worker;
    use Moose;
    extends 'presque::worker';

    sub work {
        my ($self, $job) = @_;
        say "job's done";
        ...;      # yadda yadda!
        return;
    }

    package main;
    use AnyEvent;

    my $worker =
        simple::worker->new(base_uri => 'http://localhost:5000', queue => 'foo');

    AnyEvent->condvar->recv;
#+END_SRC

A worker have to extends the presque::worker class, and implement the
method /work/. When the object is created, the class check if this
method is avalaible. You can also provide a =fail= method, which will be
called when an error occur.

** The future

I plan to add support for
[[http://en.wikipedia.org/wiki/WebSocket][websocket]], and probably
[[http://en.wikipedia.org/wiki/Xmpp][XMPP]]. More functionalities to the
worker too: logging, forking, handling many queues, ... I would like to
add priorities to queue also, and maybe scheluding job for a given date
(not sure if it's feasable with Redis).