dgafka

<?php
// Configuration on Publisher Side

public function changeBillingDetails(DistributedBus $distributedCommandBus)
{
    $distributedCommandBus->sendCommand(
        "billing", // destination
        "billing.changeDetails", // routingKey
        '["personId":"123","billingDetails":"01111"]',
        "application/json"

dgafka

<?php

class PlaceOrdernHandler
{
    // default behaviour, first parameter is payload, second are headers
    #[CommandHandler("place_order")]
    public function placeOrder(PlaceOrder $payload, array $headers) {}
}

dgafka

<?php

class MessagingConfiguration
{
    #[ServiceContext] 
    public function asyncChannel()
    {
        // this channel can be reused for different message handlers
        return AmqpBackedMessageChannelBuilder::create("async_channel");
    }

dgafka

<?php

class AddExecutorIdMiddleware
{
    /**
     * @param  mixed  $job
     * @param  callable  $next
     * @return mixed
     */
    public function handle($job, $next)

dgafka

<?php

PlaceOrder::dispatch($order)->onQueue('async_channel');

dgafka

<?php

class PlaceOrder implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable;
  
    protected Order $order;
 
    public function __construct(Order $order)
    {

dgafka

<?php

class AddExecutorIdInterceptor
{
    #[Before(pointcut: CommandHandler::class, changeHeaders: true)] 
    public function addExecutorId(PlaceOrder $payload, array $headers): array
    {
        // changeHeaders: true/false allows for changing headers or payload. 
        // you don't need to pass parameters, if you don't want to use them.
        

dgafka

<?php
$commandBus->sendWithRouting("place_order", '{"items":["milk","coffee"]}', "application/json")

(..)
    
class PlaceOrdernHandler
{
    #[CommandHandler("place_order")]
    public function placeOrder(PlaceOrder $payload) {}
}

dgafka

<?php
class PlaceOrdernHandler
{
    #[CommandHandler]
    public function placeOrder(PlaceOrder $payload) {}
}

dgafka

<?php

$bus->dispatch(new PlaceOrder(), [new ExecutorStamp($executorId)]);

(..)

class AuthorizeExecutorMiddleware implements MiddlewareInterface
{
    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {