Aeron Cluster C++ Client

A high-performance, production-ready C++ client library for Real Logic's Aeron Cluster with comprehensive SBE (Simple Binary Encoding) support, advanced pub/sub capabilities, and commit tracking for resilient message processing.

✨ Features

🚀 High Performance: >100k messages/second throughput with <1ms latency
🔄 Full SBE Compliance: Compatible with Go/Java Aeron implementations
🎯 Session Management: Automatic leader detection and failover with exponential backoff
📦 Order Publishing: Built-in support for trading order workflows
🛡️ Production Ready: Comprehensive error handling and automatic reconnection
🔧 Modern C++: C++17 with clean, extensible architecture
⚡ Easy Integration: CMake build system with minimal dependencies
📝 Commit Tracking: Resume message processing from last committed offset
🎭 Topic-Based Pub/Sub: Subscribe to specific topics with message identifier filtering
🔁 Resilient Reconnection: Smart reconnection with exponential backoff and connection state callbacks
🛠️ Developer Tools: Built-in message inspector and cluster monitor utilities

🚀 Quick Start

# Clone and build
git clone https://github.com/reverb-sys/aeron-cluster-client-cpp.git
cd aeron-cluster-client-cpp
./scripts/setup_project.sh && ./scripts/build.sh

# Run example (requires Aeron Media Driver)
./build/examples/basic_client_example

Simple Publisher Example

#include <aeron_cluster/cluster_client.hpp>

int main() {
    // Configure client using builder pattern
    auto config = aeron_cluster::ClusterClientConfigBuilder()
        .with_cluster_endpoints({"localhost:9002", "localhost:9102", "localhost:9202"})
        .with_aeron_dir("/dev/shm/aeron")
        .with_response_timeout(std::chrono::milliseconds(5000))
        .with_max_retries(3)
        .build();
    
    // Create and connect client
    aeron_cluster::ClusterClient client(config);
    if (client.connect()) {
        // Publish an order
        auto order = aeron_cluster::ClusterClient::create_sample_limit_order(
            "ETH", "USDC", "BUY", 1.0, 3500.0);
        std::string messageId = client.publish_order_to_topic(order, "order_request_topic");
        
        // Poll for responses
        client.poll_messages(10);
    }
    
    return 0;
}

Simple Subscriber Example

#include <aeron_cluster/cluster_client.hpp>

