Stop wrestling with complex infrastructure and endless boilerplate. Ecotone brings simplicity back to PHP development by letting you focus on what matters: your business logic.

• Write code that reads like your requirements "When a user registers, send a welcome email." That's not pseudocode—it's how you'll actually write it. Your code becomes self-documenting.

• Start simple, scale when ready Begin with straightforward, synchronous code. Need async processing later? Add a single attribute. No architecture overhaul required.

• Resilience built-in Reliable Workflows, automatic retries, error handling, and self-healing capabilities come standard. Fewer production fires, more peaceful nights.

• Framework agnostic You will never be forced to extend or implement any framework related classes. Ecotone seamless integrate with your application using declarative configuration using attributes. And this works beautifully with Symfony, Laravel, or any other framework using Ecotone Lite.

Your app, your way

Whether you're building a fresh startup idea or modernizing legacy code, Ecotone grows with you. From simple CRUD to event-sourced microservices, the same familiar patterns guide your way.

Read more in

How to get started

Check how to Ecotone for Symfony, Laravel or Lite.

How to use

If you're looking for a way to start and get familiar with Ecotone. Then Ecotone provides different ways to do so:

• - Tutorial will introduce you to Ecotone's fundamentals and will help you build understanding of the Messaging concepts.

Prerequisites

Before installing Ecotone, ensure you have:

• PHP 8.1 or higher

Demo

Read Blog Post

Read more about Scheduling in PHP and Ecotone

Demo

Read Blog Post

Building Blocks: Aggregates, Sagas, Event Sourcing

Demo

Read Blog Post

In this chapter we will cover process of handling and dispatching Messages with Ecotone. We will discuss topics like Commands, Events and Queries, Message Handlers, Message Buses, Aggregates and Sagas. You may be interested in theory - chapter first.

Materials

Message Bus Demo

Publishing Events Demo

Demo

Read Blog Post

Demo Message Bus

Demo Publishing Events

Ecotone comes with in-built integration for Event Sourcing which works with databases like Postgres, MySQL or MariaDB for storing Events and allows for building and storing Projected Views in any storage we decide. Read more in following chapters.

Materials

Demo

Read Blog Post

Demo

Read Blog Post

Sagas are part of the Ecotone's Workflow support. To read more refer to .

Demo

Read Blog Post

Read Aggregate Introduction sections first to get more details about Aggregates.

Aggregate Query Action

Aggregate actions are defined using public method (non-static). Ecotone will ensure loading in order to execute the query method.

#[Aggregate]
class Ticket
{
    #[Identifier]
    private Uuid $ticketId;
    private string $assignedTo;
       
    #[QueryHandler("ticket.get_assigned_person")]
    public function getAssignedTo(): string
    {
       return $this->assignedTo;
    }
}

And then we call it from Query Bus:

$this->commandBus->sendWithRouting(
    "ticket.get_assigned_person",
    // We provide instance of Ticket aggregate using metadata 
    metadata: ["aggregate.id" => $ticketId]
)

Use Event Store directly

In some cases you may want to access Event Store directly. Event Store is auto registered in your Dependency Container, so you can fetch it like any other service or inject it directly to any Handler.

