PHP code example of kraz / messenger-workflow

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

kraz / messenger-workflow example snippets

// config/bundles.php
Kraz\MessengerWorkflow\MessengerWorkflowBundle::class => ['all' => true],

final class BookController
{
    public function __construct(
        private CommandBusInterface $commandBus,
        private QueryBusInterface $queryBus,
    ) {}

    public function register(): Response
    {
        // fire-and-forget
        $this->commandBus->dispatch(new RegisterBook('978-3-16'));

        // tracked: $taskId (UUID v7) is assigned; poll it or await it
        $this->commandBus->dispatch(new RegisterBook('978-3-16'), $taskId);
        $this->commandBus->await($taskId, timeout: 30);   // void; throws TaskFailedException / TaskTimeOutException

        // queries
        $result = $this->queryBus->ask(new GetBook('978-3-16'));       // blocking round-trip
        $taskId = $this->queryBus->askAsync(new GetBook('978-3-16'));  // ...or split
        $result = $this->queryBus->await($taskId, timeout: 10);
    }
}

#[AsCommandHandler]
final class RegisterBookHandler
{
    public function __invoke(RegisterBook $command): string   // return value = task result (tracked commands)
    {
        // runs inside the inbox transaction — writes on the context's connection commit atomically
        return $bookId;
    }
}

#[AsEventHandler(fromTransport: 'book_store_events')]         // scope to this context's queue
final class BookRegisteredHandler
{
    public function __invoke(BookRegistered $event): void { /* ... */ }
}

#[AsController]
final class RegisterBookController
{
    #[Route(path: '/register', methods: ['POST'])]
    public function registerBook(#[MapRequestPayload] RegisterBookRequest $request, CommandBusInterface $commandBus): void
    {
        $commandBus->dispatch(RegisterBookCommand::fromRequest($request));
    }

    #[AsCommandHandler]
    public function registerBookHandler(RegisterBookCommand $command, BookRepositoryInterface $books): void
    {
        $books->add(Book::register($command));
    }
}

$this->outboxBus->publish($event);   // OutboxBusInterface, from messenger_workflow.messenger.outbox_buses
$this->eventBus->publish($event);    // EventBusInterface — straight to the broker

$status = $this->taskStatusProvider->getStatus($taskId);   // TaskStatusResponse: pending|completed|failed
$value = $this->taskResultProvider->getResult($taskId);    // completed command/query result value

#[AsCommandHandler]
final readonly class CompleteAuditCommandHandler
{
    public function __invoke(CompleteAuditCommand $command): void
    {
        $audit = $this->audits->get($command->auditId);

        $this->reconciliation->complete($audit, $command->completedBy, $this->clock->now());
        // No flush, no save: the aggregates are managed, and the boundary writes them.
    }
}