PHP code example of illusiard / yii2-rabbitmq

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

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

    

illusiard / yii2-rabbitmq example snippets

return [
    'components' => [
        'rabbitmq' => [
            'class' => illusiard\rabbitmq\components\RabbitMqService::class,
            'host' => '127.0.0.1',
            'port' => 5672,
            'user' => 'guest',
            'password' => 'guest',
        ],
    ],
];

return [
    'modules' => [
        'rabbitmq' => [
            'class' => illusiard\rabbitmq\Module::class,
        ],
    ],
];

'components' => [
    'rabbitmq' => [
        'class' => illusiard\\rabbitmq\\components\\RabbitMqService::class,
        'discovery' => [
            'enabled' => true,
            'paths' => [
                '@app/services/rabbitmq/consumers',
                '@app/services/rabbitmq/publishers',
                '@app/services/rabbitmq/middlewares',
                '@app/services/rabbitmq/handlers', // опционально
            ],
            'cache' => 'cache',
            'cacheTtl' => 300,
        ],
    ],
],

/** @var illusiard\rabbitmq\components\RabbitMqService $rabbit */
$rabbit = Yii::$app->get('rabbitmq');

$rabbit->publish('Hello', 'exchange', 'route.key', [
    'content_type' => 'text/plain',
], [
    'x-trace-id' => 'demo-1',
]);

return [
    'components' => [
        'rabbitmq' => [
            'class' => illusiard\rabbitmq\components\RabbitMqService::class,
            'defaultProfile' => 'default',
            'profiles' => [
                'default' => [
                    'host' => '127.0.0.1',
                    'port' => 5672,
                ],
                'backup' => [
                    'host' => '10.0.0.2',
                    'port' => 5672,
                    'vhost' => '/backup',
                ],
            ],
        ],
    ],
];

$rabbit = Yii::$app->get('rabbitmq');
$rabbit->forProfile('backup')->publish('Hello', 'exchange', 'route.key');

'components' => [
    'rabbitmq' => [
        'class' => illusiard\rabbitmq\components\RabbitMqService::class,
        'host' => 'rabbitmq.local',
        'port' => 5671,
        'ssl' => [
            'cafile' => '@app/certs/ca.pem',
            'local_cert' => '@app/certs/client.pem',
            'local_pk' => '@app/certs/client.key',
            'verify_peer' => true,
            'verify_peer_name' => true,
        ],
    ],
],

'consumeReconnectAttempts' => 3,
'consumeReconnectDelaySeconds' => 1,

'maxMessageBodyBytes' => 1024 * 1024,
'jsonDecodeDepth' => 128,
'messageLimitExceededAction' => 'reject', // reject|retry|stop

'components' => [
    'rabbitmq' => [
        'class' => illusiard\\rabbitmq\\components\\RabbitMqService::class,
        'profile' => app\\services\\rabbitmq\\RabbitMqProfile::class,
    ],
],

namespace app\\services\\rabbitmq;

use illusiard\\rabbitmq\\profile\\OptionsMerger;
use illusiard\\rabbitmq\\profile\\RabbitMqProfileInterface;

class RabbitMqProfile implements RabbitMqProfileInterface
{
    public function getConsumerDefaults(): array
    {
        return [
            'prefetch' => 10,
            'managedRetry' => true,
        ];
    }

    public function getPublisherDefaults(): array
    {
        return [
            'confirm' => true,
            'properties' => [
                'delivery_mode' => 2,
            ],
        ];
    }

    public function getMiddlewareDefaults(): array
    {
        return [
            'consumer' => ['trace', 'metrics'],
            'publisher' => ['trace'],
        ];
    }

    public function mergeOptions(array $defaults, array $overrides): array
    {
        return OptionsMerger::merge($defaults, $overrides);
    }
}

namespace app\\services\\rabbitmq\\consumers;

use illusiard\\rabbitmq\\definitions\\consumer\\ConsumerInterface;

class OrdersConsumer implements ConsumerInterface
{
    public function getQueue(): string
    {
        return 'orders';
    }

    public function getHandler()
    {
        return \\app\\queues\\RabbitMqHandler::class;
    }

    public function getOptions(): array
    {
        return [
            'prefetch' => 1,
            'managedRetry' => true,
            'consumeFailFast' => true,
            'consumeMiddlewares' => [],
        ];
    }
}

namespace app\\services\\rabbitmq\\publishers;

use illusiard\\rabbitmq\\definitions\\publisher\\AbstractPublisher;
use illusiard\\rabbitmq\\middleware\\CorrelationIdMiddleware;

class OrdersPublisher extends AbstractPublisher
{
    public function getExchange(): string
    {
        return 'orders-exchange';
    }