use Ecotone\EventSourcing\EventStore;
    
    #[QueryHandler(self::GET_CURRENT_BALANCE_QUERY)]
    public function getCurrentBalance(#[Reference] EventStore $eventStore): array
    {
        $streamName = "wallet";
        if (!$eventStore->hasStream($streamName)) {
            return [];
        }

        /** @var Event[] $event */
        $events = $eventStore->load($streamName, count: 10);

        return $events;
    }

Ecotone Plans

Ecotone comes with two plans:

• Ecotone Free comes with Apache License Version 2.0 for open features. It allows to build message-driven system in PHP, which solves resiliency and scalability at the architecture level. This covers all the features, which are not marked as Enterprise. \

• Ecotone Enterprise is based Enterprise licence. It does provides more advanced set of features aim for Enterprise usage. It does bring to the table custom features, additional integrations, and ability to optimization resource usages.

Available Enterprise Features

• - Enables integration with Kafka (Event Streaming Platform) to send, receive from Messages from topics, and to use Kafka in form of Message Channel abstraction for seamless integration into the System.

• - Provides ability to set up whole consumption process with single Attribute, and to extend it with resiliency patterns like: instant-retry, dead letter or final failure strategy.

• - Provides ability to simplify deployment strategy, adjusting asynchronous processing to business scenarios, and configure processing per Client dynamically (which is especially useful in Multi-Tenant and SAAS environments).

Materials

Links

• [Article]

Read Aggregate Introduction sections first to get more details about Aggregates.

Typicial Aggregate Flow

Repositories are used for retrieving and saving the aggregate to persistent storage. Typical flow for calling aggregate method would looks like below:

class AssignWorkerHandler
{
    private TicketRepository $ticketRepository;

    #[CommandHandler]
    public function handle(AssignWorkerCommand $command) : void
    {
       // fetch the aggregate from repository
       $ticket = $this->ticketRepository->findBy($command->getTicketId());
       // call action method
       $ticket->assignWorker($command);
       // store the aggregate in repository
       $this->ticketRepository->save($ticket);    
    }
}

$this->commandBus->send(
   new AssignWorkerCommand(
      $ticketId, $workerId,            
   )
);

By setting up Repository we provide Ecotone with functionality to fetch and store the Aggregate , so we don't need to write the above orchestration code anymore.

Ecotone's Aggregate Flow

If our class is defined as Aggregate, Ecotone will use Repository in order fetch and store it, whenever the Command is sent via Command Bus.

Now when we will send the Command, Ecotone will use ticketId from the Command to fetch related Ticket Aggregate, and will called assignWorker passing the Command. After this is completed it will use the repository to store changed Aggregate instance.

Therefore from high level nothing changes:

This way we don't need to write orchestration level code ourselves.

Get started with Ecotone

The best way to get started with Ecotone is to actually build something realistic. Therefore we will build a small back-end for Shopping System during this tutorial. The techniques we will learn in the tutorial are fundamental to building any application using Ecotone.

Lessons

The tutorial is divided into several lessons:

• , we will learn the fundamentals of Ecotone: Endpoints, Messages, Channels, and Command Query Responsibility Segregation (CQRS)

• , we will learn Tactical Domain Driven Design (DDD): Aggregates, Repositories and also Event Handlers

• , we will learn how to use Converters, therefore how to handle serialization and deserialization

As mentioned earlier, Events are stored in form of a Event Stream. Event Stream is audit of Events, which happened in the past. However to protect our business invariants, we may want to work with current state of the Aggregate to know, if given action is possible or not (business invariants).

Business Invariants

Business Invariants in short are our simple "if" statements inside the Command Handler in the Aggregate. Those protect our Aggregate from moving into incorrect state. With State-Stored Aggregates, we always have current state of the Aggregate, therefore we can check the invariants right away. With Event-Sourcing Aggregates, we store them in form of an Events, therefore we need to rebuild our Aggregate, in order to protect the invariants.

Suppose we have Ticket Event Sourcing Aggregate.

#[EventSourcingAggregate]
class Ticket
{
    use WithAggregateVersioning;

    #[Identifier]
    private string $ticketId;

    (...)

    #[CommandHandler]
    public function assign(AssignPerson $command) : array
    {
        return [new PersonWasAssigned($this->ticketId, $command->personId)];
    }
}

For this Ticket we do allow for assigning an Person to handle the Ticket. Let's suppose however, that Business asked us to allow only one Person to be assigned to the Ticket at time. With current code we could assign unlimited people to the Ticket, therefore we need to protect this invariant.

To check if whatever Ticket was already assigned to a Person, our Aggregate need to have state applied which will tell him whatever the Ticket was already assigned. To do this we use EventSourcingHandler attribute passing as first argument given Event class. This method will be called on reconstruction of this Aggregate. So when this Aggregate will be loaded, if given Event was recorded in the Event Stream, method will be called:

Then this state, can be used in the Command Handler to decide whatever we can trigger an action or not:

Read previous section to find out more about Interceptors.

Intercepting Asynchronous Endpoints

We may aswell intercept Asynchronous Endpoints pretty easily. We do it by using pointing to AsynchronousRunningEndpoint class.

class TransactionInterceptor
{
    #[Around(pointcut: AsynchronousRunningEndpoint::class)]
    public function transactional(MethodInvocation $methodInvocation)
    {
        $this->connection->beginTransaction();
        try {
            $result = $methodInvocation->proceed();

            $this->connection->commit();
        }catch (\Throwable $exception) {
            $this->connection->rollBack();

            throw $exception;
        }

        return $result;
    }
}

Inject Message's payload

As part of around intercepting, if we need Message Payload to make the decision we can simply inject that into our interceptor:

Inject Message Headers

We can also inject Message Headers into our interceptor. We could for example inject Message Consumer name in order to decide whatever to start the transaction or not:

Ecotone comes with full automation for setting up Event Sourcing for us. This we can we really easily roll out new features with Event Sourcing with just minimal or none setup at all.

Install Event Sourcing Support

Before we will start, let's first install Event Sourcing module, which will provide us with all required components:

composer require ecotone/pdo-event-sourcing

We need to configure DBAL Support in order to make use of it.

Ecotone PDO Event Sourcing does provide support for three databases:

• PostgreSQL

• MySQL

• MariaDB

Install Inbuilt Serialization Support

Ecotone provides inbuilt functionality to serialize your Events, which can be customized in case of need. This makes Ecotone take care of Event Serialization/Deserialization, and allows us to focus on the business side of the code.

We can take over this process and set up our own , however can fully do it for us, so we can simply focus on the business side of the code. To make it happen all we need to do, is to install JMS Package and we are ready to go:

Ecotone comes with inbuilt repositories, so we don't need to configure Repositories ourselves. It often happen that those are similar between projects, therefore it may be that there is no need to roll out your own.

Inbuilt Repositories

Ecotone provides inbuilt repositories to get you started quicker. This way you can enable given repository and start implementing higher level code without worrying about infrastructure part.

Doctrine ORM Support

This provides integration with . To enable it read more in .

Laravel Eloquent Support

This provides integration with . Eloquent support is available out of the box after installing .

Document Store Repository

This provides integration using relational databases. It will serialize your aggregate to json and deserialize on load using . To enable it read in .

Event Sourcing Repository

Ecotone provides inbuilt Event Sourcing Repository, which will set up Event Store and Event Streams. To enable it read .

Concurrency exceptions when multiple processes or threads access and modify shared resources simultaneously. These exceptions happen because two or more operations conflict try to change same piece of data. Ecotone provides built-in support for concurrency handling.

In order to solve concurrent access, Ecotone implements Optimistic Locking.

Optimistic Locking

Each Event Sourcing Aggregate or Event Sourcing Saga has a version property that represents the current version of the resource. When modifications are made, the version is incremented. If two concurrent processes attempt to modify the same resource with different versions, a concurrency exception is raised. This is default behaviour, if we are using inbuilt Event Sourcing support.

Custom Repositories

In case of Custom Repositories, we may use Ecotone support for optimistic locking to raise the exception in the Repository.

Version will be passed to the repository, based on #[AggregateVersion] property inside the Aggregate/Saga.

We don't need to deal with increasing those on each action. Ecotone will increase it in our Saga/Aggregate automatically. We may also use inbuilt trait to avoid adding property manually.

Self Healing

To handle concurrency exceptions and ensure the system can self-heal, Ecotone offers retry mechanisms.

In synchronous scenarios, like Command Handler being called via HTTP, can be used to recover. If a concurrency exception occurs, the Command Message will be retried immediately, minimizing any impact on the end user. This immediate retry ensures that the Message Handler can self-heal and continue processing without affecting the user experience. &#xNAN;In asynchronous scenarios, you can use still use instant retries, yet you may also provide . This means that when concurrency exception will occur, the Message will be retried after a certain delay. This as a result free the system resources from continues retries and allows for recovering after given period of delay.

Defines how to handle failures when processing messages. This is final failure strategy as it's used in case, when there is no other way to handle the failure. For example, when there is no retry policy, or when the retry policy has reached its maximum number of attempts. Also, when the destination of Error Channel is not defined, or sending to Error Channel fails.

Available Strategies

Ecotone provides three final strategies:

• RELEASE - Message is released back to the Channel for another attempt. This way order will be preserved, yet it can result in processing being blocked if the message keeps failing.

• RESEND - Message is resend back to the Channel for another attempt, as a result Message Consumer will be unblock and will be able to continue on next Messages. This way next messages can be consumed without system being stuck.

• IGNORE - Message is discarded, processing continues. Can be used for non critical message, to simply ignore failed messages.

• STOP - Consumer stops, message is preserved. This strategy can be applied when our system depends heavily on the order of the Messages to work correctly. In that case we can stop the Consumer, resulting in Message still awaiting to be consumed.

Configuration Message Channel

This can be configured on Message Channel level:

Configuration Consumer

This can also be configured at the Message Consumer level

We may define Time to Live for Messages. This way if Message will not be handled within specific amount of time, it will be automatically discarded. This is useful in scenarios like sending notifications, where given notification like One Time Password may actually have meaning only for 5 minutes.

Message Handler Time to Live

You may delay handling given asynchronous message by adding #[Delayed] attribute.

#[TimeToLive(new TimeSpan(seconds: 50))]
#[Asynchronous("notifications")]
#[EventHandler(endpointId: "welcomeEmail")]
public function sendWelcomeNotificationWhen(UserWasRegistered $event): void
{
   // handle welcome notification
}

Using Expression language

To dynamically calculate expected TTL, we can use expression language.

We could also access any object from our Dependency Container, in order to calculate the delay and pass there our Command:

Message Time to Live

We may send an Message and tell Ecotone to delay it using deliveryDelay Message Header:

We may extend Gateways functionality with asynchronocity. This way we can pass any Message via given Message Channel first.

Asynchronous Gateways

To make Gateway Asynchronous we will use Asynchronous Attribute, just like with Asynchronous Message Handlers. We may can extend any types of Gateways: Command/Event/Query Buses, or custom .

To build for example a CommandBus which will send Messages over async channel, we will simply , and add our method.

then we all Commands that will be triggered via AsynchronousCommandBus will go over async channel.

Having asynchronous CommandBus is especially useful, if given Message Handler is not meant be executed asynchronous by default.

then when using standard CommandBus above Command Handler will be executed synchronous, when using AsynchronousCommandBus it will be done asynchronously.

Outbox Event Bus

It's easy to build an Outbox pattern using this Asynchronous Gateways. Just make use of to push Messages over Database Channel.

and then register dbal channel

Then whenever we will send Events within Command Handler (which is wrapped in transaction by default while using Dbal Module). Messages will be commited as part of same transaction.

Setup for tutorial

Depending on the preferences, we may choose tutorial version for

To keep the system reliable and resilient it's important to handle errors with grace. Ecotone provides solutions to handle failures within the system that helps in:

• Self-healing the application (, , )

• Ensuring Data Consistency (, , )

Before diving into this section be sure to understand how Aggregates works in Ecotone based on .

Difference between Aggregate Types

Ecotone provides higher level abstraction to work with Event Sourcing, which is based on Event Sourced Aggregates. Event Sourced Aggregate just like normal Aggregates protect our business rules, the difference is in how they are stored.

Changes in the Application will happen. After some time we may want to refactor namespaces, change the name of Aggregate or an Event. However those kind of changes may break our system, if we already have production data which references to any of those. Therefore to make our Application to immune to future changes we need a way to decouple the code from the data in the storage, and this is what Ecotone provides.

Custom Stream Name

Our Event Stream name by default is based on the Aggregate Class name. Therefore to make it immune to changes we may provide custom Stream Name. To do it we will apply Stream attribute to the aggregate:

Then tell the projection to make use of it:

Service Name Configuration

In order for Ecotone how to route messages you need to register Service Name (Application Name).

Access attribute from interceptor

We may access attribute from the intercepted endpoint in order to perform specific action

then we would have an Message Endpoint using this Attribute:

and it can be used in the intereceptors by type hinting given parameter:

Distribution

Distribution Bus is just like CommandBus or EventBus. It creates smooth and elegant way for explicit communication between Applications (Services) without introducing any hassle and requirements for doing configurations, bindings and mappings. It make it easy for Developers to build integrations and maintain them in the long-term.

Read more in given Module section.

Ecotone comes with Support for integrating Service (Applications) together in decoupled way, for this Ecotone provides Distributed Bus.

Read more in bellow sections:

Materials

To use Ecotone's Aggregate functionality, we need a registered repository. Ecotone comes with built-in support for popular persistence options like Doctrine ORM, Eloquent, and document stores, so there's a good chance we can use what we already have without extra work. If our storage solution isn't supported yet, or if we have specific requirements, we can easily register our own custom repository by following the steps in this section. This flexibility means we're not locked into any particular database or ORM—we can use whatever fits our project best.

Repository for State-Stored Aggregate

State-Stored Aggregate are normal Aggregates, which are stored using Standard Repositories. Therefore to configure Repository for your Aggregate, create a class that extends StandardRepository interface:

1. canHandle method informs, which Aggregate Classes can be handled with this Repository. Return true, if saving specific aggregate is possible, false otherwise.

2. findBy method returns if found, existing Aggregate instance, otherwise null.

3. save method is responsible for storing given Aggregate instance.

• $identifiers are array of #[Identifier] defined within aggregate.

• $aggregate is instance of aggregate

• $metadata is array of extra information, that can be passed with Command

Set up your own Implementation

When your implementation is ready simply mark it with #[Repository] attribute:

Example implementation using Doctrine ORM

This is example implementation of Standard Repository using Doctrine ORM.

Repository:

Using Multiple Repositories

By default Ecotone when we have only one Standard and Event Sourcing Repository registered, Ecotone will use them for our Aggregate by default. This comes from simplification, as if there is only one Repository of given type, then there is nothing else to be choose from. However, if we register multiple Repositories, then we need to take over the process and tell which Repository will be used for which Aggregate.

• In case of we do it using canHandle method.

• In case of inbuilt Repositories, we should follow configuration section for given type

Repository for Event Sourced Aggregate

Custom repository for Event Sourced Aggregates is described in more details under .

Be sure to read CQRS Introduction before diving in this chapter.

Execute your Business Actions via Interface

Business Interface aims to reduce boierplate code and make your domain actions explicit. In Application we describe an Interface, which executes Business methods. Ecotone will deliver implementation for this interface, which will bind the interface with specific actions. This way we can get rid of delegation level code and focus on the things we want to achieve. For example, if we don't want to trigger action via Command/Query Bus, we can do it directly using our business interface and skip all the Middlewares that would normally trigger during Bus execution. There are different types of Business Interfaces and in this chapter we will discuss the basics of build our own Business Interface, in next sections we will dive into specific types of business interfaces: Repositories and Database Layers.

Command Interface

Let's take as an example creating new Ticket

We may define interface, that will call this Command Handler whenever will be executed.

This way we don't need to use Command Bus and we can bypass all Bus related interceptors.

The attribute #[BusinessMethod] tells Ecotone that given Interface is meant to be used as entrypoint to Messaging and which Message Handler it should send the Message to. Ecotone will provide implementation of this interface directly in our Dependency Container.

Aggregate Command Interface

We may also execute given Aggregate directly using Business Interface.

Then we define interface:

Query Interface

Defining Query Interface works exactly the same as Command Interface and we may also use it with Aggregates.

Then we may call this Query Handler using Interface

Result Conversion

If we have registered then we let Ecotone do conversion to Message Handler specific format:

Then we may call this Query Handler using Interface

Ecotone will use defined Converter to convert array to TicketDTO.

Payload Conversion

If we have registered then we let Ecotone do conversion to Message Handler specific format:

Then we may call this Query Handler using Interface

Ecotone will use defined Converter to convert array to CreateTicket command class.

Create an Aggregate by another Aggregate

There may be a scenario where the creation of an Aggregate is conditioned by the current state of another Aggregate.

Ecotone provides a possibility for that and lets you focus more on domain modeling rather than technical nuances you may face trying to implement an actual use case.

This case is supported by both Event Sourcing and State-based Aggregates.

Create a State-based Aggregate

It is possible to send a command to an Aggregate and expect a State-based Aggregate to be returned.

Create an Event Sourcing Aggregate

It is also possible to send a command to an Aggregate and expect the Event Sourcing Aggregate to be returned.

Events handling

Both of the Aggregates (called and result) can still record their Events using an Internal Recorder. Recorded Events will be published after the operation is persisted in the database.

Persisting a state change

In the case of an Event Sourcing Aggregate recording an event indicates a state change of that Aggregate.

Also, when calling a State-based Aggregate its state may be changed before returning the newly created Aggregate. E.g. you want to save a reference to the newly created Aggregate.

Ecotone will try to persist both called and returned Aggregates.

Converting Results

In rich business domains we will want to work with higher level objects than associate arrays. Suppose we have PersonNameDTO and defined Ecotone's Converter for it.

Converting to Collection of Objects

Ecotone will read the Docblock and based on that will deserialize Result Set from database to list of PersonNameDTO.

Converting to single Object

Using combination of First Row Fetch Mode, we can get first row and then use it for conversion to PersonNameDTO.

Converting Iterator

For big result set we may want to avoid fetching everything at once, as it may consume a lot of memory. In those situations we may use Iterator Fetch Mode, to fetch one by one. If we want to convert each result to given Class, we may define docblock describing the result:

Each returned row will be automatically convertered to PersonNameDTO.

Converting to specific Media Type Format

We may return the result in specific format directly. This is useful when Business Method is used on the edges of our application and we want to return the result directly.

In this example, result will be returned in application/json.\

In general having streams in need for snapshots may indicate that our model needs revisiting. We may cut the stream on some specific event and begin new one, like at the end of month from all the transactions we generate invoice and we start new stream for next month. However if cutting the stream off is not an option for any reason, we can use snapshots to avoid loading all events history for given Aggregate. Every given set of events snapshot will be taken, stored and retrieved on next calls, to fetch only events that happened after that.

Setting up

EventSourcingConfiguration provides the following interface to set up snapshots.

class EventSourcingConfiguration
{
    public const DEFAULT_SNAPSHOT_TRIGGER_THRESHOLD = 100;

    public function withSnapshotsFor(
        string $aggregateClassToSnapshot, // 1.
        int $thresholdTrigger = self::DEFAULT_SNAPSHOT_TRIGGER_THRESHOLD, // 2.
        string $documentStore = DocumentStore::class // 3.
    ): static
}

1. $aggregateClassToSnapshot - class name of an aggregate you want Ecotone to save snapshots of

2. $thresholdTrigger - amount of events for interval of taking a snapshot

3. $documentStore - reference to document store which will be used for saving/retrieving snapshots

To set up snapshots we will define configuration.

Snapshot threshold

Threshold states at which interval snapshots should be done. Therefore with below configuration:

snapshots will be done every 500 events. Then when snapshot will be loaded, it will start loading the events from event number 501 for given Aggregate instance.

Ecotone provides ability to extend any Messaging Gateway using Interceptors. We can hook into the flow and add additional behavior.

For better understanding, please read Interceptors section before going through this chapter.

Intercepting Gateways

Suppose we want to add custom logging, whenever any Command is executed. We know that CommandBus is a interface for sending Commands, therefore we need to hook into that Gateway.

Intercepting Gateways, does not differ from intercepting Message Handlers.

Building customized Gateways

We may also want to have different types of Message Buses for given Message Type. For example we could have EventBus with audit which we would use in specific cases. Therefore we want to keep the original EventBus untouched, as for other scenarios we would simply keep using it.

To do this, we will introduce our new EventBus:

That's basically enough to register our new interface. This new Gateway will be automatically registered in our DI container, so we will be able to inject it and use.

Now as this is separate interface, we can point interceptor specifically on this

Pointcut by attributes

We could of course intercept by attributes, if we would like to make audit functionality reusable

and then we pointcut based on the attribute

Asynchronous Gateways

Gateways can also be extended with asynchronous functionality on which you can read more in .

In order to optimize projections or to avoid using external storage, we may use of Projection's State. State is data that is kept between executions and can be passed to Projection's Event Handler.

Passing state inside Projection

In order to pass the state to Projection's Event Handlers we need to mark method parameter with #[ProjectionState].

Ecotone will resolve this parameter and pass the state. The returned state from the Event Handler will becomes new state for next execution. We may pass the state between all Event Handlers in given Projection.

Fetching the state from outside

You may want to fetch the state from outside to return it to the end user. In that case Ecotone brings ProjectionStateGateway.

The first parameter of the attribute is the projection name, so Ecotone can know, which state it should look for. This will automatically convert the state to your defined return type.

Demo

Setting up Projections

Projections are about deriving current state from the stream of events. Projections can be added in any moment of the application lifetime and be built up from existing history, till the current time. This is powerful concept as it allow for building views quickly and discarding them without pain, when they are not longer needed.

#[Projection("inProgressTicketList", Ticket::class] // 1
class InProgressTicketList
{
    public function __construct(private Connection $connection) {}

    #[EventHandler] // 2
    public function addTicket(TicketWasRegistered $event, array $metadata) : void
    {
        $result = $this->connection->executeStatement(<<<SQL
    INSERT INTO in_progress_tickets VALUES (?,?)
SQL, [$event->getTicketId(), $event->getTicketType()]);
    }

    #[QueryHandler("getInProgressTickets")] // 3
    public function getTickets() : array
    {
        return $this->connection->executeQuery(<<<SQL
    SELECT * FROM in_progress_tickets
SQL)->fetchAllAssociative();
    }    
}

1. This tells Ecotone that specific class is Projection. The first parameter is the name of the projection and the second is name of the stream (the default is the name of the Aggregate) that this projection subscribes to.

2. Events that this projection subscribes to

3. Optional Query Handlers for this projection. They can be placed in different classes depending on preference.

Document Store

is a great way to set up your projections. You can freely create DTO objects, or play with simple arrays and Ecotone will serialize/deserialize and store them for you.

Whenever message fails during asynchronous processing it will kept repeated till the moment it will succeed. However retry strategy with dead letter queue may be set up in order to retry message given amount of times and then move it to the storage for later review and manual retry.

This is where Ecotone Pulse kicks in, as instead of reviewing and replaying the message directly from the application's console, you may do it directly from the UI application. Besides you may connect multiple Ecotone's application to the Pulse Dashboard to have full overview of your whole system.

Installation

Enable in your service and with .

After this you may run docker image with Ecotone Pulse passing the configuration to your services and RabbitMQ connection.

Then run docker image with Ecotone Pulse passing environment variables:

SERVICES environment

Provide array of services with service name and .

AMQP_DSN environment

DSN to your RabbitMQ instance, which services are connected with .

Usage

In the dashboard you may check all the connected services. For quick overview, you will find amount of errors within given service there.

To review error messages go to specific service. From there you can review the error message, stacktrace and replay it or delete.

Demo Application

You may check demo application, where Symfony and Laravel services are connected to Ecotone pulse in .

In its lifetime events may change. In order to track those changes Ecotone provides possibility of versioning events.

use Ecotone\Modelling\Attribute\Revision;

#[Revision(2)]
class MyEvent
{
    public string $id;
}

Value given with Revision attribute will be stored by Ecotone in events metadata. Attribute is used only when event is saved in event store. In order to read it, you can access events metadata, e.g. in event handler.

[Enterprise] Accessing Metadata in Event Sourcing Handler

Depending on the version we may actually want to restore our Aggregate a bit differently. This is especially useful when we've changed the way Events are structured and introduced new version of the Event. For this we can use revision header to access the version on which given Event was stored.

Materials

Demo implementation

You may find demo implementation here.

Links

• [Article]

• [Article]

• [Article]

• [Documentation]

Read sections first to get more details about Aggregates.

Publishing Events from Aggregate

To tell Ecotone to retrieve Events from your Aggregate add trait WithEvents which contains two methods: recordThat and getRecordedEvents.

Working with Event Sourcing Aggregates

Just as with Standard Aggregate, ES Aggregates are called by Command Handlers, however what they return are Events and they do not change their internal state.

When this Aggregate will be called via Command Bus with CreateProduct Command, it will then return new ProductWasCreated Event.

Demo

Read Blog Post

All Events may contain additional Metadata. This is especially useful for storing information that are not required for Command to be handled, yet are important for auditing and projecting purposes.

Metadata

In Ecotone any communication happens via Messages, and Messages contains of Payload and Headers (Metadata).

So far we've discussed only the Payload part, for example ProductWasCreated Event Class is actually an Payload.

One of the drawback of Event Sourcing is eventual consistency. Whenever event happens we want to do two things, update our projection and inform the end user about change. However user need to be informed about the change after projection is refreshed, as otherwise he will get stale view. In order to solve this drawback Ecotone brings possibility for emitting events directly from projection. So instead of subscribing to Domain Events (Aggregate State Changed), end user may subscribe to change in the projection.

Emit the event

In order to emit the events, we are using EventStreamEmitter. Whenever we emit given events, they are stored in Projection's stream.

Demo

Read Blog Post

In case of Ecotone we don't delay whole Message, but specific Message Handler. This helps in scenarios when we have multiple Event Handler and we would like to configure the delay differently. For example may we have a case, where as a result of Order being placed, we would want to delay notification, yet to call Payment Service right away.

Choosing Event Streams for Projection

The Projection is deriving from Event Stream. There may be situations when we will want to derive the projection from multiple streams however. Let's see what options do we have:

From Single Stream

If we are interested in single stream, we can listen directly for specific aggregate

In here we are handling events from single Basket's Aggregate stream. It will contain all the events in relation to this aggregate.

From Multiple Streams

There may be situations, when we will want to handle different streams together.

From Category

In case if using we will need to use categories to target. If we would listen on Domain\Ticket stream using Stream Per Aggregate then we would not target any event, as the streams that are created are suffixed by the identifier Domain\Ticket-123.

In that case we can make use of categories in order to target Domain\Ticket-*.

Handling Events By Names

If you want to avoid storing class names of your events in the Event Store you may mark them with name.

And tell the projection to make use of it

Speeding Up Projection Restore

If projections are handling the events by names, then there is no need to deserialization of the event to the class and simple array can be used. In case of thousands of events during it will speed up the process.

This chapter will cover the basics on how to implement an Aggregate. We will be using Command Handlers in this section, so ensure reading External Command Handler section first, to understand how Command are sent and handled.

Aggregate Command Handlers

Working with Aggregate Command Handlers is the same as with External Command Handlers. We mark given method with Command Handler attribute and Ecotone will register it as Command Handler.

In most common scenarios, Command Handlers are used as boilerplate code, which fetch the aggregate, execute it and then save it.

This is non-business code that is often duplicated wit each of the Command Handler we introduce. Ecotone wants to shift the developer focus on the business part of the system, this is why this is abstracted away in form of Aggregate.

By providing Identifier attribute on top of property in your Aggregate, we state that this is identifier of this Aggregate (Entity in Symfony/Doctrine world, Model in Laravel world). This is then used by Ecotone to fetch your aggregate automatically.

However Aggregates need to be in order to be executed. When we will send an Command, Ecotone will use property with same name from the Command instance to fetch the Aggregate.

When identifier is resolved, Ecotone use repository to fetch the aggregate and then call the method and then save it. So basically do all the boilerplate for you.

State-Stored Aggregate

An Aggregate is a regular object, which contains state and methods to alter that state. It can be described as Entity, which carry set of behaviours. When creating the Aggregate object, you are creating the Aggregate Root.

1. Aggregate tells Ecotone, that this class should be registered as Aggregate Root.

2. Identifier is the external reference point to Aggregate.

   This field tells Ecotone to which Aggregate a given Command is targeted.

3. CommandHandler

Running Projection

Synchronously Event Driven Projection

By default Ecotone runs the projections synchronously with your aggregate changes. This kind of running configuration can be used to avoid eventual consistency problems or for testing purposes. However when you expect multiple accesses to your Aggregates at the same time, you may consider asynchronous projection to protect yourself from concurrency problems.

Polling Projection

You may run your projection in the background. It will query the database within constant time intervals, to look if new events have been registered. Each projection is running as separate process. To register Polling Projection make use of .

Then we can run:

Asynchronously Event Driven Projection

You may pass your projections in event driven manner using .

The difference between Polling and Event Driven projection is the way they are triggered. The Event Driven is only triggered when new event comes to the system. This avoid the pitfall of continues database access while using Polling Projection. The second strength of Asynchronously Event Driven Projection is possibility of registering multiple projections under same channel (which is same consumer).

Custom Configuration

Event Sourcing Customer Configuration

You may customize your Event Sourcing configuration with following configuration:

Projection Custom Configuration

You may configure your projection custom configuration. Take under consideration that some configuration may have sense only in case of .

load_count //Default: null

Change load batch size in each run for single projection.

cache_size //Default: 1000

The cache size is how many stream names are cached in memory, the higher the number the less queries are executed and therefore the projection runs faster, but it consumes more memory.

sleep //Default: 100000

The sleep options tells the projection to sleep that many microseconds before querying the event store again when no events were found in the last trip. This reduces the number of cpu cycles without the projection doing any real work.

persist_block_size //Default: 1000

The persist block size tells the projector to persist its changes after a given number of operations. This increases the speed of the projection a lot. When you only persist every 1000 events compared to persist on every event, then 999 write operations are saved. The higher the number, the fewer write operations are made to your system, making the projections run faster. On the other side, in case of an error, you need to redo the last operations again. If you are publishing events to the outside world within a projection, you may think of a persist block size of 1 only.

lock_timeout_ms //Default: 1000

Indicates the time (in milliseconds) the projector is locked. During this time no other projector with the same name can be started. A running projector will update the lock timeout on every loop, except you configure an update lock threshold.

update_lock_threshold //Default: 0

If update lock threshold is set to a value greater than 0 the projection won't update lock timeout until number of milliseconds have passed. Let's say your projection has a sleep interval of 100 ms and a lock timeout of 1000 ms. By default the projector updates lock timeout after each run so basically every 100 ms the lock timeout is set to: now() + 1000 ms This causes a lot of extra work for your database and in case the database is replicated this can cause a lot of network traffic, too.

gap_detection //Default: new \Prooph\EventStore\Pdo\Projection\GapDetection()

Gap Detection makes projection to wait for upcoming events if any gap occurs in your event stream. To disable Gap Detection you can set value to null.

Default flow

In default flow there is no need to fetch or store Aggregates, because this is done for us. We simply need to trigger an Command via CommandBus. However in some cases, you may want to retake orchestration flow and do it directly. For that cases Business Repository Interface or Instant Fetch Aggregate can help you.

Business Repository Interface

Special type of is Repository. This Interface allows us to simply load and store our Aggregates directly. In situations when we call Command directly in our Aggregates we won't be in need to use it. However for some specific cases, where we need to load Aggregate and store it outside of Aggregate's Command Handler, this business interface becomes useful.

Ecotone will read type hint to understand which Aggregate you would like to fetch or save.

Pure Event Sourced Repository

When using Pure Event Sourced Aggregate, instance of Aggregate does not hold recorded Events. Therefore passing aggregate instance would not contain any information about recorded events. For Pure Event Sourced Aggregates, we can use direct event passing to the repository:

Instant Fetch Aggregate

In cases where we want to fetch Aggregate directly without introducing orchestration into our code base or depending on the Repositories, we can use Fetch Aggregate Attribute.

To do instant fetch of Aggregate we will be using Fetch Attribute. Suppose we want PlaceOrder Command Handler, and we want to fetch User Aggregate:

Fetch using to evaluate the expression given inside the Attribute. For example having above "payload.userId" and following Command:

Ecotone will use userId from the Command to fetch User Aggregate instance. &#xNAN;"payload" is special variable within expression that points to our Command, therefore whatever is available within the Command is available for us to do the fetching. This provides quick way of accessing related Aggregates without the need to inject Repositories.

Allowing non existing Aggregates

By default Ecotone will throw Exception if Aggregate is not found, we can change the behaviour simply by allowing nulls in our method declaration:

Accessing Message Headers

We can also use Message Headers to fetch our related Aggregate instance:

Using External Services

In some cases we may not have enough information to provide correct Identifier, for example that may require some mapping in order to get the Identifier. For this cases we can use "reference" function to access any Service from Depedency Container in order to do the mapping.

Persistence Strategy

Describes how streams with events will be stored. Each Event Stream is separate Database Table, yet how those tables are created and what are constraints do they protect depends on the persistence strategy.

Simple Stream Strategy

This is the basics Stream Strategy which involves no constraints. This means that we can append any Events to it without providing any additional metadata.

Now as this is free append involves no could re-run this code apply exactly the same Event. This can sounds silly, but it 's make it useful for particular cases. It make it super easy to append new Events. We basically could just add this action in our code and keep applying Events to the Event Stream, we don't need to know context of what happened before.

This is useful for scenarios where we just want to store information without putting any business logic around this. It could be used to continues stream of information like:

• Temperature changes

• Counting car passing by in traffic jam

• Recording clicks and user views.

Partition Stream Strategy

This the default persistence strategy. It does creates partitioning within Event Stream to ensure that we always maintain correct history within partition. This way we can be sure that each Event contains details on like Aggregate id it does relate to, on which version it was applied, to what Aggregate it references to.

The tricky part here is that we need to know Context in order to apply the Event, as besides the Aggregate Id, we need to provide Version. To know the version we need to be aware of last previous applied Event.

This Stream Strategy is great whenever business logic is involved that need to be protected. This solves for example the problem of concurrent access on the database level, as we for example can't store Event for same Aggregate Id and Version twice in the Event Stream. We would use it in most of business scenarios where knowing previous state in order to make the decision is needed, like:

• Check if we can change Ticket based on status

• Performing invocing from previous transactions

• Decide if Order can be shipped

Stream Per Aggregate Strategy

This is similar to Partition strategy, however each Partition is actually stored in separate Table, instead of Single One.

This can be used when amount of partitions is really low and volume of events within partition is huge.

Custom Strategy

You may provide your own Customer Persistence Strategy as long as it implements PersistenceStrategy.

Setting global Persistence Strategy

To set given persistence strategy as default, we can use ServiceContext:

Multiple Persistence Strategies

Once set, the persistence strategy will apply to all streams in your application. However, you may face a situation when you need to have a different strategy for one or more of your streams.

The above will make the Simple Stream Strategy as default however, for some_stream Event Store will use the Aggregate Stream Strategy.

Ecotone allows to work with Database using DbalBusinessMethod. The goal is to create abstraction which significantly reduce the amount of boilerplate code required to implement data access layers. Thanks to Dbal based Business Methods we are able to avoid writing integration and transformation level code and focus on the Business part of the system. To make use of Dbal based Business Method, install Dbal Module first.

Write Business Methods

Let's consider scenario where we want to store new record in Persons table. To make it happen just like with Business Method we will create an Interface, yet this time we will mark it with DbalBusinessMethod.

The first parameter passed to DbalBusinessMethod is actual SQL, where we can provide set of named parameters. Ecotone will automatically bind parameters from method declaration to SQL ones by names.

Custom Parameter Name

We may bind parameter name explicitly by using DbalParameter attribute.

This can be used when we want to decouple interface parameter names from binded parameters or when name in database column is not explicit enough for being part of interface.

Returning number of records changed

If we want to return amount of the records that have been changed, we can add int type hint to our Business Method:

Query Business Methods

We may want to fetch data from the database and for this we will be using DbalQueryBusinessMethod.

The above will return result as associative array with the columns provided in SELECT statement.

Fetching Mode

To format result differently we may use different fetch modes. The default fetch Mode is associative array.

First Column Fetch Mode

This will extract the first column from each row, which allows us to return array of person Ids directly.

First Column of first row Mode

To get single variable out of Result Set we can use First Column of first row Mode.

This way we can provide simple interfaces for things Aggregate SQLs, like SUM or COUNT.

First Row Mode

To fetch first Row of given Result Set, we can use First Row Mode.

This will return array containing person_id and name.

Returning Nulls

When using First Row Mode, we may end up having no returned row at all. In this situation Dbal will return false, however if Return Type will be nullable, then Ecotone will convert false to null.

Returning Iterator

For big result set we may want to avoid fetching everything at once, as it may consume a lot of memory. In those situations we may use Iterator Fetch Mode, to fetch one by one.

Parameter Types

Each parameter may have different type and Ecotone will try to recognize specific type and set it up accordingly. If we want, we can take over and define the type explicitly.

Ecotone comes with inbuilt Event Sourcing repository after Event Sourcing package is installed. However you want to roll out your own storage for Events, or maybe you already use some event-sourcing framework and would like to integrate with it. For this you can take over the control by introducing your own Event Sourcing Repository.

Custom Event Sourcing Repository

We do start by implementing EventSourcingRepository interface:

1. canHandle - Tells whatever given Aggregate is handled by this Repository

2. findBy - Method returns previously created events for given aggregate. Which Ecotone will use to reconstruct the Aggregate.

3. save - Stores events recorded by Event Sourced Aggregate

and then we need to mark class which implements this interface as Repository

Storing Events

Ecotone provides enough information to decide how to store provided events.

Identifiers will hold array of identifiers related to given aggregate (e.g. ["orderId" ⇒ 123]). Events will be list of Ecotone's Event classes, which contains of payload and metadata, where payload is your Event class instance and metadata is specific to this event. Metadata as parameter is generic metadata available at the moment of Aggregate execution. Version before handling on other hand is the version of the Aggregate before any action was triggered on it. This can be used to protect from concurrency issues.

The structure of Events is as follows:

Core metadata

It's worth to mention about Ecotone's Events and especially about metadata part of the Event. Each metadata for given Event contains of three core Event attributes:

"_aggregate_id" - This provides aggregate identifier of related Aggregate

"_aggregate_version" - This provides version of the related Event (e.g. 1/2/3/4)

"_aggregate_type" - This provides type of the Aggregate being stored, which can be customized

Aggregate Type

If our repository stores multiple Aggregates is useful to have the information about the type of Aggregate we are storing. However keeping the class name is not best idea, as simply refactor would break our Event Stream. Therefore Ecotone provides a way to mark our Aggregate type using Attribute

This now will be passed together with Events under _aggregate_type metadata.

Named Events

In Ecotone we can name the events to avoid storing class names in the Event Stream, to do so we use NamedEvent.

then when events will be passed to save method, they will automatically provide this name under eventName property.

Snapshoting

With custom repository we still can use inbuilt . To use it for customized repository we will use BaseEventSourcingConfiguration.

Ecotone then after fetching snapshot, will load events only from this given moment using `fromAggregateVersion`.

Testing

If you want to test out your flow and storing with your custom Event Sourced Repository, you should disable default in memory repository

Ecotone comes with support for running period tasks or cron jobs.

Scheduled Method

endpointId - it's name which identifies process to run poller - Configuration how to execute this method . Above configuration tells Ecotone to execute this method every second.

After setting up Scheduled endpoint we can run the endpoint:

Scheduled Handler

You can run Scheduled for given Handler. Right now method return which is send to given routing.

requestChannelName - The channel name to which should be send.

When the Message will arrive on the Command Handler it will be automatically converted to ExchangeCommand. If you want to understand how the conversion works, you may read about it in .

Expression Language

We can also set up cron and fixed rate using expression language. This gives us ability to set it up differently based on the environment we are currently in.

Materials

Demo implementation

You may find demo implementation .

Links

Ecotone provides easy way to pass Message Headers (Metadata) with your Message and use it in your Message Handlers or Interceptors. In case of asynchronous scenarios, Message Headers will be automatically mapped and passed to through your Message Broker.

Passing headers to Bus:

Pass your metadata (headers), as second parameter.

Then you may access them directly in Message Handlers:

Converting Headers

If you have defined for given type, then you may type hint for the object and Ecotone will do the conversion:

And then we can use Classes instead of scalar types for our Headers:

Automatic Header Propagation

Ecotone by default propagate all Message Headers automatically. This as a result preserve context, which can be used on the later stage. For example we may provide executorId header and skip it in our Command Handler, however use it in resulting Event Handler.

This will execute Command Handler:

and then even so, we don't resend this Header when publishing Event, it will still be available for us:

Message Identification and Correlation

When using Messaging we may want to be able to trace where given Message came from, who was the parent how it was correlated with other Messages. In Ecotone all Messages contains of Message Id, Correlation Id and Parent Id within Metadata. Those are automatically assigned and propagated by the Framework, so from application level code we don’t need to deal manage those Messaging level Concepts.

Id

Each Message receives it's own unique Id, which is Uuid generated value. This is used by Ecotone to provide capabilities like , and Message identification for .

Parent Id

"parentId" header refers to Message that was direct ancestor of it. In our case that can be correlation of Command and Event. As a result of sending an Command, we publish an Event Message.

Parent id will always refer to the previous Message. What is important however is that, if we have multiple of the Message with same Id.

Correlation Id

Correlation Id is useful for longer flows, which can span over multiple Message Handlers. In those situations we may be interested in how our Message flow have branched:

Projection Actions

Projection initialization

As projection can be restarted, deleted and created differently. When the projection knows how to setup it itself, it's easy to rebuild it when change is needed.

And inside the projection we need to implement ProjectionInitialization to tell Ecotone what to do:

Resetting/Rebuilding the projection

In order to restart the projection in case we want to provide incompatible change, we can simply reset the projection and it will build up from the beginning.

And inside the projection we need to implement ProjectionReset to tell Ecotone what to do:

Deleting the projection

If we want to delete the projection

And inside the projection we need to implement ProjectionDelete to tell Ecotone what to do:

Manually triggering projection

If we want to manually trigger projection

Two ways of setting up Event Sourced Aggregates

There are two ways we can configure our Aggregate to record Events.

1) Pure Event Sourced Aggregate

