Understanding Message Sends

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:

• Application creates a message.
• Application calls send passing the message and channel name. 
• The engine looks up the channel and use it to resolves the physical destination on which the message will be sent.
• The engine queues the message for send until the associated state changes for the message handler are stabilized.

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 application is sent from outside of a message handler, it is referred to as an unsolicited send. 

Creating Messages

Message types generated by the application data modeler can be created via their static create() method. Application code populates message contents as a regular pojo.

Message Concurrency and Pooling

Outbound messages sent by the application may be serialized in background threads.

An application that would like to send the same message multiple times can use the copy() method on the message. The copy can be modified and used in a subsequent send call. 

Messages may also be pooled by the platform. This is the reason that message construction is done through static create methods. Once the platform has finished sending them, their backing contents can be wiped and the view reused in a subsequent sent.

Sending Messages

Using 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: 

• 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 onOnOrderEventChannelDown(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 be controlled via configuration.

Static Topics

Dynamic Topics

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. When a message is sent over a MessageChannel, all variable key components have to be resolved or an SmaException will be thrown. 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 channel's key resolution table (if non null), which is a table that can be set on the channel programmatically for key resolution.
• The key resolution table supplied to this method (if non null)
• 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 an 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 or hasXXX() returns false, channel key resolution will fall back to the key resolution table or 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 might 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 performance as it relies on standard java reflection.

Key Resolution Tables

When a variable key value is not present in a message because there is not 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 programatically 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 Provide 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 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 restriction apply:

• The table should not be modified after being set on a chanel
• 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.

Message Key Validation

When resolving topics dynamically from a messages 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 portions of the topic at the cost of additional overhead by setting the following environment properties:

Understanding Message Receipt

An application's Aep Engine creates and manages the lifecycle of the message buses that an application configures for use. When an application is configured to join one of more bus channels, subscriptions will be issued on behalf of the application to attract messages. 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 data 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 acknowledgement 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".

<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>
  	  <channels>
        <channel name="new-orders-channel">
          <qos>Guaranteed</qos>
          <key>NEWORDERS/${Region}/${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>
          </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 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. 

Channel filter take 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", it would result in the following subscriptions being 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/*/*

Registering Message View 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>
    </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

HA Considerations

Only the primary instance of an application will establish messaging connections.

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. This behavior 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),

See Also

• Unhandled Messages