pilgrimage

This is a Rust implementation of a distributed messaging system. It uses a simple design inspired by Apache Kafka. It simply records messages to local files.

Features

• Topic-based pub/sub model
• Scalability through partitioning
• Persistent messages (log file based)
• Leader/Follower Replication
• Fault Detection and Automatic Recovery
• Delivery guaranteed by acknowledgement (ACK)
• Fully implemented leader selection mechanism
• Partition Replication
• Persistent messages
• Schema Registry for managing message schemas and ensuring compatibility
• Automatic Scaling

Usage

Dependency

• Rust 1.51.0 or later

Functionality Implemented

• Message Queue: Efficient message queue implementation using Mutex and VecDeque.
• Broker: Core broker functionality including message handling, node management, and leader election.
• Consumer Groups: Support for consumer groups to allow multiple consumers to read from the same topic.
• Leader Election: Mechanism for electing a leader among brokers to manage partitions and replication.
• Storage: Persistent storage of messages using local files.
• Replication: Replication of messages across multiple brokers for fault tolerance.
• Schema Registry: Management of message schemas to ensure compatibility between producers and consumers.
• Benchmarking: Comprehensive benchmarking tests to measure performance of various components.
• Automatic Scaling: Automatically scale the number of instances based on load.

Basic usage

use pilgrimage::broker::{Broker, Node};
use std::sync::{Arc, Mutex};

fn main() {
    // Broker Creation
    let broker = Broker::new("broker1", 3, 2, "storage_path");

    // Adding a node
    let node = Node {
        data: Arc::new(Mutex::new(Vec::new())),
    };
    broker.add_node("node1".to_string(), node);

    // Send a message
    broker.send_message("Hello, world!".to_string());

    // Message received
    if let Some(message) = broker.receive_message() {
        println!("Received: {}", message);
    }
}

Multi-threaded message processing

use pilgrimage::broker::Broker;
use std::sync::Arc;
use std::thread;

fn main() {
    let broker = Arc::new(Broker::new("broker1", 3, 2, "storage_path"));

    let broker_sender = Arc::clone(&broker);
    let sender_handle = thread::spawn(move || {
        for i in 0..10 {
            let message = format!("Message {}", i);
            broker_sender.send_message(message);
        }
    });

    let broker_receiver = Arc::clone(&broker);
    let receiver_handle = thread::spawn(move || {
        for _ in 0..10 {
            if let Some(message) = broker_receiver.receive_message() {
                println!("Received: {:?}", message);
            }
        }
    });

    sender_handle.join().unwrap();
    receiver_handle.join().unwrap();
}

Fault Detection and Automatic Recovery

The system includes mechanisms for fault detection and automatic recovery. Nodes are monitored using heartbeat signals, and if a fault is detected, the system will attempt to recover automatically.

use pilgrimage::Broker;
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;

fn main() {
    let storage = Arc::new(Mutex::new(Storage::new("test_db_path").unwrap()));
    let mut broker = Broker::new("broker_id", 1, 1, "test_db_path");
    broker.storage = storage.clone();

    // Simulating a disability
    {
        let mut storage_guard = storage.lock().unwrap();
        storage_guard.available = false;
    }

    // Simulating a disability
    broker.monitor_nodes();

    // Simulating a disability
    thread::sleep(Duration::from_millis(100));
    let storage_guard = storage.lock().unwrap();
    assert!(storage_guard.is_available());
}

License

MIT

Examples

To execute a basic example, use the following command:

cargo run --example simple-send-recv
cargo run --example mulch-send-recv
cargo run --example thread-send-recv

Bench

cargo bench