This way of handling events does allow for pure functions. This means that actions called on the Aggregate returns Events and are not changing internal state of Aggregate.

1. EventSourcingAggregate and Identifier .

2. Event Sourced Aggregate must provide version. You may leave it to Ecotone using WithAggregateVersioning or you can implement it yourself.

3. CommandHandlerfor event sourcing returns events generated by specific method. This will be passed to the

2) Internal Recorder Aggregate

This way of handling events allow for similarity with . This convention requires changing internal state of Aggregate to record Events. Therefore Pure ES Aggregate is recommended as it's not require for any internal state changes in most of the scenarios. However ES Aggregate with Internal Recorder may be useful for projects migrating with other solutions, or when our team is heavily used to working this way.

1. In order to make use of alternative way of handling events, we need to provide trait WithEvents.

2. Command Handlers instead of returning events are acting the same as . All events which will be published using recordThatwill be passed to the to be stored.

In case of Ecotone we don't prioritize whole Message, but specific Message Handler. This helps in scenarios when we have multiple Event Handlers, and we would like to configure each of them differently. For example may we have a case, where we want to prioritized one notification over another.

In Ecotone the higher the priority, the quicker given Event Handler will be called.

Synchronous Prioritization

In case we publish an Event, we may want have multiple subscribing Event Handlers. In some situations we may want given action to happen before the other. This may happen for example for example, when one Event Handler updates an data, which the other is using. Therefore we may need to ensure that the Handler modifying data will be called before the one that make use of it.

