Asynchronous Message Handlers

Making message handlers asynchronous with a single attribute

Running Asynchronously

You have a synchronous SendWelcomeEmail handler that's slowing your signup endpoint to 800ms. You want to push it onto a queue without rewriting the handler, the controller, or the test. The #[Asynchronous] attribute does exactly that — same handler code, async execution.

In order to run a Command Handler asynchronously, mark it as Asynchronous.

#[Asynchronous("orders")]
#[CommandHandler(endpointId: "place_order_endpoint")]
public function placeOrder(PlaceOrderCommand $command) : void
{
   // do something with $command
}

The same way we define for Event Handlers:

#[Asynchronous("orders")]
#[EventHandler(endpointId: "order_was_placed")]
public function when(OrderWasPlaced $event) : void
{
   // do something with $event
}

We need to add endpointId on our endpoint's annotation, this will be used to route the Message in isolation to our Message Handlers.

Message Channel

The asynchronous attribute states what Channel reference we want to use:

#[Asynchronous("orders")]

The "orders" string is the name of our Message Channel. We use this name to reference which implementation we want to use—whether it's an in-memory channel for testing, a database queue, or RabbitMQ. This naming approach keeps our business code clean and independent from infrastructure choices.

To configure a specific implementation like a database channel, we use a ServiceContext class.

final readonly class EcotoneConfiguration
{
    #[ServiceContext]
    public function databaseChannel()
    {
        return DbalBackedMessageChannelBuilder::create('orders');
    }
}

That's all the configuration we need! Now whenever we reference "orders" in our handler attributes, Ecotone automatically uses this database channel. Our handlers stay exactly the same whether we're using in-memory channels for testing or database channels for production—the only difference is this single configuration change.

Running Message Consumer

We can first list all of the Message Consumers we have available for running:

Symfony

bin/console ecotone:list
+--------------------+
| Endpoint Names     |
+--------------------+
| orders             |
+--------------------+

Laravel

Lite

Then in order to run our Message Consumer, we will use ecotone:run console command:

Symfony

bin/console ecotone:run orders -vvv

Laravel

Lite

Dynamic Configuration

You may set up running configuration for given consumer while running it.

• handledMessageLimit - Amount of messages to be handled before stopping consumer
• executionTimeLimit - How long consumer should run before stopping (milliseconds)
• finishWhenNoMessages - Consumers will be running as long as there will be messages to consume
• memoryLimit - How much memory can be consumed by before stopping consumer (Megabytes)
• stopOnFailure - Stop consumer in case of exception

Symfony

bin/console ecotone:run orders 
    --handledMessageLimit=5 
    --executionTimeLimit=1000 
    --finishWhenNoMessages
    --memoryLimit=512
    --stopOnFailure

Laravel

Lite

Static Configuration

Using Service Context configuration for statically configuration.

class Configuration
{    
    #[ServiceContext]
    public function configuration() : array
    {
        return [
            PollingMetadata::create("orders")
                 ->setErrorChannelName("errorChannel")
                 ->setInitialDelayInMilliseconds(100)
                 ->setMemoryLimitInMegaBytes(100)
                 ->setHandledMessageLimit(10)
                 ->setExecutionTimeLimitInMilliseconds(100)
                 ->withFinishWhenNoMessages(true)
        ];
    }
}

Dynamic configuration overrides static

Available Providers (Types)

There are multiple different implementation which we can use:

• AMQP Support (RabbitMQ)
• Kafka Support
• DBAL Support
• SQS Support
• Redis Support
• Symfony Messenger Transport Support
• Laravel Queues
• In Memory Channels

Multiple Asynchronous Endpoints

Using single asynchronous channel we may register multiple endpoints. This allow for registering single asynchronous channel for whole Aggregate or group of related Command/Event Handlers.

#[Asynchronous("orders")]
#[EventHandler]
public function onSuccess(SuccessEvent $event) : void
{
}

#[Asynchronous("orders")]
#[EventHandler]
public function onSuccess(FailureEvent $event) : void
{
}

Asynchronous Class

You may put Asynchronous on the class, level so all the endpoints within a class will becomes asynchronous.

Intercepting asynchronous endpoint

All asynchronous endpoints are marked with special attributeEcotone\Messaging\Attribute\AsynchronousRunningEndpoint If you want to intercept all polling endpoints you should make use of annotation related point cut on this.

Asynchronous Execution Attributes (Enterprise)

This is Enterprise feature. To use it, email us at "[e-mail]" to receive trial key. Production license keys are available at https://ecotone.tech.

