Sending Messages

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

Overview

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 resolve 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 microservice's store (e.g., replication and transaction log write)

5. Once the microservice's store 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

Basic Example

To send a message, the microservice needs an AepMessageSender which can be injected via the @AppInjectionPoint annotation:

public class OrderProcessor {
    private AepMessageSender messageSender;

    @AppInjectionPoint
    public void setMessageSender(AepMessageSender messageSender) {
        this.messageSender = messageSender;
    }

    @EventHandler
    public void onNewOrder(NewOrderMessage message) {
        // Create acknowledgment message
        OrderAckMessage ack = OrderAckMessage.create();
        ack.setOrderId(message.getOrderId());
        ack.setStatus("ACCEPTED");

        // Send on the 'order-acks' channel
        messageSender.sendMessage("order-acks", ack);
    }
}

The AepMessageSender provides several send methods:

Method	Description
sendMessage(String channel, MessageView message)	Sends a message on the named channel
sendMessage(String bus, String channel, MessageView message)	Sends a message on a specific bus and channel
sendMessage(String bus, String channel, MessageView message, String key)	Sends with a caller-provided key (advanced)

Creating Messages For Send

Message types generated by ADM have no public constructors. Instances are created via the static create() factory method on the message class:

OrderAckMessage ack = OrderAckMessage.create();
ack.setOrderId(message.getOrderId());
ack.setStatus("ACCEPTED");

By routing all message creation through create() factory methods, Talon can transparently switch between pooled and non-pooled allocation of messages without requiring changes to business logic. This enables zero-garbage operation for latency-sensitive applications.

Populating Messages

Once created, messages can be populated using setter methods just like any other POJO:

@EventHandler
public void onNewOrder(NewOrderMessage message) {
    // Create and populate the message
    OrderAckMessage ack = OrderAckMessage.create();
    ack.setOrderId(message.getOrderId());
    ack.setTimestamp(System.currentTimeMillis());
    ack.setStatus("ACCEPTED");

    // Send the message
    messageSender.sendMessage("order-acks", ack);
}

For advanced encoding-specific population mechanisms (e.g., Xbuf), see Coding for Zero Garbage.

Disposing Unsent Messages

If a message is created but not sent, the application should call dispose() on the message to return it to the pool:

OrderAckMessage ack = OrderAckMessage.create();
ack.setOrderId(orderId);

// Validation fails - don't send
if (!isValid(ack)) {
    ack.dispose();  // Return to pool
    return;
}

messageSender.sendMessage("order-acks", ack);

Failing to dispose of unsent messages will not cause memory leaks, but will prevent subsequent create() calls from reusing pooled objects, potentially impacting allocation rates.

Once a message is sent, ownership transfers to the engine and the application must not modify or access the message. See Programming Fundamentals for details on message lifecycle and ownership.

Message Keys and Topic Resolution

When a message is sent, the engine resolves the physical topic or destination using the channel's configured message key. The message key can contain static values and dynamic variables that are resolved from the message being sent.

Static Keys

A static key does not contain variable substitutions:

<channel name="heartbeats">
  <key>APP/HEARTBEATS</key>
</channel>

Dynamic Keys with Variable Substitution

Variable portions of a key are denoted using ${variableName} syntax:

<channel name="order-events">
  <key>ORDERS/${region}/${orderType}</key>
</channel>

When a message is sent, the engine resolves variables by:

1. Looking up the value on the message (e.g., message.getRegion())

2. Using values from a configured key resolution table

3. Using the default value if specified in the key

Message-Based Key Resolution

The most common approach is to resolve key variables from fields on the message:

<channel name="order-events">
  <key>ORDERS/${region}/${orderType}</key>
</channel>

When sending an OrderEventMessage with:

• region = "US"

• orderType = "MARKET"

The resolved topic would be: ORDERS/US/MARKET

Key Resolution Tables

For variables not present on the message, applications can configure a key resolution table:

Configuration-Based Resolution Table:

<app name="order-processor">
  <messaging>
    <buses>
      <bus name="order-bus">
        <channels>
          <channel name="order-events" join="false">
            <keyResolutionTable>
              <entry>
                <key>region</key>
                <value>US</value>
              </entry>
              <entry>
                <key>shard</key>
                <value>1</value>
              </entry>
            </keyResolutionTable>
          </channel>
        </channels>
      </bus>
    </buses>
  </messaging>
</app>

