Reactive DDD with Akka - Reactive business processes

Introduction

In the previous post we learned how to implement a sample ordering service using the Akka-DDD framework. The service exposed the Reservation Office responsible for preparing and confirming the reservation of products, that the client added to their shopping cart. We learned that an office is a command handling business entity whose sole responsibility is to validate and process the commands. If the command validation succeeds, an appropriate event is written to the office journal.

The Reservation Office alone is obviously not able to execute the whole Ordering Process engaging activities like payment processing, invoicing and goods delivery. Therefore, to extend the functionality of the system we need to introduce two new subsystems / services: Invoicing and Shipping, that will be hosting Invoicing Office and Shipping Office respectively.

The question arises, how to employ multiple offices to work on a business process and coordinate the workflow, so that the process is executed as defined by the business. To answer this question, we will first learn how to model a business process in a distributed system. Then we will review the implementation of a sample Ordering Process developed under the ddd-leaven-akka-v2 project. Finally we will see how Akka-DDD facilitates the implementation of business processes in a distributed system.

Business processes and SOA

To deliver a business value, an enterprise needs to perform its activities in a coordinated manner. Regardless of whether it is a production line or a decision chain, activities needs to be performed in a specific order accordingly to the rules defined by the business. The business process thus defines precisely what activities, in which order and under which conditions need to be performed, so that the desired business goal gets achieved.

The coordination of the activities implies the information exchange between the collaborators. In the past, business processes were driven by the paper documents flying back and forth between process actors. Nowadays, when more and more activities get performed by computers and machines, the business process execution is realized as a message flow between services. Unfortunately though, for a variety of reasons, the implementation of the Service-Oriented Architecture (SOA), very often ended up with a Big Ball of Mud as well as scalability and reliability issues (just to name a few) in the runtime.

The recent move towards the SOA 2.0 / Event-driven SOA / "SOA done Right" / Microservices (choose your favorite buzzword) enables delivering more light-weight, reliable / fault-tolerant and scalable SOA implementations. When it comes to modeling and executing business processes, the key realization is that since business processes are event-driven, the same events that get written to the office journals could be used to trigger the start or the continuation of business processes. If so, any business process can be implemented as an event-sourced actor (called Process Manager) assuming it gets subscribed to a single stream of events (coming from an arbitrary number of offices) it is interested in. Once an event is received, the Process Manager executes an action, usually by sending a command to an office, and updates its state by writing the event to its journal. In this way, the Process Manager coordinates the work of the offices within the process. What is important is that the logic of a business process is expressed in terms of incoming events, state transitions and outgoing commands. This seems to be a quite powerful domain specific language for describing business processes. Let's take a look at the definition of an sample Ordering Process, that is written using the DSL offered by the Akka-DDD:

Ordering Process - definition

    startWhen {

        case _: ReservationConfirmed => New

    } andThen {

        case New => {

          case ReservationConfirmed(reservationId, customerId, totalAmount) =>
            WaitingForPayment {
              ⟶ (CreateInvoice(processManagerId, reservationId, customerId, totalAmount, now()))
              ⟵ (PaymentExpired(processManagerId, reservationId)) in 3.minutes
            }
        }

        case WaitingForPayment => {

          case PaymentExpired(invoiceId, orderId) =>
              ⟶ (CancelInvoice(invoiceId, orderId))

          case OrderBilled(_, orderId, _, _) =>
            DeliveryInProgress {
              ⟶ (CloseReservation(orderId))
              ⟶ (CreateShipment(UUID(), orderId))
            }

          case OrderBillingFailed(_, orderId) =>
            Failed {
              ⟶ (CancelReservation(orderId))
            }
        }

    }

Source: OrderProcessManager.scala

An order is represented as a process that is triggered by the ReservationConfirmed event published by the Reservation Office. As soon as the order is created, the CreateInvoice command is issued to the Invoicing Office and the status of the order is changed to WaitingForPayment. If the payment succeeds (the OrderBilled event is received from the Invoicing office within 3 minutes) the CreateShipment command is issued to the Shipping Office and the status of the order is changed to DeliveryInProgress. But, if the scheduled timeout message PaymentExpired is received while the order is still not billed, the CancelInvoice command is issued to the Invoicing Office and eventually the process ends with a Failed status.

I hope you agree, that the logic of the Ordering Process is easy to grasp by looking at the code above. We simply declare a set of state transitions with associated triggering events and resulting commands. Please note that ⟵ (PaymentExpired(...)) in 3.minutes gets resolved to the following command: (⟶ ScheduleEvent(PaymentExpired(...), now + 3.minutes)) that will be issued to the specialized Scheduling Office.

The Saga pattern

