README

romanfedorskij/message-bus-amphp — адаптер стратегии исполнения для romanfedorskij/message-bus v4 на базе AMPHP Fiber.

Пакет добавляет AmpFiberExecutionStrategy, которая запускает все bindings текущего flow как Amp jobs и затем ожидает их результаты.

Важно: Fiber даёт конкурентность внутри одного PHP-процесса. Это хорошо работает для Amp-compatible IO и операций, которые умеют отдавать управление event loop. Блокирующий CPU/PDO код не станет настоящим параллельным выполнением без отдельного процесса/worker pool.

Установка

composer require romanfedorskij/message-bus-amphp

Использование

<?php

declare(strict_types=1);

use Wolfcharaa\MessageBus\Amp\Execution\AmpFiberExecutionStrategy;
use Wolfcharaa\MessageBus\Flow\FlowDefinition;
use Wolfcharaa\MessageBus\Flow\FlowRegistry;

$flows = new FlowRegistry(
    FlowDefinition::sync('default'),
    FlowDefinition::sync('concurrent')
        ->strategy(AmpFiberExecutionStrategy::class),
);

После этого handlers, привязанные к flow concurrent, будут запускаться как Fiber jobs.

Ошибки

По умолчанию стратегия запускает все jobs, ожидает завершения и выбрасывает первую ошибку. Такой подход не оставляет невычитанные futures.

Если нужно получить полный HandlerExecutionResult с успешными и ошибочными handlers:

new AmpFiberExecutionStrategy(throwOnFailure: false);

Примеры

Тесты

composer test

romanfedorskij / message-bus-amphp

Maintainers

Package info

Statistics

Security