> For the complete documentation index, see [llms.txt](https://docs.xplatform.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.xplatform.com/talon/developing-applications/authoring-user-code/message-processing/processing-messages/sending-messages.md).
# 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`](https://build.neeveresearch.com/core/javadoc/LATEST/com/neeve/aep/AepMessageSender.html) which can be injected via the `@AppInjectionPoint` annotation:
```java
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:
```java
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](/talon/developing-applications/authoring-user-code/message-processing/processing-messages/handling-messages/coding-for-zero-garbage.md) for latency-sensitive applications.
### Populating Messages
Once created, messages can be populated using setter methods just like any other POJO:
```java
@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](/talon/developing-applications/authoring-user-code/message-processing/processing-messages/handling-messages/coding-for-zero-garbage.md).
### 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:
```java
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.
{% hint style="warning" %}
Once a message is sent, ownership transfers to the engine and the application must not modify or access the message. See [Programming Fundamentals](/talon/developing-applications/authoring-user-code/message-processing/processing-messages/handling-messages/programming-fundamentals.md#outbound-messages) for details on message lifecycle and ownership.
{% endhint %}
## 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:
```xml
APP/HEARTBEATS
```
### Dynamic Keys with Variable Substitution
Variable portions of a key are denoted using `${variableName}` syntax:
```xml
ORDERS/${region}/${orderType}
```
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:
```xml
ORDERS/${region}/${orderType}
```
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**:
```xml
region
US
shard
1
```
**Programmatic Resolution Table**:
```java
@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:
```java
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}`:
```xml
ORDERS/${region:US}/${orderType:LIMIT}
```
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:
```xml
ORDERS/#[shard=hash(${orderId}, 4)]
```
#### 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**:
```xml
NEWORDERS/#[ordershard=hash(${orderId}, 3)]
ordershard=1|2
```
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:
```xml
com.example.CustomChannelKeyFunctions
```
Implementation:
```java
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:
```java
@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`
{% hint style="warning" %}
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.
{% endhint %}
## 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 `` element:
```xml
true
```
### 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**:
```xml
true
true
```
{% hint style="info" %}
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](#concurrent-sends) for details.
{% endhint %}
### Accessing Sequence Numbers
Applications can access sequence numbers from received messages:
```java
@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`](https://build.neeveresearch.com/core/javadoc/LATEST/com/neeve/aep/event/AepSendStabilityEvent.html) 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**:
```xml
true
true
```
### Send Stability Semantics
An unsolicited send is considered stabilized according to the microservice's [`InboundEventAcknowledgementPolicy`](/talon/reference/configuration.md#policy-configuration):
**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**:
```java
@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:
```java
// 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:
```java
@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:
```xml
activemq
localhost
61616
Guaranteed
${filename}
true
true
```
### 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`](/talon/reference/configuration.md#policy-configuration) and [`MessageSendExceptionHandlingPolicy`](/talon/reference/configuration.md#policy-configuration):
**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:
```xml
```
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:
```xml
true
```
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](#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)
{% hint style="info" %}
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](#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.
{% endhint %}
### 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` |
{% hint style="info" %}
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.
{% endhint %}
## See Also
* [Registering Message Interest](/talon/developing-applications/configuring-messaging/registering-message-interest.md) - Receiving messages
* [Configuring Bus Connections](/talon/developing-applications/configuring-messaging/configuring-bus-connections.md) - Bus and channel configuration
* [Handling Messages](/talon/developing-applications/authoring-user-code/message-processing/processing-messages/handling-messages.md) - Message handler best practices
* [Controlling Transactions](/talon/developing-applications/authoring-user-code/message-processing/processing-messages/handling-messages/controlling-transactions.md) - Transaction management