Installation¶

1

$ composer require symfony/messenger

Alternatively, you can clone the https://github.com/symfony/messenger repository.

Concepts¶

Sender:

Responsible for serializing and sending messages to something. This something can be a message broker or a third party API for example.

Receiver:

Responsible for deserializing and forwarding messages to handler(s). This can be a message queue puller or an API endpoint for example.

Handler:

Responsible for handling messages using the business logic applicable to the messages.

Bus¶

The bus is used to dispatch messages. The behavior of the bus is in its ordered middleware stack. The component comes with a set of middleware that you can use.

When using the message bus with Symfony's FrameworkBundle, the following middleware are configured for you:

1. LoggingMiddleware (logs the processing of your messages)
2. SendMessageMiddleware (enables asynchronous processing)
3. HandleMessageMiddleware (calls the registered handle)

Example:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

use App\Message\MyMessage;
use Symfony\Component\Messenger\MessageBus;
use Symfony\Component\Messenger\Handler\Locator\HandlerLocator;
use Symfony\Component\Messenger\Middleware\HandleMessageMiddleware;

$bus = new MessageBus([
    new HandleMessageMiddleware(new HandlerLocator([
        MyMessage::class => $handler,
    ])),
]);

$result = $bus->dispatch(new MyMessage(/* ... */));

Handlers¶

Once dispatched to the bus, messages will be handled by a "message handler". A message handler is a PHP callable (i.e. a function or an instance of a class) that will do the required processing for your message:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

namespace App\MessageHandler;

use App\Message\MyMessage;

class MyMessageHandler
{
   public function __invoke(MyMessage $message)
   {
       // Message processing...
   }
}

Adding Metadata to Messages (Envelopes)¶

If you need to add metadata or some configuration to a message, wrap it with the Envelope class. For example, to set the serialization groups used when the message goes through the transport layer, use the SerializerConfiguration envelope:

1
2
3
4
5
6
7
8

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Transport\Serialization\SerializerConfiguration;

$bus->dispatch(
    (new Envelope($message))->with(new SerializerConfiguration([
        'groups' => ['my_serialization_groups'],
    ]))
);

At the moment, the Symfony Messenger has the following built-in envelopes:

1. SerializerConfiguration, to configure the serialization groups used by the transport.
2. ValidationConfiguration, to configure the validation groups used when the validation middleware is enabled.
3. ReceivedMessage, an internal item that marks the message as received from a transport.

Instead of dealing directly with the messages in the middleware you can receive the envelope by implementing the EnvelopeAwareInterface marker, like this:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

use Symfony\Component\Messenger\Asynchronous\Transport\ReceivedMessage;
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\EnvelopeAwareInterface;

class MyOwnMiddleware implements MiddlewareInterface, EnvelopeAwareInterface
{
    public function handle($envelope, callable $next)
    {
        // $envelope here is an `Envelope` object, because this middleware
        // implements the EnvelopeAwareInterface interface.

        if (null !== $envelope->get(ReceivedMessage::class)) {
            // Message just has been received...

            // You could for example add another item.
            $envelope = $envelope->with(new AnotherEnvelopeItem(/* ... */));
        }

        return $next($envelope);
    }
}

The above example will forward the message to the next middleware with an additional envelope item if the message has just been received (i.e. has the ReceivedMessage item). You can create your own items by implementing EnvelopeAwareInterface.

Transports¶

In order to send and receive messages, you will have to configure a transport. A transport will be responsible for communicating with your message broker or 3rd parties.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

namespace App\MessageSender;

use App\Message\ImportantAction;
use Symfony\Component\Messenger\Transport\SenderInterface;
use Symfony\Component\Messenger\Envelope;

class ImportantActionToEmailSender implements SenderInterface
{
   private $toEmail;
   private $mailer;

   public function __construct(\Swift_Mailer $mailer, string $toEmail)
   {
       $this->mailer = $mailer;
       $this->toEmail = $toEmail;
   }

   public function send(Envelope $envelope)
   {
       $message = $envelope->getMessage();

       if (!$message instanceof ImportantAction) {
           throw new \InvalidArgumentException(sprintf('This transport only supports "%s" messages.', ImportantAction::class));
       }

       $this->mailer->send(
           (new \Swift_Message('Important action made'))
               ->setTo($this->toEmail)
               ->setBody(
                   '<h1>Important action</h1><p>Made by '.$message->getUsername().'</p>',
                   'text/html'
               )
       );
   }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

namespace App\MessageReceiver;

use App\Message\NewOrder;
use Symfony\Component\Messenger\Transport\ReceiverInterface;
use Symfony\Component\Serializer\SerializerInterface;
use Symfony\Component\Messenger\Envelope;

class NewOrdersFromCsvFile implements ReceiverInterface
{
   private $serializer;
   private $filePath;

   public function __construct(SerializerInterface $serializer, string $filePath)
   {
       $this->serializer = $serializer;
       $this->filePath = $filePath;
   }

   public function receive(callable $handler) : void
   {
       $ordersFromCsv = $this->serializer->deserialize(file_get_contents($this->filePath), 'csv');

       foreach ($ordersFromCsv as $orderFromCsv) {
           $order = new NewOrder($orderFromCsv['id'], $orderFromCsv['account_id'], $orderFromCsv['amount']);

           $handler(new Envelope($order));
       }
   }

   public function stop(): void
   {
       // noop
   }
}