Mojo::Redis::Processor - Encapsulates the process for a Mojo app to send an expensive job to a daemon using Redis underneath and Redis SET NX and Redis Pub/Sub.


This module is specialized to help a Mojo app to send an expensive job request to be processed in parallel in a separete daemon. Communication is handled through Redis.

This is specialized for processing tasks that can be common between different running Mojo children. Race condition between children to add a new tasks is handle by Redis SET NX capability.


Mojo app which wants to send data and get stream of processed results will look like:

    use Mojo::Redis::Processor;
    use Mojolicious::Lite;

    my $rp = Mojo::Redis::Processor->new({
        data       => 'Data',
        trigger    => 'R_25',

    my $redis_channel = $rp->on_processed(
        sub {
            my ($message, $channel) = @_;
            print "Got a new result [$message]\n";


Try it like:

    $ perl -Ilib daemon

Processor daemon code will look like:

    use Mojo::Redis::Processor;
    use Parallel::ForkManager;

    use constant MAX_WORKERS  => 1;

    $pm = new Parallel::ForkManager(MAX_WORKERS);

    while (1) {
        my $pid = $pm->start and next;

        my $rp = Mojo::Redis::Processor->new;

        $next = $rp->next();
        if ($next) {
            print "next job started [$next].\n";

            $redis_channel = $rp->on_trigger(
                sub {
                    my $payload = shift;
                    print "processing payload\n";
                    return rand(100);
            print "Job done, exiting the child!\n";
        } else {
            print "no job found\n";
            sleep 1;

Try it like:

    $ perl -Ilib

Daemon needs to pick a forking method and also handle ide processes and timeouts.



This will instantiate the object for both reqeust sender and processor. Type depends on options which are passed.


Data for processing that we pass to the $pricer code.


Trigger will be a redis channel that will trigger call of pricer code.


Data for processing that we pass to the $pricer code.

redis_read, redis_write

Redis URL for read and write. Write means there is a central and replicated redis. redis_write will default to redis_read if it is not passed.

read_conn, write_conn, daemon_conn

Setting redis connections directly. daemon_conn is used to wait for trigger.


Key prefix that is used in redis. If it is not set it will default to "Redis::Processor::".


Expire time that client will set after receiving new price from price processor. Price process will continue to price as long as someone is extending this expiry.


Sleep time if there was no job available.


Retry time to wait for new job become available. If no job become available next() will return empty.

This will new the thing.


Will send the Mojo app data processing request. This is mainly a queueing job. Job will expire if no worker take it in time. If more than one app try to register the same job Redis SET NX will only assign one of them to proceed.


Mojo app will call this to register a code reference that will be triggered everytime there is a result. Results will be triggered and published based on trigger option.


Daemon will call this to start the next job. If it return empty it meam there was no job found after "retry".


Daemon will call this to register a processor code reference that will be called everytime trigger happens. The return value will be passed to Mojo apps which requested it using Redis Pub/Sub system. on_trigger will exit the loop when there is no more subscriber to the channel.

AUTHOR, <support at>


Please report any bugs or feature requests to bug-mojo-redis-processor at, or through the web interface at I will be notified, and then you'll automatically be notified of progress on your bug as I make changes.


You can find documentation for this module with the perldoc command.

    perldoc Mojo::Redis::Processor

You can also look for information at: