add p2p::subscribe_channel() and improve subscriber semantics.

src/net/acceptor.rs

@@ -76,8 +76,7 @@ impl Acceptor {
     async fn run_accept_loop(self: Arc<Self>, listener: Async<TcpListener>) -> NetResult<()> {
         loop {
             let channel = self.tick_accept(&listener).await?;
-            let channel_result = Arc::new(Ok(channel));
-            self.channel_subscriber.notify(channel_result).await;
+            self.channel_subscriber.notify(Ok(channel)).await;
         }
     }
 
@@ -86,7 +85,7 @@ impl Acceptor {
             Ok(()) => panic!("Acceptor task should never complete without error status"),
             Err(err) => {
                 // Send this error to all channel subscribers
-                let result = Arc::new(Err(err));
+                let result = Err(err);
                 self.channel_subscriber.notify(result).await;
             }
         }

src/net/channel.rs

@@ -67,9 +67,9 @@ impl Channel {
         debug!(target: "net", "Channel::stop() [START, address={}]", self.address());
         assert_eq!(self.stopped.load(Ordering::Relaxed), false);
         self.stopped.store(false, Ordering::Relaxed);
-        let stop_err = Arc::new(NetError::ChannelStopped);
-        self.stop_subscriber.notify(stop_err).await;
+        self.stop_subscriber.notify(NetError::ChannelStopped).await;
         self.receive_task.stop().await;
+        self.message_subsystem.trigger_error(NetError::ChannelStopped).await;
         debug!(target: "net", "Channel::stop() [END, address={}]", self.address());
     }
 

src/net/message_subscriber.rs

@@ -143,6 +143,8 @@ impl<M: Message> MessageDispatcherInterface for MessageDispatcher<M> {
     }
 }
 
+// NOTE: this class is a more general version of system::Subscriber which can dispatch
+// multiple different type of registered types to sub-dispatchers
 pub struct MessageSubsystem {
     dispatchers: Mutex<HashMap<&'static str, Arc<dyn MessageDispatcherInterface>>>,
 }

src/net/p2p.rs

@@ -18,7 +18,8 @@ pub type P2pPtr = Arc<P2p>;
 pub struct P2p {
     pending: PendingChannels,
     channels: ConnectedChannels<Channel>,
-    // Used internally
+    channel_subscriber: SubscriberPtr<NetResult<ChannelPtr>>,
+    // Used both internally and externally
     stop_subscriber: SubscriberPtr<NetError>,
     hosts: HostsPtr,
     settings: SettingsPtr,
@@ -30,6 +31,7 @@ impl P2p {
         Arc::new(Self {
             pending: Mutex::new(HashSet::new()),
             channels: Mutex::new(HashMap::new()),
+            channel_subscriber: Subscriber::new(),
             stop_subscriber: Subscriber::new(),
             hosts: Hosts::new(),
             settings,
@@ -77,7 +79,8 @@ impl P2p {
         self.channels
             .lock()
             .await
-            .insert(channel.address(), channel);
+            .insert(channel.address(), channel.clone());
+        self.channel_subscriber.notify(Ok(channel)).await;
     }
     pub async fn remove(&self, channel: ChannelPtr) {
         self.channels.lock().await.remove(&channel.address());
@@ -106,7 +109,11 @@ impl P2p {
         self.hosts.clone()
     }
 
-    async fn subscribe_stop(&self) -> Subscription<NetError> {
+    pub async fn subscribe_channel(&self) -> Subscription<NetResult<ChannelPtr>> {
+        self.channel_subscriber.clone().subscribe().await
+    }
+
+    pub async fn subscribe_stop(&self) -> Subscription<NetError> {
         self.stop_subscriber.clone().subscribe().await
     }
 }

src/net/sessions/inbound_session.rs

@@ -72,7 +72,7 @@ impl InboundSession {
     async fn channel_sub_loop(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
         let channel_sub = self.acceptor.clone().subscribe().await;
         loop {
-            let channel = (*channel_sub.receive().await).clone()?;
+            let channel = channel_sub.receive().await?;
             // Spawn a detached task to process the channel
             // This will just perform the channel setup then exit.
             executor

src/system/subscriber.rs

@@ -9,12 +9,12 @@ pub type SubscriptionID = u64;
 
 pub struct Subscription<T> {
     id: SubscriptionID,
-    recv_queue: async_channel::Receiver<Arc<T>>,
+    recv_queue: async_channel::Receiver<T>,
     parent: Arc<Subscriber<T>>,
 }
 
-impl<T> Subscription<T> {
-    pub async fn receive(&self) -> Arc<T> {
+impl<T: Clone> Subscription<T> {
+    pub async fn receive(&self) -> T {
         let message_result = self.recv_queue.recv().await;
 
         match message_result {
@@ -33,10 +33,10 @@ impl<T> Subscription<T> {
 
 // Simple broadcast (publish-subscribe) class
 pub struct Subscriber<T> {
-    subs: Mutex<HashMap<u64, async_channel::Sender<Arc<T>>>>,
+    subs: Mutex<HashMap<u64, async_channel::Sender<T>>>,
 }
 
-impl<T> Subscriber<T> {
+impl<T: Clone> Subscriber<T> {
     pub fn new() -> Arc<Self> {
         Arc::new(Self {
             subs: Mutex::new(HashMap::new()),
@@ -66,7 +66,7 @@ impl<T> Subscriber<T> {
         self.subs.lock().await.remove(&sub_id);
     }
 
-    pub async fn notify(&self, message_result: Arc<T>) {
+    pub async fn notify(&self, message_result: T) {
         for sub in (*self.subs.lock().await).values() {
             match sub.send(message_result.clone()).await {
                 Ok(()) => {}