Handling Messages

This guide shows you how to write message handlers - the methods where all your business logic executes in a Rumi microservice.

Note on Consensus Models: This documentation focuses on Event Sourcing, which is Rumi's primary consensus model. For State Replication examples and guidance, please refer to the Rumi documentation. Rumi is the next major version (4.x) of the Rumi and provides more robust support for State Replication, including enhanced state modeling capabilities and improved developer tooling. To understand both models conceptually, see Consensus Models.

Required Reading: Before writing message handlers, you must read Programming Fundamentals. This page covers essential rules about message immutability, single-threaded store access, and Event Sourcing determinism requirements that all handler code must follow.

Overview

Message handlers are annotated methods that process inbound messages. When a message arrives, the AEP Engine dispatches it to the appropriate handler where your business logic executes. The handler reads data from the inbound message, updates the microservice store, and sends outbound messages.

Writing a Message Handler

Here's a canonical message handler that demonstrates the key elements:

@AppHAPolicy(HAPolicy.EventSourcing)
public class OrderProcessor {

    // Injected by the platform
    private AepMessageSender messageSender;

    // Application-owned store (POJOs)
    private Map<String, Order> orders = new HashMap<>();

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

    /**
     * Message handler for new order messages
     */
    @EventHandler
    public void onNewOrder(NewOrderMessage message) {
        // 1. Read data from inbound message
        String orderId = message.getOrderId();
        String symbol = message.getSymbol();
        int quantity = message.getQuantity();

        // 2. Read and update microservice store (POJOs)
        Order order = orders.get(orderId);
        if (order == null) {
            order = new Order();
            order.setOrderId(orderId);
            orders.put(orderId, order);
        }
        order.setSymbol(symbol);
        order.setQuantity(quantity);
        order.setStatus("PENDING");

        // 3. Create and send outbound message
        OrderAckMessage ack = OrderAckMessage.create();
        ack.setOrderId(orderId);
        ack.setStatus("ACCEPTED");
        ack.setTimestamp(System.currentTimeMillis());

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

        // 4. Handler returns - transaction commits
        // The AEP Engine will:
        //   - Replicate inbound message to backup instances
        //   - Backup replays message to rebuild store
        //   - Establish consensus with cluster members
        //   - Commit the transaction
        //   - Send the outbound message
        //   - Acknowledge the inbound message
    }
}

Handler Signature

A message handler must have this signature:

@EventHandler
public void onMessageName(MessageType message) {
    // handler logic
}

Key points:

• Annotated with @EventHandler

• Must be public

• Return type must be void

• Takes exactly one parameter - the inbound message

• The message parameter type determines which messages this handler processes

Handler Method Names

Method names are not significant - the handler is matched to messages by the parameter type. However, following a naming convention like onMessageType makes code more readable.

Multiple Handlers for Same Message Type

You can have multiple handlers for the same message type:

@EventHandler
public void validateOrder(NewOrderMessage message) {
    // Validation logic
}

@EventHandler
public void recordOrder(NewOrderMessage message) {
    // Recording logic
}

Both handlers will be invoked for each NewOrderMessage. The order of invocation is deterministic but should not be relied upon - handlers should be independent.

Reading Inbound Messages

Access message fields using the generated getter methods:

@EventHandler
public void onNewOrder(NewOrderMessage message) {
    String orderId = message.getOrderId();
    String symbol = message.getSymbol();
    int quantity = message.getQuantity();
    double price = message.getPrice();

    // Process the data...
}

Important: Inbound messages are read-only. See Programming Fundamentals for rules about message immutability and lifecycle.

Accessing Microservice Store

With Event Sourcing, your microservice store consists of POJOs that you manage directly. The store is private to your application and transparent to the Rumi runtime:

@AppHAPolicy(HAPolicy.EventSourcing)
public class OrderProcessor {

    // Application-owned store (POJOs)
    private Map<String, Order> orders = new HashMap<>();
    private Map<String, Position> positions = new HashMap<>();

    @EventHandler
    public void onNewOrder(NewOrderMessage message) {
        // Access microservice store directly
        Order order = orders.get(message.getOrderId());

        if (order == null) {
            // Create new POJO
            order = new Order();
            order.setOrderId(message.getOrderId());
            orders.put(order.getOrderId(), order);
        }

        // Update data in store
        order.setQuantity(message.getQuantity());
        order.setStatus("PENDING");
    }
}

Key points:

• Store is your own POJOs, not ADM-generated

• Consensus established by replaying inbound messages on backup instances

• Store rebuilt on backup by replaying events

• Your business logic must be deterministic

Store access rules:

• Store can only be accessed from within a message handler (on the dispatch thread)

• Store changes must be deterministic - no reliance on external state like system time or random numbers

• See Programming Fundamentals for threading restrictions

• See Event Sourcing Template for determinism requirements

Sending Outbound Messages

Create and send messages using the AepMessageSender:

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

