...

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

Understanding Message Dispatch

...

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. In this case

 

Gliffy Diagram
size 750 name Inbound Message Dispatch

 

Message Acknowledgement

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, 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 features 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 the channel.Unless otherwise specified by a particular channel implementation, a channel filter takes a message channel and 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 "ORDERSNEWORDERS/${Region}/{Department}", one can specify a channel filter of "Region=NAUS|EMEA;Department=Clothing". This would join the channel on:

• ORDERSNEWORDERS/NAUS/Clothing
• ORDERSNEWORDERS/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 "ORDERSNEWORDERS/${Region}/${Department}" and a channel filter of "Region=NAUS|EMEA", it would result in the following subscriptions being issued during join:

• ORDERSNEWORDERS/NAUS/*
• ORDERSNEWORDERS/EMEA/*

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

• ORDERSNEWORDERS/*/*

Registering Message View Factories

...

<apps>
  <app name="MyApp">
    <messaging>
      <factories>
		<factory name="com.examplesample.messages.MyMessageFactoryOrderMessagesFactory" />
		<factory name="com.example.messages.other.MyOtherMessageFactory" />       </factories>
    </messaging>
  </app>
</app>

...

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

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 for the case where an engine shuts down with an error (e.g. AepEngine.stop(Exception) with a non null cause. In this case channel joins are left in tact 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

• Processing Unhandled Messages

...