Programmatic Resolution Table:

@EventHandler
public void onChannelUp(AepChannelUpEvent event) {
    MessageChannel channel = event.getMessageChannel();

    // Properties-based table
    Properties krt = new Properties();
    krt.setProperty("region", "EMEA");
    krt.setProperty("shard", "1");
    channel.setKeyResolutionTable(krt);
}

Raw (Zero Garbage) Resolution Table:

For performance-sensitive applications using Xbuf encoding:

private final RawKeyResolutionTable globalKeyResolutionTable =
    RawKeyResolutionTable.create(64);

@AppEventHandlerInit
public void init() {
    krt.put("Region", XString.create("EMEA"));
    krt.put("Shard", XString.create("1"));
}

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

When using a RawKeyResolutionTable:

• The table should not be modified after being set on a channel

• XString values must not be modified after insertion

• Cannot use both Raw and Properties-based tables on the same channel

Default Values

Variables can specify default values using the syntax ${variableName:defaultValue}:

<channel name="order-events">
  <key>ORDERS/${region:US}/${orderType:LIMIT}</key>
</channel>

If region cannot be resolved from the message or key resolution table, "US" will be used.

Channel Key Functions

Channel key functions enable dynamic computation of key portions at runtime. Functions are declared using #[variableName = functionName(arg1, argN)] syntax:

<channel name="order-events">
  <key>ORDERS/#[shard=hash(${orderId}, 4)]</key>
</channel>

Built-In Functions

Function	Description	Arguments
hash	Hashes value into N partitions	Arg1: value to hash Arg2: number of partitions Returns: 1 through N
env	Looks up runtime property	Arg1: property name Arg2 (optional): default value
concat	Concatenates two strings	Arg1: string1 Arg2: string2

Hash Function Example:

<buses>
  <bus name="order-bus" descriptor="solace://solhost:55555">
    <channels>
      <channel name="new-orders">
        <key>NEWORDERS/#[ordershard=hash(${orderId}, 3)]</key>
      </channel>
    </channels>
  </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 this example:

• The sender sends on topics: NEWORDERS/1, NEWORDERS/2, or NEWORDERS/3

• The receiver subscribes to: NEWORDERS/1 and NEWORDERS/2 (but not 3)

Custom Channel Key Functions

Applications can define custom channel key functions by setting nv.sma.channelkeyfunctioncontainers to a comma-separated list of classes containing channel key functions.

Requirements for Custom Functions:

• Must be declared public static

• First argument must be the key being resolved (to which the function appends)

• May accept additional XString arguments representing function arguments from the channel key

Example:

Configuration:

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

Implementation:

public class CustomChannelKeyFunctions {
    /**
     * A custom key function that substitutes 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 message fields, it's important to validate that values don't introduce illegal characters or empty topic levels. Configure validation using these properties:

Property	Default	Description
nv.sma.maxresolvedkeylength	0	Maximum resolved key length. When > 0, validates key length. Set to lowest max length across all bindings for portability.
nv.sma.cleanmessagekey	false	When true, replaces non-alphanumeric characters in variable values with underscore. For example, "Asia/Pac" becomes "Asia_Pac". Does not apply to channel key functions.
nv.sma.allowemptykeyfield	false	When false, empty string values in key variables cause resolution to fail. Does not apply to channel key functions.
nv.sma.treatemptykeyfieldasnull	false	When true, treats empty strings as null, allowing fallback to other sources or default values. Takes precedence over allowemptykeyfield.
nv.sma.validatemessagekey	false	When true, enables message key validation before sending.

Caller-Provided Keys

Applications can provide the resolved key directly in the send call:

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

    // Provide custom topic directly
    messageSender.sendMessage("order-processing-bus",
                             "order-event-channel",
                             event,
                             "CUSTOMORDEREVENTS/" + message.getOrderType());
}

When using caller-provided keys:

• Variable substitution is not performed on the provided key

• Key validation is performed

• The key will be prefixed with the channel name if topic_starts_with_channel=true

Caller-provided keys should be used with care as they break the ability to change topics at runtime and receivers won't know how to subscribe to such sends.

Message Sequencing

The AEP Engine assigns monotonically increasing sequence numbers to outbound messages to enable receivers to detect message loss or out-of-order delivery. Sequence numbers are managed per bus+channel+qos combination.

Enabling Sequence Numbers

Sequence numbers are enabled per channel using the <sequenceMessages> element:

<app name="order-processor">
  <messaging>
    <buses>
      <bus name="order-bus">
        <channels>
          <channel name="order-events" join="false">
            <sequenceMessages>true</sequenceMessages>
          </channel>
        </channels>
      </bus>
    </buses>
  </messaging>
</app>

Solicited vs Unsolicited Sends

By default, sequence numbering behavior differs between solicited and unsolicited sends:

Solicited sends (in-transaction sends):

• Sequence numbers are set by default

• Controlled by setOutboundSequenceNumbers (default: true)

Unsolicited sends (out-of-transaction sends):

• Sequence numbers are not set by default

• Controlled by both setOutboundSequenceNumbers (default: true) and sequenceUnsolicitedSends (default: false)

• To enable sequence numbers on unsolicited sends, set sequenceUnsolicitedSends=true

Configuration example:

<app name="gateway-app">
  <messaging>
    <!-- messaging config -->
  </messaging>

  <!-- Enable sequence numbers on unsolicited sends -->
  <sequenceUnsolicitedSends>true</sequenceUnsolicitedSends>

  <!-- Global control for all outbound sequence numbers (default: true) -->
  <setOutboundSequenceNumbers>true</setOutboundSequenceNumbers>
</app>

When using sequenceUnsolicitedWithSolicitedSends=true for concurrent send safety, unsolicited sends are converted to solicited sends and thus follow solicited send sequencing behavior (controlled only by setOutboundSequenceNumbers). See Concurrent Sends for details.

Accessing Sequence Numbers

Applications can access sequence numbers from received messages:

@EventHandler
public void onOrderEvent(OrderEventMessage message) {
    long seqNo = message.getMessageMetadata().getSequenceNumber();
    // Check for gaps or out-of-order delivery
}

Unsolicited Sends

An unsolicited send is a message sent from outside of a message handler - not in response to a received message. Unsolicited sends are common in gateway applications that input messages from external sources.

Overview

When messages are sent from within a message handler (solicited sends), the AEP Engine automatically manages send stability and ensures zero-loss delivery across failovers. For unsolicited sends, additional configuration and application participation may be required to ensure zero-loss delivery.

Send Stability Tracking

The AEP Engine provides AepSendStabilityEvent to notify applications when an unsolicited send has been stabilized and is guaranteed to be delivered. This allows gateway applications to track which messages have been successfully delivered.

Enable Send Stability Events:

<app name="file-gateway" mainClass="com.example.FileGateway">
  <messaging>
    <!-- ... -->
  </messaging>
  <dispatchSendStabilityEvents>true</dispatchSendStabilityEvents>
  <sequenceUnsolicitedWithSolicitedSends>true</sequenceUnsolicitedWithSolicitedSends>
</app>

Send Stability Semantics

An unsolicited send is considered stabilized according to the microservice's InboundEventAcknowledgementPolicy:

Sent in transaction stream (sequenceUnsolicitedWithSolicitedSends=true):

• OnSendStability: Stabilized when all messages in the transaction (and prior transactions) are acknowledged

• OnStoreStability: Stabilized when the transaction has been replicated to backup and/or transaction log

Sent outside transaction stream (sequenceUnsolicitedWithSolicitedSends=false and replicateUnsolicitedSends=false):

• Stabilized on receipt of acknowledgment from the message bus

Correlating Send Stability

Applications need to correlate stability events with the source data that triggered the send. Use one of these approaches:

1. Use Message Fields:

@EventHandler
public void onSendStability(AepSendStabilityEvent event) {
    FileLineMessage message = (FileLineMessage) event.getMessage();
    long lineNumber = message.getLineNumber();
    // Update cursor file, acknowledge upstream, etc.
}

2. Use Message Attachments:

When correlation data shouldn't be in the message payload:

// During send
FileLineMessage message = FileLineMessage.create();
message.setText(line);
message.setAttachment(new Long(lineNumber));
messageSender.sendMessage("line-input", message);

// On stability
@EventHandler
public void onSendStability(AepSendStabilityEvent event) {
    FileLineMessage message = (FileLineMessage) event.getMessage();
    long lineNumber = (Long) message.getAttachment();
    // Process stability for this line number
}

Example: File Gateway with Send Stability

The following example shows a file gateway that reads lines from a file and publishes each line. It tracks send stability to maintain a cursor file:

@AppHAPolicy(value = HAPolicy.StateReplication)
public class FileInputGateway {
    Tracer tracer = Tracer.create("file-tailer", Level.INFO);
    private volatile AepMessageSender messageSender;
    private volatile AepEngine engine;

    @Configured(property = "gateway.filename")
    private volatile String filename;

    private RandomAccessFile cursor;
    private volatile long stabilizedLine = 0;

    @AppInjectionPoint
    public void injectMessageSender(AepMessageSender messageSender) {
        this.messageSender = messageSender;
    }

    @AppInjectionPoint
    public void injectAepEngine(AepEngine engine) {
        this.engine = engine;
    }

    @AppMain
    public void run(String[] args) throws Exception {
        BufferedReader reader = new BufferedReader(new FileReader(filename));
        try {
            cursor = new RandomAccessFile(filename + ".cursor", "rwd");
            cursor.setLength(8);
            cursor.seek(0);

            // Wait for messaging to start
            engine.waitForMessagingToStart();

            // Read and send lines
            String line = null;
            stabilizedLine = cursor.readLong();
            int lineNumber = 0;

            while ((line = reader.readLine()) != null) {
                if (++lineNumber <= stabilizedLine) {
                    continue;
                }

                FileLineMessage message = FileLineMessage.create();
                message.setText(line);
                message.setFileName(filename);
                message.setLineNumber(lineNumber);
                message.setAttachment(new Long(message.getLineNumber()));
                messageSender.sendMessage("line-input", message);
            }

            // Wait for send stability before exiting
            while (stabilizedLine < lineNumber) {
                Thread.sleep(100);
            }
        }
        finally {
            reader.close();
            cursor.close();
        }
    }

    @EventHandler
    public final void onSendStability(AepSendStabilityEvent event)
            throws IOException {
        FileLineMessage message = (FileLineMessage) event.getMessage();
        stabilizedLine = (Long) message.getAttachment();
        cursor.seek(0);
        cursor.writeLong(stabilizedLine);
        cursor.getFD().sync();
        tracer.log("Got stability for: " + stabilizedLine, Level.INFO);
    }

    @AppStat(name = "Lines Sent")
    public long getLinesSent() {
        return stabilizedLine;
    }
}

Configuration:

<model xmlns="http://www.neeveresearch.com/schema/x-ddl">
  <buses>
    <bus name="file-connector">
      <provider>activemq</provider>
      <address>localhost</address>
      <port>61616</port>
      <channels>
        <channel name="line-input">
          <qos>Guaranteed</qos>
          <key>${filename}</key>
        </channel>
      </channels>
    </bus>
  </buses>

  <apps>
    <app name="file-input-gateway"
         mainClass="com.sample.FileInputGateway">
      <messaging>
        <factories>
          <factory name="com.sample.messages.FileGatewayMessageFactory"/>
        </factories>
        <buses>
          <bus name="file-connector">
            <channels>
              <channel name="line-input" join="false"/>
            </channels>
          </bus>
        </buses>
      </messaging>
      <dispatchSendStabilityEvents>true</dispatchSendStabilityEvents>
      <sequenceUnsolicitedWithSolicitedSends>true</sequenceUnsolicitedWithSolicitedSends>
    </app>
  </apps>
</model>

Send Exception Handling

Exceptions can occur during send calls or asynchronously via negative acknowledgments (nacks) from the messaging provider. Handling depends on configuration:

Sent outside transaction stream:

• Send Exception: Thrown to caller, no AepSendStabilityEvent dispatched

• Nack: Reported in AepSendStabilityEvent via non-null getStatus()

Sent in transaction stream:

Exception handling depends on InboundEventAcknowledgementPolicy and MessageSendExceptionHandlingPolicy:

OnSendStability:

• Send failure with LogExceptionAndContinue: Exception logged and reported in AepSendStabilityEvent

• Send failure with TreatAsFatal: Engine shuts down, exception in AepEngineStoppedEvent

• Nack: Engine shuts down, exception in AepEngineStoppedEvent

OnStoreStability:

• AepSendStabilityEvent dispatched after replication

• Send failure with LogExceptionAndContinue: Exception logged (not in stability event as it was already dispatched)

• Send failure with TreatAsFatal: Engine shuts down

• Nack: Engine shuts down

HA Considerations for Unsolicited Sends

• EventSourcing: Unsolicited sends are not replicated or persisted in Event Sourcing mode. Use State Replication for unsolicited sends requiring durability.

• Failover: Engine does not dispatch AepSendStabilityEvent for recovered sends after failover

• At Least Once Delivery: Unsolicited sends provide at-least-once delivery semantics. Downstream applications must handle potential duplicates.

Additional Considerations

• Threading: AepSendStabilityEvent is dispatched on the engine's multiplexer thread. Applications performing expensive or blocking operations should use a separate thread.

• Event Ordering:

  • With sequenceUnsolicitedWithSolicitedSends=true: Stability events issued in send order

  • With sequenceUnsolicitedWithSolicitedSends=false: Stability events issued in acknowledgment order (usually ordered per bus+channel+qos)

Concurrent Sends

The Talon runtime ensures single-threaded operation for solicited sends (in-transaction sends). However, unsolicited sends may introduce concurrency that requires additional configuration when:

• Multiple unsolicited sends are performed concurrently using multiple threads, or

• Unsolicited sends are performed concurrently with solicited sends

Thread Safety Configuration

Talon provides two alternative mechanisms for ensuring concurrent send safety:

Option 1: Bus-Level Concurrent Sends (Recommended)

Configure the message bus binding for concurrent sends using the enable_concurrent_sends connection descriptor property:

<buses>
  <bus name="sample-bus" descriptor="activemq://localhost:61616?enable_concurrent_sends=true">
    <channels>
      <!-- channel configuration -->
    </channels>
  </bus>
</buses>

This is the preferred, modern approach applicable to all message bus bindings supported by Talon. It ensures correct operation for both concurrent send scenarios:

• Multiple unsolicited sends performed concurrently using multiple threads

• Unsolicited sends performed concurrently with solicited sends

Option 2: Transaction Pipeline Integration (Legacy)

Use the sequenceUnsolicitedWithSolicitedSends configuration parameter:

<model>
  <buses>
    <bus name="sample-bus" descriptor="activemq://localhost:61616">
      <channels>
        <!-- channel configuration -->
      </channels>
    </bus>
  </buses>

  <apps>
    <app name="sample-app" mainClass="com.sample.SampleApp">
      <messaging>
        <!-- messaging config -->
      </messaging>

      <!-- Alternative: inject unsolicited sends into transaction pipeline -->
      <sequenceUnsolicitedWithSolicitedSends>true</sequenceUnsolicitedWithSolicitedSends>
    </app>
  </apps>
</model>

This legacy approach injects unsolicited sends into the engine's transaction pipeline, effectively converting them to solicited sends. This achieves concurrent send safety as a side effect of transaction pipeline integration.

Key difference between the two approaches:

The approaches differ in how sequence numbers are applied to unsolicited sends. See Message Sequencing for details on sequence number configuration.

With enable_concurrent_sends:

• Unsolicited sends remain out-of-transaction

• Follow normal unsolicited send sequencing behavior (no sequence numbers by default)

With sequenceUnsolicitedWithSolicitedSends:

• Unsolicited sends are converted to solicited sends (injected into transaction pipeline)

• Follow solicited send sequencing behavior (sequence numbers set by default)

Both enable_concurrent_sends and sequenceUnsolicitedWithSolicitedSends are alternative mechanisms for achieving concurrent send safety. They are equivalent except for the default behavior of sequence numbering on unsolicited sends (see Message Sequencing). Setting both is safe but not necessary. The enable_concurrent_sends approach is preferred for new applications, while sequenceUnsolicitedWithSolicitedSends remains available for applications that prefer the transaction injection approach.

Configuration Summary

The following table summarizes how to configure Talon for concurrent sends in various scenarios:

Solicited Sends	Concurrent Unsolicited Sends	Sequence Number in Solicited Sends	Sequence Number in Unsolicited Sends	Configuration
No	Yes	N/A	Yes or No	enable_concurrent_sends=true
Yes	Yes	Yes	No	enable_concurrent_sends=true
Yes	Yes	No	Yes	enable_concurrent_sends=true
Yes	Yes	No	No	enable_concurrent_sends=true
Yes	Yes	Yes	Yes	sequenceUnsolicitedWithSolicitedSends=true

When using sequence numbers with concurrent sends, Talon ensures that messages are transmitted on the wire in the same sequence as the sequence numbers assigned to them.

See Also

• Registering Message Interest - Receiving messages

• Configuring Bus Connections - Bus and channel configuration

• Handling Messages - Message handler best practices

• Controlling Transactions - Transaction management