#pub-sub #events #queue #mpmc #emmit

rust-pubsub

A thread-safe, in-memory publish-subscribe library for Rust with flexible subscription modes

2 releases

Uses new Rust 2024

new 0.1.1 Apr 25, 2025
0.1.0 Apr 25, 2025

#340 in Concurrency

Download history 239/week @ 2025-04-23

239 downloads per month

MIT/Apache

30KB
316 lines

Rust PubSub

A thread-safe, in-memory publish-subscribe library for Rust, designed for efficient and flexible inter-thread communication. Each topic supports multiple publishers and subscribers, arbitrary message formats, cross-file communication, customizable queue behavior for each subscriber, and the option to process callbacks in dedicated threads. Each subscriber uses a separate crossbeam-channel, allowing different behaviors for the same topic (such as channel depth, behavior when the channel is full, and message processing method - either in a separate thread (subscribe) or in the local thread (subscribe_manual)), making it ideal for modular Rust projects.

中文文档

Features

  • Thread-Safe: Built with crossbeam-channel for safe message passing between threads.
  • Multi-Publisher and Multi-Subscriber: Supports multiple publishers and subscribers for the same topic.
  • Modular Design: Enables publishing and subscribing across files and modules via a singleton PubSub instance.
  • Arbitrary Message Formats: Supports any type implementing Send + Sync + Clone + 'static, including custom structs.
  • Customizable Queue Behavior: Each subscriber can choose to overwrite old messages or stop writing when the queue is full.
  • Callback Processing: Callbacks run in dedicated threads for non-blocking operation.
  • Flexible Subscriptions: Offers manual receiving and callback-based subscriptions.
  • Timeout Support: Provides blocking, non-blocking, and timeout-based operations.

Installation

Add to your Cargo.toml:

[dependencies]
rust-pubsub = "0.1.0"

Usage Examples

Below are examples demonstrating multi-publisher and multi-subscriber communication, arbitrary message formats (using a custom struct), customizable queue-full behavior, callback processing in dedicated threads, and modular design across files.

1. Multi-Publisher and Multi-Subscriber Across Files with Custom Struct

This example shows two publishers and two subscribers in separate files, using a custom struct CustomMessage as the message type. One subscriber uses a callback in a dedicated thread, highlighting modular design and arbitrary message formats.

src/main.rs

// Custom message struct
#[derive(Clone)]
pub struct CustomMessage {
    pub id: u32,
    pub content: String,
}

mod publisher;
mod subscriber;

fn main() {
    // Start two subscribers
    subscriber::manual_subscriber();
    subscriber::callback_subscriber();

    // Start two publishers
    publisher::publisher_one();
    publisher::publisher_two();

    loop {
        std::thread::sleep(std::time::Duration::from_millis(1000));
    }
}

src/publisher.rs

use rust_pubsub::PubSub;

use crate::CustomMessage;

pub fn publisher_one() {
    let pubsub = PubSub::instance();
    let topic = "multi_topic";
    let topic_id = pubsub.create_publisher(topic);

    std::thread::spawn(move || {
        loop {
            pubsub.publish(
                topic_id,
                CustomMessage {
                    id: 1,
                    content: "Message from publisher 1".to_string(),
                },
            );
            std::thread::sleep(std::time::Duration::from_millis(1000));
        }
    });
}

pub fn publisher_two() {
    let pubsub = PubSub::instance();
    let topic = "multi_topic";
    let topic_id = pubsub.create_publisher(topic);
    std::thread::spawn(move || {
        loop {
            pubsub.publish(
                topic_id,
                CustomMessage {
                    id: 2,
                    content: "Message from publisher 2".to_string(),
                },
            );

            std::thread::sleep(std::time::Duration::from_millis(1000));
        }
    });
}

src/subscriber.rs

use rust_pubsub::{PubSub, TopicConfig};
use std::thread;

use crate::CustomMessage;

pub fn manual_subscriber() {
    let pubsub = PubSub::instance();
    let topic = "multi_topic";
    let config = TopicConfig::new(10, true); // Overwrite when queue is full
    let receiver = pubsub.subscribe_manual::<CustomMessage>(topic, config);
    thread::spawn(move || {
        loop {
            if let Some(msg) = receiver.recv() {
                println!("Manual subscriber received: ID={}, Content={}", msg.id, msg.content);
            }
        }
    });
}

pub fn callback_subscriber() {
    let pubsub = PubSub::instance();
    let topic = "multi_topic";
    let config = TopicConfig::new(10, false); // Stop writing when queue is full
    pubsub.subscribe::<CustomMessage, _>(topic, config, |msg: &CustomMessage| {
        println!("Callback subscriber received: ID={}, Content={}", msg.id, msg.content); // The closure will execute in a separate internal thread
    });
}

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.


