Why Queues?

Queues make it easy to remove a costly or slow process from any synchronous call. The most common example is sending mail—doing so can be slow, and you don’t want your users to have to wait for mail to send in response to their actions. Instead, trigger a “send mail” queued job and let the users get on with their day. Sometimes you may not just want to save your users time, but you might have a process like a cron job or a webhook that has a lot of work to process; rather than letting it all run at once (and potentially time out), you may choose to queue its individual pieces and let the queue worker process them one at a time.

Additionally, if you have some heavy processing that’s more than your server can handle, you can spin up more than one queue worker to work through your queue faster than your normal application server could on its own.

Basic Queue Configuration

Like many other Laravel features that abstract multiple providers, queues have their own dedicated config file (config/queue.php) that allows you to set up multiple drivers and define which will be the default. This is also where you’ll store your SQS, Redis, or beanstalkd authentication information.

Queued Jobs

Remember our bank analogy? Each person in the bank “queue” (line) is, in programming terms, a job. This job could be shaped any way; it could just be a string, or an array, or an object. In Laravel, it’s a collection of information containing the job name, the data payload, the number of attempts that have been made so far to process this job, and some other simple metadata.

But you don’t need to worry about that in your interactions with Laravel. Laravel provides a structure called a Job, which is intended to encapsulate a single task—a behavior that your application can be commanded to do—and allow it to be added to and pulled from a queue. There are also simple helpers to make it easy to queue Artisan commands and mail.

Let’s start with an example where, every time a user changes his plan with your SaaS app, you want to rerun some calculations about your overall profit.

php artisan make:job CrunchReports

<?php

namespace App\Jobs;

use Illuminate\Bus\Queueable;
use Illuminate\Queue\SerializesModels;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;

class CrunchReports implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    /**
     * Create a new job instance.
     *
     * @return void
     */
    public function __construct()
    {
        //
    }

    /**
     * Execute the job.
     *
     * @return void
     */
    public function handle()
    {
        //
    }
}

...
use App\ReportGenerator;

class CrunchReports implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    protected $user;

    public function __construct($user)
    {
        $this->user = $user;
    }

    public function handle(ReportGenerator $generator)
    {
        $generator->generateReportsForUser($this->user);

        Log::info('Generated reports.');
    }
}

$user = auth()->user();
$daysToCrunch = 7;
\App\Jobs\CrunchReports::dispatch($user, $daysToCrunch);

DoThingJob::dispatch()->onConnection('redis');

DoThingJob::dispatch()->onQueue('high');

// Delays five minutes before releasing the job to queue workers
$delay = now()->addMinutes(5);
DoThingJob::dispatch()->delay($delay);

Running a Queue Worker

So what is a queue worker, and how does it work? In Laravel, it’s an Artisan command that stays running forever (until it’s stopped manually) and takes the responsibility for pulling down jobs from your queue and running them:

php artisan queue:work

This command starts a daemon “listening” to your queue; every time there are jobs on the queue, it will pull down the first job, handle it, delete it, and move on to the next. If at any point there are no jobs, it “sleeps” for a configurable amount of time before checking again to see if there are any more jobs.

You can define how many seconds a job should be able to run before the queue listener stops it (--timeout), how many seconds the listener should “sleep” when there are no jobs left (--sleep), how many tries each job should be allowed before being deleted (--tries), which connection the worker should listen to (the first parameter after queue:work), and which queues it should listen to (--queue=):

php artisan queue:work redis --timeout=60 --sleep=15 --tries=3
--queue=high,medium

You can also process just a single job with php artisan queue:work.

Handling Errors

So, what happens when something goes wrong with your job when it’s in the middle of processing?

php artisan queue:work --tries=3

public function handle()
{
    ...
    if ($this->attempts() > 3) {
        //
    }
}

php artisan queue:failed-table
php artisan migrate

...
class CrunchReports implements ShouldQueue
{
    ...

    public function failed()
    {
        // Do whatever you want, like notify an admin
    }
}