When database transactions are globally enabled for a message channel, all async handlers on that channel are wrapped in a transaction. However, some handlers may not need a transaction — for example, a handler that only calls a 3rd party API, sends an email, or triggers a webhook. Wrapping such handlers in an unnecessary database transaction wastes resources and holds connections open longer than needed.

With asynchronousExecution attributes on the #[Asynchronous] attribute, you can configure runtime behavior per-handler — applied when the polling consumer processes the Message, not at the synchronous bus call. Attributes must implement the AsynchronousEndpointAttribute interface.

use Ecotone\Messaging\Attribute\Asynchronous;
use Ecotone\Messaging\Attribute\WithoutDatabaseTransaction;

#[Asynchronous("orders", asynchronousExecution: [new WithoutDatabaseTransaction()])]
#[CommandHandler(endpointId: "send_order_confirmation")]
public function sendConfirmation(SendOrderConfirmation $command, EmailService $emailService): void
{
    // This handler only sends an email — no database work needed.
    // The global transaction is skipped for this specific handler,
    // while other handlers on the "orders" channel still use transactions.
    $emailService->send($command->email, $command->orderId);
}

The attributes are resolved at runtime when the message is consumed, and are available to interceptors targeting AsynchronousRunningEndpoint.

Built-in Asynchronous Execution Attributes

Attribute	Effect
WithoutDatabaseTransaction	Skips the global DBAL transaction interceptor for this handler
WithoutMessageCollector	Skips message collection — events are sent directly to channels during handler execution instead of being buffered and released after completion
ErrorChannel	Routes failures from this handler to a per-handler Error Channel — see Per-Handler Error Channel
DelayedRetry	Retries the handler with the configured delay/backoff, then routes to a dead letter channel on exhaustion — see Per-Handler Delayed Retry. Mutually exclusive with ErrorChannel.

Combining Asynchronous Execution Attributes

Multiple attributes compose on a single handler. A common combination is WithoutDatabaseTransaction together with ErrorChannel: the handler doesn't need a DB transaction because it only calls a 3rd party API, but you still want failures captured to a dedicated error channel for retry or review.

use Ecotone\Messaging\Attribute\Asynchronous;
use Ecotone\Messaging\Attribute\ErrorChannel;
use Ecotone\Messaging\Attribute\WithoutDatabaseTransaction;

#[Asynchronous(
    "orders",
    asynchronousExecution: [
        new WithoutDatabaseTransaction(),
        new ErrorChannel("emailDeliveryErrors"),
    ]
)]
#[CommandHandler(endpointId: "send_order_confirmation")]
public function sendConfirmation(SendOrderConfirmation $command, EmailService $emailService): void
{
    // No DB transaction is opened around this handler — the global transaction interceptor is skipped.
    // If the email API throws, the failed Message is captured to "emailDeliveryErrors" instead of
    // propagating up the polling consumer; other handlers on the "orders" channel keep their defaults.
    $emailService->send($command->email, $command->orderId);
}

The attributes are independent — each one is read by the interceptor it concerns (WithoutDatabaseTransaction by the DBAL transaction interceptor, ErrorChannel by the polling consumer's error interceptor). Order in the array does not matter.

Custom Asynchronous Execution Attributes

You can create your own attributes and inject them into interceptors:

use Ecotone\Messaging\Attribute\AsynchronousEndpointAttribute;

#[Attribute]
class CustomDelayedRetry implements AsynchronousEndpointAttribute
{
    public function __construct(public int $maxRetries = 3) {}
}

#[Asynchronous("orders", asynchronousExecution: [new CustomDelayedRetry(maxRetries: 5)])]
#[CommandHandler(endpointId: "place_order_endpoint")]
public function placeOrder(PlaceOrderCommand $command): void {}

Then access it in an interceptor:

#[Around(pointcut: AsynchronousRunningEndpoint::class)]
public function retry(
    MethodInvocation $invocation,
    ?CustomDelayedRetry $policy = null
): mixed {
    // $policy is injected from the handler's asynchronousExecution attributes
    return $invocation->proceed();
}

Endpoint Id

Each Asynchronous Message Handler requires us to define "endpointId". It's unique identifier of your Message Handler.

#[Asynchronous("orders")]
#[EventHandler(endpointId: "order_was_placed")] // Your important endpoint Id
public function when(OrderWasPlaced $event) : void {}

The Endpoint ID travels with your message as part of the headers to your message channel. Once we consume the message from the Message Channel, Ecotone uses this ID to route it to the correct Message Handler. This completely decouples our messages from specific handler classes and methods—we can refactor, rename, or move our handlers around without breaking message routing.