Laravel Queues with Other Languages

Author

Name: Chris Fidao

Languages translating eachother. — Image by Annie Ruygt

Need to run some Laravel workers? Deploy your Laravel application on Fly.io, you’ll be up and running in minutes!

Learning multiple programming languages is good — there are a lot of benefits!

One such benefit is the ability to use what a language is strongest at. Some top-of-mind examples include Python’s various data-munging libraries, or Go’s concurrency tooling.

Now, you may want to use another language — but you likely aren’t about to rewrite your app. Luckily, I have some ideas! They look a bit like micro-services, but aren’t. Mostly.

What I like to do is sprinkle in other languages as part of my codebase’s “plumbing” — stuff that happens unseen, in the background.

One way to go about this is with gRPC. This lets you call a function, which (in the background) makes a network call to some other codebase, and gets a response. It looks like you’re making just any old function call, but it actually might be talking to another application. This is fun but gets complicated, relative to the 2nd idea.

That second idea: using queues! Queues are a tidy abstraction layer between languages because your codebase(s) can push or pull a job from a queue, and doesn’t need to care about who is reading or writing those jobs.

Let’s Do Some Queueing

Let’s see how to make use of Laravel’s queues with multiple languages.

There are 2 strategies you might employ:

Push to a queue from Laravel and read jobs in another language
Push to a queue from another language and read jobs in Laravel

The tricky bit is that Laravel creates (and consumes) jobs of a specific format. We need to know a job’s “shape” so we can read or write them correctly.

What’s In a Laravel Job

If you dispatch a job from Laravel into a queue, the payload looks something like this:

    {
    "uuid": "f69d79dc-2e0f-495a-9f70-62de0b4c7299",
    "displayName": "App\\Jobs\\SomeFancyJob",
    "job": "Illuminate\\Queue\\CallQueuedHandler@call",
    "maxTries": null,
    "maxExceptions": null,
    "failOnTimeout": false,
    "backoff": null,
    "timeout": null,
    "retryUntil": null,
    "data": {
        "commandName": "App\\Jobs\\SomeFancyJob",
        "command": "O:21:\"App\\Jobs\\SomeFancyJob\":1:{s:4:\"user\";O:45:\"Illuminate\\Contracts\\Database\\ModelIdentifier\":5:{s:5:\"class\";s:15:\"App\\Models\\User\";s:2:\"id\";i:1;s:9:\"relations\";a:0:{}s:10:\"connection\";s:6:\"sqlite\";s:15:\"collectionClass\";N;}}"
    }
}

  

The fun thing happening here is that the job key is specifying a class and method! In this case, it’s Illuminate\\Queue\\CallQueuedHandler@call. This is Laravel’s default handler - it does a bit of magic to rehydrate the job class (SomeFancyJob in our case) and run the handle() method on that. It’s a wrapper around our SomeFancyJob class.

That job key is the main bit — it defines what class and method Laravel should call when processing that job.

Finally, note that the command is serialized PHP - an instance of the SomeFancyJob class. More on that in a bit.

A Note on Queue Drivers

Unfortunately, we need to care about the queue driver. I tend to use Amazon’s SQS, since it’s the most legible — you can inspect jobs fairly easily and see the data within them.

Furthermore, SQS is API driven, which lets us avoid some thorny issues. Laravel does a lot of work for us in terms of ensuring we only pull one job at a time (redis, database, etc). With SQS, we’re just making calls to an API — SQS deals with the harder stuff. This means our “other language” code doesn’t need to care about locking mechanisms.

Processing Third-Party Jobs

Let’s see how to produce a job (from another language) that Laravel will later process via queue:work (or queue:listen).

We could mimic Laravel’s jobs ourselves and push a job that Laravel will handle without any further work. However, one thing I hate doing is asking another language to produce serialized PHP. It’s fragile! Luckily we aren’t forced to do that (although you still could, if you wanted).