As the business process participants are distributed and communicate asynchronously (just like the human actors in the real world!) the only way to deal with a failure is to incorporate it into the business process logic. If a failure happens (a command rejected by the office, a command not processed at all (office stopped), an event not received within the configured timeout), the counteraction, called compensation, must be executed. For example, the creation of an invoice is compensated by its cancellation (see the Ordering Process above). Following this rule, we break the long running conversation (the business process) into multiple smaller actions and counteractions that can be coordinated in the distributed environment without the global / distributed transaction.

This pattern for reaching the distributed consensus without a distributed transaction is called the Saga pattern and was first introduced by the Hector Garcia-Molina in the 1987. A Saga pattern can be implemented with or without the central component (coordinator) (see: Orchestration vs. Choreography). The implementation of the Ordering Process follows the Orchestration pattern - the Ordering Process is managed by an actor, that is external to all process participants.

Process Managers and the Coordination Office

The execution of the logic of a particular business process instance is handled by the Process Manager actor. The Process Manager is a stateful / event-sourced actor, just like the regular Aggregate Root actor, except it receives events instead of commands. Just like the Aggregate Root actors, Process Manager actors work in the offices. Both the Command Office (hosting Aggregate Roots) and the Coordination Office (hosting Process Managers) can be started using the OfficeFactory#office method:

    office[Scheduler]           // start Scheduling Command Office

    office[OrderProcessManager] // start Ordering Coordination Office

Source: HeadquartersApp.scala

Let the events flow

The Coordination Office is expected to correlate the received events with the business process instances using the CorrelationID meta-attribute of the event. Therefore to stream the events from the event store to a particular Coordination Office we need to create a Source (see: Reading events from a journal) emitting only these events that:

1) belong to the domain of the particular business process,
2) were assigned the CorrelationID meta-attribute.

One way to address the first requirement is to create a journal (aggregated business process journal), that aggregates the events belonging to the domain of a particular business process. Some Akka Persistence journal providers may support the automatic creation of journals that group events by tags. Unfortunately, this functionality is not yet supported by the Event Store plugin, that is used by the Akka-DDD. Luckily though, using the Event Store projection mechanism, we can create the journal of the Ordering Process by activating the following projection:

fromStreams(['$ce-Reservation', '$ce-Invoice', 'currentDeadlines-global']).
    when({
        'ecommerce.sales.ReservationConfirmed' : function(s,e) {
            linkTo('order', e);
        },
        'ecommerce.invoicing.OrderBilled' : function(s,e) {
            linkTo('order', e);
        },
        'ecommerce.invoicing.OrderBillingFailed' : function(s,e) {
            linkTo('order', e);
        },
        'ecommerce.invoicing.PaymentExpired' : function(s,e) {
            linkTo('order', e);
        }
    });

Source: order-process.js

Now, the events from the order journal need to be assigned the CorrelationID and directed to the responsible Coordination Office. What is important and challenging though, is to ensure that the events get delivered in a reliable manner.

Reliable events propagation

By reliable delivery I mean effectively-once delivery, which takes place when:

• message is delivered at least once,
• message is processed (by the destination actor) exactly-once,
• messages get processed (by the destination actor) in the order they were stored in the source journal.

Effectively-once delivery (a.k.a. The Business Handshake Pattern) can easily be accomplished if the sender keeps track of the "in delivery" messages and the receiver keeps track of the processed messages. The implementation of the pattern becomes straightforward if both the sender and the receiver are event-sourced actors.

For more detailed explanation of how the reliable delivery gets achieved, please refer to the Reliable Delivery Akka-DDD wiki page.

In our scenario, we already have the event-sourced Process Manager on the receiving side (the Coordination Office is just a transparent proxy). The missing component is the event-sourced actor on the sending side. As it is an infrastructure level component, it will automatically be created by the Akka-DDD framework, just after the Coordination Office gets started. The overall behavior of the actor that needs to be created matches the concept of the Receptor — a sensor that reacts to a signal (stimulus), transforms it and propagates (stimulus transduction). The Akka-DDD provides the implementation of the Receptor that supports reliable event propagation, including the back-pressure mechanism.

Receptor

A Receptor gets created by the factory method based on the provided configuration object. The configuration can be built using the simple Receptor DSL.

A Receptor actor is a durable subscriber. During the initialization, it is subscribing (by itself) to the event journal of the business entity that was provided as the stimuliSource in the configuration object. After the event has been received and stored in the receptor journal, the transformation function gets called and the result gets sent to the configured receiver. If the receiver address is to be obtained from an event, that gets propagated, then the receiverResolver function should be provided in the configuration.

One might question the fact that the same events get written twice into the event store (first time to the office journal, second time to the receptor journal). I would like to clarify, that in fact this does not happen. The receptor is by default configured to use an in-memory journal and the only messages that get persisted are the snapshots of the receptor state. The snapshots get written to the snapshot store on a regular basis (every n events, where n is configurable) and contain the messages awaiting the delivery receipt.

