README

消息总线Symfony进程基础设施

安装

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

使用方法

进程工厂

要使用ProcessPublisher或ProcessConsumer实例，需要ProcessFactory。

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())
            )
        );
    }
}

同步进程启动器

要同步运行进程，请使用SynchronousProcessLauncher

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

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

异步进程启动器

要异步运行进程，请使用AsynchronousProcessLauncher

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
);

发布者集成

配置的ProcessLauncher可以与ProcessPublisher一起使用

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

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

消费者集成

配置的ProcessLauncher可以与ProcessConsumer一起使用

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

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

运行测试

使用composer安装依赖项

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