// Some service provider
use Illuminate\Support\Facades\Queue;
use Illuminate\Queue\Events\JobFailed;
...
    public function boot()
    {
        Queue::failing(function (JobFailed $event) {
            // $event->connectionName
            // $event->job
            // $event->exception
        });
    }

php artisan queue:failed

+----+------------+---------+----------------------+---------------------+
| ID | Connection | Queue   | Class                | Failed At           |
+----+------------+---------+----------------------+---------------------+
| 9  | database   | default | App\Jobs\AlwaysFails | 2018-08-26 03:42:55 |
+----+------------+---------+----------------------+---------------------+

php artisan queue:retry 9

php artisan queue:retry all

php artisan queue:forget 5

php artisan queue:flush

Controlling the Queue

Sometimes, from within the handling of a job, you’ll want to add conditions that will potentially either release the job to be restarted later or delete the job forever.

To release a job back into the queue, use the release() command, as in Example 16-7.

public function handle()
{
    ...
    if (condition) {
        $this->release($numberOfSecondsToDelayBeforeRetrying);
    }
}

If you want to delete a job during its handling, you can just return at any point, as seen in Example 16-8; that’s the signal to the queue that the job was handled appropriately and should not be returned to the queue.

public function handle()
{
    ...
    if ($jobShouldBeDeleted) {
        return;
    }
}

Queues Supporting Other Functions

The primary use for queues is to push jobs onto them, but you can also queue mail using the Mail::queue functionality. You can learn more about this in “queue()”. You can also queue Artisan commands, which we covered in Chapter 8.

Firing an Event

There are three ways to fire an event. You can use the Event facade, inject the Dispatcher, or use the event() global helper:

Event::fire(new UserSubscribed($user, $plan));
// or
$dispatcher = app(Illuminate\Contracts\Events\Dispatcher::class);
$dispatcher->fire(new UserSubscribed($user, $plan));
// or
event(new UserSubscribed($user, $plan));

If in doubt, I’d recommend using the global helper function.

To create an event to fire, use the make:event Artisan command:

php artisan make:event UserSubscribed

That’ll make a file that looks something like Example 16-9.

<?php

namespace App\Events;

use Illuminate\Broadcasting\Channel;
use Illuminate\Queue\SerializesModels;
use Illuminate\Broadcasting\PrivateChannel;
use Illuminate\Broadcasting\PresenceChannel;
use Illuminate\Foundation\Events\Dispatchable;
use Illuminate\Broadcasting\InteractsWithSockets;
use Illuminate\Contracts\Broadcasting\ShouldBroadcast;

class UserSubscribed
{
    use Dispatchable, InteractsWithSockets, SerializesModels;

    /**
     * Create a new event instance.
     *
     * @return void
     */
    public function __construct()
    {
        //
    }

    /**
     * Get the channels the event should be broadcast on.
     *
     * @return \Illuminate\Broadcasting\Channel|array
     */
    public function broadcastOn()
    {
        return new PrivateChannel('channel-name');
    }
}

Let’s take a look at what we get here. SerializesModels works just like with jobs; it allows you to accept Eloquent models as parameters. InteractsWithSockets, ShouldBroadcast, and the broadcastOn() method provide the backing functionality for broadcasting events using WebSockets, which we’ll cover in a bit.

It might seem strange that there’s no handle() or fire() method here. But remember, this object exists not to determine a particular action, but just to encapsulate some data. The first piece of data is its name; UserSubscribed tells us that a particular event happened (a user subscribed). The rest of the data is any data we pass into the constructor and associate with this entity.

Example 16-10 shows what we might want to do with our UserSubscribed event.

...
class UserSubscribed
{
    use InteractsWithSockets, SerializesModels;

    public $user;
    public $plan;

    public function __construct($user, $plan)
    {
        $this->user = $user;
        $this->plan = $plan;
    }
}

Now we have an object that appropriately represents the event that happened: $event->user subscribed to the $event->plan plan. Remember, firing this event is as simple as event(new UserSubscribed($user, $plan)).

Listening for an Event

We have an event, and the ability to fire it. Now let’s look at how to listen for it.

First, we’ll create an event listener. Let’s say we want to email the app’s owner every time a new user subscribes:

php artisan make:listener EmailOwnerAboutSubscription --event=UserSubscribed

That gives us the file in Example 16-11.

<?php

namespace App\Listeners;

use App\Events\UserSubscribed;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Contracts\Queue\ShouldQueue;

class EmailOwnerAboutSubscription
{
    /**
     * Create the event listener.
     *
     * @return void
     */
    public function __construct()
    {
        //
    }

    /**
     * Handle the event.
     *
     * @param  UserSubscribed  $event
     * @return void
     */
    public function handle(UserSubscribed $event)
    {
        //
    }
}

This is where the action happens—where the handle() method lives. This method expects to be passed an event of type UserSubscribed and act in response to it.

So, let’s make it send an email (Example 16-12).

...
use App\Mail\UserSubscribed as UserSubscribedMessage;

class EmailOwnerAboutSubscription
{
    public function handle(UserSubscribed $event)
    {
        Log::info('Emailed owner about new user: ' . $event->user->email);

        Mail::to(config('app.owner-email'))
            ->send(new UserSubscribedMessage($event->user, $event->plan);
    }
}

Great! Now, one last task: we need to set this listener up to listen to the UserSubscribed event. We’ll set that up in the $listen property of the EventServiceProvider class (see Example 16-13).

class EventServiceProvider extends ServiceProvider
{
    protected $listen = [
        \App\Events\UserSubscribed::class => [
            \App\Listeners\EmailOwnerAboutSubscription::class,
        ],
    ];

As you can see, the key of each array entry is the class name of the event, and the value is an array of listener class names. We can add as many class names as we want under the UserSubscribed key and they will all listen and respond to each UserSubscribed event.

<?php

namespace App\Listeners;

class UserEventSubscriber
{
    public function onUserSubscription($event)
    {
        // Handles the UserSubscribed event
    }

    public function onUserCancellation($event)
    {
        // Handles the UserCancelled event
    }

    public function subscribe($events)
    {
        $events->listen(
            \App\Events\UserSubscribed::class,
            'App\Listeners\UserEventSubscriber@onUserSubscription'
        );

        $events->listen(
            \App\Events\UserCancelled::class,
            'App\Listeners\UserEventSubscriber@onUserCancellation'
        );
    }
}

...
class EventServiceProvider extends ServiceProvider
{
    ...
    protected $subscribe = [
        \App\Listeners\UserEventSubscriber::class
    ];
}

Configuration and Setup

Take a look at config/broadcasting.php to find the configuration settings for your event broadcasting. Laravel supports three drivers for broadcasting: Pusher, a paid SaaS offering; Redis, for locally running WebSocket servers; and log, for local development and debugging.

Broadcasting an Event

To broadcast an event, you need to mark that event as a broadcast event by having it implement the Illuminate\Contracts\Broadcasting\ShouldBroadcast interface. This interface requires you to add the broadcastOn() method, which will return an array of either strings or Channel objects, each representing a WebSocket channel.

Example 16-16 shows our UserSubscribed event, modified to broadcast on two channels: one for the user (to confirm the user’s subscription) and one for admins (to notify them of a new subscription).

...
use Illuminate\Contracts\Broadcasting\ShouldBroadcast;

class UserSubscribed implements ShouldBroadcast
{
    use Dispatchable, InteractsWithSockets, SerializesModels;

    public $user;
    public $plan;

    public function __construct($user, $plan)
    {
        $this->user = $user;
        $this->plan = $plan;
    }

