Messaging System

Language: English | 한국어

Modern Messaging Infrastructure with Advanced Patterns

Complete pub/sub, request/reply, event streaming, and message pipeline support

📚 Documentation | 🔗 Quick Start | 📖 Examples | 📊 Architecture | 🔄 Contributing

---

✨ Overview

A modern messaging system built on C++20 with comprehensive support for enterprise messaging patterns. The system provides a complete messaging infrastructure with pluggable backends, advanced routing capabilities, and robust reliability features.

🎯 Key Features

Core Messaging Message Bus - Central pub/sub coordinator Topic Router - Wildcard pattern matching (*, #) Message Queue - Thread-safe priority queues Message Serialization - Container-based payloads Trace Context - Distributed tracing support	Advanced Patterns Pub/Sub - Publisher and subscriber helpers Request/Reply - Synchronous RPC over async messaging Event Streaming - Event sourcing with replay Message Pipeline - Pipes-and-filters processing DI Container - Dependency injection support
Backend Support Standalone - Self-contained execution Integration - Thread pool integration Auto-detection - Runtime backend selection Pluggable Executors - IExecutor abstraction Mock Support - Testing and development	Production Quality Thread-safe - Lock-based and atomic operations Type-safe - Result error handling Well-tested - 100+ unit/integration tests Benchmarked - Performance validated Documented - Comprehensive API docs

---

📦 What's Included

The messaging system is fully implemented:

✅ Core Components (Complete)

• Message Types - Structured messages with metadata, priority, TTL
• Message Builder - Fluent API for message construction
• Message Queue - Thread-safe FIFO and priority queues
• Topic Router - Pattern-based routing with wildcards
• Message Bus - Async/sync pub/sub coordination
• Message Broker - Advanced routing and filtering

✅ Patterns (Complete)

• Pub/Sub Pattern - Publisher and Subscriber classes
• Request/Reply Pattern - Request client and server
• Event Streaming - Event sourcing with replay capability
• Message Pipeline - Stage-based message processing

✅ Infrastructure (Complete)

• Backend Interface - Pluggable execution backends
• Standalone Backend - Self-contained thread pool
• Integration Backend - External thread pool integration
• DI Container - Service registration and resolution
• Error Codes - Centralized error handling (-700 to -799)

✅ Testing & Benchmarks (Complete)

• Unit Tests - 100+ tests across all components
• Integration Tests - End-to-end messaging scenarios
• Pattern Tests - Comprehensive pattern validation
• Performance Benchmarks - Throughput and latency testing

✅ Documentation (Complete)

• API Reference - Complete API documentation
• Migration Guide - Upgrade instructions
• Patterns API - Pattern usage guide
• Design Patterns - Architecture documentation

---

🏗️ Architecture

System Overview

┌─────────────────────────────────────────────────────────────┐
│                    Messaging Patterns                        │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐   │
│  │ Pub/Sub  │  │ Req/Rep  │  │ Pipeline │  │ Stream   │   │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘   │
└─────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│                      Message Bus                             │
│  ┌──────────────────┐  ┌──────────────────┐                │
│  │ Message Broker   │  │ Topic Router     │                │
│  └──────────────────┘  └──────────────────┘                │
└─────────────────────────────────────────────────────────────┘
                              │
                ┌─────────────┴─────────────┐
                ▼                           ▼
┌───────────────────────────┐   ┌───────────────────────────┐
│   Message Queue           │   │  Subscription Manager     │
│  ┌─────────────────────┐  │   │  ┌─────────────────────┐ │
│  │ Priority Queue      │  │   │  │ Subscriber Registry │ │
│  │ Dead Letter Queue   │  │   │  │ Filter Chain        │ │
│  └─────────────────────┘  │   │  └─────────────────────┘ │
└───────────────────────────┘   └───────────────────────────┘
                              │
                ┌─────────────┴─────────────┐
                ▼                           ▼
┌───────────────────────────┐   ┌───────────────────────────┐
│  Integration Layer        │   │   Backend Selection       │
│  ┌──────────────────┐     │   │  ┌──────────────────┐    │
│  │ Logger Adapter   │     │   │  │ Standalone       │    │
│  │ Monitor Adapter  │     │   │  │ Integration      │    │
│  │ Thread Adapter   │     │   │  │ Auto-detect      │    │
│  └──────────────────┘     │   │  └──────────────────┘    │
└───────────────────────────┘   └───────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────┐
│              Foundation (common_system)                      │
│   Result<T>, error_info, event_bus, interfaces              │
└─────────────────────────────────────────────────────────────┘

Message Flow

Publisher → Message Bus → Topic Router → Subscribers
    ↓           ↓              ↓             ↓
  Message   Priority      Pattern      Callbacks
  Builder    Queue        Matching     (async)

---

🚀 Quick Start

Prerequisites

• C++20 compiler (GCC 11+, Clang 14+, MSVC 2022+, Apple Clang 14+)
• CMake 3.20+
• vcpkg (for dependency management)

Installation

git clone https://github.com/kcenon/messaging_system.git
cd messaging_system

# Recommended: Use local sibling directories (for unified_system development)
# Requires all system projects in the same parent directory:
#   Sources/common_system/
#   Sources/thread_system/
#   Sources/messaging_system/
#   ...
cmake -B build -DMESSAGING_USE_LOCAL_SYSTEMS=ON
cmake --build build -j

# Alternative: Build with vcpkg
cmake -B build -DCMAKE_TOOLCHAIN_FILE=/path/to/vcpkg/scripts/buildsystems/vcpkg.cmake
cmake --build build -j

C++20 Module Build (Experimental)

For C++20 module support (requires CMake 3.28+ and compatible compiler):

# Build with C++20 modules enabled
cmake -B build -G Ninja -DCMAKE_BUILD_TYPE=Release \
    -DMESSAGING_BUILD_MODULES=ON \
    -DMESSAGING_USE_LOCAL_SYSTEMS=ON
cmake --build build -j

# Use the module in your project:
# import kcenon.messaging;

Requirements:

• CMake 3.28+
• Ninja or Ninja Multi-Config generator
• Clang 16+ or GCC 14+ or MSVC 17.4+

Module Structure:

• kcenon.messaging - Primary module interface
• kcenon.messaging:core - Core messaging (message, queue, bus, broker)
• kcenon.messaging:patterns - Messaging patterns (pub/sub, request/reply)
• kcenon.messaging:task - Distributed task queue system
• kcenon.messaging:integration - Transports, backends, DI

Dependency Management

The build system uses UnifiedDependencies.cmake module which supports three modes:

Mode	CMake Option	Description
LOCAL	-DMESSAGING_USE_LOCAL_SYSTEMS=ON	Use sibling directories (recommended for development)
FetchContent	-DMESSAGING_USE_FETCHCONTENT=ON	Fetch from GitHub automatically
find_package	(default)	Use installed system packages

The unified module provides:

• Automatic target mapping for different naming conventions
• Consistent dependency resolution across all modes
• Simplified CMakeLists.txt (~300 lines vs ~920 lines previously)

Basic Example

#include <kcenon/messaging/core/message_bus.h>
#include <kcenon/messaging/core/message.h>
#include <kcenon/messaging/backends/standalone_backend.h>

using namespace kcenon::messaging;

int main() {
    // Create standalone backend
    auto backend = std::make_shared<standalone_backend>(4);
    backend->initialize();

    // Create message bus
    message_bus_config config;
    config.worker_threads = 4;
    auto bus = std::make_shared<message_bus>(backend, config);
    bus->start();

    // Subscribe to topic
    bus->subscribe("user.created", [](const message& msg) {
        std::cout << "User created: " << msg.metadata().id << std::endl;
        return common::VoidResult::ok();
    });

    // Publish message
    auto msg = message_builder()
        .topic("user.created")
        .source("app")
        .type(message_type::event)
        .build();

    if (msg.is_ok()) {
        bus->publish(msg.value());
    }

    // Cleanup
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    bus->stop();
    backend->shutdown();

    return 0;
}

---

📖 Usage Examples

Pub/Sub Pattern

#include <kcenon/messaging/patterns/pub_sub.h>

using namespace kcenon::messaging::patterns;

// Create publisher
auto publisher = std::make_shared<patterns::publisher>(bus, "events");

// Publish events
auto msg = message_builder()
    .topic("events.user.login")
    .type(message_type::event)
    .build();
publisher->publish(msg.value());

// Create subscriber
auto subscriber = std::make_shared<patterns::subscriber>(bus);
subscriber->subscribe("events.user.*", [](const message& msg) {
    // Handle user events
    return common::VoidResult::ok();
});

Request/Reply Pattern

#include <kcenon/messaging/patterns/request_reply.h>

using namespace kcenon::messaging::patterns;

// Server side
auto server = std::make_shared<request_server>(bus, "calculator");
server->register_handler([](const message& request) -> common::Result<message> {
    auto a = request.payload().get_value("a").value<int>();
    auto b = request.payload().get_value("b").value<int>();

    auto response = message_builder()
        .topic("calculator.response")
        .type(message_type::reply)
        .build();
    response.value().payload().set_value("result", a + b);

    return response;
});

// Client side
auto client = std::make_shared<request_client>(bus);
auto request = message_builder()
    .topic("calculator.add")
    .type(message_type::query)
    .build();
request.value().payload().set_value("a", 10);
request.value().payload().set_value("b", 20);

auto response = client->request(request.value(), std::chrono::seconds(5));
if (response.is_ok()) {
    auto result = response.value().payload().get_value("result").value<int>();
    std::cout << "Result: " << result << std::endl;  // 30
}

Event Streaming

#include <kcenon/messaging/patterns/event_streaming.h>

using namespace kcenon::messaging::patterns;

// Create event stream
auto stream = std::make_shared<event_stream>(bus, "orders");

// Publish events
stream->publish_event("order.created", order_data);
stream->publish_event("order.paid", payment_data);
stream->publish_event("order.shipped", shipping_data);

// Replay events
stream->replay_from(std::chrono::system_clock::now() - std::chrono::hours(24),
    [](const message& event) {
        // Process historical events
        return common::VoidResult::ok();
    });

Message Pipeline

#include <kcenon/messaging/patterns/message_pipeline.h>

using namespace kcenon::messaging::patterns;

// Build processing pipeline
auto pipeline = pipeline_builder()
    .add_stage("validate", [](message& msg) {
        // Validation logic
        return common::VoidResult::ok();
    })
    .add_stage("transform", [](message& msg) {
        // Transformation logic
        return common::VoidResult::ok();
    })
    .add_stage("enrich", [](message& msg) {
        // Enrichment logic
        return common::VoidResult::ok();
    })
    .build();

// Process message through pipeline
auto msg = message_builder().topic("data").build();
auto result = pipeline.process(msg.value());

---

🌟 Integration with Base Systems

The messaging system integrates seamlessly with specialized base systems:

Required Systems

• common_system - Result pattern, error handling, base interfaces
• thread_system - High-performance thread pools for message dispatch
• container_system - Type-safe message payloads and serialization

Optional Systems

• network_system - Distributed messaging over TCP/IP
• monitoring_system - Real-time metrics and performance telemetry
• database_system - Message persistence and audit logging

Logging Configuration

Logging is provided through common_system's ILogger interface with runtime binding via GlobalLoggerRegistry. This allows flexible logger implementation injection:

#include <kcenon/common/logging/log_functions.h>
#include <kcenon/common/interfaces/global_logger_registry.h>

// Register a logger with the global registry (typically at application startup)
auto& registry = kcenon::common::interfaces::GlobalLoggerRegistry::instance();
registry.set_default_logger(my_logger_implementation);

// Logging is now available throughout the messaging system
// No explicit logger parameter needed in messaging components

Integration Example

#include <kcenon/messaging/backends/integration_backend.h>
#include <kcenon/thread/core/thread_pool.h>
#include <kcenon/common/interfaces/global_logger_registry.h>

// Create thread pool from thread_system
auto thread_pool = std::make_shared<thread::thread_pool>(8);

// Get logger from common_system's GlobalLoggerRegistry
auto logger = kcenon::common::interfaces::get_logger();

// Create integration backend
auto backend = std::make_shared<integration_backend>(
    thread_pool,
    logger,
    nullptr  // optional monitoring
);

// Create message bus
auto bus = std::make_shared<message_bus>(backend);
bus->start();

// Messages are now dispatched via thread_system
// And logged via common_system's ILogger interface

---

📊 Performance

Benchmarks

Performance characteristics based on benchmark results:

Message Throughput

• Message creation: ~5M messages/sec
• Queue operations: ~2M operations/sec
• Topic routing: ~500K routes/sec
• Pub/Sub: ~100K messages/sec
• Request/Reply: ~50K requests/sec

Latency (p99)

• Message creation: < 1 μs
• Queue enqueue/dequeue: < 2 μs
• Topic matching: < 5 μs
• End-to-end pub/sub: < 100 μs
• Request/reply: < 1 ms

Running Benchmarks

# Build with benchmarks
cmake -B build -DMESSAGING_BUILD_BENCHMARKS=ON
cmake --build build -j

# Run benchmarks
./build/test/benchmarks/bench_message_creation
./build/test/benchmarks/bench_message_queue
./build/test/benchmarks/bench_topic_router
./build/test/benchmarks/bench_pub_sub_throughput
./build/test/benchmarks/bench_request_reply_latency

---

🧪 Testing

The system includes comprehensive test coverage:

Test Suites

• Core Tests - Message, queue, router, bus (40+ tests)
• Backend Tests - Standalone and integration backends (27 tests)
• Pattern Tests - All messaging patterns (80+ tests)
• Integration Tests - End-to-end scenarios (4 test suites)

Running Tests

# Build with tests
cmake -B build -DMESSAGING_BUILD_TESTS=ON
cmake --build build -j

# Run all tests
cd build
ctest --output-on-failure

# Run specific test suite
./test/unit/core/test_message_bus
./test/unit/patterns/test_pub_sub
./test/integration_tests/test_full_integration

Coverage Report

# Generate coverage report
cmake -B build -DCMAKE_BUILD_TYPE=Coverage
cmake --build build -j
cd build
ctest
lcov --capture --directory . --output-file coverage.info
genhtml coverage.info --output-directory coverage_html

---

📚 Documentation

Core Documentation

• Benchmarks - Performance characteristics and measurements
• Features - Complete feature documentation
• Production Quality - Quality assurance and reliability
• Project Structure - Codebase organization
• API Reference - Complete API documentation
• Migration Guide - Upgrade instructions
• Patterns API - Messaging patterns guide
• Design Patterns - Architecture patterns

Additional Resources

• Examples - Working code examples
• Integration Tests - End-to-end test scenarios
• Performance Benchmarks - Performance benchmark suite

---

🔧 Build Options

Option	Default	Description
MESSAGING_USE_LOCAL_SYSTEMS	OFF	Use sibling directories for dependencies
MESSAGING_USE_FETCHCONTENT	OFF	Auto-fetch dependencies from GitHub
MESSAGING_BUILD_TESTS	ON	Build unit and integration tests
MESSAGING_BUILD_EXAMPLES	ON	Build example programs
MESSAGING_BUILD_BENCHMARKS	OFF	Build performance benchmarks
KCENON_WITH_NETWORK_SYSTEM	ON	Enable network_system integration for transport implementations
KCENON_WITH_MONITORING_SYSTEM	OFF	Enable monitoring_system integration for metrics collection

Optional Dependency Matrix

The messaging_system has optional dependencies that can be enabled or disabled at compile time:

Dependency	Guard Macro	Components Affected	Fallback Behavior
network_system	KCENON_WITH_NETWORK_SYSTEM	http_transport, websocket_transport	Returns error::not_supported
monitoring_system	KCENON_WITH_MONITORING_SYSTEM	message_bus_collector	No-op (silently ignores operations)

Each optional component provides a compile-time is_available constant:

#include <kcenon/messaging/adapters/http_transport.h>
#include <kcenon/messaging/adapters/websocket_transport.h>
#include <kcenon/messaging/collectors/message_bus_collector.h>

// Check at compile time
static_assert(http_transport::is_available || !http_transport::is_available,
              "Compile-time check example");

// Or check at runtime via feature flags
#include <kcenon/messaging/config/feature_flags.h>

if (kcenon::messaging::config::has_network_system()) {
    // Use full transport functionality
} else {
    // Handle fallback case
}

Network System Integration

The KCENON_WITH_NETWORK_SYSTEM option controls whether websocket_transport and http_transport use the actual network_system implementation:

# Full transport functionality (default)
cmake -B build -DKCENON_WITH_NETWORK_SYSTEM=ON

# Stub mode - transports return not_supported errors
cmake -B build -DKCENON_WITH_NETWORK_SYSTEM=OFF

When disabled, transport classes compile but return error::not_supported for all operations. This allows building the messaging system without the network_system dependency.

Isolated Build (No Optional Dependencies)

To verify that fallback implementations work correctly, use the isolated build preset:

# Configure and build in isolated mode
cmake --preset=isolated
cmake --build --preset=isolated

# Run tests in isolated mode
ctest --preset=isolated

# Or use the verification script
./scripts/verify_isolated_build.sh

The isolated preset disables all optional dependencies (KCENON_WITH_NETWORK_SYSTEM=OFF, KCENON_WITH_MONITORING_SYSTEM=OFF) and runs tests to verify fallback implementations.

Example Builds

# Development build with local systems
cmake -B build -DMESSAGING_USE_LOCAL_SYSTEMS=ON -DMESSAGING_BUILD_TESTS=ON
cmake --build build -j

# Production build with FetchContent
cmake -B build -DMESSAGING_USE_FETCHCONTENT=ON -DCMAKE_BUILD_TYPE=Release
cmake --build build -j
sudo cmake --install build

# Debug build with all features
cmake -B build -DCMAKE_BUILD_TYPE=Debug \
    -DMESSAGING_BUILD_TESTS=ON \
    -DMESSAGING_BUILD_EXAMPLES=ON \
    -DMESSAGING_BUILD_BENCHMARKS=ON
cmake --build build -j

---

🌐 Platform Support

Platform	Compiler	Status
Ubuntu 22.04	GCC 11+	✅ Tested
Ubuntu 22.04	Clang 14+	✅ Tested
macOS 13+	Apple Clang 14+	✅ Tested
Windows 10+	MSVC 2022+	✅ Tested

---

🤝 Contributing

Contributions are welcome!

How to Contribute

1. Fork the repository
2. Create a feature branch
3. Make your changes with tests
4. Ensure all tests pass
5. Submit a pull request

Development Workflow

# Clone repository
git clone https://github.com/kcenon/messaging_system.git
cd messaging_system

# Create feature branch
git checkout -b feature/amazing-feature

# Build and test
cmake -B build -DMESSAGING_BUILD_TESTS=ON
cmake --build build -j
cd build && ctest

# Commit and push
git commit -m "feat: add amazing feature"
git push origin feature/amazing-feature

---

📄 License

BSD 3-Clause License

Copyright (c) 2021-2025, Messaging System Contributors All rights reserved.

See LICENSE file for full license text.

---

📬 Contact & Support

• Author: kcenon (@kcenon)
• Email: kcenon@gmail.com
• Issues: GitHub Issues
• Discussions: GitHub Discussions

---

Built on specialized systems

common_system • thread_system • monitoring_system • container_system • database_system • network_system

Made with ❤️ by the Open Source Community