Initial implementation

Cargo.toml

@@ -7,11 +7,16 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+futures = "0.3.8"
+log = "0.4"
 
 [dependencies.tokio]
 version = "1"
 features = ["sync"]
 
+[dev-dependencies]
+env_logger = "0.8"
+
 [dev-dependencies.tokio]
 version = "1"
-features = ["rt", "macros"]
+features = ["rt", "rt-multi-thread", "macros"]

src/broker_task.rs

@@ -0,0 +1,194 @@
+use log::trace;
+use std::{
+    any::{Any, TypeId},
+    collections::HashMap,
+    marker::PhantomData,
+};
+use tokio::sync::{broadcast, mpsc, oneshot};
+
+use crate::{message::Message, publication::Publication, subscription::Subscription};
+
+pub type ErasedSender = Box<dyn Any + Send + Sync>;
+
+pub trait MessageType {
+    fn message_type_id(&self) -> TypeId;
+
+    fn create_sender(&self) -> ErasedSender;
+}
+
+pub struct BasicMessageType<T> {
+    _phantom: PhantomData<T>,
+}
+
+impl<T> Default for BasicMessageType<T> {
+    fn default() -> Self {
+        BasicMessageType {
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<T: Message> MessageType for BasicMessageType<T> {
+    fn message_type_id(&self) -> TypeId {
+        TypeId::of::<T>()
+    }
+
+    fn create_sender(&self) -> ErasedSender {
+        trace!(
+            "Creating sender for {} ({:?})",
+            std::any::type_name::<T>(),
+            MessageType::type_id(self)
+        );
+        // TODO: configurable queue size (per message?)
+        let (sender, _) = broadcast::channel::<T>(8);
+        let sender: ErasedSender = Box::new(sender);
+        sender
+    }
+}
+
+pub trait SubscribeRequest {
+    fn message_type(&self) -> &dyn MessageType;
+
+    /// `sender` must be `tokio::sync::broadcast::Sender<T>` where
+    /// `MessageType::get_message_type` returns the `TypeId` of `T`.
+    unsafe fn send_subscribe_response(self: Box<Self>, sender: &ErasedSender);
+}
+
+pub type SubscribeRequestBox = Box<dyn SubscribeRequest + Send + Sync>;
+pub type SubscribeRequestSender = mpsc::Sender<SubscribeRequestBox>;
+
+pub struct BasicSubscribeRequest<T> {
+    msg_type: BasicMessageType<T>,
+    response_tx: oneshot::Sender<Subscription<T>>,
+}
+
+impl<T: Message> SubscribeRequest for BasicSubscribeRequest<T> {
+    fn message_type(&self) -> &dyn MessageType {
+        &self.msg_type
+    }
+
+    unsafe fn send_subscribe_response(self: Box<Self>, sender: &ErasedSender) {
+        let sender = &*(&**sender as *const dyn Any as *const broadcast::Sender<T>);
+        // let sender = (**sender).downcast_ref::<broadcast::Sender<T>>().unwrap();
+        let receiver = sender.subscribe();
+        let subscription = Subscription::new(receiver);
+        let _ = self.response_tx.send(subscription);
+    }
+}
+
+impl<T: Message> BasicSubscribeRequest<T> {
+    pub(crate) fn new() -> (Self, oneshot::Receiver<Subscription<T>>) {
+        let (response_tx, response_rx) = oneshot::channel();
+        (
+            Self {
+                msg_type: Default::default(),
+                response_tx,
+            },
+            response_rx,
+        )
+    }
+}
+
+pub trait AdvertiseRequest {
+    fn message_type(&self) -> &dyn MessageType;
+
+    /// `sender` must be `tokio::sync::broadcast::Sender<T>` where
+    /// `MessageType::get_message_type` returns the `TypeId` of `T`.
+    unsafe fn send_advertise_response(self: Box<Self>, sender: &ErasedSender);
+}
+
+pub type AdvertiseRequestBox = Box<dyn AdvertiseRequest + Send + Sync>;
+pub type AdvertiseRequestSender = mpsc::Sender<AdvertiseRequestBox>;
+
+pub struct BasicAdvertiseRequest<T> {
+    msg_type: BasicMessageType<T>,
+    response_tx: oneshot::Sender<Publication<T>>,
+}
+
+impl<T: Message> AdvertiseRequest for BasicAdvertiseRequest<T> {
+    fn message_type(&self) -> &dyn MessageType {
+        &self.msg_type
+    }
+
+    unsafe fn send_advertise_response(self: Box<Self>, sender: &ErasedSender) {
+        let sender = &*(&**sender as *const dyn Any as *const broadcast::Sender<T>);
+        // let sender = (**sender).downcast_ref::<broadcast::Sender<T>>().unwrap();
+        let publication = Publication::new(sender.clone());
+        let _ = self.response_tx.send(publication);
+    }
+}
+
+impl<T: Message> BasicAdvertiseRequest<T> {
+    pub(crate) fn new() -> (Self, oneshot::Receiver<Publication<T>>) {
+        let (response_tx, response_rx) = oneshot::channel();
+        (
+            Self {
+                msg_type: Default::default(),
+                response_tx,
+            },
+            response_rx,
+        )
+    }
+}
+
+struct Registry {
+    senders: HashMap<TypeId, ErasedSender>,
+}
+
+impl Default for Registry {
+    fn default() -> Self {
+        Registry {
+            senders: HashMap::new(),
+        }
+    }
+}
+
+impl Registry {
+    fn get_sender_for_type(&mut self, message_type: &dyn MessageType) -> &ErasedSender {
+        let type_id = message_type.message_type_id();
+        let sender_entry = self.senders.entry(type_id);
+        sender_entry.or_insert_with(|| message_type.create_sender())
+    }
+
+    fn handle_subscribe(&mut self, subscribe_request: SubscribeRequestBox) {
+        let sender = self.get_sender_for_type(subscribe_request.message_type());
+        unsafe { subscribe_request.send_subscribe_response(sender) }
+    }
+
+    fn handle_advertise(&mut self, advertise_request: AdvertiseRequestBox) {
+        let sender = self.get_sender_for_type(advertise_request.message_type());
+        unsafe { advertise_request.send_advertise_response(sender) }
+    }
+}
+
+pub struct BrokerTask {
+    pub(crate) subscribe_rx: mpsc::Receiver<SubscribeRequestBox>,
+    pub(crate) advertise_rx: mpsc::Receiver<AdvertiseRequestBox>,
+}
+
+impl BrokerTask {
+    pub(crate) async fn run(mut self) {
+        trace!("BrokerTask starting");
+
+        let mut registry = Registry::default();
+
+        loop {
+            tokio::select! {
+                Some(subscribe_req) = self.subscribe_rx.recv() => {
+                    registry.handle_subscribe(subscribe_req)
+                }
+                Some(advertise_req) = self.advertise_rx.recv() => {
+                    registry.handle_advertise(advertise_req)
+                }
+            }
+        }
+
+        // trace!("BrokerTask exiting");
+    }
+}
+
+impl Drop for BrokerTask {
+    fn drop(&mut self) {
+        trace!("BrokerTask dropped");
+    }
+}

src/lib.rs

@@ -1,40 +1,79 @@
+mod broker_task;
 mod message;
 mod publication;
 mod subscription;
 
+use broker_task::{
+    AdvertiseRequestSender, BasicAdvertiseRequest, BasicSubscribeRequest, BrokerTask,
+    SubscribeRequestSender,
+};
 pub use message::Message;
 pub use publication::{Publication, SendError};
 pub use subscription::{RecvError, Subscription};
 
-#[derive(Clone, Debug, PartialEq)]
+use futures::executor::block_on;
-#[non_exhaustive]
+use tokio::sync::mpsc;
-pub enum SubscribeError {}
 
 #[derive(Clone, Debug, PartialEq)]
 #[non_exhaustive]
-pub enum AdvertiseError {}
+pub enum OrsbError {
+    BrokerClosed,
+    NoResponse,
+}
 
-pub struct Orsb {}
+#[derive(Debug, Clone)]
+pub struct Orsb {
+    subscribe_tx: SubscribeRequestSender,
+    advertise_tx: AdvertiseRequestSender,
+}
 
 impl Orsb {
-    pub fn new() -> Self {
+    pub fn start_new() -> Self {
-        todo!()
+        Self::start_new2().0
     }
 
-    pub async fn subscribe<T: Message>(&mut self) -> Result<Subscription<T>, SubscribeError> {
+    pub(crate) fn start_new2() -> (Self, tokio::task::JoinHandle<()>) {
-        todo!()
+        let (subscribe_tx, subscribe_rx) = mpsc::channel(8);
+        let (advertise_tx, advertise_rx) = mpsc::channel(8);
+
+        let broker = BrokerTask {
+            subscribe_rx,
+            advertise_rx,
+        };
+        let join_handle = tokio::spawn(broker.run());
+
+        let orsb = Orsb {
+            subscribe_tx,
+            advertise_tx,
+        };
+
+        (orsb, join_handle)
+    }
+
+    pub async fn subscribe<T: Message>(&mut self) -> Result<Subscription<T>, OrsbError> {
+        let (subscribe_request, response_rx) = BasicSubscribeRequest::<T>::new();
+        self.subscribe_tx
+            .send(Box::new(subscribe_request))
+            .await
+            .or(Err(OrsbError::BrokerClosed))?;
+        response_rx.await.or(Err(OrsbError::NoResponse))
     }
 
-    pub async fn advertise<T: Message>(&mut self) -> Result<Publication<T>, AdvertiseError> {
+    pub async fn advertise<T: Message>(&mut self) -> Result<Publication<T>, OrsbError> {
-        todo!()
+        let (advertise_request, response_rx) = BasicAdvertiseRequest::<T>::new();
+        self.advertise_tx
+            .send(Box::new(advertise_request))
+            .await
+            .or(Err(OrsbError::BrokerClosed))?;
+        response_rx.await.or(Err(OrsbError::NoResponse))
     }
 
-    pub fn subscribe_blocking<T: Message>(&mut self) -> Result<Subscription<T>, SubscribeError> {
+    pub fn subscribe_blocking<T: Message>(&mut self) -> Result<Subscription<T>, OrsbError> {
-        todo!()
+        block_on(self.subscribe::<T>())
     }
 
-    pub fn advertise_blocking<T: Message>(&mut self) -> Result<Publication<T>, AdvertiseError> {
+    pub fn advertise_blocking<T: Message>(&mut self) -> Result<Publication<T>, OrsbError> {
-        todo!()
+        block_on(self.advertise::<T>())
     }
 }
 
@@ -45,10 +84,17 @@ mod test {
     #[derive(Clone, Debug, PartialEq)]
     struct TestMsg(u8);
 
-    #[test]
+    fn init_logger() {
-    fn test_sync() {
+        let _ = env_logger::try_init();
-        let mut orsb = Orsb::new();
+    }
+
+    #[tokio::test]
+    async fn test_sync() {
+        init_logger();
 
+        let mut orsb = Orsb::start_new();
+
+        tokio::task::spawn_blocking(move || {
             let mut sub = orsb.subscribe_blocking::<TestMsg>().unwrap();
             let mut publ = orsb.advertise_blocking::<TestMsg>().unwrap();
 
@@ -56,9 +102,31 @@ mod test {
             publ.send(TestMsg(20)).unwrap();
             publ.send(TestMsg(30)).unwrap();
 
-        assert_eq!(sub.recv_blocking(), Ok(TestMsg(10)));
+            assert_eq!(sub.try_recv(), Ok(TestMsg(10)));
-        assert_eq!(sub.recv_blocking(), Ok(TestMsg(20)));
+            assert_eq!(sub.try_recv(), Ok(TestMsg(20)));
-        assert_eq!(sub.recv_blocking(), Ok(TestMsg(30)));
+            assert_eq!(sub.try_recv(), Ok(TestMsg(30)));
+            assert_eq!(sub.try_recv(), Err(RecvError::Empty));
+        })
+        .await
+        .unwrap();
+    }
+
+    #[tokio::test]
+    async fn test_async() {
+        init_logger();
+
+        let mut orsb = Orsb::start_new();
+
+        let mut sub = orsb.subscribe::<TestMsg>().await.unwrap();
+        let mut publ = orsb.advertise::<TestMsg>().await.unwrap();
+
+        publ.send(TestMsg(10)).unwrap();
+        publ.send(TestMsg(20)).unwrap();
+        publ.send(TestMsg(30)).unwrap();
+
+        assert_eq!(sub.recv().await, Ok(TestMsg(10)));
+        assert_eq!(sub.recv().await, Ok(TestMsg(20)));
+        assert_eq!(sub.recv().await, Ok(TestMsg(30)));
         assert_eq!(sub.try_recv(), Err(RecvError::Empty));
     }
 }

src/message.rs

@@ -1,3 +1,5 @@
-pub trait Message: 'static + Clone + Send + Sync {}
+use std::any::Any;
+
+pub trait Message: 'static + Any + Clone + Send + Sync {}
 
 impl<T> Message for T where T: 'static + Clone + Send + Sync {}

src/publication.rs

@@ -1,20 +1,25 @@
-use std::marker::PhantomData;
+use tokio::sync::broadcast;
 
 use crate::message::Message;
 
 #[derive(Clone, Debug, PartialEq)]
 #[non_exhaustive]
-pub enum SendError {
+pub enum SendError {}
-    NoListeners,
-}
 
+#[derive(Debug)]
 pub struct Publication<T> {
-    _phantom: PhantomData<T>,
+    sender: broadcast::Sender<T>,
 }
 
 impl<T: Message> Publication<T> {
-    pub fn send(&mut self, message: T) -> Result<(), SendError> {
+    pub(crate) fn new(sender: broadcast::Sender<T>) -> Self {
-        let _ = message;
+        Publication { sender }
-        todo!()
+    }
+
+    pub fn send(&mut self, message: T) -> Result<usize, SendError> {
+        match self.sender.send(message) {
+            Ok(subscribers) => Ok(subscribers),
+            Err(_) => Ok(0),
+        }
     }
 }

src/subscription.rs

@@ -1,4 +1,5 @@
-use std::marker::PhantomData;
+use futures::executor::block_on;
+use tokio::sync::broadcast;
 
 use crate::message::Message;
 
@@ -10,20 +11,46 @@ pub enum RecvError {
     Lagged,
 }
 
+impl From<broadcast::error::RecvError> for RecvError {
+    fn from(err: broadcast::error::RecvError) -> Self {
+        match err {
+            broadcast::error::RecvError::Closed => RecvError::Closed,
+            broadcast::error::RecvError::Lagged(_) => RecvError::Lagged,
+        }
+    }
+}
+
+impl From<broadcast::error::TryRecvError> for RecvError {
+    fn from(err: broadcast::error::TryRecvError) -> Self {
+        match err {
+            broadcast::error::TryRecvError::Empty => RecvError::Empty,
+            broadcast::error::TryRecvError::Closed => RecvError::Closed,
+            broadcast::error::TryRecvError::Lagged(_) => RecvError::Lagged,
+        }
+    }
+}
+
+#[derive(Debug)]
 pub struct Subscription<T> {
-    _phantom: PhantomData<T>,
+    receiver: broadcast::Receiver<T>,
+}
+
+impl<T> Subscription<T> {
+    pub(crate) fn new(receiver: broadcast::Receiver<T>) -> Self {
+        Subscription { receiver }
+    }
 }
 
 impl<T: Message> Subscription<T> {
     pub async fn recv(&mut self) -> Result<T, RecvError> {
-        todo!()
+        Ok(self.receiver.recv().await?)
     }
 
     pub fn recv_blocking(&mut self) -> Result<T, RecvError> {
-        todo!()
+        block_on(self.recv())
     }
 
     pub fn try_recv(&mut self) -> Result<T, RecvError> {
-        todo!()
+        Ok(self.receiver.try_recv()?)
     }
 }