RabbitMQ Support

Asynchronous PHP RabbitMQ
Installation
composer require ecotone/amqp
Module Powered By
​Enqueue solid and powerful abstraction over asynchronous queues.
Configuration
In order to use AMQP Support we need to add ConnectionFactory to our Dependency Container. 
Symfony
Laravel
Lite
# config/services.yaml
# You need to have RabbitMQ instance running on your localhost, or change DSN
    Enqueue\AmqpExt\AmqpConnectionFactory:
        class: Enqueue\AmqpExt\AmqpConnectionFactory
        arguments:
            - "amqp://guest:[email protected]:5672//"
# Register AMQP Service in Provider
​
use Enqueue\AmqpExt\AmqpConnectionFactory;
​
public function register()
{
     $this->app->singleton(AmqpConnectionFactory::class, function () {
         return new AmqpConnectionFactory("amqp+lib://guest:[email protected]:5672//");
     });
}
use Enqueue\AmqpExt\AmqpConnectionFactory;
​
$application = EcotoneLiteApplication::boostrap(
    [
        AmqpConnectionFactory::class => new AmqpConnectionFactory("amqp+lib://guest:[email protected]:5672//")
    ]
);
We register our AmqpConnection under the class name Enqueue\AmqpExt\AmqpConnectionFactory. This will help Ecotone resolve it automatically, without any additional configuration.
Message Channel
To create AMQP Backed Message Channel (RabbitMQ Channel), we need to create Service Context. 
class MessagingConfiguration
{
    #[ServiceContext] 
    public function orderChannel()
    {
        return AmqpBackedMessageChannelBuilder::create("orders");
    }
}
Now orders channel will be available in our Messaging System. 
Distributed Publisher and Consumer
To create distributed publisher or consumer provide Service Context.
Distributed Publisher
class MessagingConfiguration
{
    #[ServiceContext] 
    public function distributedPublisher()
    {
        return AmqpDistributedBusConfiguration::createPublisher();
    }
}
Distributed Consumer
class MessagingConfiguration
{
    #[ServiceContext] 
    public function distributedConsumer()
    {
        return AmqpDistributedBusConfiguration::createConsumer();
    }
}
Custom Message Handling
​Message Channels are way for Ecotone to take over of creating Message Publisher and Consumer for you. All you need to do is to state that given Message Handler should make use of it. 
On other hand for cases where you need to more custom configuration, you may take over and create your own Publisher and Consumer.
Message Publisher
If you want to publish Message directly to Exchange, you may use of Publisher.
class AMQPConfiguration
{
    #[ServiceContext] 
    public function registerAmqpConfig()
    {
        return 
            AmqpMessagePublisherConfiguration::create(
                MessagePublisher::class, // 1
                "delivery", // 2
                "application/json" // 3
            );
    }
}
1.Reference name -  Name under which it will be available in Dependency Container.
2.Exchange name - Name of exchange where Message should be publisher
3.Default Conversion [Optional] - Default type, payload will be converted to.
Publisher is a special type of Gateway, which implements Publisher interface.
It will be available in your Dependency Container under passed Reference name.
In case interface name MessagePublisher:class is used, it will be available using auto-wire.
#[EventHandler] 
public function whenOrderWasPlaced(OrderWasPlaced $event, MessagePublisher $publisher) : void
{
    $publisher->convertAndSendWithMetadata(
        $event,
        [
            "system.executor" => "Johny"
        ]
    );
}
Available actions
interface MessagePublisher
{
​
    // 1
    public function send(string $data, string $sourceMediaType = MediaType::TEXT_PLAIN) : void;
​
    // 2
    public function sendWithMetadata(string $data, array $metadata, string $sourceMediaType = MediaType::TEXT_PLAIN) : void;
​
    // 3
    public function convertAndSend(object|array $data) : void;
​
​
    // 4
    public function convertAndSendWithMetadata(object|array $data, array $metadata) : void;
}
1.send - Send a string type via Publisher. It does not need any conversion, you may add additional Media Type of $data.
2.sendWithMetadata - Does the same as send, allows for sending additional Meta data.
3.convertAndSend - Allow for sending types, which needs conversion. Allow for sending objects and array, Ecotone make use of Conversion system to convert $data.
4.convertAndSendWithMetadata - Does the same as convertAndSend, allow for sending additional Meta data.
Publisher Configuration
AmqpMessagePublisherConfiguration::create(
    MessagePublisher::class,
    "delivery"
)
    ->withDefaultPersistentDelivery(true) // 1
    ->withDefaultRoutingKey("someKey") // 2
    ->withRoutingKeyFromHeader("routingKey") // 3
    ->withHeaderMapper("application.*") // 4