Coordination Office Receptor

Having learned how to build a receptor, it should be easy to understand the behavior of the Coordination Office receptor by examining its configuration. As we can see, the receptor reacts to the events, coming from the aggregated process journal, adds the CorrelationID meta-attribute to the event message and propagates the event message to the Coordination Office representative actor. The name of the aggregated process journal and the CorrelationID resolver function get retrieved from the ProcessConfig object — an implicit parameter of the office factory method.

To summarize, the Coordination Office receptor gets automatically created by the Akka-DDD framework, based on the configuration of the business process.

So, let's take a look at the OrderProcessConfiguration:

  implicit object OrderProcessConfig extends ProcessConfig[OrderProcessManager]("order", department) {
    def correlationIdResolver = {
      case ReservationConfirmed(reservationId, _, _) => reservationId // orderId
      case OrderBilled(_, orderId, _, _) => orderId
      case OrderBillingFailed(_, orderId) => orderId
      case PaymentExpired(_, orderId) => orderId
    }
  }

The aggregated process journal is given the name: order. The correlationIdResolver function returns the CorrelationID from the orderId attribute of the event, for all events, except the ReservationConfirmed. For the ReservationConfirmed event, the CorrelationID / order ID must be generated, because the Ordering Process is not yet started, while the ReservationConfirmed event is being processed.

Message flow - the complete picture

After the initial event has been received and processed by the Process Manager, the business process instance gets started. The business process will continue, driven by the events. The events will be emitted as soon as the commands, issued by the Process Manager, get processed in the responsible Command Offices.

The following diagram visualizes the flow of the commands and events within the system that occurs when the business process is running.

Business process journal

Before reacting to an event, the Process Manager writes the event to its journal. The journal of a Process Manager is de facto a journal of a business process instance - it keeps the events related to particular business process instance in order they were processed by the Process Manager.

Ordering Process - an alternative implementation (Event Choreography)

Instead of giving the responsibility for the business process execution to the single, external business entity (the Ordering Process is managed by a Process Manager, operating in the Headquarters subsystem), we could let the process emerge by allowing more direct communication between the offices. For example, to send the CreateShipment command to the Shipping Office in reaction to the OrderBilled event coming from the Invoicing Office we could register a simple Receptor in the Shipping subsystem. In the same way, we could implement other interactions required by the Ordering Process and thus make the Headquarters subsystem obsolete.

The overall event flow could be modeled as presented on the following diagram:

We can find this alternative approach to the implementation of the Ordering Process in the previous version of the ddd-leaven-akka-v2 project. The Headquarters subsystem did not exist back then. In the Shipping subsystem, you can find the implementation of the Payment Receptor, that I described above.

Summary

Once we start modeling the interactions that shape our business domain, using the language of the commands and the events, we are on a good way towards creating a model of a system that is simple to understand by the business people and also simple to express in the code.

Although, we could implement the event-centric model on the basis of the monolith architecture, we should not be afraid of the distributed architecture, especially if we want the system to be scalable and fault-tolerant.

As I demonstrated above, the coordination of the activities between the event-sourced business entities can be performed in the distributed environment efficiently and reliably.

Reactive DDD with Akka - integrating the Event Store

Introduction

It has been a while since I wrote the last episode in my series: "Reactive DDD with Akka". In that time, in 2015, I managed to release two new projects:

• Akka-DDD - project that contains reusable artifacts for building applications on top of the Akka platform, following CQRS/DDDD-based approach,
• ddd-leaven-akka-v2 - follow-up project of the ddd-leaven-akka that makes usage of Akka-DDD artifacts

As both projects are in good shape now, it is high time for me to resume the series and hopefully get more feedback from the community. But first, let's recall what we have learned so far.

Previously we discovered that the Akka with its Akka Persistence module provided a solid platform for building micro-services, using DDD/CQRS design principles and patterns. We learned:

• how to implement event-sourced Aggregate Roots as persistent actors (aka clerks) [1],
• how the command dispatching is performed by offices in the standalone and the distributed environment and
• how the events produced by the Aggregate Roots can be transmitted reliably to other actors.

[1] Please see Don't call me, call my office for explanation of the office & clerk concept

Further, I noticed the lack of support of the query side of the system in the Akka Persistence, and we found out how to overcome this limitation by using an external message broker. We also noticed that if we used the Event Store as an underlying event store, we could reliably feed the query side of the system by executing the Event Store Projections (thus avoiding the necessity of introducing a message broker into the system). In my view this idea was so interesting that I decided to give it a try and that's why I started the Akka-DDD project.

In this episode we will learn how to create two kinds of journals using the Event Store Projections mechanism:

• office journals - containing events emitted by a concrete office
• aggregated business process journals - containing events related to a concrete business process

From the point of view of the business application requirements, these two kinds of journals are much more useful than the journals of single Aggregate Roots (the clerk journals). We will learn how the specialized services such as View Updaters and Receptors can subscribe to these journals using the Event Store JVM Client.

Recently a new version of the Akka Persistence has been released that contains Query Support on its feature list. So consequently, do we still need to use the Event Store API directly? We will answer this question at the end.

However, now, we will begin with introduction to the Event Store (as a component that we want to integrate with the Akka-DDD) by looking at it from the two opposite sides:

• 'the write side' - the Event Store as a journal provider
• 'the read side' - the Event Store as the event bus

The Event Store as a journal provider

The clerk's journal

The EventStore Akka Persistence is a storage plug-in (journal provider) implemented for the Event Store. Under the hood the plug-in uses Event Store JVM Client. When a persistent actor requests the persisting of an event message, the journal tells the Event Store to write the given event message to a stream with a streamId equal to the actor's persistenceId. If a stream with the given ID does not exist, it is automatically created by the Event Store. The Akka-DDD defines persistenceId of a persistent actor as a concatenation of officeId and clerkId (separated by a dash), where officeId must be implicitly provided for each Aggregate Root class (via OfficeId type class) and clerkId is defined as actor's ID [2]. The [officeId]-[clerkId] is thus the ID of a stream in the Event Store being a journal of a clerk (identified by clerkId) within an office (identified by officeId).

[2] The ID of a persistent and sharded actor is extracted from the first command message it receives.

The journal entry format

The Event Store requires the following data to create a journal entry:

• EventId - used internally by the Event Store for idempotency
• EventType - an arbitrary string - commonly used for events selection (see Projections)
• Data - actual data in the serialized form (according to the ContentType)
• ContentType - for instance json, binary, etc
• Metadata (optional) - additional data associated with the entry in a serialized form (according to the MetadataContentType)
• MetadataContentType (optional) - for instance json, binary, etc.

Transforming an event to a journal entry

Before a domain event is written to an actor's journal, it is first wrapped by the Akka-DDD in an EventMessage envelope (see: AggregateRoot#raise) and then the EventMessage is wrapped by the Akka in a PersistentRepr envelope. Eventually the journal plug-in is executed to store the PersistentRepr in the actor's journal.

Event (Domain) → EventMessage (Akka-DDD) → PersistentRepr (Akka) → Journal entry (EventStore plug-in)

The EventMessage envelope allows adding an arbitrary number of meta attributes. The Akka-DDD takes care of handling the following meta attributes:

• id - the application level message ID (independent from the internal ID used by the Event Store)
• timestamp - the record of the time the event was created at
• causationId - the application level ID of some other message, that caused this message (ie. commandId for the events raised by the clerks)
• correlationId (optional) - the application level ID of a business process, that the event message is associated with
• _deliveryId (optional) - it is used between actors that communicate using At-Least-Once delivery semantics

The actual serialization of the PersistentRepr to the journal entry format is performed by a specialized serializer that is automatically registered by the Akka-DDD. The serializer uses json format taking advantage of the fact that the Event Store natively supports json. The EventType attribute is set to a name of an event class. The serializer also takes care of the serializing of the metadata defined in the EventMessage. Here is an example of a journal entry for a ReservationCreated event, read from Reservation-57d868 stream (office id: Reservation, clerk id: 57d868):



As you can see, although the Data element contains a json representation of an object instance of a PersistentImpl class (a class implementing a PersistentRepr trait), the EventType element contains a value that refers to the actual event. The event itself is stored under the payload attribute of the PersistentImpl object.

Please notice also the sequenceNr attribute of the PersistentImpl. It is generated by the Akka and represents a position of the entry in the actor's journal. Since the sequenceNr is available only after an event message has been stored, the Akka-DDD distinguishes between the EventMessage - the event message to be stored in the journal and the OfficeEventMessage - the event message fetched from the journal.

The Event Store as the event bus

In general, the event bus is a layer that allows the 'publish - subscribe' style communication between components without requiring the components to explicitly register with one another. So far we have discussed how events get published (written to the journals) from 'the write side' of the system. Now it is a time to describe the event subscribers - the actors that want to be notified about the published events. There are two types of the event subscribers available in the Akka-DDD:

• View Updaters - responsible for the updating of 'the read side' of the system. They are interested in the events from the particular office journal.
• Receptors - responsible for the event-driven interaction between the subsystems (event choreography), including long-running processes (sagas). They are interested in the events from a particular office journal or a particular aggregated business process journal.

For the next paragraph, let's leave the question how the office journals and the aggregated business process journals get created. Let's assume, they can be created.

