Message Publisher

Message Publisher for publishing to external message brokers

You're integrating with a third-party service that listens on RabbitMQ — they don't run Ecotone, they have their own wire format. You don't want your business code to know about the AMQP client, the topic name, or the serialization format. Message Publisher is a typed interface that publishes raw messages to an external broker; if you swap brokers later, the call sites don't change.

Modules Providing Support

• RabbitMQ Module

Message Publisher Interface

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.

How to use Message Publisher

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"
        ]
    );
}

High Throughput Publishing (Enterprise)

When the Publisher is configured with withHighThroughputPublishing(), messages published within the same scope are batched into single Broker writes:

#[ServiceContext]
public function messagePublisher()
{
    return AmqpMessagePublisherConfiguration::create()
                ->withDefaultRoutingKey("orders")
                ->withHighThroughputPublishing();
}

On Brokers that can confirm a delivery after the fact - RabbitMQ, Kafka and SQS - the Publisher additionally exposes publishDeferred, which fires the message without blocking on Broker confirmation and returns a Future:

$future = $publisher->publishDeferred($orderData);

$future->resolve(); // awaits delivery confirmation

DBAL and Redis Publishers batch, yet cannot defer confirmation, as their write blocks until the store confirms it. Publish batches there with convertAndSend.

Combined with BatchMessage, this allows publishing thousands of messages in single batched Broker operations. Read more in Non-blocking Batched Delivery.