feat: asynchronous message publishing with batching and pre-commit delivery guarantees (Enterprise)

[dgafka]

Why is this change proposed?

Publishing to a message broker blocks the handler on one full round-trip per message — a
handler emitting many messages pays N sequential broker round-trips, which caps publishing
throughput as systems scale.

Ecotone now provides an Enterprise feature that makes publishing scale, while still keeping
all the delivery guarantees end users rely on: every message is still confirmed by the
broker, a failed delivery still fails the business operation or is routed to the error
channel, and every failure points to the exact message that failed — nothing is ever lost
silently.

This is achieved by firing messages to the broker immediately, batching them natively per
provider, and deferring confirmation awaiting to the last responsible moment — messages
travel to the broker while the handler keeps working, and all confirmations are collected
before the operation completes.

Benchmarks

Publishing 1,000 messages, relative to each provider's previous message-by-message
synchronous publishing (= 100%). Mode of 10 iterations, all delivery confirmations awaited:

Provider	Feature used	Time vs previous	Speed-up
Kafka	Batching + async sending	9%	~11×
SQS	Batching + async sending	9%	~11×
DBAL	Batching	10%	~10×
AMQP	Batching + async sending	32%	~3×
Redis	Batching	36%	~3×
Kafka	Async sending (single messages)	36%	~3×
SQS	Async sending (single messages)	65%	~1.5×

Batching + async sending publishes the workload as provider-native batches with
confirmation awaiting deferred; async sending fires individual messages without waiting.
DBAL and Redis write synchronously, so batching is their applicable feature.

Resulting flow

graph TD
    A[Handler emits messages] --> B[Collector gathers them]
    B -->|one BatchMessage| C[Adapter: provider-native batch publish, no wait]
    C --> D[Registry: PendingDelivery per publish]
    D --> E[Waiter interceptor: await all confirmations]
    E -->|all confirmed| F[Operation completes]
    E -->|some failed| G[Only failed messages to error channel, or operation fails]

Loading

Enabling

ServiceConfiguration::createWithDefaults()
    ->withExtensionObjects([
        // Message Channel: messages published from handlers are gathered,
        // sent as one batch and confirmed before the operation completes
        AmqpBackedMessageChannelBuilder::create('orders')
            ->withAsyncPublishing(),
        KafkaMessageChannelBuilder::create('orders')
            ->withAsyncPublishing(timeoutInMilliseconds: 12000),

        // Message Publisher: enables MessagePublisher::asyncPublish()
        AmqpMessagePublisherConfiguration::create()
            ->withAsyncPublishing(),
    ]);

Example

#[CommandHandler]
public function place(PlaceOrder $command, EventBus $eventBus): void
{
    // events are batched to the broker and their confirmations
    // awaited automatically before the operation completes
    $eventBus->publish(new OrderWasPlaced($command->orderId));
}

// explicit async publishing with a Future
$future = $messagePublisher->asyncPublish(
    BatchMessage::constructEmpty()
        ->append($firstOrder)
        ->append($secondOrder)
);

$future->resolve(); // throws AsyncPublishingFailedException listing only the messages that failed

Pull Request Contribution Terms

• I have read and agree to the contribution terms outlined in CONTRIBUTING.