1.withDefaultPersistentDelivery - should AMQP messages be persistent.
2.withDefaultRoutingKey - default routing key added to AMQP message 
3.withRoutingKeyFromHeader - should routing key be retrieved from header with name
4.withHeaderMapper - On default headers are not send with AMQP message. You map provide mapping for headers that should be mapped to AMQP Message
Consumer
Consumer allow use to customize experience to fetch from chosen queue:
class Consumer
{
    #[MessageConsumer("consumer")]
    public function execute(string $message) : void
    {
        // do something with Message
        // if you have converter registered you can type hint exact type you expect
    }
}
To connect consumer directly to a AMQP Queue, we need to provide Ecotone with information, how the Queue is configured. 
class AmqpConfiguration
{
    #[ServiceContext] 
    public function registerAmqpConfig(): array
    {
        return [
            AmqpQueue::createWith("orders"), // 1
            AmqpExchange::createDirectExchange("system"), // 2
            AmqpBinding::createFromNames("system", "orders", "placeOrder"), // 3
            AmqpMessageConsumerConfiguration::create("consumer", "orders") // 4
        ];
    }
}
1.AmqpQueue::createWith(string $name) - Registers Queue with specific name
2.AmqpExchange::create*(string $name) - Registers of given type with specific name
3.AmqpBinding::createFromName(string $exchangeName, string $queueName, string $routingKey)- Registering binding between exchange and queue
4.Provides Consumer that will be registered at given name "consumer" and will be polling "orders" queue
Exchange configuration
$amqpExchange = AmqpExchange::createDirectExchange
$amqpExchange = AmqpExchange::createFanoutExchange
$amqpExchange = AmqpExchange::createTopicExchange
$amqpExchange = AmqpExchange::createHeadersExchange
​
$amqpExchange = $amqpExchange
    ->withDurability(true) // exchanges survive broker restart
    ->withAutoDeletion() // exchange is deleted when last queue is unbound from it
Queue Configuration
$amqpQueue = AmqpQueue::createDirectExchange
                ->withDurability(true) // the queue will survive a broker restart
                ->withExclusivity() // used by only one connection and the queue will be deleted when that connection closes
                ->withAutoDeletion() // queue that has had at least one consumer is deleted when last consumer unsubscribes
                ->withDeadLetterExchangeTarget($amqpExchange);
Publisher Transactions
Ecotone AMQP comes with support for RabbitMQ Transaction for published messages. 
This means that, if you send more than one message at time, it will be commited together.
If you want to enable/disable for all Asynchronous Endpoints or specific for Command Bus. You may use of ServiceContext. 
By default all transactions are enabled
class ChannelConfiguration
{
    #[ServiceContext]
    public function registerTransactions() : array
    {
        return [
            AmqpConfiguration::createWithDefaults()
                ->withTransactionOnAsynchronousEndpoints(false)
                ->withTransactionOnCommandBus(false)
        ];
    }
​
}
To enable transactions on specific endpoint if default is disabled, mark consumer with Ecotone\Amqp\AmqpTransaction\AmqpTransaction annotation.
    #[AmqpTransaction)]
    #[MessageConsumer("consumer")]
    public function execute(string $message) : void
    {
        // do something with Message
    }
Modules - Previous
JMS Converter
Next - Modules
DBAL Support
Last modified 1mo ago