@EventHandler
public void onNewOrder(NewOrderMessage message) {
    // Process order...
    Order order = orders.get(message.getOrderId());
    // ... update store ...

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

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

When messages are sent: Outbound messages are not immediately sent when you call sendMessage(). Instead:

1. The message is queued

2. The handler returns

3. The inbound message is replicated to backup instances

4. Backup instances replay the message to rebuild store

5. Consensus is established

6. The transaction commits

7. Then the outbound message is sent

This ensures that messages are only sent if the transaction succeeds, providing exactly-once semantics.

See Sending Messages for details on message keys, channels, and unsolicited sends.

Transaction Lifecycle

When a message handler executes, it runs within a transaction:

@EventHandler
public void onNewOrder(NewOrderMessage message) {
    // Transaction starts (automatically)

    // 1. Read message data
    String orderId = message.getOrderId();

    // 2. Update store (POJO)
    Order order = new Order();
    order.setOrderId(orderId);
    orders.put(orderId, order);

    // 3. Queue outbound messages
    OrderAckMessage ack = OrderAckMessage.create();
    ack.setOrderId(orderId);
    messageSender.sendMessage("order-acks", ack);

    // Handler returns

    // 4. Transaction commits (automatically):
    //    - Inbound message replicated to cluster
    //    - Backup replays message to rebuild store
    //    - Consensus established
    //    - Outbound messages sent
    //    - Inbound message acknowledged
}

For details on how consensus works, see Cluster Consensus.

For advanced transaction control, see Controlling Transactions.

Common Patterns

Pattern: Lookup or Create

@EventHandler
public void onOrderUpdate(OrderUpdateMessage message) {
    String orderId = message.getOrderId();

    // Get existing or create new
    Order order = orders.get(orderId);
    if (order == null) {
        order = new Order();
        order.setOrderId(orderId);
        orders.put(orderId, order);
    }

    // Update
    order.setQuantity(message.getQuantity());
}

Pattern: Conditional Send

@EventHandler
public void onOrderUpdate(OrderUpdateMessage message) {
    Order order = orders.get(message.getOrderId());

    if (order != null) {
        order.setQuantity(message.getQuantity());

        // Send notification only if quantity exceeds threshold
        if (order.getQuantity() > 1000) {
            LargeOrderAlert alert = LargeOrderAlert.create();
            alert.setOrderId(order.getOrderId());
            alert.setQuantity(order.getQuantity());
            messageSender.sendMessage("alerts", alert);
        }
    }
}

Pattern: Aggregate and Send

@EventHandler
public void onTrade(TradeMessage message) {
    String symbol = message.getSymbol();

    // Update running totals in store
    DailyStats stats = dailyStats.get(symbol);
    if (stats == null) {
        stats = new DailyStats();
        stats.setSymbol(symbol);
        dailyStats.put(symbol, stats);
    }

    stats.setVolume(stats.getVolume() + message.getQuantity());
    stats.setTradeCount(stats.getTradeCount() + 1);

    // Send periodic snapshot
    if (stats.getTradeCount() % 100 == 0) {
        StatsSnapshot snapshot = StatsSnapshot.create();
        snapshot.setSymbol(symbol);
        snapshot.setVolume(stats.getVolume());
        snapshot.setTradeCount(stats.getTradeCount());
        messageSender.sendMessage("stats-snapshots", snapshot);
    }
}

Pattern: Forwarding Messages

You cannot resend an inbound message directly. To forward a message, copy it first:

@EventHandler
public void onOrder(OrderMessage message) {
    // Cannot do this:
    // messageSender.send("mirror", message); // ERROR!

    // Must copy first:
    OrderMessage copy = message.copy();
    messageSender.sendMessage("mirror", copy);
}

See Programming Fundamentals for details.

Advanced Topics

Using Savepoints

For long handlers or handlers that may fail partway through, you can use savepoints to commit work incrementally:

@EventHandler
public void onBatch(BatchMessage message, MessageView view) {
    for (int i = 0; i < message.getItemCount(); i++) {
        // Process item
        processItem(message.getItem(i));

        // Savepoint every 100 items
        if (i % 100 == 0) {
            view.setSavePoint();
        }
    }
}

See Using Savepoints for details.

Zero Garbage Programming

For ultra-low-latency applications, you can eliminate garbage collection pauses using zero-garbage techniques:

@EventHandler
public void onOrder(OrderMessage message) {
    // Use XStrings instead of Strings
    XString symbol = message.getSymbol();  // No allocation

    // Use iterators instead of for-each
    XIterator<OrderLine> iter = message.iterateLines();
    while (iter.hasNext()) {
        OrderLine line = iter.next();  // No allocation
        // Process line...
    }
}

See Coding for Zero Garbage for details.

See Also

• Programming Fundamentals - Core rules for message handlers

• Sending Messages - Creating and sending outbound messages

• Controlling Transactions - Advanced transaction control

• Coding for Zero Garbage - Ultra-low-latency techniques

• Event Sourcing Template - Event Sourcing model and requirements

• Cluster Consensus - How consensus works (conceptual)