Processes

Search…
Processes
XOOM LATTICE long-running process orchestrations, aka Sagas, with guided steps to results.
Processes can be used to manage complex, coarse-grained, long-running business transactions. Each step in the process is orchestrated until a given outcome is reached, which may be a successful completion.
Some processes may never end in that they may repeat indefinitely, producing one or more smaller subsets of successful outcomes with each iteration.
Because such a process runs longer than for a single transaction and may spread across any number of Bounded Contexts (i.e. microservices), and because the process may at some point be temporarily evicted from memory, the process itself must be persistent. The XOOM LATTICE tools assist in managing such processes, whether by means of object persistence, stateful command models, or sourced with events and commands.
Note that a long-running process is often referred to as Saga. Even so, we here acknowledge that this use of the name Saga to refer to such a process is not typically according to the original pattern definition, although it may be designed as such. Primarily this would be accomplished by emulating a rollback by applying compensating transactions to the point where any previous transactions are essentially undone. This is known as backward recovery, but there is also a second choice known as forward recovery, and the two may be combined.
How Processes Work
As a general rule, a process is informed of an outcome, and in reaction to that outcome causes another action to be taken. This will repeat as the process is informed of the outcome of the previous action that it caused, and then cause yet another action to be taken.
Typically a DomainEvent will trigger the process to start. The process will then emit a Command, which is to cause the next step in the process. When the DomainEvent from that step outcome is seen, process will again emit a Command, which is to cause the next step in the process. This repeats until the process has completed (if in fact it is meant to complete). The process tracks the current state of the state machine, which along with the next DomainEvent it sees, will determine the next step to be taken.
A process that spans three different Bounded Contexts.
The above diagram shows how the process starts by a user submitting a command and then proceeds through five total steps. Note that a DomainEvent is seen by the process as a stimulus, and the process then emits a Command to cause the next action. This pattern repeats until completion is reached on the far right.
Process Types
A process is a finite state machine. XOOM LATTICE provides a protocol for processes, the Process.
The Process and three abstract types, each corresponding to a model entity type.
There are three Process standard implementors that are abstract base types. These base classes are meant to be extended by concrete implementors.
Type
Description
SourcedProcess
Uses Command Sourcing and/or Event Sourcing to maintain the state of the process and to direct next steps.
ObjectProcess
Use a persistent object-relational mapped object to maintain the state of the state machine. The extender must still emit commands or events to direct next steps.
StatefulProcess
Use a persistent key-value to maintain the state of the state machine. The extender must still emit commands or events to direct next steps.
These are all documented in detail below.
Especially in the case of implementing a Process, the SourcedProcess is generally the best choice. The use of Command and/or DomainEvent types to direct the steps of the process are a natural fit. You may still, however, accomplish the same by using Command and/or DomainEvent types within an ObjectProcess and StatefulProcess. It may be more a matter of whether you prefer to always persist state snapshots, which you must always do with these two process types.
Process Protocol
The protocol for processes is defined in XOOM LATTICE.
1
package io.vlingo.lattice.model.process;
2
​
3
import java.util.List;
4
import java.util.function.Supplier;
5
​
6
import io.vlingo.common.Completes;
7
import io.vlingo.lattice.model.Command;
8
import io.vlingo.lattice.model.DomainEvent;
9
import io.vlingo.symbio.Source;
10
​
11
public interface Process<T> {
12
​
13
  Chronicle<T> chronicle();
14
​
15
  String id();
16
​
17
  void process(final Command command);
18
​
19
  <R> Completes<R> process(final Command command, final Supplier<R> andThen);
20
​
21
  void process(final DomainEvent event);
22
​
23
  <R> Completes<R> process(final DomainEvent event, final Supplier<R> andThen);
24
​
25
  <C> void processAll(final List<Source<C>> sources);
26
​
27
  <C,R> Completes<R> processAll(final List<Source<C>> sources, final Supplier<R> andThen);
28
​
29
  void send(final Command command);
30
​
31
  void send(final DomainEvent event);
32
}
Copied!
As expected with the VLINGO XOOM platform, the protocol is quite simple. The operations work as described in the following subsections.
Chronicle<T> chronicle()
Returns the current state of the process as a Chronicle<S>, where S is the type of the state. Since a Process is a state machine, the current state held by the Chronicle<S> is the state of the state machine.
1
public class Chronicle<S> {
2
  public final S state;
3
​
4
  public Chronicle(final S state) {
5
    this.state = state;
6
  }
7
​
8
  public Chronicle<S> transitionTo(final S state) {
9
    return new Chronicle<>(state);
10
  }
11
}
Copied!
The API makes no assumptions that the full state itself will be persisted, but makes it possible. Since the Process may be CommandSourced or EventSourced, or a combination of both, it could well be that there is no need to use the Chronicle<S> for full-state persistence. In such cases the Process may require minimal state transitions, and thus any state-representing stream may not be large enough to justify the use of snapshots.
You use transitionTo() to create a new Chronicle<S> that wraps the new state of the state machine.
It is meant for you to extend the Chronicle<S> to manage your more fine-grained state transitions.
String id()
Returns the global unique identity of the Process, which should likely be the id() or streamName() of the underlying base types: SourcedProcess, ObjectProcess, StatefulProcess.
void process(Command command)
Wraps the Command inside a ProcessMessage and persists it to the underlying storage, and then sends it through the configured message exchange.
<R> Completes<R> process(Command command, Supplier<R> andThen)
Performs as does process(Command), and also returns the Completes<R> that provides the eventual outcome of the Supplier<R> parameter andThen. After the Command is safely persisted, the andThen is executed and its value is given as outcome of the Completes<R>.
void process(DomainEvent event)
Wraps the DomainEvent inside a ProcessMessage and persists it to the underlying storage, and then sends it through the configured message exchange.
<R> Completes<R> process(DomainEvent event, Supplier<R> andThen)
Performs as does process(DomainEvent), and also returns the Completes<R> that provides the eventual outcome of the Supplier<R> parameter andThen. After the DomainEvent is safely persisted, the andThen is executed and its value is given as outcome of the Completes<R>.
<C> void processAll(List<Source<C>> sources)
Wraps each of the Source<C> instances inside its own ProcessMessage and persists all of them to the underlying storage, and then sends each through the configured message exchange.
<C,R> Completes<R> processAll(List<Source<C>> sources, Supplier<R> andThen)
Performs as does processAll(List<Source<C>>), and also returns the Completes<R> that provides the eventual outcome of the Supplier<R> parameter andThen. After all Source<C> instances are safely persisted, the andThen is executed and its value is given as outcome of the Completes<R>.
void send(Command command)
Sends the Command through the configured message exchange without first persisting it.
void send(DomainEvent event)
Sends the DomainEvent through the configured message exchange without first persisting it.
Using the Process Types
The abstract base Process types—SourcedProcess, ObjectProcess, and StatefulProcess—are described in the following subsections.
SourcedProcess
To use a Command Sourced and/or Event Sourced Process type, extend the SourcedProcess. Recall from the above diagram that generally a process is stimulated by a DomainEvent and then causes the next action by emitting a Command. The fact that Command instances are persisted to represent the process state, this makes the process Command Sourced, as in sourced by commands.
1
public class ManagedOrderProcess extends SourcedProcess<OrderState> implements OrderProcess {
2
  private OrderProcessState state;
3
  
4
  public ManagedOrderProcess() {
5
    super(); // uses address id as streamName
6
  }
7
  
8
  @Override
9
  public void orderPlaced(Order order) {
10
    state = OrderProcessState.with(order);
11
    process(new AuthorizePayment(state.order.paymentInfo));
12
  }
13
  
14
  @Override
15
  public void paymentAuthorized(PaymentAuthorization paymentAuthorization) {
16
    state = state.with(paymentAuthorization);
17
    process(new PickItems(state.order.items));
18
  }
19
  
20
  ...
21
}
Copied!
Note that the DomainEvent instances are not received directly by the process. Instead an exchange listener receives the DomainEvent, looks up the process, and dispatches one of the process protocol messages, such as orderPlaced() and paymentAuthorized().
StatefulProcess
The StatefulProcess is very similar to the SourcedProcess, except that you must implement a few overrides necessary for the StatefulProcess and underlying StatefulEntity abstract base type.
1
public class ManagedOrderProcess extends StatefulProcess<OrderState> implements OrderProcess {
2
  private Chronicle<OrderProcessState> chronicle;
3
  
4
  public ManagedOrderProcess() {
5
    super(); // uses address id as streamName
6
  }
7
  
8
  @Override
9
  public Chronicle<OrderProcessState> chronicle() {
10
    return chronicle;
11
  }
12
  
13
  @Override
14
  public void orderPlaced(Order order) {
15
    state = OrderProcessState.with(order);
16
    process(new AuthorizePayment(state.order.paymentInfo));
17
  }
18
  
19
  @Override
20
  public void paymentAuthorized(PaymentAuthorization paymentAuthorization) {
21
    state = state.with(paymentAuthorization);
22
    process(new PickItems(state.order.items));
23
  }
24
​
25
  ...
26
​
27
  @Override
28
  protected void state(final OrderProcessState state) {
29
    this.chronicle = chronicle.transitionTo(state);
30
  }
31
​
32
  @Override
33
  protected Class<OrderProcessState> stateType() {
34
    return OrderProcessState.class;
35
  }
36
}
Copied!
The chronicle() is required by the StatefulProcess, and the state() and stateType() are required by the StatefulEntity.
ObjectProcess
The ObjectProcess is very similar to the SourcedProcess, except that you must implement a few overrides necessary for the ObjectProcess and underlying ObjectEntity abstract base type.
1
public class ManagedOrderProcess extends ObjectProcess<OrderState> implements OrderProcess {
2
  private Chronicle<OrderProcessState> chronicle;
3
  
4
  public ManagedOrderProcess() {
5
    super(); // uses address id as streamName
6
  }
7
  
8
  @Override
9
  public Chronicle<OrderProcessState> chronicle() {
10
    return chronicle;
11
  }
12
  
13
  @Override
14
  public void orderPlaced(Order order) {
15
    chronicle = new Chronicle(OrderProcessState.with(order));
16
    process(new AuthorizePayment(state.order.paymentInfo));
17
  }
18
  
19
  @Override
20
  public void paymentAuthorized(PaymentAuthorization paymentAuthorization) {
21
    chronicle.state.setPaymentAuthorization(paymentAuthorization);
22
    process(new PickItems(state.order.items));
23
  }
24
​
25
  ...
26
​
27
  @Override
28
  protected OrderProcessState stateObject() {
29
    return chronicle.state;
30
  }
31
​
32
  @Override
33
  protected void state(final OrderProcessState state) {
34
    this.chronicle = chronicle.transitionTo(state);
35
  }
36
​
37
  @Override
38
  protected Class<OrderProcessState> stateType() {
39
    return OrderProcessState.class;
40
  }
41
}
Copied!
The chronicle() is required by the ObjectProcess, and the stateObject(), state(), and stateType() are required by the ObjectEntity.