Knowing the ID of a stream, we can use the Event Store API to register a subscriber, interested in getting events from that stream. A subscription can be defined as a 'live-only' (only new events get pushed to the subscriber) or a 'catch-up' one. A 'catch-up' subscription works in a very similar way to a 'live-only' subscription, with one notable difference: subscriber specifies the position, from which events will get pushed. A 'catch-up' subscription thus allows creating the durable subscribers. Such subscribers can resume processing the events as long as they are able to record position of the last processed event. The subscribers resume the processing after they were stopped or terminated and then restarted e.g. as the result of system crash. In the next episode, we will learn how to implement the View Updaters and the Receptors using the 'catch-up' subscriptions.

But now let's see how we can create new streams using the Event Store Projections.

Creating streams using the Event Store Projections

The Event Store is able to execute the built-in or a user defined projection - a chunk of javascript code containing the following elements:

• The identifier(s) of input event stream(s)
• The event(s) selection(s)
• The function(s) that accepts an event and a state as a parameter. The function can call the linkTo(streamId, event) function to write the input event into an arbitrary stream or it can call the emit(streamId, eventType, event) function to emit a new event into an arbitrary stream.

A projection will be stopped automatically after all historical events are processed unless it is started in the continuous mode. If so, also the new events will be processed as they are added to the input stream(s).

Let's see an example. The projection below will watch the built-in $stats-[ip:host] stream containing low level system statistics for events of the type: $statsCollected and will emit a new event (of the type: heavyCpuFound) to the new heavycpu stream whenever a value of the sys-cpu statistic, read from the caught event, would exceed 40:

Creating an office journal