Default Synchronous Prioritization

There may be cases when we will use Synchronous Projections and then we would like to use this data in Event Handler. In that case, if standard Event Handler would be called first, we would lack the data. Therefore by default when Standard Event Handler has same priority as Projection Event Handler, Projection will be called first.

This also works for Aggregates to be prioritized before Standard Event Handlers. Therefore the ordering when same priority level is used, is as follows:

1. Projection Event Handlers

2. Aggregate/Sagas Event Handlers

3. Standard Event Handlers

Asynchronous Prioritization

Ecotone does allow to prioritize an handling of given Message before another one. This way we can handle quicker Messages that have been published to Asynchronous Message Channel later. For example we may send a lot of different notifications, however when Customer asks for One-Time Password we want to deliver it immediately. Therefore for this scenario we would setup higher priority for Authentication Token notification handling, than for other Notifications.

Message Handler Priority

You may prioritize handling given asynchronous message by adding #[Priority] attribute.

Message Priority

We may send an Message and tell Ecotone to prioritize it using priority Message Header:

Dbal Dead Letter

Ecotone comes with full support for managing full life cycle of a error message by using Dbal Module.

• Store failed Message with all details about the exception

• Allow for reviewing error Messages

• Allow for deleting and replaying error Message back to the

Installation

To make use of Dead Letter, we need to have installed.

