Overview

This section assumes you have read Working with Messaging to understand the basics of SMA messaging.

As a quick refresher: Talon's Simple Message Abstraction layer SMA abstracts the concept of a message provider using MessageBuses. MessageBuses define MessageChannels over which Messages are sent and received. Message channels defined a named conduit that allows mapping to destinations in a messaging provider. Applications thus collaborate with one another by passing messages over the set of buses and channels that they share. An application that will receive messages from a Message Channel will declare an interest in 'joining' the channel which will cause the subscriptions to be issued for the channel on behalf of the application.

The sections below provide details on how applications send and receive messages.

Understanding Bus Configuration

Before diving into the details of send and receive, it is helpful to understand how message buses are configured for an application. Message buses are resources that are shared between applications. In configuration message buses are defined independently of applications to provide the common connection details and the set of channels that they provide for applications. Applications will reference the buses that they will use to collaborate with other applications along with a subset of the channels that they will work with. Applications work abstractly with buses and channel to send and receive messages, allowing configuration changes at runtime to augment:

• The messaging provider (for example to switch from ActiveMQ to Solace)
• The message bus connection details in the form of an address and optionally a port (for example to switch from a message broker in a test environment to one in production). 
• Bus connection properties, a combination of properties that are common to all bus providers and configuration specific to the type of bus provider being used. 
• Topic structure by modifying the channels' configured keys.
• Application subscriptions by modifying the applications' configured filters. 
  
  

<model>
 
  <!-- Define buses and channels that will be shared between applications-->
  <buses>
    <bus name="sample-bus">
	  <provider>activemq</provider>
      <address>localhost</address>
      <port>61616</port>
      <properties>
        <set_key_on_receipt>true</set_key_on_receipt>
      <properties>
  	  <channels>
        <channel name="new-orders-channel">
          <qos>Guaranteed</qos>
          <key>NEWORDERS/${Region}/${Department}
        </channel>
        <channel name="order-events">
          <qos>Guaranteed</qos>
          <key>ORDEREVENTS/${Region}/${EventType}/${Department}
        </channel>
      <channels>
    </bus>
  </buses>
  
  <!-- Apps will reference the buses they use in their messaging config -->
  <apps>
    <app name="sample-app" mainClass="com.sample.SampleApp">
      <messaging>
		<factories>
          <factory name="com.sample.messages.OrderMessagesFactory" />
        </factories>
		<buses>
          <bus name="sample-bus">
			<channel name="new-orders-channel" join="true">
     		  <filter>Region=US|Canada</filter>
            </channel>
          </bus>
        </buses>
      <messaging>
    </app>
  </apps>
</model>

The above configuration is an example of configuring a bus in its decomposed property form. A message bus can alternatively be configured using the bus descriptor attribute as shown in the following example. 

<buses>
  <bus name="sample-bus" descriptor="activemq://localhost:61616&set_key_on_receipt=true">
    <channels>
      ...
    </channels>
  </bus>
</buses>

The bus descriptor format is useful in cases where the descriptor name may be supplied as an external configuration property that is substituted. In the following example, the bus descriptor defaults to using a loopback bus which is useful for unit testing and configured with the myBusDescriptor environment property in other environments. 

<buses>
  <bus name="sample-bus" descriptor="${myBusDescriptor::loopback://mybus">
    <channels>
      ...
    </channels>
  </bus>
</buses>

Common Bus Configuration Properties

The following properties are common to all bus implementations (unless specifically noted otherwise) and can be specified in a bus descriptor either as parameters in a descriptor URL or as properties in the buses' configuration settings when specified in its decomposed form. 

Most bus implementations support additional properties specified to the messaging provider. Consult the documentation for the bus provider to learn about the properties it supports. 

Sending Messages

Messages are sent over message bus channels in a fire and forget fashion - the application can rely on the underlying AepEngine to deliver the message according to the quality of service configured for a channel. An application views message channels as a logical, ordered conduit over which messages flow. 

