PHP code example of gpalyan / laravel-outbox

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

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

    

gpalyan / laravel-outbox example snippets

use TransactionalOutbox\Models\OutboxMessage;

DB::transaction(function () use ($order) {
    $order->save();

    OutboxMessage::store(
        channel: 'orders.created',
        payload: json_encode($order),
        deduplicationKey: "order.created:{$order->id}",
        headers: ['X-Type' => 'OrderCreated'], // optional
    );
});

use TransactionalOutbox\Data\OutboxDraft;

OutboxMessage::storeBatch([
    new OutboxDraft(
        channel: 'orders.created',
        payload: json_encode($order1),
        deduplicationKey: "order.created:{$order1->id}",
    ),
    new OutboxDraft(
        channel: 'orders.updated',
        payload: json_encode($order2),
        deduplicationKey: "order.updated:{$order2->id}:{$order2->version}",
    ),
]);

namespace App\Messaging;

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;
use RuntimeException;
use TransactionalOutbox\Contracts\OutboxPublisherInterface;
use TransactionalOutbox\Models\OutboxMessage;

final class RabbitMqOutboxPublisher implements OutboxPublisherInterface
{
    public function __construct(private AMQPStreamConnection $connection) {}

    public function publish(OutboxMessage $message): void
    {
        $route = config("rabbitmq.routes.{$message->channel}");

        if (! $route) {
            throw new RuntimeException("No RabbitMQ route configured for channel '{$message->channel}'");
        }

        $amqpMessage = new AMQPMessage(
            $message->payload,
            [
                'content_type' => 'application/octet-stream',
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                'message_id' => $message->id,
                'application_headers' => new AMQPTable($message->headers ?? []),
            ],
        );

        $channel = $this->connection->channel();
        $channel->basic_publish(
            msg: $amqpMessage,
            exchange: $route['exchange'],
            routing_key: $route['routing_key'],
        );
        // Any AMQP exception here propagates up; the worker catches it and reschedules with backoff.
    }
}

// config/rabbitmq.php
return [
    'routes' => [
        'orders.created' => ['exchange' => 'orders', 'routing_key' => 'orders.created'],
        'orders.updated' => ['exchange' => 'orders', 'routing_key' => 'orders.updated'],
        'payments.processed' => ['exchange' => 'payments', 'routing_key' => 'payments.processed'],
    ],
];

// AppServiceProvider::register()
$this->app->singleton(AMQPStreamConnection::class, fn () => new AMQPStreamConnection(
    host: config('rabbitmq.host'),
    port: config('rabbitmq.port'),
    user: config('rabbitmq.user'),
    password: config('rabbitmq.password'),
));

$this->app->bind(
    \TransactionalOutbox\Contracts\OutboxPublisherInterface::class,
    \App\Messaging\RabbitMqOutboxPublisher::class,
);

// routes/console.php
use Illuminate\Support\Facades\Schedule;

Schedule::command('transactional-outbox:process outbox')
    ->everyMinute()
    ->withoutOverlapping();

namespace App\Console\Commands;

use Illuminate\Console\Command;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use Throwable;
use TransactionalOutbox\Events\MessageConsumed;

final class RabbitMqListenCommand extends Command
{
    protected $signature = 'app:rabbitmq-listen {queue} {channel}';

    public function handle(AMQPStreamConnection $connection): int
    {
        $amqpChannel = $connection->channel();
        $amqpChannel->basic_qos(prefetch_size: 0, prefetch_count: 10, a_global: false);

        $amqpChannel->basic_consume(
            queue: $this->argument('queue'),
            consumer_tag: '',
            no_local: false,
            no_ack: false,             // <- manual ack; critical for at-least-once
            exclusive: false,
            nowait: false,
            callback: function (AMQPMessage $msg) {
                try {
                    event(new MessageConsumed(
                        channel: $this->argument('channel'),
                        payload: $msg->getBody(),
                        deduplicationKey: $msg->get('message_id'), // broker message id
                        headers: $this->extractHeaders($msg),
                    ));

                    $msg->ack();
                } catch (Throwable $e) {
                    // event() failed (DB unavailable, etc.) — return the message to the broker.
                    $msg->reject(requeue: true);
                    report($e);
                }
            },
        );

        while ($amqpChannel->is_consuming()) {
            $amqpChannel->wait();
        }

        return self::SUCCESS;
    }

    private function extractHeaders(AMQPMessage $msg): array
    {
        $headers = $msg->get_properties()['application_headers'] ?? null;

        return $headers?->getNativeData() ?? [];
    }
}

use TransactionalOutbox\Contracts\InboxHandlerInterface;
use TransactionalOutbox\Models\InboxMessage;

final class PaymentProcessedHandler implements InboxHandlerInterface
{
    public function handle(InboxMessage $message): void
    {
        $data = json_decode($message->payload, true, flags: JSON_THROW_ON_ERROR);
        // process $data... throw on failure — the job will retry automatically
    }
}

// AppServiceProvider::register()
$this->app->bind('payments.processed', PaymentProcessedHandler::class);

// routes/console.php
Schedule::command('transactional-outbox:process inbox')
    ->everyMinute()
    ->withoutOverlapping();

// routes/console.php
use TransactionalOutbox\Models\OutboxMessage;
use TransactionalOutbox\Models\InboxMessage;

Schedule::command('model:prune', ['--model' => [OutboxMessage::class, InboxMessage::class]])
    ->daily();

bash
php artisan vendor:publish --tag=transactional-outbox-config
php artisan vendor:publish --tag=transactional-outbox-migrations
php artisan migrate

php artisan app:rabbitmq-listen orders.queue orders.created