What we can do instead is hook into the job key and have a class of our choosing handle processing the job! Our “other language” could add a job to the queue with a payload (a JSON string) like this:

    {
    "uuid": "f69d79dc-2e0f-495a-9f70-62de0b4c7299",
    "displayName": "App\\Jobs\\HandleThirdPartyJob",
    "job": "App\\Jobs\\HandleThirdPartyJob@call",
    "maxTries": 3,
    "maxExceptions": 1,
    "failOnTimeout": false,
    "backoff": null,
    "timeout": null,
    "retryUntil": null,
    "data": {
        "userId": 1,
        "foo": "bar"
    }
}

  

In this case, the class that will handle this message is App\Jobs\HandleThirdPartyJob (via its call() method).

The trade off here is that our HandleThirdPartyJob class needs to handle queue retries/failures (and so on) itself. We don’t get the benefit of the Illuminate\\Queue\\CallQueuedHandler@call class wrapping the job and doing much of that work for us.

A working example of HandleThirdPartyJob:

    <?php

namespace App\Jobs;

use App\Models\User;
use Illuminate\Queue\Jobs\SqsJob;
use Illuminate\Support\Facades\Log;

class HandleThirdPartyJob
{
    public function call(SqsJob $job, array $data): void
    {
        try {
            $u = User::findOrFail($data['userId'] ?? null);
        } catch(\Exception $e) {
            // Alternatively, to retry this job:
            // $job->release();
            return $job->delete();
        }

        Log::info('The user to perform stuff on is :' . $u->email);
        $job->delete();
    }
}

  

When to use this?

Your use cases will vary wildly. My use case has been for Chipper CI, where a little Golang service accepts streamed build output. For each “chunk” of build output, it creates an SQS (FIFO, in my case) message, which Laravel consumes - saving the build output and streaming it to the browser.

Using Golang to Push a Job

Here’s a bit of what it looks like for a Golang app to push a job into SQS in the format Laravel can read:

    // SendMessage takes req and converts it to an SQS job
// req is data (a Go struct) that can get turned into a JSON string
func SendMessage(req *SendMessageRequest) error {
  // Aws.Sess needs creating
  sqs := sqs.New(Aws.Sess)

  // Turn this data into a JSON string
  messageBody, err := json.Marshal(SqsMessageBody{
    DisplayName: "App\\Jobs\\HandleThirdPartyJob",
    Job:         "App\\Jobs\\HandleThirdPartyJob@call",
    MaxTries:    0,
    Timeout:     0,
    Data:        req, // data to turn into JSON
  })

  if err != nil {
    return fmt.Errorf("could not create job JSON for SQS: %w", err)
  }

  _, err = sqs.SendMessage(&sqs.SendMessageInput{
    MessageBody:  aws.String(string(messageBody)),
    QueueUrl:     aws.String(Aws.config.SqsUrl),
  })

  if err != nil {
    return fmt.Errorf("could not push message to SQS: %w", err)
  }

  return nil
}

  

Pushing to Third-Party Workers

This use case is a bit more interesting. What if Laravel pushed a job into a queue, and another language processed that job?

This is useful for all sorts of things — each language has their own strengths. Maybe you use Python to do some data munging (Python is really good at this). Maybe you use Golang to make use of concurrency while processing a job (sending lots of HTTP requests). There’s a lot of possible use cases!

Mohamed (who worked at Laravel) recently wrote up an excellent example of this.

What’s this look like in Laravel? Well, it turns out that Laravel can push a bunch of ad-hoc data into a queue:

    # These both work
Queue::push(['foo' => 'bar']);
Queue::push("nonsense", ['foo' => 'bar']);

  

And the respective payloads from those calls look like this:

    {
  "uuid": "8c7c2034-05e4-4321-8a69-d892d6f6703d",
  "displayName": null,
  "job": {
    "foo": "bar"
  },
  "maxTries": null,
  "maxExceptions": null,
  "failOnTimeout": false,
  "backoff": null,
  "timeout": null,
  "data": ""
}