    public function getRoutingKey(): string
    {
        return 'orders.created';
    }

    public function getOptions(): array
    {
        return [
            'type' => 'order.created',
            'properties' => [
                'delivery_mode' => 2,
            ],
            'headers' => [
                'x-source' => 'checkout',
            ],
        ];
    }

    public function getMiddlewares(): array
    {
        return [
            CorrelationIdMiddleware::class,
        ];
    }
}

$rabbit = Yii::$app->get('rabbitmq');
$rabbit->publishById('orders', ['orderId' => 123], [
    'correlationId' => 'corr-1',
    'headers' => [
        'x-trace-id' => 'trace-1',
    ],
]);

return [
    'components' => [
        'rabbitmq' => [
            'class' => illusiard\\rabbitmq\\components\\RabbitMqService::class,
            'topology' => [
                'exchanges' => [
                    [
                        'name' => 'orders-exchange',
                        'type' => 'direct',
                        'durable' => true,
                    ],
                ],
                'queues' => [
                    [
                        'name' => 'orders',
                        'durable' => true,
                        'arguments' => [
                            'x-dead-letter-exchange' => 'orders-retry-exchange',
                            'x-dead-letter-routing-key' => 'orders.retry',
                        ],
                    ],
                    [
                        'name' => 'orders.retry',
                        'durable' => true,
                        'arguments' => [
                            'x-message-ttl' => 60000,
                            'x-dead-letter-exchange' => 'orders-exchange',
                            'x-dead-letter-routing-key' => 'orders',
                        ],
                    ],
                ],
                'bindings' => [
                    [
                        'exchange' => 'orders-exchange',
                        'queue' => 'orders',
                        'routingKey' => 'orders',
                    ],
                ],
            ],
        ],
    ],
];

'components' => [
    'rabbitmq' => [
        'class' => illusiard\rabbitmq\components\RabbitMqService::class,
        'mandatory' => true,
        'returnSink' => illusiard\rabbitmq\amqp\InMemoryReturnSink::class,
        'returnSinkEnabled' => true,
    ],
],

$rabbit = Yii::$app->get('rabbitmq');
$rabbit->publish('payload', 'exchange', 'rk');

// Снять события возвратов и ACK/NACK (если включён confirm).
$rabbit->tick(0.1);
$returns = $rabbit->drainReturns();

'managedRetry' => true,
'retryPolicy' => [
    'maxAttempts' => 3,
    'retryExchange' => 'retry-exchange',
    'retryQueues' => [
        ['name' => 'orders.retry.5s', 'ttlMs' => 5000],
        ['name' => 'orders.retry.30s', 'ttlMs' => 30000],
    ],
    'deadQueue' => 'orders.dead',
],

use illusiard\rabbitmq\message\Envelope;

$env = new Envelope(
    ['orderId' => 123],
    ['x-trace-id' => 'trace-1'],
    [],
    'order.created',
    'corr-1'
);

$rabbit->publishEnvelope($env, 'orders-exchange', 'orders.created');

$rabbit->publishJson(
    ['orderId' => 123],
    'orders-exchange',
    'orders.created',
    [
        'type' => 'order.created',
        'correlationId' => 'corr-1',
        'headers' => ['x-trace-id' => 'trace-1'],
    ]
);

$env = Yii::$app->get('rabbitmq')->decodeEnvelope($meta['body'], $meta);

'publishMiddlewares' => [
    illusiard\\rabbitmq\\middleware\\CorrelationIdMiddleware::class,
    illusiard\\rabbitmq\\middleware\\PublishLoggingMiddleware::class,
],
'consumeMiddlewares' => [
    illusiard\\rabbitmq\\middleware\\ConsumeLoggingMiddleware::class,
],

use illusiard\rabbitmq\message\Envelope;
use illusiard\rabbitmq\rpc\RpcClient;

$client = new RpcClient(Yii::$app->get('rabbitmq'));
$response = $client->call(
    new Envelope(['ping' => true], [], [], 'rpc.ping'),
    'rpc-exchange',
    'rpc.ping',
    5
);

use illusiard\rabbitmq\message\Envelope;
use illusiard\rabbitmq\rpc\RpcServer;

$server = new RpcServer(Yii::$app->get('rabbitmq'));
$server->serve('rpc.queue', function (Envelope $request): Envelope {
    return new Envelope(['pong' => true], [], [], 'rpc.pong', $request->getCorrelationId());
});

services/rabbitmq/consumers/OrdersConsumer.php
services/rabbitmq/publishers/OrdersPublisher.php
services/rabbitmq/middlewares/TraceIdMiddleware.php
services/rabbitmq/handlers/OrdersHandler.php