Storing Messages in Dead Letter

If we configure default error channel to point to "dbal_dead_letter" then all Error Messages will land there directly

Dead Letter with Delayed Retries

We may also want to try to recover before we consider Message to be stored in Dead Letter:

and then we use inbuilt Retry Strategy:

Dead Letter Console Commands

Help

Get more details about existing commands

Listing Error Messages

Listing current error messages

Show Details About Error Message

Get more details about given error message

Replay Error Message

Replay error message. It will return to previous channel for consumer to pick it up and handle again.

Replay All Messages

Replaying all the error messages.

Delete Message

Delete given error message

Turn off Dbal Dead Letter

Managing Multiple Ecotone Applications

The above solution requires running Console Line Commands. If we want however, we can manage all our Error Messages from one place using .

This is especially useful when we've multiple Applications, so we can go to single place and see if any Application have failed to process Message.

Outbox Pattern

To ensure full level of data consistency, we may decide to store messages along side with data changes. This way we work only with single data storage, avoiding completely problem with persisting Message in two sources at once. To make it happen Ecotone implements so called Outbox pattern.

When loading Aggregates or Sagas we need to know what Identifier should be used for that. This depending on the business feature we work may require different approaches. In this section we will dive into different solutions which we can use.

Auto-Mapping from the Command/Event

Ecotone resolves the mapping automatically, when Identifier in the Aggregate/Saga is named the same as the property in the Command/Event.

then, if Message has productId, it will be used for Mapping:

Installation

In order to use Deduplication, install .

Default Idempotent Message Consumer

Be sure to read before diving in this chapter.

Handling Queries

External Query Handlers are Services available in your dependency container, which are defined to handle Queries.

Queries are Plain Old PHP Objects:

To send an Query we will be using

Whenever we use more than one storage during single action, storing to first storage may end up with success, yet the second may not. This can happen when we store data in database and then send Messages to Message Broker. If failure happen it can be that we will send some Message to Broker, yet fail to store related data or vice versa. Ecotone provide you with tools to help solve this problem in order to make sending Messages to Message Broker resilient.

Message Collector

Ecotone by default enables Message Collector. Collector collect messages that are about to be send to asynchronous channels in order to send them just before the transaction is committed. This way it help avoids bellow pitfalls: