Registering Message Interest

The AEP engine creates and manages the lifecycle of the message buses that an application configures for use. When a microservice is configured to join one or more bus channels, the engine issues appropriate topic subscriptions on behalf of the microservice.

Inbound Message Dispatch

The above depicts how messages drawn into the microservice via the issued subscriptions are processed:

• The SMA message bus binding implementation receives a provider specific message from the underlying message bus. The provider specific message is comprised of an application specific data payload and SMA metadata.

• The binding uses an application registered message view factory (generated by ADM) and the SMA message metadata present on the derlying bus message to convert the bus specific message into a message object (called a "Message View" and is also ADM generated).

• The binding uses the SMA message metadata to identify the message channel on which to dispatch the received message.

• The binding wraps the message view in a MessageEvent with a reference to the message channel and dispatches it to the application's AEP Engine, where it is enqueued for processing.

• The AEP Engine picks up the message event and dispatches to each application event handler that has a signature matching the message type.

• Once the AEP Engine stabilizes the results of the application's message processing, a message acknowledgment is dispatched back to the binding.

Registering Message 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,

• define an EventHandler for the message.

Configuring Channels For Join

For the AEP Engine to issue subscriptions for the message channel on which a message is sent, the channel 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>

Channel Filters

A channel filter is used to filter what is received over a message channel. In other words, 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

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/*/*

Cleaning Channel Filters

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:

Property	Default	Description
nv.sma.cleanchannelfilter	false*	Controls whether or not channel filter values are sanitized by replacing any non-letter or digit character with a '_'. For example, if the channel key is specified as /Orders/${Region} and a filter of Region=Asia/Pac is given, then the filter will match all messages with the resolved key value of /Orders/Asia_Pac (rather than /Orders/Asia/Pac). *Default Value: In X 3.7, the value defaults to the value specified for nv.sma.cleanmessagekey In X 3.8 onwards the default value is false.

As of the X 3.8, 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.

• Loopback Binding: A filter level of "..." will not be replaced nor will a '*' found anywhere in the topic level.

• JMS Binding: Depends on the provider instance.

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

Registering Message Factories

A Talon message bus bindings receive messages from the underlying bus in provider specific form but with SMA metadata that is tagged by the corresponding binding in upstream Talon microservices that sent the messages. In order to convert the provider specific message to a message view (object), the receiving binding needs the following:

• The id of the received message

• The id of the received message's factory.

• The factory registered with the Talon SMA runtime.

The first two above are present in the SMA metadata tagged to the inbound message. However, the message factory needs to be registered with the SMA runtime for it to find and use to convert the inbound message to object form. This registration can be done declaratively or programmatically.

Declarative Registration

Message factories are declaratively registered in the applications DDL configuration as follows:

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

Programmatic Registration

Registration can also be done programmatically via the AEP Engine. A common way to do this is to provide an AppInjectionPoint for the AEP Engine 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());
  }
}

Adding a Message Handler

When a message is received by a message bus, it is enqueued into the AEP Engine's inbound event queue for dispatch. A message handler is defined as a method that is annotated with the @EventHandler annotation and has a method signature that accepts just a message type (for State Replicated and Event Sourced microservices) or a message type and the entity type of the microservice's store root. The following are examples of these types of handlers

Single Parameter Handler

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

Two Parameter Handler

The AEP Engine 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 be executed once and exactly once in the life of a microservice.

Preserving Interest on Shutdown

By default, when an AEP 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:

Preserving Interest via configuration

<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)

Removing All Interest Programmatically

@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 Interest 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:

Preserving Per Channel Channel via Configuration

Preserving Interest for a specific channel via configuration

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

Preserving Per Channel Channel Programmatically

Preserving Subscriptions for a specific channel programmatically

@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);
  }
}