The diagram below depicts the path of a message sent through an engine. The basic flow of sending is as follows:

1. The application creates a message.
2. The application calls send, passing the message and channel name. 
3. The engine looks up the channel and uses it to resolves the physical destination on which the message will be sent via the configured channel key.
4. The engine queues the message for sending until the associated state changes for the message handler are stabilized to the application's store (e.g replication and transaction log write).
5. Once the application state changes have been stabilized, the enqueued message is then sent. This involves the following sub-steps
   1. The message is serialized
   2. The serialized contents and message metadata are then packaged in a message bus provider specific message 
   3. The message is sent through the underlying transport.

Solicited vs Unsolicited Sends

When an outbound message is sent from within an application event handler, it is referred to as a solicited send. When an outbound message is sent from outside of a message handler, it is referred to as an unsolicited send. The steps listed above pertain to Solicited sends.

Working With Messages

Message types generated by the application data modeler have no public constructors, but instances can be created via the static create() factory method on the message class. By routing all message creation through the create() factory methods, X is able to transparently switch between pooled and unpooled allocation of messages without the need for any changes to business logic. Once a message has been created, application code can populate field values by calling setter methods, as though it were any other POJO.

For example:

MyMessage message = MyMessage.create();
message.setStringField("test");
message.setIntField(123);

Outbound messages sent by the application may be serialized on background threads. Because of this, message instances must not be used by the application after being sent nor should they be resent. This includes any nested entities within the message. If the application needs to send the same message multiple times, use the copy() method on the message to create a new deep copy of the message:

MyMessage original = MyMessage.create();
original.setName("Neeve");
original.setYear(2017);
 
MyMessage copy = original.copy();
copy.setYear(2018);
 