int main() {
    auto config = aeron_cluster::ClusterClientConfigBuilder()
        .with_cluster_endpoints({"localhost:9002", "localhost:9102", "localhost:9202"})
        .with_aeron_dir("/dev/shm/aeron")
        .build();
    
    aeron_cluster::ClusterClient client(config);
    
    // Set up message callback
    client.set_message_callback([](const aeron_cluster::ParseResult& result) {
        if (result.is_order_message()) {
            std::cout << "Received order: " << result.payload << std::endl;
        }
    });
    
    if (client.connect()) {
        // Subscribe to topic with message identifier
        client.send_subscription_request("order_notification_topic", 
                                        "MY_IDENTIFIER", 
                                        "LAST_COMMIT", 
                                        "instance_1");
        
        // Poll for messages
        while (client.is_connected()) {
            client.poll_messages(10);
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
    }
    
    return 0;
}

💡 Why This Library?

Production-Ready from Day One

Unlike basic Aeron examples, this library provides:

✅ Battle-tested reconnection logic with exponential backoff
✅ Message deduplication to handle at-least-once delivery
✅ Commit tracking for resilient message processing
✅ Comprehensive error handling for production deployments
✅ Built-in monitoring tools for debugging and operations

Complete Feature Set

📝 5 working examples covering basic to advanced patterns
🛠️ 2 developer tools for debugging and monitoring
📊 Connection statistics and performance tracking
🔁 Automatic reconnection with connection state callbacks
🎯 Topic-based routing with message identifier filtering
📖 Inline documentation in all header files

Compatible & Extensible

🔄 SBE compliant - Works with Go/Java Aeron implementations
🏗️ Clean architecture - Easy to extend and customize
⚡ Modern C++17 - Type-safe and performant
🎨 Builder pattern - Intuitive configuration API

📚 Documentation

Resource	Description
Getting Started	Step-by-step setup guide with platform-specific instructions
Examples	Five comprehensive examples with detailed documentation
Header Documentation	Inline API documentation in header files
Basic Example	Start here - Perfect for understanding the code flow
Pub/Sub Test	Advanced reconnection and filtering patterns
Tools Documentation	Message inspector and cluster monitor usage

Quick Reference

// Core APIs
ClusterClient client(config);              // Create client
client.connect()                           // Connect to cluster
client.publish_order_to_topic(order, topic) // Publish message
client.send_subscription_request(...)      // Subscribe to topic
client.set_message_callback(callback)      // Set message handler
client.poll_messages(n)                    // Poll for messages
client.get_connection_stats()              // Get statistics
client.disconnect()                        // Clean disconnect

// Commit Management
client.commit_message(...)                 // Commit message offset
client.get_last_commit(topic, identifier)  // Get last commit
client.resume_from_last_commit(...)        // Resume from commit

// Configuration Builder
auto config = ClusterClientConfigBuilder()
    .with_cluster_endpoints({...})
    .with_aeron_dir(path)
    .with_response_timeout(ms)
    .with_max_retries(n)
    .build();

🏗️ Installation

Prerequisites

Compiler: GCC 7+, Clang 6+, or MSVC 2019+
CMake: 3.16 or later
Dependencies: Aeron C++, JsonCpp

Ubuntu/Debian

sudo apt-get install cmake build-essential libjsoncpp-dev
git clone https://github.com/reverb-sys/aeron-cluster-client-cpp.git
cd aeron-cluster-client-cpp
./scripts/setup_project.sh && ./scripts/build.sh

macOS

brew install cmake jsoncpp
git clone https://github.com/reverb-sys/aeron-cluster-client-cpp.git
cd aeron-cluster-client-cpp
./scripts/setup_project.sh && ./scripts/build.sh

Build Output

After building, you'll find:

build/
├── libaeron-cluster-cpp.a          # Main static library
├── examples/
│   ├── basic_client_example        # Basic publisher/subscriber
│   ├── pubsub_reconnect_test       # Reconnection test suite
│   ├── commit_resume_example       # Commit tracking demo
│   ├── order_publishing_example    # Order publishing
│   └── advanced_features_example   # Advanced features
└── tools/
    ├── message_inspector            # SBE message debugger
    └── cluster_monitor              # Cluster health monitor

Build Options

# Build with examples (default: ON)
./scripts/build.sh

# Build with tests
./scripts/build.sh --with-tests

# Build in debug mode
cmake -DCMAKE_BUILD_TYPE=Debug -B build
cmake --build build

# Build production optimized
./scripts/build_production.sh

See GETTING_STARTED.md for detailed installation instructions on all platforms.

🎯 Examples

The library includes comprehensive examples demonstrating various use cases:

1. Basic Client Example

File: examples/basic_client_example.cpp

A complete example showing connection, order publishing, and subscription modes. This example is perfect for testing basic code flow and understanding the client lifecycle.

# Run as publisher (default)
./build/examples/basic_client_example --orders 5 --interval 1000

# Run as subscriber
./build/examples/basic_client_example --client-type subscriber

# Check cluster connectivity
./build/examples/basic_client_example --check-only

Features demonstrated:

Pre-flight checks for Aeron Media Driver
Publisher and subscriber modes
Message callbacks and acknowledgment handling
Connection statistics tracking
Graceful shutdown and error handling

2. Pub/Sub Reconnection Test

File: examples/pubsub_reconnect_test.cpp

Advanced test suite for validating reconnection behavior and message identifier filtering.

# Reconnection test (default)
./build/examples/pubsub_reconnect_test --messages 100 --disconnect-at 30

# Message identifier filtering test
./build/examples/pubsub_reconnect_test --test-mode identifier-filter \
    --subscribe-to IDENTIFIER_A \
    --identifiers IDENTIFIER_A,IDENTIFIER_B,IDENTIFIER_C

Two test modes:

Reconnect Test: Validates that subscribers can disconnect, reconnect, and receive all messages
Identifier Filter Test: Ensures subscribers only receive messages for their specific identifier

Features demonstrated:

Automatic reconnection with exponential backoff
Message deduplication tracking
Multi-identifier publishing
Connection state callbacks
Commit-based message replay

3. Commit/Resume Example

File: examples/commit_resume_example.cpp

Demonstrates commit tracking for reliable message processing.

./build/examples/commit_resume_example

Features demonstrated:

Committing message offsets
Resuming from last committed position
Topic subscription/unsubscription
Manual vs automatic commit handling

4. Order Publishing Example

File: examples/order_publishing_example.cpp

Focused example for publishing trading orders with various order types.

./build/examples/order_publishing_example

5. Advanced Features Example

File: examples/advanced_features_example.cpp

Showcases advanced client capabilities and performance optimizations.

./build/examples/advanced_features_example

All examples support common options:

--help              # Show detailed help
--endpoints LIST    # Cluster endpoints (default: localhost:9002,localhost:9102,localhost:9202)
--aeron-dir PATH    # Aeron directory (default: /dev/shm/aeron)
--debug             # Enable debug logging

🧪 Testing

# Build with tests
./scripts/build.sh --with-tests

# Run test suite
cd build && ctest -V

# Run specific tests
./tests/test_sbe_encoding
./tests/test_session_management

🔧 Developer Tools

The library includes powerful debugging and monitoring tools built on the same client infrastructure:

Message Inspector

Debug and analyze SBE-encoded messages with ease.

# Inspect binary message file
./build/tools/message_inspector --file message.bin

# Decode hex string
./build/tools/message_inspector --hex "0A1B2C3D4E5F..."

# Test encoding/decoding
./build/tools/message_inspector --test-encoding --verbose

# Generate sample messages
./build/tools/message_inspector --generate topic
./build/tools/message_inspector --generate session-connect

Features:

Parse and validate SBE message format
Hex dump visualization
Round-trip encoding/decoding tests
Sample message generation
Extract readable strings from binary data

Cluster Monitor

Real-time monitoring of cluster health and connectivity.

# Continuous monitoring (default)
./build/tools/cluster_monitor --continuous --interval 5

# Single check
./build/tools/cluster_monitor --once

# Custom endpoints and settings
./build/tools/cluster_monitor \
    --endpoints localhost:9002,remote:9002,backup:9002 \
    --interval 10 \
    --timeout 15 \
    --verbose

Features:

Test connectivity to all cluster members
Detect current leader
Track response times and success rates
Real-time cluster health assessment
Connection statistics and uptime tracking
Visual status dashboard with color-coded output

📊 Performance

Benchmarks on modern hardware (Intel i7, 32GB RAM):

Metric	Value
Connection Time	<100ms
Message Throughput	>100k orders/sec
End-to-End Latency	<1ms (localhost)
Memory Usage	<50MB typical
Reconnection Time	<5s with exponential backoff
Message Deduplication	O(1) with hash-based tracking

🚀 Advanced Features

Commit Tracking & Resume

The library includes a sophisticated commit management system for reliable message processing:

// Automatic commit on message processing
client.set_message_callback([&](const aeron_cluster::ParseResult& result) {
    // Process message
    process_order(result.payload);
    
    // Commit is tracked automatically with sequence number
    // Messages can be replayed from last commit on reconnection
});

// Manual commit control
client.commit_message(topic, message_identifier, message_id, timestamp, sequence);

// Resume from last committed position
auto last_commit = client.get_last_commit(topic, message_identifier);
if (last_commit) {
    std::cout << "Resuming from sequence: " << last_commit->sequence_number << std::endl;
}

// Subscribe with resume strategy
client.send_subscription_request(topic, identifier, "LAST_COMMIT", instance_id);

Key benefits:

At-least-once delivery semantics
Automatic message replay after disconnection
Per-topic, per-identifier commit tracking
Support for multiple concurrent subscriptions

Connection State Management

Robust connection handling with automatic recovery:

// Monitor connection state changes
client.set_connection_state_callback([](
    aeron_cluster::ConnectionState old_state, 
    aeron_cluster::ConnectionState new_state) {
    
    if (new_state == aeron_cluster::ConnectionState::CONNECTED) {
        std::cout << "Connected to cluster!" << std::endl;
    } else if (new_state == aeron_cluster::ConnectionState::DISCONNECTED) {
        std::cout << "Connection lost, will reconnect..." << std::endl;
    }
});

// Exponential backoff reconnection (automatic)
// - Initial delay: 5 seconds
// - Maximum delay: 30 seconds
// - Jitter: ±1 second to prevent thundering herd
// - Infinite retries (configurable)

Message Identifier Filtering

Subscribe to specific message streams using identifiers:

// Publisher sends to multiple identifiers
publisher.publish_message_to_topic(messageType, payload, headers, topic);

// Subscriber filters by identifier
subscriber.send_subscription_request(
    "order_notification_topic",
    "TRADER_A",  // Only receive messages for TRADER_A
    "LAST_COMMIT",
    "instance_1"
);

Use cases:

Multi-tenant message routing
Load balancing across consumer instances
Selective message consumption
Testing and development isolation

Session Management

Automatic leader detection and failover:

// Client automatically handles:
// - Leader detection
// - Session establishment
// - Leader changes/failover
// - Session ID tracking

int64_t session_id = client.get_session_id();
int32_t leader_member = client.get_leader_member_id();

// Connection statistics
auto stats = client.get_connection_stats();
std::cout << "Messages sent: " << stats.messages_sent << std::endl;
std::cout << "Messages received: " << stats.messages_received << std::endl;
std::cout << "Leader redirects: " << stats.leader_redirects << std::endl;

🏛️ Architecture

The library is organized into several key components:

Core Components

Component	Header	Description
ClusterClient	`cluster_client.hpp`	Main client interface for publishing and subscribing
CommitManager	`commit_manager.hpp`	Message offset tracking and resume functionality
SessionManager	`session_manager.hpp`	Session lifecycle and leader detection
MessageHandler	`message_handler.hpp`	Message parsing and callback routing
SBE Encoder	`sbe_messages.hpp`	Simple Binary Encoding message serialization
Config	`config.hpp`	Configuration builder with validation
Logger	`logging.hpp`	Flexible logging with multiple levels and outputs

Supporting Classes

Order Types (order_types.hpp): Trading order data structures and serialization
Topic Router (topic_router.hpp): Topic classification and routing logic
Subscription (subscription.hpp): Topic subscription message building
Performance Config (performance_config.hpp): Performance tuning and optimization settings
Protocol (protocol.hpp): SBE protocol constants and utilities

Message Flow

┌─────────────────┐
│   Application   │
└────────┬────────┘
         │
         ▼
┌─────────────────┐         ┌──────────────────┐
│ ClusterClient   │────────▶│ SessionManager   │
│                 │         │ (Leader detect)  │
└────────┬────────┘         └──────────────────┘
         │
         ├─ publish ─────▶ SBE Encoder ────▶ Aeron Cluster
         │
         └─ subscribe ───▶ MessageHandler ──▶ Callbacks
                              │
                              ▼
                         CommitManager
                         (Track offsets)

🤝 Contributing

We welcome contributions from the community! Please see our Contributing Guide for details.

Development Setup

git clone https://github.com/reverb-sys/aeron-cluster-client-cpp.git
cd aeron-cluster-client-cpp
./scripts/setup_project.sh --dev
./scripts/build.sh --with-tests --build-type Debug

Code Style

We use clang-format for consistent formatting:

./scripts/format.sh

🐛 Issues and Support

🐛 Bug Reports: GitHub Issues
💬 Discussions: GitHub Discussions
📖 Documentation: Browse the examples and header files with inline documentation
❓ Questions: Use GitHub Discussions for usage questions

When reporting issues, please include:

Operating system and version
Compiler version
Aeron version (check /home/ubuntu/aeron)
Complete error messages with stack traces
Minimal reproduction code
Steps to reproduce

Common Issues

Q: Connection timeout when connecting to cluster

Verify Aeron Media Driver is running: check for /dev/shm/aeron/cnc.dat
Confirm cluster endpoints are correct and accessible
Try the basic example with --check-only flag first

Q: Build fails with "Aeron not found"

Run ./scripts/setup_project.sh to install Aeron
Check that Aeron is built at /home/ubuntu/aeron
Verify AERON_ROOT in CMakeLists.txt points to correct location

Q: Messages not being received

Ensure you've called send_subscription_request() before polling
Check that message identifier matches between publisher and subscriber
Use --debug flag to see detailed message flow
Verify session is established: get_session_id() returns valid ID

Q: How do I test without a cluster?

Use the basic_client_example --check-only to test setup
The message_inspector tool can validate SBE encoding
Refer to GETTING_STARTED.md for cluster setup

🗺️ Roadmap

✅ Recently Completed

Commit Tracking: Message offset management and resume functionality
Advanced Pub/Sub: Topic-based messaging with identifier filtering
Reconnection Logic: Exponential backoff with connection state callbacks
Developer Tools: Message inspector and cluster monitor utilities
Enhanced Examples: Comprehensive test suite for reconnection and filtering
Connection State Management: Real-time connection monitoring
Performance Optimizations: Message batching and buffer pooling

🚧 In Progress

Enhanced Protocol Support: Additional SBE message types for trading workflows
Persistent Commit Storage: File-based commit tracking across restarts
Metrics Integration: Prometheus/OpenTelemetry support for production monitoring

🔮 Future Plans

Connection Pooling: Multi-session management for high throughput
Language Bindings: Python and Node.js wrappers
Docker Examples: Containerized deployment with docker-compose
Zero-Copy Operations: Advanced memory management optimizations
Admin API: Cluster management and monitoring REST API
Message Compression: Optional compression for large payloads

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

📈 Project Status

This library is actively maintained and used in production environments. We follow semantic versioning and provide migration guides for breaking changes.

Stability: Production ready
Maintenance: Active development
Community: Growing user base
Testing: Comprehensive test coverage

⭐ Star this repository if you find it useful!

Report Bug • Request Feature • Join Discussion

This library is not affiliated with Real Logic Ltd. It is an independent implementation based on the open Aeron Cluster protocol.

Name		Name	Last commit message	Last commit date
Latest commit History 80 Commits
cmake		cmake
examples		examples
include		include
scripts		scripts
src		src
tests		tests
tools		tools
.clang-format		.clang-format
.gitignore		.gitignore
CMakeLists.txt		CMakeLists.txt
GETTING_STARTED.md		GETTING_STARTED.md
LICENSE		LICENSE
README.md		README.md

Uh oh!

Folders and files

Latest commit

History

Repository files navigation

Aeron Cluster C++ Client

✨ Features

🚀 Quick Start

Simple Publisher Example

Simple Subscriber Example

💡 Why This Library?

Production-Ready from Day One

Complete Feature Set

Compatible & Extensible

📚 Documentation

Quick Reference

🏗️ Installation

Prerequisites

Ubuntu/Debian

macOS

Build Output

Build Options

🎯 Examples

1. Basic Client Example

2. Pub/Sub Reconnection Test

3. Commit/Resume Example

4. Order Publishing Example

5. Advanced Features Example

🧪 Testing

🔧 Developer Tools

Message Inspector

Cluster Monitor

📊 Performance

🚀 Advanced Features

Commit Tracking & Resume

Connection State Management

Message Identifier Filtering

Session Management

🏛️ Architecture

Core Components

Supporting Classes

Message Flow

🤝 Contributing

Development Setup

Code Style

🐛 Issues and Support

Common Issues

🗺️ Roadmap

✅ Recently Completed

🚧 In Progress

🔮 Future Plans

📄 License

📈 Project Status

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages