webit/message-bus-sf-process

Message Bus - Symfony Process Infrastructure

Symfony Process infrastructure for Message Bus, _(*1)

Installation

composer require webit/message-bus-sf-process=^1.0.0

Usage

ProcessFactory

To use both ProcessPublisher or ProcessConsumer instance of ProcessFactory is needed., _(*2)

use Webit\MessageBus\Message;
use Symfony\Component\Process\Process;

class MyProcessFactory implements ProcessFactory
{
    /**
     * @inheritdoc
     */
    public function create(Message $message)
    {
        return new Process(
            sprintf(
                '/usr/local/my-binary.php %s %s',
                escapeshellarg($message->type()),
                escapeshellarg($message->content())
            )
        );
    }
}

Synchronous ProcessLauncher

To run process synchronously use SynchronousProcessLauncher, _(*3)

use Webit\MessageBus\Infrastructure\Symfony\Process\Launcher\SynchronousProcessLauncher;

$myFactory = new MyProcessFactory();
$launcher = new SynchronousProcessLauncher($myFactory);

Asynchronous ProcessLauncher

To run process asynchronously use AsynchronousProcessLauncher, _(*4)

use Webit\MessageBus\Infrastructure\Symfony\Process\Launcher\ParallelProcessManager;
use Webit\MessageBus\Infrastructure\Symfony\Process\Launcher\AsynchronousProcessLauncher;

$myFactory = new MyProcessFactory();
$launcher = new AsynchronousProcessLauncher(
    $myFactory,
    new ParallelProcessManager($maxParallelProcessNumber = 5) // to run at most 5 parallel processes
);

Publisher integration

Configured ProcessLauncher can be used with ProcessPublisher, _(*5)

use Webit\MessageBus\Infrastructure\Symfony\Process\ProcessPublisher;
use Webit\MessageBus\Message;

$publisher = new ProcessPublisher($launcher);
$publisher->publish(new Message('type', 'content'));

Consumer integration

Configured ProcessLauncher can be used with ProcessConsumer, _(*6)

use Webit\MessageBus\Infrastructure\Symfony\Process\ProcessConsumer;
use Webit\MessageBus\Message;

$consumer = new ProcessConsumer($launcher);
$consumer->consume(new Message('type', 'content'));

Running tests

Install dependencies with composer, _(*7)

docker-compose run --rm composer
docker-compose run --rm spec