Message Buffering#

The OUCH 5.0 library provides optional message buffering for egress messages (Client → Exchange), enabling retransmission of unacknowledged messages after reconnection.

Overview#

Message buffering allows you to:

  • Buffer egress messages before sending

  • Track which messages have been acknowledged

  • Automatically retransmit unacknowledged messages after reconnection

  • Implement “benign retransmission” for robust recovery

Enabling Buffering#

Configuration#

Enable message buffering in session settings:

Settings settings;
// ... other settings ...

OutboundMessageBufferSettings buffer_settings;
buffer_settings.max_messages_ = 10000;        // Maximum number of messages to buffer
buffer_settings.max_bytes_ = 10 * 1024 * 1024; // Maximum buffer size (10 MB)
buffer_settings.track_acknowledgments_ = true;  // Track which messages are acknowledged

settings.buffer_settings_ = buffer_settings;
settings.auto_retransmit_on_reconnect_ = true;  // Auto-retransmit on reconnect

Session session(io_ctx, settings);

Buffer Settings#

struct OutboundMessageBufferSettings {
    size_t max_messages_ = 10000;              // Maximum messages to buffer
    size_t max_bytes_ = 10 * 1024 * 1024;     // Maximum buffer size in bytes
    bool track_acknowledgments_ = false;       // Enable acknowledgment tracking (default: false)
};

Checking Buffer Status#

Is Buffering Enabled?#

if (session.is_buffering_enabled()) {
    // Buffering is enabled
}

Get Buffer Instance#

if (auto* buffer = session.get_message_buffer()) {
    size_t message_count = buffer->size();
    size_t bytes_used = buffer->bytes_used();
    size_t unacknowledged = buffer->unacknowledged_count();
}

Acknowledgment Tracking#

When buffering is enabled, you can mark messages as acknowledged when you receive confirmation from the exchange. This is optional - according to the OUCH 5.0 specification, the exchange will ignore retransmitted messages with UserRefNum lower than the last processed one (benign retransmission). Acknowledgment tracking is used to optimize retransmission volume by avoiding retransmission of messages that have already been processed.

Marking Messages as Acknowledged#

void on_order_accepted(Session* s, std::uint64_t seq, const OrderAccepted* msg) {
    if (auto* buffer = s->get_message_buffer()) {
        buffer->mark_acknowledged(msg->get_user_ref_num());
    }
}

void on_order_replaced(Session* s, std::uint64_t seq, const OrderReplaced* msg) {
    if (auto* buffer = s->get_message_buffer()) {
        buffer->mark_acknowledged(msg->get_user_ref_num());
    }
}

void on_order_canceled(Session* s, std::uint64_t seq, const OrderCanceled* msg) {
    if (auto* buffer = s->get_message_buffer()) {
        buffer->mark_acknowledged(msg->get_user_ref_num());
    }
}

Automatic Retransmission#

When auto_retransmit_on_reconnect_ is enabled, unacknowledged messages are automatically retransmitted after reconnection.

How It Works#

  1. Messages are buffered when sent

  2. (Optional) Messages are marked as acknowledged when responses are received

  3. On reconnection, unacknowledged messages are automatically retransmitted

  4. UserRefNum values are preserved (messages are retransmitted with original UserRefNum)

Note: If acknowledgment tracking is disabled or messages are not marked as acknowledged, all buffered messages will be retransmitted. The exchange will ignore messages with UserRefNum lower than the last processed one (benign retransmission per OUCH 5.0 specification), so this is safe but less efficient.

Manual Retransmission#

If you need more control, you can disable automatic retransmission and handle it manually:

settings.auto_retransmit_on_reconnect_ = false;

// After reconnection, manually retransmit
void on_established(Session* s) {
    // Retransmit unacknowledged messages
    std::error_code ec = s->retransmit_unacknowledged();
    if (ec) {
        std::cerr << "Retransmission error: " << ec.message() << std::endl;
    }
}

Buffer Management#

Buffer Size Limits#

The buffer has two limits:

  • Message count limit: Maximum number of messages (max_messages_)

  • Byte size limit: Maximum buffer size in bytes (max_bytes_)

When either limit is reached, the oldest messages may be removed to make room for new ones.

Checking Buffer Status#

if (auto* buffer = session.get_message_buffer()) {
    std::cout << "Buffer status:" << std::endl;
    std::cout << "  Messages: " << buffer->size() << std::endl;
    std::cout << "  Bytes used: " << buffer->bytes_used() << std::endl;
    std::cout << "  Unacknowledged: " << buffer->unacknowledged_count() << std::endl;
}

Best Practices#

  1. Enable for production: Use message buffering in production applications that need retransmission

  2. Mark acknowledgments: Always mark messages as acknowledged when you receive responses

  3. Monitor buffer size: Keep an eye on buffer size to avoid memory issues

  4. Set appropriate limits: Configure max_messages_ and max_bytes_ based on your application’s needs

  5. Handle buffer full: Be prepared to handle cases where the buffer is full

Example: Complete Buffering Setup#

// Configure buffering
OutboundMessageBufferSettings buffer_settings;
buffer_settings.max_messages_ = 10000;
buffer_settings.max_bytes_ = 10 * 1024 * 1024;
buffer_settings.track_acknowledgments_ = true;

Settings settings;
settings.buffer_settings_ = buffer_settings;
settings.auto_retransmit_on_reconnect_ = true;

Session session(io_ctx, settings);

// Attach listeners that mark acknowledgments
session.attach_listener(
    [](Session* s, std::uint64_t seq, const OrderAccepted* msg) {
        if (auto* buffer = s->get_message_buffer()) {
            buffer->mark_acknowledged(msg->get_user_ref_num());
        }
        // ... handle order accepted ...
    },
    [](Session* s, std::uint64_t seq, const OrderReplaced* msg) {
        if (auto* buffer = s->get_message_buffer()) {
            buffer->mark_acknowledged(msg->get_user_ref_num());
        }
        // ... handle order replaced ...
    },
    [](Session* s, std::uint64_t seq, const OrderCanceled* msg) {
        if (auto* buffer = s->get_message_buffer()) {
            buffer->mark_acknowledged(msg->get_user_ref_num());
        }
        // ... handle order canceled ...
    }
);

// Send messages (automatically buffered)
EnterOrder order;
// ... set fields ...
session.send_message(&order);  // Automatically buffered

Thread Safety#

Important: OutboundMessageBuffer is not thread-safe. All methods must be called from the same thread that owns the associated Session’s IOContext.

  • Buffer operations are performed automatically by Session when send_message() is called (which requires IOContext’s thread)

  • Message handlers run on IOContext’s thread, so mark_acknowledged() is safe to call from handlers

  • get_unacknowledged_messages() and retransmit_unacknowledged() must be called from IOContext’s thread

Best Practices#

  1. Enable acknowledgment tracking for production applications to optimize retransmission

  2. Mark messages as acknowledged when you receive confirmations (optional but recommended)

  3. Use auto-retransmit for automatic recovery after reconnection

  4. Set appropriate buffer limits based on your message volume

  5. Monitor buffer status to detect potential issues

  6. Thread safety: All buffer operations are automatically safe when called from IOContext’s thread (same as Session operations)