We learned that the stream with ID: Reservation-57d868 represents a journal of a clerk 57d868 working in the Reservation office. Now we want to create a journal that contains the events emitted by all clerks working in the Reservation office. To accomplish this, we need to use the $by_category system (built-in) projection. It turns out that the Event Store is able to extract a category of a stream from its ID (treating dash (-) as a category separator). The $by_category projection, once enabled (all system projections are disabled by default), will detect the Reservation category and will create a $ce-Reservation journal for the Reservation office automatically. Similarly it will create appropriate office journals for all the other offices already existing in the system or created in the future (all the system projections are running in continuous mode so we don't need to restart them in the future anymore).

Creating an aggregated business process journal

Now, when we know how to create the office journals, we can use them as input streams for the aggregated business process journals. For example let's create an invoicing journal that will contain all the events related to the invoicing business process:



This time we have defined journals of the two offices (Reservation and Invoice) as input streams. Then for each type of an event, relevant to the invoicing business process, we have defined a function that simply "inserts" the original event into an invoicing stream [3].

[3] In fact, when using linkTo function, the event inserted into the output stream is not the original event (or its copy), but a special link event containing only a pointer to the original event.

Please notice that the invoicing stream does not represent a journal of a concrete instance of the invoicing business process (an invoicing process for a concrete customer/order). Once we learn about the Coordination Office (in the upcoming episode in the series) we will also learn how a special Receptor takes care of forwarding the events, read from the invoicing stream to the Invoicing Coordination Office that in turn forwards/routes them to the concrete clerks responsible for the management of the single business process instances. The clerks then decide which events to store in their own journals - the journals representing the single process instances.

Reading events from a journal

It is a time to learn how to implement a durable event subscriber using the Akka-DDD framework. The trait we will need to use is EventSourceProvider located in the eventstore-akka-persistence module. The trait exposes the eventSource(esConnection, observable, fromPositionExclusive): Source[EventMessageRecord, Unit] method. As the method's signature suggests, it accepts some observable object (next to the Event Store connection object and the start position) and returns an object that is a source of the EventMessageEntry objects. The Source class is Akka's representation of Publisher as defined by the Reactive Streams standard.

The eventSource method takes an observable BusinessEntity and obtains streamId from it by calling the StreamIdResolver. The StreamIdResolver knows how to resolve a streamId regardless whether the given entity is a clerk, an office or a coordination office. The method then uses the obtained streamId to create a Publisher by calling the streamPublisher(streamId, position, ...) method provided by the EventStore JVM Client Reactive Streams API. Finally the method converts the Publisher object to a Source object that is instructed to emit the event messages (discussed previously) wrapped into an EventMessageEntry envelope.

The Akka-DDD makes use of the EventSourceProvider trait to implement two types of the durable subscribers: the View Update Service and the Receptor. We will not dive into the implementation details of these services in this article, but as you can imagine, the stream processing is the preferred pattern used there.

The Akka Persistence Query

As stated in the docs, since version 2.4, the Akka Persistence provides a universal asynchronous stream based query interface that various journal plug-ins can implement in order to expose their query capabilities. The interface exposes the ReadJournal trait family that provides two groups of methods for reading events from the journal:

The methods that return a source, that is emitting the historical events:

• currentEventsByPersistenceId(id)
• currentEventsByTag(tag)

The methods that return a "live" source, that is emitting both the historical and the upcoming events:

• eventsByPersistenceId(id)
• eventsByTag(tag)

The journal plug-ins are not obliged to support all types of queries so they must explicitly document which types of queries they support.

As you can see, the interface supports not only queries for events from a single journal but also queries for "tagged" events from an arbitrary number of journals.

Some journal plug-ins may support the EventsByTag queries out of the box by requiring events to be wrapped in a akka.persistence.journal.Tagged before they get written to the journal (Such a wrapping could be implemented using Event Adapters). Other plug-ins may treat tags as identifiers of the arbitrary event journals such as office journals or business process journals. These journals could be managed externally (for example using the Projections in case of the Event Store (as we have seen above)).

Going back to the Akka-DDD, would it be possible to use the Akka Persistence Query instead of the EventStore JVM Client and thus to gain more interoperability? Well, currently this is not possible, because the Eventstore Akka Persistence plug-in supports only the queries for the events from a single journal (the EventsByTag queries are not supported). So the following code will not work unfortunately:

Conclusion

Although the Akka Query is marked as "experimental" and the Event Store Projections are still in "Beta" version, I think, they both are worth considering when thinking about developing a new system, that is implementing the DDD/EDA/CQRS (or Microservices, if you like) architecture. Being able to easily create the arbitrary streams of events to which the interested actors can subscribe using the standard protocol (Reactive Streams) is great when heading for a loose coupling and reactiveness.

Reactive DDD with Akka - lesson 3 (Projections)

Don't miss: ddd-leaven-akka-v2 - runnable sample e-commerce application built from Akka-DDD artifacts described in this series.

Let's continue our adventure with Akka and DDD. So far we have been concentrated on write/command side of the system. We have covered topics like command processing, event storage, sharding etc. but haven't touched the other side of the system: the query side. Our goal of this lesson is to implement necessary infrastructure supporting creation of components responsible for feeding query-side database: projections. For query side we will choose sql database although there are other technologies worth considering (document or graph databases might be a better choice depending on requirements).

Our first task is to check if Akka (akka-persistence in current, still experimental version to be precise) supports concept of projections. And as it turns out, Akka provides View trait that defines an actor capable of receiving events from journal of particular processor (aggregate root - AR) instance. This sounds good but unfortunately the concept of View is insufficient for building projections that typically listen to aggregated stream of events rather then events of particular AR instance. A workaround would be to register a view for a processor that receives events from all ARs of given type, but in such scenario, events would be persisted twice and reliable event delivery (more on that later) would had to be ensured between ARs and aggregating processor.

What we should require from "ideal" event store implementation is a dsl for event streams transformation and aggregation.

EventStore should be mentioned here as it provides impressive set of methods for handling streams of events: Event Store as a Read Model including stream aggregation (per category/AR type) . You can configure EventStore Journal for Akka Persistence and try experiment with projections (be prepared to write some javascript and json).

But what if we don't want to depend on any particular journal provider? Well, composing streams of events would be possible if journals were exposed through some functional (composable) interface rather than View. And this is one of the goals already established by akka team with introduction of akka-stream module. Akka-stream is implementation of recently announced standard for asynchronous, composable and non-blocking stream processing on JVM: http://www.reactive-streams.org/ that models stream publishers and subscribers (as defined by the specification) as akka actors. It also already provides simple dsl (PersistentFlow) for creating event stream publishers that are backed by Views. As you can see here  event streams can be aggregated by merging stream publishers (although aggregation by category is not yet supported). Once the akka-streams is released we should definitely make use of it and see how to implement projections as reactive components (subscribers of event streams). For now we need to find a different solution. The only possibility left is to introduce third-party broker and forward events to topics or queues configured within the broker. As we will see in following section, integrating akka with external broker is much simpler that one could imagine. This approach will require enriching AR with Publisher component (trait) responsible for sending persisted events to configured target actor. This is a perfect opportunity to learn more about message delivery in distributed systems.

Reliable event delivery

Sending events generated by AR to any other actor (being a broker gateway or aggregating processor (already mentioned before)) should be straightforward to implement, right? Yes, it is, if you accept that in case of node crash some messages might be lost (despite the fact that corresponding commands had been acknowledged to the client!). Akka does not guarantee that message will be delivered (this is so called at-most-once delivery semantics) because reliable delivery does not come for free in distributed systems. To obtain at-least-once delivery semantics (message is guaranteed to be delivered, but duplicates may occur) acknowledgment protocol must be used. Sender must store the message before sending and receiver must acknowledge the message once it is received.  If acknowledgment is received, sender must mark previously stored message as confirmed. If acknowledgment is not received within configured time period or sender is restarted (for example as the result of crash) sender retries delivery. Because acknowledgment message might be lost as well, sender might resend message already delivered. Akka-persistence provides at-least-once delivery semantics in its core. Journal plugin api defines methods for storing confirmation entries for messaged that have been acknowledged. Acknowledgment and retry mechanism is encapsulated within Channel component that must be used to send events from within processor (AR) to destination actor. Events must be wrapped in Persistent envelop so that receiver can acknowledge the message by simply calling Persistent.acknowledge method.

ReliablePublisher

Thanks to traits composability we can put all the behavior required by reliable AR (sender) to separate component: ReliablePublisher. It extends from abstract EventPublisher that in turns extends from abstract EventHandler. We mix in AggregateRoot with EventHandler making event handling in context of AR configurable as much as possible (useful for testing purposes). Please notice that it is possible to stop redelivery when configured number of retries is exceeded. For that purpose RedeliverFailureListener has been registered on the channel to be notified if redelivery fails. The listener actor throws RedeliveryFailedException exception that results in restart of parent actor (AR) (to make it work supervisorStrategy of listener actor had to be adjusted). Inside preRestart method (that RedeliveryPublisher overrides) we can trigger compensation action and/or mark failing event as deleted (to prevent continuation of redelivery after AR is restarted).

To provide destination for ReliablePublisher its abstract member target must be defined on creation time (AggregateRootActorFactory should take care of this as shown here: ProductPublishingSpec). Finally we should verify if ReliablePublisher works as expected and is really reliable. You can find test here: ReliablePublisherSpec.

Message exchange over queue

Now we will configure infrastructure for events transmission over durable queue. We will use Apache Camel to arrange in-only message exchange  between a component writing to the queue (producer) and projection component reading from the queue (consumer). Thanks to akka-camel both components (producer and consumer) can be represented as ... (surprise! ;-) actors. EventMessageConfirmableProducer is the producer that receives event messages (notice that Events are published inside EventMessage envelope) coming from ReliablePublisher (AR) and forwards them to the configured camel endpoint (for example jms queue or topic) (see transformOutgoingMessage method). Once event message is accepted by the queue (and persisted if the queue is durable) producer acknowledges event reception to the publisher (ReliablePublisher) (see routeResponse method). Please notice that we choosed to unwrap EventMessage from ConfirmablePersistent envelope before putting it into the queue (so that consumer does not have to do unwrapping itself). ConfirmablePersistent still needs to be attached to the EventMessage so we convert it to meta attribute.

Finally we can implement projection as Consumer actor, an actor that consumes messages from camel endpoint. Projection actor simply applies provided projection specification (ProjectionSpec) to received event and finalizes message exchange by either sending acknowledgment or failure. To prevent processing of duplicated events concrete implementation of ProjectionSpec must return sequence number of last processed event for given aggregareId (see currentVersion method):

Before pulling all the pieces together we need to register concrete broker as component of Camel. We will use ActiveMQ. Configuration of the ActiveMQ component is straightforward (see ActiveMQMessaging trait) And of course we need a runner class (EmbeddedActiveMQRunner) that starts the broker as embedded service.

Implementing projection 

Now we can implement concrete projection specification. Let's model ProductCatalog service inside Sales module (context) that maintains list of products with price. Whenever new Product is added to inventory (within Inventory module/context) it should also be added to product catalog in Sales module. Thus we need projection that listens to InventoryQueue and updates product catalog. We will skip implementation details of ProductCatalog as accessing relational db is rather not very interesting topic (we use Slick for that purpose). InventoryProjection just calls insert method of ProductCatalog providing Product data copied from the ProductAdded event and empty price.

An integration test is available here: EventsPublishingClusterSpec . From first node we send two AddProduct commands. Commands are handled by Product AR within inventory context and generated events are published (reliably) to InventoryQueue that is available from all nodes in the cluster. On the second node we register InvetoryProjection and wait for 1 second (so that published events have time to reach the queue) and then send GetProduct query message to ProductFinder to check if expected product has been added to product catalog.

And surprisingly test succeeds ;)!


