ZeroMQ message dispatcher for ProophServiceBus

, _(*1)

Use ZeroMQ as message producer for Prooph Service Bus. Works together with bus types: CommandBus, EventBus., _(*2)

Important

This library will receive support until December 31, 2019 and will then be deprecated., _(*3)

For further information see the official announcement here: https://www.sasaprolic.com/2018/08/the-future-of-prooph-components.html, _(*4)

Requirements

PHP doesn't come with native support for ZeroMQ however there is an extension ext-zmq instructions are available on the ZMQ website for the PHP bindings., _(*5)

http://zeromq.org/bindings:php, _(*6)

Installation

So after ext-zmq is installed on your server/development machine you're ready for the next step! Composer will be able to install prooph zeromq producer in seconds if not quicker. Run the following command to install via composer., _(*7)

composer require prooph/psb-zeromq-producer:~0.2, _(*8)

Command/Event Bus (PUB/SUB)

To construct your Command/Event bus you'll need to have a server running ZMQ with ZMQ::SOCKET_SUB this will then receive the messages from the producer., _(*9)

For basic tutorial on PUB/SUB: http://zguide.zeromq.org/page:all#Getting-the-Message-Out, _(*10)

Usage Examples

$container = new Container;
$container['config'] = [
    'prooph' => [
        'zeromq_producer' => [
            'dsn' => 'tcp://127.0.0.1:5555', // ZMQ Server Address.
            'persistent_id' => 'example', // ZMQ Persistent ID to keep connections alive between requests.
            'rpc' => false, // Use as Query Bus.
        ]
    ]
];

$factory = Prooph\ServiceBus\Message\ZeroMQ\Container\ZeroMQMessageProducerFactory;
$zmqProducer = $factory($container);

// Setup complete, now to add it to the prooph service bus.

$commandBus = new Prooph\ServiceBus\CommandBus();
$router = new Prooph\ServiceBus\Plugin\Router\CommandRouter();
$router->route('ExampleCommand')
    ->to($zmqProducer);

$commandBus->utilize($router);
$echoText = new ExampleCommand('It works');
$commandBus->dispatch($echoText);

// Now check your server to make sure it received this command.

Query Bus (REQ/REP)

To construct your Query bus you'll need to have a ZMQ server running with ZMQ::SOCKET_REP this will then receive the messages from the producer and MUST reply as part of the REQ/REP specification., _(*11)

For basic tutorial on REQ/REP: http://zguide.zeromq.org/page:all#Ask-and-Ye-Shall-Receive, _(*12)

Usage Examples

// file: CLIENT.php

$container = new Container;
$container['config'] = [
    'prooph' => [
        'zeromq_producer' => [
            'dsn' => 'tcp://127.0.0.1:5556', // ZMQ Server Address.
            'persistent_id' => 'example', // ZMQ Persistent ID to keep connections alive between requests.
            'rpc' => true, // Use as Query Bus.
        ]
    ]
];

$factory = Prooph\ServiceBus\Message\ZeroMQ\Container\ZeroMQMessageProducerFactory;
$zmqProducer = $factory($container);

// Setup complete, now to add it to the prooph service bus.

$queryBus = new Prooph\ServiceBus\QueryBus();
$router = new Prooph\ServiceBus\Plugin\Router\QueryRouter();
$router->route('ExampleQuery')
    ->to($zmqProducer);

$queryBus->utilize($router);
$getText = new ExampleQuery('Hello Server.');
$promise = $queryBus->dispatch($getText);

$promise->then(function ($response) {
    var_dump($response); // string "Hello Client."
});

exit(0);

// file: SERVER.php
<?php

$context = new ZMQContext;
$socket = new ZMQSocket($context, ZMQ::SOCKET_REP);
$socket->bind('tcp://127.0.0.1:5556');

echo "ZMQ Stub Server Started.";

while ($message = $socket->recv()) {
    if ('Hello Server.' === $message) {
        $socket->send('Hello Client.');
    }
}

Support

Ask questions on Stack Overflow tagged with #prooph.
File issues at https://github.com/prooph/psb-zeromq-producer/issues.
Say hello in the prooph gitter chat.

Contribute

Please feel free to fork and extend existing or add new features and send a pull request with your changes! To establish a consistent code quality, please provide unit tests for all your changes and may adapt the documentation., _(*13)