PMG Queue¶

Example Message¶

<?php
use PMG\Queue\Message;

final class SendAlert implements Message
{
    private $userId;

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

    public function getUserId()
    {
        return $this->userId;
    }
}

Because messages are serialized to be put in a persistent backend they shouldn’t include objects that require state. In the example above the message just contains a user’s identifier rather than the full object. The handler would then look up the user.

See Consumers and Producers for more information about handlers and messages fit into the system as a whole.

Routers¶

pmg/queue is built with multi-queue support in in mind. To accomplish that on the producer side of things an implementation of PMG\Queue\Router is used.

interface Router¶

Namespace:	PMG\Queue

queueFor(PMG\Queue\Message $message)¶

Looks a queue name for a given message.

Parameters:	$message – the message to route
Returns:	A string queue name if found, null otherwise.
Return type:	string or null

<?php
use PMG\Queue\Router\SimpleRouter;

// all message will go in the "queueName" queue
$router = new SimpleRouter('queueName');

<?php

use PMG\Queue\Router\MappingRouter;

$router = new MappingRouter([
    // the `SendAlert` message will go into the `Alerts` queue
    'SendAlert' => 'Alerts',
]);

<?php

use PMG\Queue\DefaultProducer;
use PMG\Queue\SimpleMessage;
use PMG\Queue\Router\FallbackRouter;
use PMG\Queue\Router\MappingRouter;

$router = new FallbackRouter(new MappingRouter([
    'SendAlert' => 'Alerts',
]), 'defaultQueue');

$producer = new DefaultProducer($driver, $router);

// goes into the `Alerts` queue
$producer->send(new SimpleMessage('SendAlert'));

// goes into `defaultQueue`
$producer->send(new SimpleMessage('OtherThing'));

Retrying Messages¶

When a message fails – by throwing an exception or returns false from a MessageHandler – the consumer puts it back in the queue to retry up to 5 times by default. This behavior can be adjusted by providing a RetrySpec as the third argument to DefaultConsumers constructor. pmg/queue provides a few by default.

Retry specs look at PMG\Queue\Envelope instances, not raw messages. See the internals documentation for more info about them.

interface RetrySpec¶

Namespace:	PMG\Queue

canRetry(PMG\Queue\Envelope $env)¶

Inspects an envelop to see if it can retry again.

Parameters:	$env – The message envelope to check
Returns:	true if the message can be retried, false otherwise.
Return type:	boolean

<?php

use PMG\Queue\DefaultConsumer;
use PMG\Queue\Retry\LimitedSpec;

// five retries by default. This is what the consumer does automatically
$retry = new LimitedSpec();

// Or limit to a specific number of retries
$retry = new LimitedSpec(2);

// $driver and $handler as above
$consumer = new DefaultConsumer($driver, $handler, $retry);

<?php

use PMG\Queue\DefaultConsumer;
use PMG\Queue\Retry\NeverSpec;

$retry = new NeverSpec();

// $driver and $handler as above
$consumer = new DefaultConsumer($driver, $handler, $retry);

Logging¶

When something goes wrong DefaultConsumer logs it with a PSR-3 Logger implementation. The default is to use a NullLogger, but you can provide your own logger as the fourth argument to DefaultConsumer‘s constructor.

<?php

use PMG\Queue\DefaultConsumer;

$monolog = new Monolog\Logger('yourApp');

// $driver, $handler, $retry as above
$consumer = new DefaultConsumer($driver, $handler, $retry, $monolog);

Build Custom Consumers¶

Extend PMG\\Queue\\AbstractConsumer to make things easy and only have to implement the once method. Here’s an example that decorates another Consumer with events.

<?php

use PMG\Queue\AbstractConsumer;
use PMG\Queue\Consumer;
use PMG\Queue\Message;
use Symfony\Component\EventDispatcher\Event;
use Symfony\Component\EventDispatcher\EventDispatcherInterface;

final class EventingConsumer extends AbstractConsumer
{
    /** @var Consumer */
    private $wrapped;

    /** @var EventDispatcherInterface $events */

    // constructor that takes a consumer and dispatcher to set the props ^

    public function once($queueName)
    {
        $this->events->dispatch('queue:before_once', new Event());
        $this->wrapped->once($queueName);
        $this->events->disaptch('queue:after_once', new Event());
    }
}

Callable Handler¶

The simplest handler could just be a callable that invokes the provided callback with the message.

<?php

use PMG\Queue\DefaultConsumer;
use PMG\Queue\Message;
use PMG\Queue\Driver\MemoryDriver;
use PMG\Queue\Handler\CallableHandler;

$handler = new CallableHandler(function (Message $msg) {
    switch ($msg->getName()) {
        case 'SendAlert':
            sendAnAlertSomehow($msg);
            break;
        case 'OtherMessage':
            handleOtherMessageSomehow($msg);
            break;
    }
});

$consumer = new DefaultConsumer(new MemoryDriver(), $handler);

Multiple Handlers with Mapping Handler¶

The above switch statement is a lot of boilerplaint, so PMG provies a mapping handler that looks up callables for a message based on its name. For example, here’s a callable for the send alert message.

<?php

final class SendAlertHandler
{
    private $users;
    private $mailer;

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

    public function __invoke(SendAlert $message)
    {
        $user = $this->users->getByIdentifierOrError($message->getUserId());

        $this->mailer->send(
            \Swift_Message::newInstance()
                ->setTo([$user->getEmail()])
                ->setFrom(['help@example.com'])
                ->setSubject('Hello')
                ->setBody('World')
        );
    }
}