Last but not least, I added experimental feature to our application:
If command sender wants to be notified once the view has been updated he can request special delivery receipt!

Please see ProductAcknowledgedPublicationSpec:

     

As homework you can find out how the feature is implemented and let me know your opinion about the idea :)

Reactive DDD with Akka

Don't miss: ddd-leaven-akka-v2 - runnable sample e-commerce application built from Akka-DDD artifacts described in this series.

When comparing traditional and DDD/CQRS architectures (see my previous article) I said that the goal of DDD model is to decompose complex business domain into manageable pieces taking into account scalability and consistency requirements. What it means is that by bringing concepts like bounded contexts, transaction boundaries and event based communication DDD and CQRS are great enablers for building scalable software. But so far, example services I have presented, were supposed to be run on top of relational database and within global transactions. This is very limiting architecture not suited for building scalable software. Continuous consistency of underlying data that is guaranteed by global transactions should not be perceived as standard requirement of any (including enterprise-class) system. It's an artificial requirement that we all got used to but does not addresses real requirements of the customer. To fully benefit from DDD/CQRS architecture we should change the underlying technology. Today we have a choice. There is a lot of NoSQL databases and there are a few platforms that address scalability as first concern. For JVM, Akka (part of Typesafe platform) is the most robust open-sourced platform for building event-driven applications. Recently akka-persistence module has been released that takes care of handling long-running/persistable processes (this is what Aggregate Roots and Sagas are all about). This is a great feature that allows thinking of Akka as complete platform for building enterprise applications.

