Extending Laravel with Managers

Most Laravel libraries that have drivers use a Manager class.

A “driver” is an implementation of some library, but (generally) using a different underlying technology.

A few example Laravel libraries with drivers (multiple implementations) are: Queue, Database, Cache, and Session.

To make that a bit more concrete: the drivers you can choose between for the Queue library (out of the box) are Sync, Database, Beanstalkd, SQS, Redis, and Null.

From our perspective (as developers), the queue system is interacted with in the same way. Each driver, however, handles its specific implementation details as dictated by the underlying technology used.

1. sync runs the jobs immediately (no background work done at all)
2. database uses database tables and has to care about database-specific locking mechanisms
3. sqs uses AWS’s code (their HTTP API)
4. redis uses LUA scripts to run commands against a Redis instance

Managers Manage Implementations

For any library, the aforementioned Manager classes manage the various drivers available. The Manager acts as a place to register and create the implementations of a library.

Let’s look at the DB facade. This facade resolves to whatever is registered as, simply, db. If we check DatabaseServiceProvider, we see that db resolves to an instance of DatabaseManager:

$this->app->singleton('db', function ($app) {
    return new DatabaseManager($app, $app['db.factory']);
});

So, when we use DB::foo(), we’re interacting with the DatabaseManager class.

The DB manager is able to give us registered implementations. For example, two ways to get a DB connection are DB::connection() (to get the default connection) and DB::connection('sqlite') (to get a connection named sqlite).

The connection classes returned have the same set of methods available on them - they implement a shared interface. This means you can call $connection->table('users')->get() on all database connection types.

The DatabaseManager class, like most Manager classes, also has an extend() method that we can use to add our own database drivers.

This is common to all Manager classes within Laravel. They are responsible for letting us extend a library and also generating implementations (acting as a Factory).

Bring Your Own Driver

You can extend a lot of Laravel with your own drivers - and managers are how!

Let’s see how to create an alternative Queue driver. In our case, we’ll extend and tweak the SQS queue driver.

It basically boils down to this:

# Somewhere within a service provider...
use Illuminate\Queue\QueueManager;

QueueManager::extend('sqs-poll', function() {
    return new SqsLongPollerQueue;
});

This adds a new SQS queue driver named sqs-poll. Why do this? Because I want to add long-polling.

For SQS, long polling keeps the SQS connection open for X seconds, waiting for new jobs to come in. The end result is that we make less API calls to SQS. Since we’re billed on SQS calls, this reduces cost. SQS is pretty cheap, so this is only really useful at scale.

You should also keep your queue timeout option at the set number of seconds (or higher) to avoid queue timeout errors polluting your error logs.

Read up on long-polling vs short-polling! Long polling is nice as workers often retrieve jobs sooner - the default “short polling” only asks a subset of SQS servers to see if there is a job! This can delay jobs being processed, even at smaller scales.

Most manager classes have an extend() method that works just like above - you can add a “driver” and define the implementation of it. Then your application can be configured to use it!

To use the above queue driver, I’d update config/queue.php and add a new item to the connections array with array key sqs-poll. In our case, we can copy the sqs driver details for our new sqs-poll connection.

Our Implementation

For the sake of completeness, let’s see what the SqsLongPollerQueue implementation looks like.

To accomplish what we want, we can extend the SqsQueue class that comes with Laravel, and tweak the pop() method (which is used to get a message from the queue).

<?php

namespace App\Queue;

use Illuminate\Queue\SqsQueue;
use Illuminate\Queue\Jobs\SqsJob;

class SqsLongPollerQueue extends SqsQueue
{
    /**
     * Pop the next job off of the queue.
     *
     * @param  string|null  $queue
     * @return \Illuminate\Contracts\Queue\Job|null
     */
    public function pop($queue = null)
    {
        $response = $this->sqs->receiveMessage([
            'QueueUrl' => $queue = $this->getQueue($queue),
            'AttributeNames' => ['ApproximateReceiveCount'],
            'WaitTimeSeconds' => 20, // We added this line
        ]);

        if (! is_null($response['Messages']) && count($response['Messages']) > 0) {
            return new SqsJob(
                $this->container, $this->sqs, $response['Messages'][0],
                $this->connectionName, $queue
            );
        }
    }
}

A Real Example

This method of extension is how the SQS FIFO Laravel package works - it adds the driver (“connection”) sqs-fifo. FIFO stands for “first in, first out” - in other words, unlike with regular SQS, FIFO queues guarantee that jobs are released in the order they are created.

This is useful for workloads that need to be run in a specific order. I use this on Chipper CI to ensure builds are run in the order they are received.

Now, if jobs are run in order, how do you process multiple jobs at the same time? To enable running jobs in parallel, there is a concept of “groups”. A group is just a string used to identify a group of queue jobs. Jobs with group ID “xyz” are all processed in order, while jobs with group ID “abc” will be processed in their own order. Groups “xyz” and “abc” can be run in parallel.

So, back to the FIFO package - the SqsFifoQueue class is the class added via the QueueManager. This class extends SqsQueue (just like our example above) and tweaks what’s needed for FIFO queues to work - allowing us to add job groups and de-duplication checks.