bin/ircd: fix bugs and clean up

bin/ircd/src/main.rs

@@ -87,6 +87,11 @@ impl Ircd {
         let (reader, writer) = stream.split();
 
         let mut reader = BufReader::new(reader);
+
+        // New subscriber
+        let receiver = self.senders.clone().subscribe().await;
+
+        // New irc connection
         let mut conn = IrcServerConnection::new(
             writer,
             peer_addr,
@@ -96,11 +101,10 @@ impl Ircd {
             self.configured_chans.clone(),
             self.p2p.clone(),
             self.senders.clone(),
+            receiver.get_id(),
         );
 
-        let receiver = self.senders.clone().subscribe().await;
-
-        // send messages history
+        // Send messages in buffer
         for msg in self.privmsgs_buffer.lock().await.to_vec() {
             receiver.self_notify(msg).await;
         }
@@ -134,6 +138,7 @@ impl Ircd {
                     if let Err(e) = result {
                         warn!("Close connection for clinet {}: {}", peer_addr, e);
                         receiver.unsubscribe().await;
+                        break
                     }
                 }
             })
@@ -237,7 +242,7 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
                 let result = ircd.process(executor_cloned.clone(), stream, peer_addr).await;
 
                 if let Err(e) = result {
-                    error!("failed process the {} connections: {}", peer_addr, e);
+                    error!("Failed processing connection {}: {}", peer_addr, e);
                     continue
                 };
 

bin/ircd/src/server.rs

@@ -36,6 +36,7 @@ pub struct IrcServerConnection {
     // p2p
     p2p: P2pPtr,
     senders: SubscriberPtr<Privmsg>,
+    subscriber_id: u64,
 }
 
 impl IrcServerConnection {
@@ -48,6 +49,7 @@ impl IrcServerConnection {
         configured_chans: FxHashMap<String, ChannelInfo>,
         p2p: P2pPtr,
         senders: SubscriberPtr<Privmsg>,
+        subscriber_id: u64,
     ) -> Self {
         Self {
             write_stream,
@@ -62,6 +64,7 @@ impl IrcServerConnection {
             configured_chans,
             p2p,
             senders,
+            subscriber_id,
         }
     }
 
@@ -213,7 +216,9 @@ impl IrcServerConnection {
                             (*self.privmsgs_buffer.lock().await).push(protocol_msg.clone())
                         }
 
-                        self.senders.notify(protocol_msg.clone()).await;
+                        self.senders
+                            .notify_with_exclude(protocol_msg.clone(), &[self.subscriber_id])
+                            .await;
 
                         debug!(target: "ircd", "PRIVMSG to be sent: {:?}", protocol_msg);
                         self.p2p.broadcast(protocol_msg).await?;