Now pull in the mapping handler with composer require pmg/queue-mapping-handler and we can integrate the callable above with it.

<?php

use PMG\Queue\DefaultConsumer;
use PMG\Queue\Handler\MappingHandler;

$handler = MappingHandler::fromArray([
    'SendAlert' => new SendAlertHandler(/*...*/),
    //'OtherMessage' => new OtherMessageHandler()
    // etc
]);

/** @var PMG\Queue\Driver $driver */
$consumer = new DefaultConsumer($driver, $handler);

Using Tactician to Handle Messages¶

Tactician is a command bus from The PHP League. You can use it to do message handling with the queue.

composer install pmg/queue-tactician

Use the same command bus with each message.

<?php

use League\Tactician\CommandBus;
use PMG\Queue\DefaultConsumer;
use PMG\Queue\Handler\TaticianHandler;

$handler = new TacticianHandler(new CommandBus(/* ... */));

/** @var PMG\Queue\Driver $driver */
$consumer = new DefaultConsumer($driver, $handler);

Alternative, you can create a new command bus to handle each message with CreatingTacticianHandler. This is useful if you’re using forking child processes to handle messages.

<?php

use League\Tactician\CommandBus;
use PMG\Queue\DefaultConsumer;
use PMG\Queue\Handler\CreatingTaticianHandler;

$handler = new TacticianHandler(function () {
    return new CommandBus(/* ... */);
});

/** @var PMG\Queue\Driver $driver */
$consumer = new DefaultConsumer($driver, $handler);

Handling Messages in Separate Processes¶

To handle messages in a forked process use the PcntlForkingHandler decorator.

<?php

use PMG\Queue\Handler\MappingHandler;
use PMG\Queue\Handler\PcntlForkingHandler;

// create an actual handler
$realHandler = MappingHandler::fromArray([
    // ...
]);

// decorate it with the forking handler
$handler = new PcntlForkingHandler($realHandler);

Forking is useful for memory management, but requires some consideration. For instance, database connections might need to be re-opened in the forked process. In such cases, the best bet is to simply create the resources on demand. that’s why the TaticianHandler above takes a factory callable by default.

In cases where a process fails to fork, a PMG\Queue\Exception\CouldNotFork exception will be thrown and the consumer will exit with an unsuccessful status code. Your process manager (supervisord, upstart, systemd, etc) should be configured to restart the consumer when that happens.

Drivers¶

Drivers are the queue backend hidden behind the PMG\Queue\Driver interface. pmg/queue comes with two drivers built in: memory and pheanstalk (beanstalkd).

Drivers have method for enqueuing and dequeueing messages as well as methods for acknowledging a message is complete, retrying a message, or marking a message as failed.

Envelopes¶

Envelopes wrap up messages to allow drivers to add additional metadata. One example of such metadata is a retry count that the consumers may use to determine if a message should be retried. The pheanstalk driver implements its own envelop class so it can track the beanstalkd job identifier for the message.

Drivers are free to do whatever they need to do as long as their envelope implements PMG\Queue\Envelope.

Driver Implementations¶

The core pmg/queue library provides a in memory driver and PMG maintains a driver for beanstalkd that uses the pheanstalk library.

<?php
use PMG\Queue\DefaultConsumer;
use PMG\Queue\Driver\MemoryDriver;

// ...

$driver = new MemoryDriver();

// $handler instanceof PMG\Queue\MessageHandler
$consumer = new DefaultConsumer($driver, $handler);

<?php
use PMG\Queue\Driver\MemoryDriver;

class SomeTest extends \PHPUnit_Framework_TestCase
{
    const TESTQ = 'TestQueue';

    /** @var MemoryDriver $driver */
    private $driver;

    public function testSomething()
    {
        // imagine some stuff happened before this, now we need to verify that

        $envelope = $this->driver->dequeue(self::TESTQ);

        $this->assertNotNull($envelope);
        $msg = $envelope->unwrap();
        $this->assertInstanceOf(SendAlert::class, $msg);
        $this->assertEquals(123, $msg->getUserId());
    }

}

<?php
use Pheanstalk\Pheanstalk;
use PMG\Queue\Driver\PheanstalkDriver;
use PMG\Queue\Driver\Serializer\NativeSerializer;

$driver = new PheanstalkDriver(
    new Pheanstalk('localhost', 11300),
    new NativeSerializer('this is a key used to sign messages')
);

Serializers¶

Persistent drivers require some translation from envelopes and messages to something the persistent backend can store. Similarly, whatever is stored in the queue backend needs to be turned back into a message. Serializers make that happen.

All serializers implements PMG\Queue\Serializer\Serializer and one implementation is provied by default: NativeSerializer.

NativeSerializer uses PHP’s build in serialize and unserialize functions. Serialized envelopes are base64 encoded and signed (via an HMAC) with a key given to NativeSerializer in its constructor. The signature is a way to authenticate the message (make sure it came from a source known to use).

<?php
use PMG\Queue\Serializer\NativeSerializer;

$serializer = new NativeSerializer('this is the key');

// ...

<?php
use PMG\Queue\Serializer\NativeSerializer;
use PMG\Queue\Driver\PheanstalkDriver;

$serializer = new NativeSerializer('YourSecretKeyHere', array_merge([
    // your message classes
    SendAlert::class,
    // ...
], PheanstalkDriver::allowedClasses()));

Implementing Your Own Drivers¶

Persistent drivers are not required to use serializers (or anything else), but if they do PMG\Queue\Driver\AbstractPersistanceDriver provides helpers for the usage of serializers.