AWS and Kubernetes certified
AWS Advanced Networking Specialty
CKAD

Using Symfony Messenger with AWS SQS

Symfony 4.1 adds a new Messenger component which allows us to handle a message now or later (by pushing it to a queue system). I wrote a post a long time ago to show some use cases and implementation of how to use Symfony with sqs. Now, with the new Messenger component, thing becomes much easier.

Let's start with a use case

When Emily submits the contact form, my website will send an email to me. I want my website postpone the process of sending the email but show Emily a Thank you pop-up as soon as possible.

The Steps

  1. Assume we have a Symfony 4.1+ flex application, Run

composer require messenger enqueue/messenger-adapter  enqueue/sqs

Then you will see some errors, don’t worry about them now, it’s because we haven’t specified the value of the ENQUEUE_DSN environment variable. Let’s edit and clean both .env and .env.dist file. Make sure you only have one ENQUEUE_DSN variable in both files.

//.env
ENQUEUE_DSN=sqs:?key=[aws_key]&secret=[aws_secret]&region=[aws_region]

update messenger.yaml

framework:
    messenger:
        transports:
              sqs: enqueue://default?&topic[name]=topic&queue[name]=queue_name

The topic[name] option is not necessary for sqs, but it is required by the messenger-adapter bundle for now, see related Git Issue

  1. According to the official tutorial, we need to define three things.
    Firstly, the message object. A message object is simply a PHP object that can be handled by a handler.
    In my example, I use a Data transfer object (DTO) with Symfony form to handle form submission. The DTO can also be used as a Message object as well, The Message object looks like

<?php

namespace App\Requests;

use Symfony\Component\Validator\Constraints as Assert;

class Contact
{
    /**
     * @Assert\NotBlank()
     */
    protected $name;

    /**
     * @Assert\NotBlank()
     * @Assert\Email()
     */
    protected $email;

    /**
     * @Assert\NotBlank()
     */
    protected $subject;

    /**
     * @Assert\NotBlank()
     */
    protected $message;

    public function getName(): ?string
    {
        return $this->name;
    }

    public function setName(string $name): void
    {
        $this->name = $name;
    }

    public function getEmail(): ?string
    {
        return $this->email;
    }

    public function setEmail(string $email): void
    {
        $this->email = $email;
    }

    public function getSubject(): ?string
    {
        return $this->subject;
    }

    public function setSubject(string $subject): void
    {
        $this->subject = $subject;
    }

    public function getMessage(): ?string
    {
        return $this->message;
    }

    public function setMessage(string $message): void
    {
        $this->message = $message;
    }
}

Then, let’s dispatch the message in the controller

/**
 * @Route("/api/contact", name="contact")
 */
public function postContact(Request $request, MessageBusInterface $bus): Response
{
    $form = $this->createForm(ContactType::class, new Contact());
    $this->processForm($request, $form);

    if (!$form->isValid()) {
        $this->throwApiProblemValidationException($form);
    }

    $contact = $form->getData();

    $bus->dispatch($contact);

    return $this->createApiResponse($contact, 200);
}

Secondly, we need to create a handler to handle the message (a handler that sends the email in our case). A handler can be as easy as

<?php
declare(strict_types=1);

namespace App\MessageHandler;

use App\Requests\Contact;
use App\Service\Mailer;
use Swift_Message;

class ContactHandler
{
    private $adminEmail;
    private $hostEmail;
    private $mailer;

    public function __construct(string $adminEmail, string $hostEmail, Mailer $mailer)
    {
        $this->adminEmail = $adminEmail;
        $this->hostEmail = $hostEmail;
        $this->mailer = $mailer;
    }

    public function __invoke(Contact $contact)
    {
        $content = \/sprintf/((
            "Email from %s \r\nMessage: \r\n%s",
            $contact->getEmail(),
            $contact->getMessage()
        );

        $message = (new Swift_Message($contact->getSubject()))
            ->setFrom($this->hostEmail)
            ->setTo($this->adminEmail)
            ->setBody($content, Mailer::/CONTENT_TEXT/);

        $this->mailer->sendEmail($message);
    }
}

Then, according to the offical document, we need to tag our handler in the services.yaml like

App\MessageHandler\ContactHandler:
    tags: [messenger.message_handler]

However, we can leverage the auto-configuration feature (this is a missing part of the official doc now) to avoid adding lines in services.yaml. Simply, we implement the MessageHandlerInterface in our ContactHandler.php

use Symfony\Component\Messenger\Handler\MessageHandlerInterface;
class ContactHandler implements MessageHandlerInterface
{
  //...
}

Now, the message should be handled synchronously, having said that, when Emily clicks the submit button, the email will be processed immediately.

We want to postpone sending the email, so the last step is to config the transport. This is the beautify of the Messenger component as we don’t need to touch the source code to decide whether to process the message immediately or push it to the queue and process it later.

How? Let’s edit the messenger.yaml file

framework:
    messenger:
        transports:
            sqs: enqueue://default&topic[name]=your_topic&queue[name]=your_queue

        routing:
            'App\Requests\Contact': sqs

In order to process the message, we simply run the command

php bin/console messenger:consume sqs

Tricks

Sending email
As the command php bin/console messenger:consume sqs is a worker (a long running PHP cli command) , it won’t send the email immediately if we enabled email spool in swiftmailer.yaml.

//config/packages/swiftmailer.yaml
swiftmailer:
    url: '%env(MAILER_URL)%'
    spool: { type: 'memory' }
    url: '%env(MAILER_URL)%'

The quick solution is to remove the line spool: { type: 'memory' }
Deployment
I use supervisord to manage the queue worker. I found it is super hard to use environment vars as I have to edit them in three places: nginx conf (for web server), .bashrc or bash_profile (for normal Symfony commands) and supervisor conf (workers, managed by supervisord are not able to inherit
env vars defined by bash_profile or bashrc).

It is a nightmare. In the end, life is saved by just using .env instead of env vars

The working example and source code

The contact form submission of http://julianli.co/ follows exact the same process described by this post.

Check out the Source Code !

Edit on 24/12/2018

Issues I had when I was upgrading symfony4.1 to symfony 4.2

  1. There is a need to update the enqueue/sqs dependencies to >0.9 and we also need to update enqueue.yaml file, checkout the change

  2. I don't know why there is a weird decision made by the enqueue team but the new enqueue dependencies break the php bin/console messenger:consume sqs command. here is the fix . The root cause is if you don't set receiveTimeout option it defaults to 30 sec, However, in the SqsConsumer.php, the max time out is 20 sec. So if you only use the default option, it will throw an exception.