sender.send("destination1", original);
sender.send('destination2", copy);

If message pooling is enabled, sent messages will be wiped and reused in subsequent sends.

Using the AEP Message Sender

The simplest way to send a message is to add an AppInjectionPoint annotation that allows the XVM to inject a message sender. The application then supplies the bus and channel name along with the message. 

public class MyApp {
  private volatile AepMessageSender messageSender;
 
  @AppInjectionPoint
  public void injectMessageSender(AepMessageSender messageSender) {
    this.messageSender = messageSender;
  }
 
  @EventHandler
  public void onNewOrder(NewOrderMessage message) {
    // create the message
	OrderEventMessage event = OrderEventMessage.create();
	event.setOrderId(message.getOrderId());


    // send message through the aep sender
    // on the 'order-event-channel' of the 'order-processing-bus'  
    messageSender.sendMessage("order-processing-bus"
                              "order-event-channel", 
                              event);
  }
} 

See also: 

• Talon Annotations
• AepMessageSender (javadoc)

Sending via the AepEngine

When an application does not use the AepMessageSender, it must provide the MessageChannel in the AepEngine's sendMessage call. When using this pattern, the application can register event handlers for channel up and down events which are dispatched to the application when messaging is started or stopped. The application can qualify the EventHandler annotation to identify a particular channelName@busName to specify the channel of interest.

public class MyApp {
  private volatile MessageChannel orderEventChannel;
  private volatile AepEngine aepEngine;
 
  @AppInjectionPoint
  public void onEngineInjected(AepEngine aepEngine) {
    this.aepEngine = aepEngine;
  }
  @EventHandler(source = "order-event-channel@order-processing-bus")
  public void onOrderEventChannelUp(AepChannelUpEvent channelUpEvent) {
    this.orderEventChannel= channelUpEvent.getChannel();
  }

  @EventHandler(source = "new-orders-channel@order-processing-bus")
  public void onOrderEventChannelDown(AepChannelDownEvent channelDownEvent) {
    this.orderEventChannel= null;
  }
 
  @EventHandler
  public void onNewOrder(NewOrderMessage message) {
	OrderEventMessage event = OrderEventMessage.create();
	event.setOrderId(message.getOrderId());

    // The channel name and bus must be set on the message
    // if using Event Sourcing because handlers are invoked
    // on the backup and during recovery before messaging is
    // started. 
    // 
    // This is not required when using State Replication.
    event.setMessageChannel("order-event-channel");
    event.setMessageBus("order-processing-bus");
 
    aepEngine.sendMessage(orderEventChannel, event);
  }
} 

Channel Keys and Topic Resolution

The message channel encapsulates the physical destination on which the message will be sent using a channel key. The channel key abstraction allows the actual message provider destination to be controlled via configuration. Channel keys can be either static or dynamic:

1. A static key uses the channel key as the actual topic on which to send the message. For example, "Orders/USA".
2. A dynamic key contains variable components that are resolved at runtime either at startup or during a send call to resolve the specific topic. For example, "Orders/${Region}".
3. Channel key functions allow portions of the key to be resolved using a function. For example, "Orders/#[orderProcessorShard = hash(${orderId}, 2)]", would resolve to Orders/1 or Orders/2 based on the hashcode of the orderId field in the message.

By default, a message channel will prepend the channel name to the beginning of the topic as the first level in the topic name. This mechanism prevents multiple channel definitions from conflicting with one another. This behavior can be disabled by configuring the bus provider with the bus property topic_starts_with_channel=false.

Keyless Channels

When no channel key is specified for a defined channel, the platform's default behavior is to use the channel name as the destination for the message. For this case, it is not possible to set topic_starts_with_channel=false.

Static Keys

A message channel that is configured with no variable key components is said to be a static key. On every send call, the same key will be used as is for the topic (again, the channel name will be prepended unless topic_starts_with_channel is set to false in the bus descriptor).

Dynamic Keys

A channel key consists of static and variable portions with the variable portions consisting identified in the form of ${propName[::defaultValue]}. For example, a channel key of 'ORDERS/${salesRegion::US}/${productId}/Purchases' is interpreted as having variable key components of 'salesRegion' and '${productId}', with 'salesRegion' specifying a default value of 'US' in cases where it is unspecified. Variable key components are resolved from the following sources in this order:

• The message contents via message reflection. The channel will look for a field accessor matching the variable key component (e.g. a getSalesRegion() method) if the message supports message reflection.
• The key resolution table supplied to an AepEngine.sendMessage method or AepMessageSender.sendMessage method (if non-null).
• The channel's key resolution table (if non-null), which is a table that can be set on the channel programmatically for key resolution when handling an AepChannelUpEvent (see restrictions below for EventSourcing considerations).
• Default values encoded into the channel key itself. In the above example, this would be 'US' for 'salesRegion'.

If a variable key segment cannot be resolved by any of the above means at send time it results in a SMAException being thrown from the send call.

Message Reflection

Message reflection allows content-based routing of the message. To use message reflection the channel key variable should match the name of the getter for the field in the message starting with a lower case character. For example for a message with a getter of:

public final String getSalesRegion();

The channel key variable should be defined as ${salesRegion}

Valid Field Types

The following message field types are supported for message key reflection and will be used as substitution values when the hasXXX() method corresponding to the field returns true.

If the field value is null or hasXXX() returns false, channel key resolution will fall back to the key resolution table or the default value if present. If the field value is not reflectable or the field value is null and there is the value is not found in a key resolution table, then an exception is thrown.

Nested Field Reflection

Key values can come from embedded entities in the message by specifying the bean path of the field. For example, if the message is a CustomerProfileUpdateMessage with an embedded entity Address field one could specify a channel key variable ${address.zipCode} which would be substituted with getAddress().getZipCode().

Nested field key substitution is not necessarily a zero garbage operation and may be less performant as it relies on standard java reflection.

Key Resolution Tables

When a variable key value is not present in a message because there is no field matching the name, the field is not set or not reflectable, key resolution will fall back to a channel key resolution table if available.

Channel Key Resolution Table

In many cases, substitution values for a dynamic key come from the application environment or configuration. Message channels can be configured with a key resolution table to allow substitution of key variables that don't come from the message being sent. Channel key resolution tables can be supplied programmatically by registering an event handler for the AepChannelUpEvent and setting the key resolution table. The following shows an example of setting a global key resolution table that will be configured for every channel in the app:

public class MyApp {
  private final Properties globalKeyResolutionTable = new Properties();


  public MyApp() {
    krt.put("Region", "EMEA");
    krt.put("Shard", "1");
  }

  @EventHandler()
  public void onOrderEventChannelUp(AepChannelUpEvent channelUpEvent) {
    channelUpEvent.getChannel.setKeyResolutionTable(globalKeyResolutionTable);
  }
} 

Caller Provided Key Resolution Table

AEP send methods allow the caller to pass in a key resolution with the send method call to augment key resolution on a case by case basis.

RawKeyResolutionTable

In addition to specifying a key resolution table using a java.util.Properties map, a RawKeyResolutionTable can be configured instead. A RawKeyResolutionTable stores the substitution value in an XString which can improve performance because the substitution value can be stored as pre-encoded bytes and eliminate character encoding costs. The following example shows how a RawKeyResolutionTable can be used instead of Properties:

public class MyApp {
  private final RawKeyResolutionTable globalKeyResolutionTable = MessageBusBindingFactory.createRawKeyResolutionTable()

  public MyApp() {
    krt.put("Region", XString.create("EMEA"));
    krt.put("Shard",  XString.create("1"));
  }

  @EventHandler()
  public void onOrderEventChannelUp(AepChannelUpEvent channelUpEvent) {
    channelUpEvent.getChannel.setRawKeyResolutionTable(globalKeyResolutionTable);
  }
} 

When using a RawKeyResolutionTable the following restrictions apply:

• The table should not be modified after being set on a channel
• The XString values placed in the table must not be modified after inserting into the table.
• It is not permissible to set both a RawKeyResolutionTable and Properties based key resolution table on the same channel.

Channel Key Functions

Channel key functions can be used to define a variable portion of the channel key using a function that is evaluated at runtime. Channel key functions are #[] enclosed and take the following form:

#[variableName = functionName(arg1, argN)]

where

• • variableName: Names the variable portion of the key. The naming of the variable portion defined by the function is important as it allows the portion to be referenced when defining channel filters (see channel filters below)
  • functionName: The name of the static registered channel key function to use. 
  • args: The arguments to pass to the function. The arguments can either be static values, or ${variable} values sourced from the message or key resolution tables. 

Built-In Functions

The following table describes the currently available set of built-in channel key functions provided by the platform: 

(Math.abs(value.hashCode()) % partitions.getValueAsLong() + 1)

Example:

<buses>  
  <bus name="order-bus" descriptor="solace://solhost:55555>
    <channels>
      <channel name="new-orders">
        <key>NEWORDERS/#[ordershard=hash(${orderId}, 3)]</key>
      </channel>
    </channel>
  </bus>
</buses>
 
<apps>
  <app name="sender">
    <messaging>
      <bus name="order-bus">
        <channels>
          <channel name="new-orders" join="false">
        </channels>
      </bus>
    </messaging>
  </app>
  <app name="receiver-1">
    <messaging>
      <bus name="order-bus">
        <channels>
          <channel name="new-orders" join="true">
            <filter>ordershard=1|2<filter>
          </channel>
        </channels>
      </bus>
    </messaging>
  </app>
</apps>

In the above example:

• The 'new-orders' channel key second topic level is defined as a variable, ordershard that hashes the value of the orderId field in the message. 
• Senders will thus send on a topic of either NEWORDERS/1, NEWORDERS/3 or NEWORDERS/3. 
• The receiver declares that it will join the channel for receipt and specifies a filter on the ordershard variable of 1 or 2 causing it to issue subscriptions on NEWORDERS/1 NEWORDERS/2 but not NEWORDERS/3.

Registering Additional Functions

An application may define its own channel key functions by setting the value of  `nv.sma.channelkeyfunctioncontainers` to a comma-separated list of classes that contain additional channel key functions. 

Custom Key Method Requirements

A custom key function must:

• Be declared as public static.
• Accept the key being resolved as its first argument (the key function appends the resolved value to the key). 
• May accept one or more additional XString arguments representing the function argument declared in the channel key. 

A custom key function may throw a RuntimeException if the arguments provided are invalid. A channel key function may append nothing to the resolved key, but implementors should be careful that doing so would not produce an invalid topic. 

Custom Key Function Example

<env>
  <nv>
    <sma>
      <channelkeyfunctioncontainers>com.example.CustomChannelKeyFunctions</channelkeyfunctioncontainers>
    </sma>
  </nv>
</env>
 
 

public class CustomChannelKeyFunctions {
  
  /**
   * A custom key function that substituteds the first character of the provided 
   * value into the message key. 
   */
  public static final void abbreviate(final XString messageKey, XString value) {
    messageKey.append(value.charAt(0));
  }
}

Message Key Validation

When resolving topics dynamically from a message's fields it may be important to validate that value from the message field doesn't introduce additional levels in the topic or illegal characters or result in empty topic levels. To prevent this a message bus can be configured to 'clean' the dynamic variable portions of the topic at the cost of additional overhead by setting the following environment properties:

Caller Provided Key

It is also possible for the caller to provide the message key directly in the send call. In this case, the supplied key will be used as is for the topic. This can be useful in certain situations but should be used with care because it breaks the ability for the topic to be changed at runtime and receivers will not have knowledge of how to subscribe to such sends.

@EventHandler
public void onNewOrder(NewOrderMessage message) {
  // create the message
  OrderEventMessage event = OrderEventMessage.create();
  event.setOrderId(message.getOrderId());

  // send message through the aep sender
  // on the 'order-event-channel' of the 'order-processing-bus'  
  messageSender.sendMessage("order-processing-bus"
                            "order-event-channel", 
                            event,
                            "CUSTOMORDEREVENTS/" + message.getOrderType());
} 

When using caller supplied topic:

• Variable substitution is not done on the provided key.
• Key validation is done on the topic
• The provided key will be prepended with the channel name if topic_starts_with_channel=true is set for the bus.

Understanding Message Receipt

An application's AepEngine creates and manages the lifecycle of the message buses that an application configures for use. When an application is configured to join one or more bus channels, appropriate topic subscriptions will be issued on behalf of the application. Message dispatch occurs as follows:

• The message bus binding implementation receives a serialized, message provider specific message.
• Using an application supplied message view factory (generated by ADM), and using SMA message metadata also transported with the provider-specific message, the bus wraps the serialized message in a view object (also ADM generated).
• Using message metadata transported with the message, the binding looks up the message channel on which to dispatch the received message.
• The message view is wrapped in a MessageEvent along with its message channel and is dispatched to the application's AepEngine, where it is enqueued for processing. 
• The AepEngine picks up the message event and dispatches to each application event handler that has a signature matching the message type. 
• Once the AepEngine stabilizes the results of the application's message processing, a message acknowledgment is dispatched back to the binding. 

Expressing Interest

For an application to receive messages, it must:

• join the message channels on which the message is sent,
• register message factories for the bus provider to deserialize the message,
• and define an EventHandler for the message.

Configuring Channels For Join

In order for the application's AepEninge to issue subscriptions for the message channel on which a message is sent, it must be joined. Buses and channels are configured via the platform's configuration DDL. The below configuration snippet demonstrates:

• Defining a bus named "sample-bus" with a "new-orders-channel".
• An application that uses the "sample-bus" and joins the "new-orders-channel" (note the usage of join="true")

<?xml version="1.0"?>
<model xmlns="http://www.neeveresearch.com/schema/x-ddl" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
 
  <!-- Define buses and channels that will be shared between applications-->
  <buses>
    <bus name="sample-bus">
	  <provider>activemq</provider>
      <address>localhost</address>
      <port>61616</port>
  	  <channels>
        <channel name="new-orders-channel">
          <qos>Guaranteed</qos>
          <key>NEWORDERS/${Region}/${Department}</key>
        </channel>
      <channels>
    </bus>
  </buses>
  
  <!-- Apps will reference the buses they use in their messaging config -->
  <apps>
    <app name="sample-app" mainClass="com.sample.SampleApp">
      <messaging>
		<factories>
          <factory name="com.sample.messages.OrderMessagesFactory" />
        </factories>
		<buses>
			<bus name="sample-bus">
				<channels>
					<channel name="new-orders-channel" join="true">
						<filter>Region=US|Canada</filter>
					</channel>
				</channels>
	        </bus>
		</buses>
      <messaging>
    </app>
  </apps>
</model>

Adding an EventHandler

When a message is received by a message bus, it is enqueued into the application's Inbound Event Queue to be queued for dispatch, which the engine will pick up.

@EventHandler
public void onNewOrder(NewOrderMessage message) {
  /*... do some work ...*/
}

The application's underlying AepEngine will ensure that the message is acknowledged once state changes made by the handler have been stabilized. That, coupled with the engine's message deduplication feature, ensures that even in the event of a failover, the handler will only be executed once. 

Channel Filters

A channel filter filters variable parts of a channel key to filter what is received over a message channel. It is used to determine the subscriptions issued on behalf of the application. See the Configuring Channels for Join section above for an example of using channel filters. In particular, pay attention to the "key" and "filter" elements.

Channel filter syntax takes the following form: 

var1=val1[|val2][;var2=val3]

For example, given a channel key of "NEWORDERS/${Region}/{Department}", one can specify a channel filter of "Region=US|EMEA;Department=Clothing". This would join the channel on:

• NEWORDERS/US/Clothing
• NEWORDERS/EMEA/Clothing

If a variable portion of the channel key is omitted in a filter, it will result in the subscription being joined in a wildcard fashion, assuming the underlying bus implementation supports wildcards. So given a channel key of "NEWORDERS/${Region}/${Department}" and a channel filter of "Region=US|EMEA", the following subscriptions would be issued during join:

• NEWORDERS/US/*
• NEWORDERS/EMEA/*

Finally, if the channel filter is set to null for the channel key in the example above, then the resulting subscription would be:

• NEWORDERS/*/*

Channel Filter Cleaning

When the global configuration setting nv.sma.cleanchannelfilter is set to true, non-alphanumeric characters in the configured filter values will be replaced by underscores in order to match the keys used on the send side. The below configuration setting can be used to opt-out of this behavior, but typically the default behavior is more desirable:

As of the 3.8 Release, channel filter cleaning has been enhanced to not replace certain wildcard characters that are legal for use in subscriptions. 

• Solace Binding: A '>' will not be replaced if it represents the whole topic level and a '*' will not be replaced if it is the last character in the topic level. See also: Solace Binding
• Loopback Binding: A filter level of "..." will not be replaced nor will a '*' found anywhere in the topic level. See also Loopback Binding
• JMS Binding: Depends on the provider instance ... see JMS Binding

Otherwise, any characters that are not alpha-numeric will be replaced. Prior to 3.8, any non-alphanumeric character was replaced included the wildcard combinations described above. 

Handling Received Messages

Messages dispatched to applications are read-only, it is illegal for an application to modify a received message.

As Messages may be pooled by the platform it is also not legal for an application to hold onto a message beyond the scope of its handler unless it first calls acquire() on the message. If the application does acquire() the message it should later dispose() of the message to allow the platform to reuse it. In most cases holding on to the message beyond the scope of the handler is not the best idea from a performance perspective. Instead, applications will typically copy the data they need out of the message into either application state objects or into outbound messages.

Inbound messages may not be resent as outbound messages. Applications that need to forward an inbound message as an outbound message should first copy() the message and send the message copy:

@EventHandler
public void onMyMessageReceived(MyMessage message) {
  MyMessage copy = message.copy();
  messageSender.send("mirror", copy);
}

Registering MessageView Factories

Message bus binding implementations receive messages in serialized form and wrap them with a MessageView that is passed to the application to work with. MessageViews are wrapped by locating the message MessageViewFactory for the message which is typically generated by ADM. To locate the factory and message type, a binding consults Message Metadata that is transported along with the serialized message. An application must, therefore, register the message factories it intends to use so that bus binding can find the factories. This can be done by registration or by programming.

Registration via Config DDL

Most often, applications will list message view factories that they use in their DDL Config.

<apps>
  <app name="MyApp">
    <messaging>
      <factories>
		<factory name="com.sample.messages.OrderMessagesFactory" />
      </factories>
      <buses>
        <!-- ... -->
      </buses>
    </messaging>
  </app>
</app>

Programmatically

Registration can also be done programmatically via the AepEngine. A common way to do this is to provide an AppInjectionPoint for the AepEngine in the application.

public class MyApp {
  
  @AppInjectionPoint
  public void onEngineCreated(AepEngine engine) {
	engine.registerFactory(new com.example.messages.MyMessageFactory());
	engine.registerFactory(new com.example.messages.other.MyOtherMessageFactory());
  }
}

Message Sequencing And Duplicate Detection

To avoid duplicates and provide exactly once delivery semantics each message sent through an AepEngine is assigned a message sender id derived from the hashcode of the application name and a monotonically increasing sequence number. The following table describes the fields that the engine sets on outbound messages to enable downstream duplicate detection. These properties are transported on a message bus binding as part of the message metadata sent with the serialized message.

Message Field	Type	Description
MessageSender	int	A globally unique id that identifies the sender of a message. By default, an AepEngine uses the hashcode of the engine name as the sender id.
MessageSno	long	A monotonically increasing sequence number.   A sequence number of <=0 indicates that the message is not sequenced. Receivers should not perform duplicate checking on it. A sequence number of 1 indicates that the start (or restart of a stream). A receiver that receives sequence numbers 1,2,3,1,2,3 should not consider the 4th message a duplicate. Otherwise, receivers should consider a sequence number not greater than the previous sequence number as a duplicate.
MessageFlow	int	Indicates the flow to which the message belongs. Flows allow for partitioning of message traffic in conjunction with application state and allow intra-application state partitioning. Usage of flows is not currently supported for end users. External senders should always set the flow to 0.

As an AepEngine processes messages, it keeps track of the last seen sequence number for each sender that has sent to it as part of its HA state. Consequently, when an application fails over to a backup or is restarted it will remember the last message seen from senders sending to it and can filter duplicates on behalf of the application. When an application's state is wiped or reinitialized it will restart its sending stream sequence to 1 which alerts downstream applications not to consider its newly lowered sequence numbers as duplicates.

By default any message sent from a message handler is sequenced by an AepEngine, the application should not set the sequence number or sender itself. Messages sent from outside of a message handler (unsolicited sends) are not sequenced by default. Sequencing of unsolicited sends must be enabled via configuration discussed below.

Configuring Duplicate Detection

For applications that are tolerant of duplicates or can perform duplicate detection on their own, it is possible to disable duplicate detection for the engine. It is also possible to enable sequencing of messages that are sent from outside of a message handler can be done as well.

<model>
  
  <!-- Apps will reference the buses they use in their messaging config -->
  <apps>
    <app name="sample-app" mainClass="com.sample.SampleApp">
      <messaging>
		...
      <messaging>

      <!-- 
      <performDuplicateChecking>true</performDuplicateChecking>
      <setOutboundSequenceNumbers>true</setOutboundSequenceNumbers>
      <sequenceUnsolicitedSends>false</sequenceUnsolicitedSends>
      <sequenceUnsolicitedWithSolicitedSends>false</sequenceUnsolicitedWithSolicitedSends>
    </app>
  </apps>
</model>

The following table summarizes AEP engine configuration properties that have an impact on duplicate detection:

HA Considerations

It is worth noting that for a cluster application only the primary instance of the application establishes message bus connections. In the event of a failure, a backup member will be elected and reestablish messaging connections. From a programming standpoint, this has the biggest impact on applications using the EventSourcing HA Policy as it means that even though their handlers are being invoked message channels will not have been established for the applications.

During live operation, the primary application keeps track of the last transaction for which outbound messages were acknowledged. This id, known as the stableTransactionId is persisted in the applications transaction log and replicated to backups. After failure or recovery backup applications will retransmit outbound messages that are part of transactions that have not stabilized and rely on downstream applications to use duplicate detection to eliminate any duplicates.

Preserving Subscriptions on Shutdown

By default, when an engine is stopped without an error, bus channels that were 'joined' will be 'left', meaning that any subscriptions or interests created by the message bus will be unsubscribed or unregistered. For many applications, it is desirable to preserve subscriptions if an application is being gracefully shutdown for maintenance reasons – one may want messages to be queued for the application while it is down. For such cases the default behavior or unsubscribing on graceful shutdown can be overridden by configuring an application to preserve channel joins on stop:

<app name="sample-app" mainClass="com.sample.SampleApp">
  <messaging>
	...
  <messaging>
  <preserveChannelJoinsOnStop>true</preserveChannelJoinsOnStop>
</app>

Note that this property has no effect when an engine shuts down with an error (e.g. AepEngine.stop(Exception) with a non-null cause. In this case, channel joins are left intact, allowing a backup to take over. 

This behavior can also be overridden programmatically on a case by case basis by an EventHandler for the AepEngineStoppingEvent setting AepEngineStoppingEvent.setPreserveChannelJoins(boolean):

@EventHandler
public void onEngineStopping(final AepEngineStoppingEvent event) {
  if(event.getPreserveChannelJoins()) {
    tracer.log("Overriding unsubscribe behavior on application stop to remove subscriptions.");
    event.setPreserveChannelJoins(false);
  }
}

In the above case, the value set programmatically overrides the configured value for the application.

Per Channel Subscription Preservation

Subscription preservation or removal can also be configured more granularly at the channel level. Like the application level configuration setting, this per channel configuration setting only applies to a graceful close. 

The following example shows an application configuring subscription preservation on a per channel basis using the preserveJoinsOnClose configuration property: 

<app name="sample-app" mainClass="com.sample.SampleApp">
  <preserveChannelJoinsOnStop>true</preserveChannelJoinsOnStop>
  <messaging>
    <buses>
      <bus name="orders-bus">
        <channels>
          <channel name="canceled-orders" join="true">
            <filter>Region=US</filter>
            <preserveJoinsOnClose>Default</preserveJoinsOnClose>
          </channel>
          <channel name="new-orders" join="true">
            <filter>Region=US</filter>
            <preserveJoinsOnClose>Preserve</preserveJoinsOnClose>
          </channel name="app-ping" join="true">
            <filter>Region=US</filter>
            <preserveJoinsOnClose>Leave</preserveJoinsOnClose>
          </channel>
        </channels>
      </bus>
    </buses>
  <messaging>
</app>

In the above case, if the application is stopped gracefully:

• Subscriptions for the canceled-orders channel would be preserved (because preserveChannelJoinsOnStop=true at the application level).

• Subscriptions for the new-orders channel would be preserved regardless of the configured value for preserveChannelJoinsOnStop. 
• Subscriptions for the app-ping channel would be unsubscribed regardless of the configured value for preserveChannelJoinsOnStop. 

One can also configure per channel subscription preservation programmatically via the message channel's MessageChannelDescriptor. A programmatically set value will override that set via DDL configuration, and can be set at any time before the channel is closed:

@EventHandler(source = "channel4@aeptest1")
public void onChannel4UpEventHandler(final AepChannelUpEvent event) {
  MessageChannel channel = event.getMessageChannel();
  if(channel.getName().equals("app-ping")) {
    channel .getDescriptor().setPreserveJoinsOnClose(PreserveJoinPolicy.Leave);
  }
}