Lets then start building event-driven, scalable, resilient and responsive (in short reactive) application using Akka and other goodies from Typesafe platform.
I have already started a project on Github. You are welcome to contribute!

Below is the first lesson I learned from the project and wanted to share it with you. Hopefully more lessons will come (see open issues).

Lesson 1 - Aggregate root is an actor

The source code for lesson 1 is available here: https://github.com/pawelkaczor/ddd-leaven-akka/tree/Lesson1

The goal of lesson 1 is to learn how to build event sourced Aggregate Root (AR) with Akka.
The idea is simple. Aggregate Root should be modeled as stateful actor that accepts Commands and produces Events. Because actor is message driven, we can send Command messages directly to Aggregate Root avoiding "command to method call" transformation.

As already mentioned akka-persistence provides necessary artifacts for building persistable/stateful actors. The key component is akka.persistence.Processor trait. Processor is an actor capable of restoring its state (aka recovering) during reincarnation (start or restart of the actor). The type of underlying storage is append-only pluggable journal.

Command sourcing

Any message of type Persistent that comes to a processor is stored in a journal before it is processed. During recovery, persistent messages are replayed to the processor so that it can restore internal state from these messages.
This pattern (called Command sourcing) is not particularly applicable for Aggregate Roots because replying of command that has not yet been validated is not desired.

Event sourcing

To build AR we need to extend from EventsourcedProcessor that adds event sourcing capability (Eventsourced trait) to Processor trait - only produced events will be stored in the journal. This means we need to explicitly invoke persist(event) method of Eventsourced trait to store produced event in the journal after command message has been validated (by validation I mean ensuring AR's invariants will not be compromised by the command). Since persist method persists events asynchronously (does not block the current thread) it accepts a callback (event handler) as the second argument. Main responsibility of event sourced AR is to provide event handler that will update internal state of AR and handle the event by publishing it and/or sending a response to the client/command sender. Handling of event should be customizable.

AggregateRoot trait

Let's see how to build abstract event sourced AggregateRoot class.

Abstract AggregateRoot keeps state using private variable member of type AggregateState (abstract) and takes care of updating this variable whenever an event is produced/raised (raise method) or replayed (receiveRecovery method). State itself (concrete implementation of AbstractState) should be immutable class implementing method apply that defines state transitions for each event (except initialization). Initialization of the state is performed by AggregateRootFactory - the abstract member of AR that must be overridden in concrete implementation of AR. Initialization is event-driven as well which means that AggregateRootFactory creates initial state from an event. To complete the picture, the raise(event) method calls persist method and, after event is persisted, it either calls default handler or handler provided as the second (optional) argument of the raise method. Default handler publishes an event to event bus (provided by Akka) and sends Acknowledged message back to the sender.

Reservation AR

Please take a look at implementation of concrete Aggregate Root (Reservation). The code should be self explanatory. Command processing consists of validation and raising an event.

ReservationSpec verifies if Reservation AR is in fact stateful component, capable of handling reservation process. The test just simply sends several commands to Reservation AR in valid order and verifies if expected events have been persisted. In the middle of the process Reservation actor is restarted to verify if it preserves the state. And in fact it is since subsequent commands are handled successfully.

Errors handling

By default if any exception of type java.lang.Exception is thrown by the actor the actor is restarted by its supervisor (this is defined in default SupervisionStrategy). Exceptions are not propagated to the command sender automatically as you might expect. We can either catch exception and send them back to the sender from within receiveCommand method or send the exception from within preRestart method that takes exception as reason argument. Overriding preRestart method seems to be a simpler approach. Now we can test if exceptions are returned to the sender: ReservationFailuresSpec

In next lesson...

Currently the client needs to get a reference to particular instance of Aggregate Root before sending the command. It would be much easier for him if he could just send the command to some command gateway. This will be the topic of the next lesson.