Provide subscription handler interface (adnanademovic#186)

Author: adnanademovic

rosrust/src/api/handlers.rs

@@ -0,0 +1,53 @@
+use crate::Message;
+use std::collections::HashMap;
+use std::marker::PhantomData;
+
+/// Handles all calls involved with a subscription
+///
+/// Each subscription is done in one thread, so there is no synchronization necessary between
+/// these calls.
+pub trait SubscriptionHandler<T>: Send + 'static {
+    /// Called before any message is accepted from a certain caller ID
+    ///
+    /// Contains the headers for handling the specific connection
+    fn connection(&mut self, headers: HashMap<String, String>);
+
+    /// Called upon receiving any message
+    fn message(&mut self, message: T, callerid: &str);
+}
+
+pub struct CallbackSubscriptionHandler<T, F, G> {
+    on_message: F,
+    on_connect: G,
+    _phantom: PhantomData<T>,
+}
+
+impl<T, F, G> CallbackSubscriptionHandler<T, F, G>
+where
+    T: Message,
+    F: Fn(T, &str) + Send + 'static,
+    G: Fn(HashMap<String, String>) + Send + 'static,
+{
+    pub fn new(on_message: F, on_connect: G) -> Self {
+        Self {
+            on_message,
+            on_connect,
+            _phantom: PhantomData,
+        }
+    }
+}
+
+impl<T, F, G> SubscriptionHandler<T> for CallbackSubscriptionHandler<T, F, G>
+where
+    T: Message,
+    F: Fn(T, &str) + Send + 'static,
+    G: Fn(HashMap<String, String>) + Send + 'static,
+{
+    fn connection(&mut self, headers: HashMap<String, String>) {
+        (self.on_connect)(headers)
+    }
+
+    fn message(&mut self, message: T, callerid: &str) {
+        (self.on_message)(message, callerid)
+    }
+}

rosrust/src/api/mod.rs

@@ -11,6 +11,7 @@ pub mod raii;
 pub mod resolve;
 mod ros;
 mod slave;