    public function broadcastOn()
    {
        // String syntax
        return [
            'users.' . $this->user->id,
            'admins'
        ];

        // Channel object syntax
        return [
            new Channel('users.' . $this->user->id),
            new Channel('admins'),
            // If it were a private channel: new PrivateChannel('admins'),
            // If it were a presence channel: new PresenceChannel('admins'),
        ];
    }
}

By default, any public properties of your event will be serialized as JSON and sent along as the data of your broadcast event. That means the data of one of our broadcast UserSubscribed events might look like Example 16-17.

{
    'user': {
        'id': 5,
        'name': 'Fred McFeely',
        ...
    },
    'plan': 'silver'
}

You can override this by returning an array of data from the broadcastWith() method on your event, as in Example 16-18.

public function broadcastWith()
{
    return [
        'userId' => $this->user->id,
        'plan' => $this->plan
    ];
}

You can customize which queue your event is pushed onto with its broadcastQueue() method, as in Example 16-19. You may choose to do this so you can keep other queue items from slowing down your event broadcast; real-time WebSockets aren’t much fun if a long-running job that’s higher in the queue keeps the events from going out in time.

public $broadcastQueue = 'websockets-for-faster-processing';

You can also force a given event to skip the queue entirely (using the “sync” queue driver, which is processed by the current PHP thread), by having it implement the ShouldBroadcastNow contract.

use Illuminate\Contracts\Broadcasting\ShouldBroadcastNow;

class UserSubscribed implements ShouldBroadcastNow
{
    //
}

And, finally, you can choose to customize whether a given event should be broadcast at all by giving it a broadcastWhen() method:

public function broadcastWhen()
{
    // Notify me only when users sign up from the White House
    return str_contains($this->user->email, 'whitehouse.gov');
}

Receiving the Message

If you choose to host your own Redis WebSockets server, the Laravel docs have a great walkthrough on how to set that up using socket.io and ioredis.

As of this book’s publication, the most common solution Laravel developers use is Pusher. Plans over a certain size cost money, but there’s a generous free plan. Pusher makes it incredibly simple to set up a simple WebSocket server, and its JavaScript SDK handles all of the authentication and channel management with almost no work on your part. SDKs are available for iOS, Android, and many more platforms, languages, and frameworks.

There’s also a new tool announced just before the publication of this book called Laravel WebSockets that lets you host your own Laravel-based Pusher-compatible WebSockets server. You can install the package into your current Laravel app (the same app you’re broadcasting from) or into a separate microservice.

If you choose to work with a Laravel WebSockets server, you’ll follow all of the directions in this book as if you were working with Pusher, but your configuration settings will be a bit different.

...
<script src="https://js.pusher.com/4.3/pusher.min.js"></script>
<script>
// Enable pusher logging - don't include this in production
Pusher.logToConsole = true;

// Globally, perhaps; just a sample of how to get data in
var App = {
    'userId': {{ auth()->id() }},
    'pusherKey': '{{ config('broadcasting.connections.pusher.key') }}'
};

// Locally
var pusher = new Pusher(App.pusherKey, {
    cluster: '{{ config('broadcasting.connections.pusher.options.cluster') }}',
    encrypted: {{ config('broadcasting.connections.pusher.options.encrypted') }}
});

var pusherChannel = pusher.subscribe('users.' + App.userId);

pusherChannel.bind('App\\Events\\UserSubscribed', (data) => {
    console.log(data.user, data.plan);
});
</script>

Advanced Broadcasting Tools

Laravel has a few more tools to make it possible to perform more complex interactions in event broadcasting. These tools, a combination of framework features and a JavaScript library, are called Laravel Echo.

These framework features work best when you use Laravel Echo in your JavaScript frontend (which we’ll cover in “Laravel Echo (the JavaScript Side)”), but you can still enjoy some of the benefits of Echo without using the JavaScript components. Echo will work with both Pusher and Redis, but I’m going to use Pusher for any examples.

// Run this right after you initialize echo
// With Axios
window.axios.defaults.headers.common['X-Socket-Id'] = Echo.socketId();

// jQuery
$.ajaxSetup({
    headers: {
        'X-Socket-Id': Echo.socketId()
    }
});

broadcast(new UserSubscribed($user, $plan))->toOthers();

...
// routes/channels.php

// Define how to authenticate a private channel
Broadcast::channel('teams.{teamId}', function ($user, $teamId) {
    return (int) $user->team_id === (int) $teamId;
});

// Define how to authenticate a presence channel; return any data
// you want the app to have about the user in the channel
Broadcast::channel('rooms.{roomId}', function ($user, $roomId) {
    if ($user->rooms->contains($roomId)) {
        return [
            'name' => $user->name
        ];
    }
});

var pusher = new Pusher(App.pusherKey, {
    authEndpoint: '/broadcasting/auth'
});

...
<script src="https://js.pusher.com/4.3/pusher.min.js"></script>
<script>
    // Enable pusher logging - don't include this in production
    Pusher.logToConsole = true;

    // Globally, perhaps; just a sample of how to get data in
    var App = {
        'userId': {{ auth()->id() }},
        'pusherKey': '{{ config('broadcasting.connections.pusher.key') }}'
    };

    // Locally
    var pusher = new Pusher(App.pusherKey, {
        cluster: '{{ config('broadcasting.connections.pusher.options.cluster') }}',
        encrypted: {{ config('broadcasting.connections.pusher.options.encrypted') }},
        authEndpoint: '/broadcasting/auth'
    });

    // Private channel
    var privateChannel = pusher.subscribe('private-teams.1');

    privateChannel.bind('App\\Events\\UserSubscribed', (data) => {
        console.log(data.user, data.plan);
    });

    // Presence channel
    var presenceChannel = pusher.subscribe('presence-rooms.5');

    console.log(presenceChannel.members);
</script>

Laravel Echo (the JavaScript Side)

Laravel Echo is comprised of two pieces: the advanced framework features we just covered, and a JavaScript package that takes advantage of those features and drastically reduces the amount of boilerplate code you need to write powerful WebSocket-based frontends. The Echo JavaScript package makes it easy to handle authentication, authorization, and subscribing to private and presence channels. Echo can be used with the SDKs for either Pusher JS (for Pusher or a custom Pusher-compatible server) or socket.io (for Redis).

npm install pusher-js laravel-echo --save

let mix = require('laravel-mix');

mix.js('resources/assets/js/app.js', 'public/js');

// app.js
require('./bootstrap');

// ... lots of Vue stuff ...

// Add your Echo bindings here

// bootstrap.js
import Echo from "laravel-echo";

window.Echo = new Echo({
    broadcaster: 'pusher',
    key: process.env.MIX_PUSHER_APP_KEY,
    cluster: process.env.MIX_PUSHER_APP_CLUSTER
});

<meta name="csrf-token" content="{{ csrf_token() }}">

<script src="{{ asset('js/app.js') }}"></script>

var currentTeamId = 5; // Likely set elsewhere

Echo.channel(`teams.${currentTeamId}`)
    .listen('UserSubscribed', (data) => {
        console.log(data);
    });

Echo.channel(`teams.${currentTeamId}`)
    .listen('UserSubscribed', (data) => {
        console.log(data);
    })
    .listen('UserCanceled', (data) => {
        console.log(data);
    });

var currentTeamId = 5; // Likely set elsewhere

Echo.private(`teams.${currentTeamId}`)
    .listen('UserSubscribed', (data) => {
        console.log(data);
    });

var currentTeamId = 5; // Likely set elsewhere

Echo.join(`teams.${currentTeamId}`)
    .here((members) => {
        console.log(members);
    });

var currentTeamId = 5; // Likely set elsewhere

Echo.join('teams.' + currentTeamId)
    .here((members) => {
        // runs when you join
        console.table(members);
    })
    .joining((joiningMember, members) => {
        // runs when another member joins
        console.table(joiningMember);
    })
    .leaving((leavingMember, members) => {
        // runs when another member leaves
        console.table(leavingMember);
    });

Echo.private(`App.User.${userId}`)
    .notification((notification) => {
        console.log(notification.type);
    });

Echo.private('room')
    .whisper('typing', {
        name: this.user.name
    });

Echo.private('room')
    .listenForWhisper('typing', (e) => {
        console.log(e.name);
    });

Available Task Types

First, let’s take a look at the simplest option: a closure, run every minute (Example 16-32). That means that, every time the cron job hits the schedule:run command, it will call this closure.

// app/Consoles/Kernel.php
public function schedule(Schedule $schedule)
{
    $schedule->call(function () {
        CalculateTotals::dispatch();
    })->everyMinute();
}

There are two other types of tasks you can schedule: Artisan and shell commands.

You can schedule Artisan commands by passing their syntax exactly as you would call them from the command line:

$schedule->command('scores:tally --reset-cache')->everyMinute();

And you can run any shell commands that you could run with PHP’s exec() method:

$schedule->exec('/home/myapp.com/bin/build.sh')->everyMinute();

Available Time Frames

The beauty of the scheduler isn’t just that you can define your tasks in code; it’s that you can schedule them in code, too. Laravel keeps track of time passing and evaluates whether it’s time for any given task to run. That’s easy with everyMinute() because the answer is always simple: run the task. But Laravel keeps the rest simple for you, too, even for the most complex of requests.

Let’s take a look at your options by starting with a monstrous definition that’s simple in Laravel:

$schedule->call(function () {
    // Runs once a week on Sunday at 23:50
})->weekly()->sundays()->at('23:50');

Notice that we can chain times together: we can define frequency and specify the day of the week and the time, and of course we can do so much more.

Table 16-1 shows a list of potential date/time modifiers for use when scheduling a job.

Most of these can be chained one after another, but of course, any combinations that don’t make sense chained can’t be chained.

Example 16-33 shows a few combinations you could consider.

// Both run weekly on Sunday at 23:50
$schedule->command('do:thing')->weeklyOn(0, '23:50');
$schedule->command('do:thing')->weekly()->sundays()->at('23:50');

// Run once per hour, weekdays, 8am-5pm
$schedule->command('do:thing')->weekdays()->hourly()->when(function () {
    return date('H') >= 8 && date('H') <= 17;
});

// Run once per hour, weekdays, 8am-5pm using the "between" method
$schedule->command('do:thing')->weekdays()->hourly()->between('8:00', '17:00');

// Run every thirty minutes except when directed to by the SkipDetector
$schedule->command('do:thing')->everyThirtyMinutes()->skip(function () {
    return app('SkipDetector')->shouldSkip();
});

Blocking and Overlap

If you want to avoid your tasks overlapping each other—for example, if you have a task running every minute that may sometimes take longer than a minute to run—end the schedule chain with the withoutOverlapping() method. This method skips a task if the previous instance of that task is still running:

$schedule->command('do:thing')->everyMinute()->withoutOverlapping();

Handling Task Output

Sometimes the output from your scheduled task is important, whether for logging, notifications, or just ensuring that the task ran.

If you want to write the returned output of a task to a file, use sendOutputTo():

$schedule->command('do:thing')->daily()->sendOutputTo($filePath);

If you want to append it to a file instead, use appendOutputTo():

$schedule->command('do:thing')->daily()->appendOutputTo($filePath);

And if you want to email the output to a designated recipient, write it to a file first and then add emailOutputTo():

$schedule->command('do:thing')
    ->daily()
    ->sendOutputTo($filePath)
    ->emailOutputTo('me@myapp.com');

Make sure that your email settings are configured correctly in Laravel’s basic email configuration.

You may also want to send some output to a webhook to verify that your tasks ran correctly. There are a few services that provide this sort of uptime monitoring, most significantly Laravel Envoyer (a zero-downtime deployment service that also provides cron uptime monitoring) and Dead Man’s Snitch, a tool designed purely for monitoring cron job uptime.

These services don’t expect something to be emailed to them, but rather expect an HTTP “ping,” so Laravel makes that easy with pingBefore() and thenPing():

$schedule->command('do:thing')
    ->daily()
    ->pingBefore($beforeUrl)
    ->thenPing($afterUrl);

If you want to use the ping features, you’ll need to pull in Guzzle using Composer: composer require guzzlehttp/guzzle.

Task Hooks

Speaking of running something before and after your task, there are hooks for that, with before() and after():

$schedule->command('do_thing')
    ->daily()
    ->before(function () {
        // Prepare
    })
    ->after(function () {
        // Cleanup
    });