dgafka

<?php

class CommandController 
{
    public function execute(Request $request, CommandBus $commandBus): Response 
    {
        $routingKey = $request->headers->get('X-Routing-Key');
        
        $commandBus->sendWithRouting(
            routingKey: $routingKey,

dgafka

/** RabbitMQ Consumer example */
#[RabbitConsumer(
    endpointId: 'transaction_handler', 
    queueName: 'transactions', 
    finalFailureStrategy: FinalFailureStrategy::STOP
)]
public function processTransactionEvent(TransactionEvent $event): void
{
    // Handle Transaction Event
}

dgafka

/** RabbitMQ Consumer example */
#[InstantRetry(retryTimes: 2, exceptions: [NetworkException::class])]
#[RabbitConsumer(endpointId: 'transaction_handler', queueName: 'transactions')]
public function processTransactionEvent(TransactionEvent $event): void
{
    // handle
}

/** Kafka Consumer example */
#[InstantRetry(retryTimes: 2)] // no exceptions given, meaing retry on all exceptions

dgafka

#[InstantRetry(retryTimes: 2, exceptions: [NetworkException::class])]
#[RabbitConsumer(endpointId: 'transaction_handler', queueName: 'transactions')]
public function processTransactionEvent(TransactionEvent $event): void
{
    // handle
}

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