{
  "uuid": "ade43fc5-c530-4bb3-b6ba-b0fbc6cf6245",
  "displayName": "nonsense",
  "job": "nonsense",
  "maxTries": null,
  "maxExceptions": null,
  "failOnTimeout": false,
  "backoff": null,
  "timeout": null,
  "data": {
    "foo": "bar"
  }
}

  

Since whatever language we choose to process the job won’t care about the Laravel-specific data, we can decide if we like our relevant data being within job or data keys (or even both).

Here we can receive said message. Note that the SQS message.Body property is a string — the JSON string that we’ve been talking about above.

    // Here, we expect the json "data" key to 
// be in a Nonsense struct, which
// in turn expects to be created from
// JSON in format {"foo": "some-string"}
type LaravelMessage struct {
  Uuid string `json:"uuid"`
  Data Nonsense `json:"data"`
}

type Nonsense struct {
  Foo string `json:"foo"`
}

func ReceiveMessage() (*Nonsense, error) {
  // Aws.Sess needs creating
  sqs := sqs.New(Aws.Sess)

  // Get one message at a time
  msgResult, err := sqs.ReceiveMessage(&sqs.ReceiveMessageInput{
    QueueUrl:            someQueueURL,
    MaxNumberOfMessages: aws.Int64(1), // one message
    VisibilityTimeout:   someTimeout,
  })

  if len(msgResult.Messages) > 0 {
    msg := msgResult.Messages[0]

    // Convert the SQS message body into the Go struct
    // defined above, which expects a Laravel-formatted job.
    // Note: message.Body is the JSON string containing
    // uuid, data, and other Laravel job data
    laravelMsg := &LaravelMessage{}
    err := json.Unmarshal([]byte(*message.Body), laravelMsg)
    if err != nil {
      return nil, fmt.Errorf("could not parse SQS message body: %v", err)
    }

    return laravelMsg, nil
  }

  return nil, fmt.Errorf("no sqs message found in queue: %s", someQueueUrl)
}

  

Golang, being strictly typed (and being Golang), is a bit verbose because of how it needs you to predefine what data you expect to get in a JSON string.

All we’re doing here is grabbing the SQS message Body and converting it to an object. We start by assuming the message body is a Laravel-generated JSON string. Then we grab just the bits we care about (the uuid and the data). Our Nonsense struct definition lines up with the data found in the JSON string, and so Golang is able to create an object from the JSON.

After that, our Go program can do whatever it needs with the data!

Fly.io ❤️ Laravel

Fly your servers close to your users—and marvel at the speed of close proximity. Deploy globally on Fly in minutes!
Deploy your Laravel app! →

Other Queue Drivers

As mentioned, I like using SQS because pushing and popping jobs (messages) off the queues just involve some API work — gory details of “at most once” or “at least once” guarantees are taken care of for you.

For other drivers, we need to care about how we push/pop jobs to the queue storage.

Pushing a job into the queue for other drivers isn’t too hard — depending on the driver (sqs, redis, database, etc).

Redis push method is here (it’s a LUA script)
Database push method is here (it’s just an insert!)

Popping (pulling/getting) a job gets a bit trickier, as you need to care about the driver’s locking mechanisms (to ensure multiple workers don’t consume a job more than once).

Redis pop method is here (again, a bit of LUA)
Database pop’s relevant method is here
- It uses some locking, but the exact locking mechanism changes per database driver via the getLockForPopping() method

SQS is the easiest there. Redis is probably okay-ish (just steal some LUA from the Laravel framework core), and Database is the most annoying, since the syntax and locking mechanism changes per database type.

It’s totally doable to use these techniques with other drivers, it’s just a bit more annoying — API work is generally the easiest (and SQS is cheap).

Next post ↑: Laravel, React, and Inertia SSR on Fly.io
Previous post ↓: Improving Client Side Pagination with Livewire