中文文档

Rust PubSub 是一个线程安全的、基于内存的发布-订阅库,专为 Rust 设计,旨在实现高效的线程间通信。每个topic支持多个发布者和订阅者、任意消息格式、跨文件通信、每个订阅者可自定义队列满行为以及是否在专用线程中处理回调,每个订阅者使用单独的crossbeam-channel,因此每个订阅者可对同一个topci指定不同的行为(如通道深度,通道满时的行为,消息处理方式是在单独的线程中处理(subscribe)还是在本地线程处理(subscribe_manual)),非常适合模块化的 Rust 项目。

English Documentation

功能

  • 线程安全:使用 crossbeam-channel 实现线程间安全消息传递。
  • 多发布者和多订阅者:支持同一主题的多个发布者和订阅者。
  • 模块化设计:通过单例 PubSub 实例,支持跨文件和模块的发布和订阅。
  • 任意消息格式:支持任何实现 Send + Sync + Clone + 'static 的类型,包括自定义结构体。
  • 可自定义队列行为:每个订阅者可配置队列满时覆盖旧消息或停止写入。
  • 回调处理:回调在单独线程中处理,确保非阻塞。
  • 灵活订阅:支持手动接收和基于回调的订阅。
  • 超时支持:提供阻塞、非阻塞和带超时操作。

安装

Cargo.toml 中添加:

[dependencies]
rust-pubsub = "0.1.0"

使用示例

以下示例展示了多发布者和多订阅者通信、任意消息格式(使用自定义结构体)、可自定义队列满行为、在专用线程中处理回调以及跨文件模块化设计。

1. 跨文件的多发布者和多订阅者(使用自定义结构体)

此示例展示在不同文件中定义的两个发布者和两个订阅者,使用自定义结构体 CustomMessage 作为消息类型。一个订阅者使用回调,在内部专用线程中处理。

src/main.rs

// Custom message struct
#[derive(Clone)]
pub struct CustomMessage {
    pub id: u32,
    pub content: String,
}

mod publisher;
mod subscriber;

fn main() {
    // Start two subscribers
    subscriber::manual_subscriber();
    subscriber::callback_subscriber();

    // Start two publishers
    publisher::publisher_one();
    publisher::publisher_two();

    loop {
        std::thread::sleep(std::time::Duration::from_millis(1000));
    }
}

src/publisher.rs

use rust_pubsub::PubSub;

use crate::CustomMessage;

pub fn publisher_one() {
    let pubsub = PubSub::instance();
    let topic = "multi_topic";
    let topic_id = pubsub.create_publisher(topic);

    std::thread::spawn(move || {
        loop {
            pubsub.publish(
                topic_id,
                CustomMessage {
                    id: 1,
                    content: "Message from publisher 1".to_string(),
                },
            );
            std::thread::sleep(std::time::Duration::from_millis(1000));
        }
    });
}

pub fn publisher_two() {
    let pubsub = PubSub::instance();
    let topic = "multi_topic";
    let topic_id = pubsub.create_publisher(topic);
    std::thread::spawn(move || {
        loop {
            pubsub.publish(
                topic_id,
                CustomMessage {
                    id: 2,
                    content: "Message from publisher 2".to_string(),
                },
            );

            std::thread::sleep(std::time::Duration::from_millis(1000));
        }
    });
}

src/subscriber.rs

use rust_pubsub::{PubSub, TopicConfig};
use std::thread;

use crate::CustomMessage;

pub fn manual_subscriber() {
    let pubsub = PubSub::instance();
    let topic = "multi_topic";
    let config = TopicConfig::new(10, true); // Overwrite when queue is full
    let receiver = pubsub.subscribe_manual::<CustomMessage>(topic, config);
    thread::spawn(move || {
        loop {
            if let Some(msg) = receiver.recv() {
                println!("Manual subscriber received: ID={}, Content={}", msg.id, msg.content);
            }
        }
    });
}

pub fn callback_subscriber() {
    let pubsub = PubSub::instance();
    let topic = "multi_topic";
    let config = TopicConfig::new(10, false); // Stop writing when queue is full
    pubsub.subscribe::<CustomMessage, _>(topic, config, |msg: &CustomMessage| {
        println!("Callback subscriber received: ID={}, Content={}", msg.id, msg.content); // The closure will execute in a separate internal thread
    });
}

许可证

本项目采用 Apache 2.0 或 MIT 许可证,供您选择。

Dependencies

~600KB