puzzle/amqp

AMQP implementation based on PECL extension

8.3 2024-12-23 16:03 UTC

Requires

• php: >=8.3
• puzzle/configuration: ~8.3
• puzzle/pieces: ~3.0
• ramsey/uuid: ^4.0
• swarrot/swarrot: ^4.0
• symfony/console: ^5.4|^6.0
• symfony/event-dispatcher: ~5.4|^6.0

Requires (Dev)

• behat/behat: ~3.3
• doctrine/collections: <=1.4.0
• groall/rabbitmq-http-api-client-php: ^0.2.4
• knplabs/gaufrette: ~0.2
• odolbeau/rabbit-mq-admin-toolkit: ~5.2.2
• phpunit/phpunit: ~11.0
• pimple/pimple: ~3.0
• puzzle/assert: ~1.1
• symfony/debug: ~3.2

Suggests

• ext-amqp: PECL AMQP extension is required
• ext-zip: PECL ZLib extension is required for GZip processor

MIT

• Nicolas Le Nardou

• AMQP
• asynchronous

README

PHP 5.6 & 7.x users, please use < 5.x versions.

QA

Configuration

# amqp.yml
broker:
    host: myRabbit
    port: 5672
    login: guest
    password: guest
    vhost: /
global:
    disallowSilentDropping: false

# app.yml
id: myApp

Usage

Sending a message

<?php

require '../vendor/autoload.php';

use Puzzle\Configuration\Memory;
use Puzzle\AMQP\Clients\Pecl;
use Puzzle\AMQP\Messages\Message;

$configuration = new Memory(array(
    'amqp/broker/host' => 'myRabbit',
    'amqp/broker/login' => 'guest',
    'amqp/broker/password' => 'guest',
    'amqp/broker/vhost' => '/',
    'app/id' => 'myApp',
));

$client = new Pecl($configuration);

$message = new Message('my.routing.key');
$message->setJson([
    'key' => 'value',
    'key2' => 'value2',
]);

$client->publish('myExchange', $message);

Consuming a message

Worker declaration :

<?php

use Puzzle\AMQP\Consumers;
use Puzzle\AMQP\Clients;
use Puzzle\AMQP\Workers\ProcessorInterfaceAdapter;
use Puzzle\Configuration\Memory;

$configuration = new Memory(array(
    'amqp/broker/host' => 'rabbitmq',
    'amqp/broker/login' => 'guest',
    'amqp/broker/password' => 'guest',
    'amqp/broker/vhost' => '/',
    'app/id' => 'myApp',
));

$consumer = new Consumers\Simple();

$worker = new ExampleWorker();

$consumer->consume(
    new ProcessorInterfaceAdapter($worker),
    new Clients\Pecl($configuration),
    'queue.name'
);

Worker example :

<?php

use Psr\Log\LoggerAwareTrait;
use Psr\Log\NullLogger;
use Puzzle\AMQP\ReadableMessage;
use Puzzle\AMQP\Workers\Worker;

class ExampleWorker implements Worker
{
    use LoggerAwareTrait;

    public function __construct()
    {
        $this->logger = new NullLogger();
    }

    public function process(ReadableMessage $message): void
    {
        // your code here
    }
}

BC Breaks changelog

4.x -> 5.x

• Drop support for php 5.6 & 7.x
• Worker now catch Throwable, not only Exceptions
• Consumer / Worker / WorkerContext signature change

3.x -> 4.x

• Chunk management introduced in 3.1 has been refactored and made easier : just use Streamed* bodies and same client as usual

2.x -> 3.x

• Message hooks has been removed
• Raw & Json implementations of WritableMessage has been replaced by Message + implementations of Body (Text, Json, Binary)
• Message interface has been renamed into MessageMetadata
• Some specific features removed from WorkerContext
• Getter for specific headers removed from MessageAdapter
• Message InMemory implementation (for unit testing purpose) has been changed
• MessageAdapter must not be directly constructed anymore, use MessageAdapterFactory instead

1.x -> 2.x

• Drop support for Silex 1.x (in favor of Silex 2.x)