Package org.axonframework.messaging.core.sequencing

Interface SequencingPolicy<M extends Message>

Type Parameters:

M - the type of message to sequence

All Known Implementing Classes:

ExtractionSequencingPolicy, FallbackSequencingPolicy, FullConcurrencyPolicy, HierarchicalSequencingPolicy, MetadataSequencingPolicy, NoOpSequencingPolicy, PropertySequencingPolicy, RoutingKeySequencingPolicy, SequentialPerAggregatePolicy, SequentialPolicy

Functional Interface:

This is a functional interface and can therefore be used as the assignment target for a lambda expression or method reference.

@FunctionalInterface
public interface SequencingPolicy<M extends Message>

Interface to a policy definition for concurrent message processing.

Some implementations are provided by default:

• SequentialPolicy: Requires that all messages are handled in the order they arrive. For event handling this also means that at most 1 thread is processing messages for this processor at any time.
• FullConcurrencyPolicy: Allows each message to be handled independently of any other message. Message processing will typically start in the same order the messages were scheduled in, although no guarantees can be given about the actual processing order.
• SequentialPerAggregatePolicy: Default event processing policy. Will force events (only supports event handling tasks) generated by the same aggregate to be handled sequentially. At most one thread will be processing events of a single aggregate at any time.
• RoutingKeySequencingPolicy: Default command processing policy. Will force commands (only supports command handling tasks) routed for the same routing key to be handled sequentially.
• ExtractionSequencingPolicy: Message processing policy that allows to extract the sequence identifier from the message payload.
• MetadataSequencingPolicy: Message processing policy that allows to extract the sequence identifier from the message metadata.
• PropertySequencingPolicy: Message processing policy that extracts the sequence identifier from the message payload based on a given property
• FallbackSequencingPolicy: Message processing policy that allows two policies in an exception-based fallback behavior for sequence identifier extraction.
• HierarchicalSequencingPolicy: Message processing policy that allows to combine two policies in a fallback pattern (similar to FallbackSequencingPolicy, but not exception based).

Since:

0.3

Author:

Allard Buijze

Field Summary

Fields

Modifier and Type	Field	Description
static final Object	BROADCAST	Well-known sentinel sequence identifier signaling that a message should be handled by every handler eligible for it, rather than being routed to a single one based on its sequence identifier.

Method Summary

Modifier and Type	Method	Description
Optional<Object>	sequenceIdentifierFor(M message, @Nullable ProcessingContext context)	Returns the sequence identifier for the given message.

Field Details

BROADCAST

static final Object BROADCAST

Well-known sentinel sequence identifier signaling that a message should be handled by every handler eligible for it, rather than being routed to a single one based on its sequence identifier.

When the sequence identifier resolved for a message is this exact instance, that message is delivered to every eligible handler instead of only the one its identifier would otherwise select. Such messages remain sequenced relative to one another wherever sequencing applies.

Note: this is a unique instance recognized by identity, not by value. A policy requests a broadcast only by returning SequencingPolicy.BROADCAST itself. Sequence identifiers extracted from a message - a metadata value, a payload property, or an entity identifier - can never request a broadcast by accident, whatever their value.

Equality is identity, while Object.hashCode() is deliberately a fixed value rather than the identity hash. Identity equality is what makes an accidental broadcast impossible. The fixed hash is only a safety net: every recognition site skips segment routing for this instance, but a site that hashed the sentinel into a segment regardless would then pick the same segment on every run instead of a different arbitrary one after each restart. Sharing a hash bucket with the String "BROADCAST" is harmless, since identity equality still keeps the two apart wherever sequence identifiers are collected or keyed.

Since:

5.3.0

Method Details

sequenceIdentifierFor

Optional<Object> sequenceIdentifierFor(M message,
 @Nullable ProcessingContext context)

Returns the sequence identifier for the given message. When two messages have the same identifier (as defined by their equals method), they will be executed sequentially. A Optional#empty() value indicates that there are no sequencing requirements for the handling of this message.

The Optional#empty() should only be returned when the policy cannot determine a sequence identifier for the given message. This typically happens when the policy is not applicable for the specific message type. When Optional#empty() is returned, it is up to the component using this policy to provide a default behavior, use another policy, throw an exception or react in any other way - as appropriate.

Parameters:

message - the message for which to get the sequencing identifier.

context - the processing context in which the message is being handled. There might be limitations to the instance of the ProcessingContext depending on where the SequencingPolicy is being applied. When handling Events for example, the ProcessingContext doesn't allow you to register phases actions by for example ProcessingLifecycle.on(org.axonframework.messaging.core.unitofwork.ProcessingLifecycle.Phase, java.util.function.Function<org.axonframework.messaging.core.unitofwork.ProcessingContext, java.util.concurrent.CompletableFuture<?>>) or retrieving components by ApplicationContext.component(java.lang.Class<C>).

Returns:

a sequence identifier for the given message, or Optional#empty() if this policy cannot determine a sequence identifier for the given message.