+pub mod handlers;
 
 pub struct ShutdownManager {
     should_shutdown: AtomicBool,

rosrust/src/api/raii.rs

@@ -4,9 +4,8 @@ use super::master::Master;
 use super::slave::Slave;
 use crate::rosxmlrpc::Response;
 use crate::tcpros::{Message, PublisherStream, ServicePair, ServiceResult};
-use crate::RawMessageDescription;
+use crate::{RawMessageDescription, SubscriptionHandler};
 use log::error;
-use std::collections::HashMap;
 use std::sync::atomic::AtomicUsize;
 use std::sync::Arc;
 
@@ -101,20 +100,18 @@ pub struct Subscriber {
 }
 
 impl Subscriber {
-    pub(crate) fn new<T, F, G>(
+    pub(crate) fn new<T, H>(
         master: Arc<Master>,
         slave: Arc<Slave>,
         name: &str,
         queue_size: usize,
-        on_message: F,
-        on_connect: G,
+        handler: H,
     ) -> Result<Self>
     where
         T: Message,
-        F: Fn(T, &str) + Send + 'static,
-        G: Fn(HashMap<String, String>) + Send + 'static,
+        H: SubscriptionHandler<T>,
     {
-        let id = slave.add_subscription::<T, F, G>(name, queue_size, on_message, on_connect)?;
+        let id = slave.add_subscription::<T, H>(name, queue_size, handler)?;
 
         let info = Arc::new(InteractorRaii::new(SubscriberInfo {
             master,

rosrust/src/api/ros.rs

@@ -7,11 +7,12 @@ use super::raii::{Publisher, Service, Subscriber};
 use super::resolve;
 use super::slave::Slave;
 use crate::api::clock::Delay;
+use crate::api::handlers::CallbackSubscriptionHandler;
 use crate::api::ShutdownManager;
 use crate::msg::rosgraph_msgs::{Clock as ClockMsg, Log};
 use crate::msg::std_msgs::Header;
 use crate::tcpros::{Client, Message, ServicePair, ServiceResult};
-use crate::{RawMessage, RawMessageDescription};
+use crate::{RawMessage, RawMessageDescription, SubscriptionHandler};
 use error_chain::bail;
 use log::error;
 use ros_message::{Duration, Time};
@@ -293,13 +294,35 @@ impl Ros {
             queue_size = usize::max_value();
         }
         let name = self.resolver.translate(topic)?;
-        Subscriber::new::<T, F, G>(
+        Subscriber::new::<T, _>(
             Arc::clone(&self.master),
             Arc::clone(&self.slave),
             &name,
             queue_size,
-            on_message,
-            on_connect,
+            CallbackSubscriptionHandler::new(on_message, on_connect),
+        )
+    }
+
+    pub fn subscribe_with<T, H>(
+        &self,
+        topic: &str,
+        mut queue_size: usize,
+        handler: H,
+    ) -> Result<Subscriber>
+    where
+        T: Message,
+        H: SubscriptionHandler<T>,
+    {
+        if queue_size == 0 {
+            queue_size = usize::max_value();
+        }
+        let name = self.resolver.translate(topic)?;
+        Subscriber::new::<T, H>(
+            Arc::clone(&self.master),
+            Arc::clone(&self.slave),
+            &name,
+            queue_size,
+            handler,
         )
     }
 

rosrust/src/api/slave/mod.rs

@@ -7,7 +7,7 @@ use super::error::{self, ErrorKind, Result};
 use crate::api::ShutdownManager;
 use crate::tcpros::{Message, PublisherStream, Service, ServicePair, ServiceResult};
 use crate::util::{kill, FAILED_TO_LOCK};
-use crate::RawMessageDescription;
+use crate::{RawMessageDescription, SubscriptionHandler};
 use crossbeam::channel::TryRecvError;
 use error_chain::bail;
 use log::error;
@@ -148,20 +148,18 @@ impl Slave {
     }
 
     #[inline]
-    pub fn add_subscription<T, F, G>(
+    pub fn add_subscription<T, H>(
         &self,
         topic: &str,
         queue_size: usize,
-        on_message: F,
-        on_connect: G,
+        handler: H,
     ) -> Result<usize>
     where
         T: Message,
-        F: Fn(T, &str) + Send + 'static,
-        G: Fn(HashMap<String, String>) + Send + 'static,
+        H: SubscriptionHandler<T>,
     {
         self.subscriptions
-            .add(&self.name, topic, queue_size, on_message, on_connect)
+            .add(&self.name, topic, queue_size, handler)
     }
 
     #[inline]

rosrust/src/api/slave/subscriptions.rs

@@ -1,7 +1,7 @@
 use crate::api::error::{self, ErrorKind, Result};
 use crate::tcpros::{SubscriberRosConnection, Topic};
 use crate::util::FAILED_TO_LOCK;
-use crate::Message;
+use crate::{Message, SubscriptionHandler};
 use error_chain::bail;
 use log::error;
 use std::collections::{BTreeSet, HashMap};
@@ -51,18 +51,10 @@ impl SubscriptionsTracker {
             .collect()
     }
 
-    pub fn add<T, F, G>(
-        &self,
-        name: &str,
-        topic: &str,
-        queue_size: usize,
-        on_message: F,
-        on_connect: G,
-    ) -> Result<usize>
+    pub fn add<T, H>(&self, name: &str, topic: &str, queue_size: usize, handler: H) -> Result<usize>
     where
         T: Message,
-        F: Fn(T, &str) + Send + 'static,
-        G: Fn(HashMap<String, String>) + Send + 'static,
+        H: SubscriptionHandler<T>,
     {
         let msg_definition = T::msg_definition();
         let msg_type = T::msg_type();
@@ -92,7 +84,7 @@ impl SubscriptionsTracker {
             )
             .into())
         } else {
-            Ok(connection.add_subscriber(queue_size, on_message, on_connect))
+            Ok(connection.add_subscriber(queue_size, handler))
         }
     }
 

rosrust/src/lib.rs

@@ -1,6 +1,7 @@
 #![recursion_limit = "1024"]
 
 pub use crate::api::raii::{Publisher, Service, Subscriber};
+pub use crate::api::handlers::{SubscriptionHandler};
 pub use crate::api::{error, Clock, Parameter};
 pub use crate::raw_message::{RawMessage, RawMessageDescription};
 #[doc(hidden)]

rosrust/src/singleton.rs

@@ -5,7 +5,7 @@ use crate::error::{ErrorKind, Result};
 use crate::rosxmlrpc::Response;
 use crate::tcpros::{Client, Message, ServicePair, ServiceResult};
 use crate::util::FAILED_TO_LOCK;
-use crate::RawMessageDescription;
+use crate::{RawMessageDescription, SubscriptionHandler};
 use crossbeam::sync::ShardedLock;
 use ctrlc;
 use error_chain::bail;
@@ -198,6 +198,15 @@ where
     ros!().subscribe_with_ids_and_headers::<T, F, G>(topic, queue_size, on_message, on_connect)
 }
 
+#[inline]
+pub fn subscribe_with<T, H>(topic: &str, queue_size: usize, handler: H) -> Result<Subscriber>
+where
+    T: Message,
+    H: SubscriptionHandler<T>,
+{
+    ros!().subscribe_with::<T, H>(topic, queue_size, handler)
+}
+
 #[inline]
 pub fn publish<T>(topic: &str, queue_size: usize) -> Result<Publisher<T>>
 where

rosrust/src/tcpros/subscriber.rs

@@ -3,6 +3,7 @@ use super::header::{decode, encode, match_field};
 use super::{Message, Topic};
 use crate::rosmsg::RosMsg;
 use crate::util::lossy_channel::{lossy_channel, LossyReceiver, LossySender};
+use crate::SubscriptionHandler;
 use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt};
 use crossbeam::channel::{bounded, select, Receiver, Sender, TrySendError};
 use log::error;
@@ -77,16 +78,10 @@ impl SubscriberRosConnection {
     // This creates a new thread to call on_message. Next API change should
     // allow subscribing with either callback or inline handler of the queue.
     // The queue is lossy, so it wouldn't be blocking.
-    pub fn add_subscriber<T, F, G>(
-        &mut self,
-        queue_size: usize,
-        on_message: F,
-        on_connect: G,
-    ) -> usize
+    pub fn add_subscriber<T, H>(&mut self, queue_size: usize, handler: H) -> usize
     where
         T: Message,
-        F: Fn(T, &str) + Send + 'static,
-        G: Fn(HashMap<String, String>) + Send + 'static,
+        H: SubscriptionHandler<T>,
     {
         let data_stream_id = self.next_data_stream_id;
         self.connected_ids.insert(data_stream_id);
@@ -105,9 +100,7 @@ impl SubscriberRosConnection {
             // TODO: we might want to panic here
             error!("Subscriber failed to connect to data stream");
         }
-        thread::spawn(move || {
-            handle_data::<T, F, G>(data_rx, connection_rx, on_message, on_connect)
-        });
+        thread::spawn(move || handle_data::<T, H>(data_rx, connection_rx, handler));
         data_stream_id
     }
 
@@ -176,29 +169,27 @@ impl SubscriberRosConnection {
     }
 }
 
-fn handle_data<T, F, G>(
+fn handle_data<T, H>(
     data: LossyReceiver<MessageInfo>,
     connections: Receiver<HashMap<String, String>>,
-    on_message: F,
-    on_connect: G,
+    mut handler: H,
 ) where
     T: Message,
-    F: Fn(T, &str),
-    G: Fn(HashMap<String, String>) + Send + 'static,
+    H: SubscriptionHandler<T>,
 {
     loop {
         select! {
             recv(data.kill_rx.kill_rx) -> _ => break,
             recv(data.data_rx) -> msg => match msg {
                 Err(_) => break,
                 Ok(buffer) => match RosMsg::decode_slice(&buffer.data) {
-                    Ok(value) => on_message(value, &buffer.caller_id),
+                    Ok(value) => handler.message(value, &buffer.caller_id),
                     Err(err) => error!("Failed to decode message: {}", err),
                 },
             },
             recv(connections) -> msg => match msg {
                 Err(_) => break,
-                Ok(conn) => on_connect(conn),
+                Ok(conn) => handler.connection(conn),
             },
         }
     }