photon_ring/
typed_bus.rs

1// Copyright 2026 Photon Ring Contributors
2// SPDX-License-Identifier: Apache-2.0
3
4use alloc::boxed::Box;
5use alloc::string::{String, ToString};
6use core::any::{Any, TypeId};
7use hashbrown::HashMap;
8use spin::Mutex;
9
10use crate::channel::{self, Publisher, Subscribable, Subscriber};
11
12/// Wrapper that erases the concrete `T` behind `dyn Any` while preserving
13/// the `TypeId` and a human-readable type name for diagnostics.
14struct TopicSlot {
15    type_id: TypeId,
16    type_name: &'static str,
17    inner: Box<dyn Any + Send + Sync>,
18}
19
20/// The concrete (generic) payload stored inside a [`TopicSlot`].
21struct TypedEntry<T: Copy + Send + 'static> {
22    subscribable: Subscribable<T>,
23    publisher: Option<Publisher<T>>,
24}
25
26// Safety: `TypedEntry<T>` is `Send + Sync` when `T: Copy + Send + 'static`
27// because `Subscribable<T>` is `Send + Sync` and `Publisher<T>` is `Send`.
28// The `Option` wrapper and the fact that the publisher is only accessed
29// under the outer `Mutex` make this safe.
30unsafe impl<T: Copy + Send + 'static> Send for TypedEntry<T> {}
31unsafe impl<T: Copy + Send + 'static> Sync for TypedEntry<T> {}
32
33/// A topic bus that supports different message types per topic.
34///
35/// Unlike [`Photon<T>`](crate::Photon) which requires a single message type
36/// across all topics, `TypedBus` allows each topic to have its own
37/// `T: Copy + Send + 'static`.
38///
39/// # Example
40///
41/// ```
42/// use photon_ring::TypedBus;
43///
44/// let bus = TypedBus::new(1024);
45///
46/// // Different types per topic
47/// let mut price_pub = bus.publisher::<f64>("prices");
48/// let mut vol_pub = bus.publisher::<u32>("volumes");
49///
50/// let mut price_sub = bus.subscribe::<f64>("prices");
51/// let mut vol_sub = bus.subscribe::<u32>("volumes");
52///
53/// price_pub.publish(42.5);
54/// vol_pub.publish(1000);
55///
56/// assert_eq!(price_sub.try_recv(), Ok(42.5));
57/// assert_eq!(vol_sub.try_recv(), Ok(1000));
58/// ```
59pub struct TypedBus {
60    topics: Mutex<HashMap<String, TopicSlot>>,
61    default_capacity: usize,
62}
63
64impl TypedBus {
65    /// Create a bus. `capacity` is the ring size for each topic (power of two).
66    pub fn new(capacity: usize) -> Self {
67        TypedBus {
68            topics: Mutex::new(HashMap::new()),
69            default_capacity: capacity,
70        }
71    }
72
73    /// Take the publisher for a topic. Creates the topic if it doesn't exist.
74    ///
75    /// # Panics
76    ///
77    /// - Panics if the topic already exists with a different type `T`.
78    /// - Panics if the publisher for this topic was already taken.
79    pub fn publisher<T: Copy + Send + 'static>(&self, topic: &str) -> Publisher<T> {
80        let mut topics = self.topics.lock();
81        if !topics.contains_key(topic) {
82            topics.insert(
83                topic.to_string(),
84                Self::make_slot::<T>(self.default_capacity),
85            );
86        }
87        let slot = topics.get_mut(topic).unwrap();
88        let entry = Self::downcast_mut::<T>(slot, topic);
89        entry
90            .publisher
91            .take()
92            .unwrap_or_else(|| panic!("publisher already taken for topic '{topic}'"))
93    }
94
95    /// Subscribe to a topic (future messages only). Creates the topic if needed.
96    ///
97    /// # Panics
98    ///
99    /// Panics if the topic already exists with a different type `T`.
100    pub fn subscribe<T: Copy + Send + 'static>(&self, topic: &str) -> Subscriber<T> {
101        let mut topics = self.topics.lock();
102        if !topics.contains_key(topic) {
103            topics.insert(
104                topic.to_string(),
105                Self::make_slot::<T>(self.default_capacity),
106            );
107        }
108        let slot = topics.get_mut(topic).unwrap();
109        let entry = Self::downcast_mut::<T>(slot, topic);
110        entry.subscribable.subscribe()
111    }
112
113    /// Get the clone-able subscriber factory for a topic.
114    ///
115    /// # Panics
116    ///
117    /// Panics if the topic already exists with a different type `T`.
118    pub fn subscribable<T: Copy + Send + 'static>(&self, topic: &str) -> Subscribable<T> {
119        let mut topics = self.topics.lock();
120        if !topics.contains_key(topic) {
121            topics.insert(
122                topic.to_string(),
123                Self::make_slot::<T>(self.default_capacity),
124            );
125        }
126        let slot = topics.get_mut(topic).unwrap();
127        let entry = Self::downcast_mut::<T>(slot, topic);
128        entry.subscribable.clone()
129    }
130
131    /// Downcast the erased `TopicSlot` to `TypedEntry<T>`, panicking with a
132    /// clear message on type mismatch.
133    fn downcast_mut<'a, T: Copy + Send + 'static>(
134        slot: &'a mut TopicSlot,
135        topic: &str,
136    ) -> &'a mut TypedEntry<T> {
137        let requested = TypeId::of::<T>();
138        if slot.type_id != requested {
139            panic!(
140                "topic '{topic}' exists with type '{}', cannot access as '{}'",
141                slot.type_name,
142                core::any::type_name::<T>(),
143            );
144        }
145        slot.inner
146            .downcast_mut::<TypedEntry<T>>()
147            .expect("TypeId matched but downcast failed (this is a bug)")
148    }
149
150    fn make_slot<T: Copy + Send + 'static>(capacity: usize) -> TopicSlot {
151        let (pub_, sub_) = channel::channel::<T>(capacity);
152        TopicSlot {
153            type_id: TypeId::of::<T>(),
154            type_name: core::any::type_name::<T>(),
155            inner: Box::new(TypedEntry {
156                subscribable: sub_,
157                publisher: Some(pub_),
158            }),
159        }
160    }
161}
photon_ring/typed_bus.rs

photon_ring/
typed_bus.rs
