share the list of seen privmsg ids so that locally sent messages can be added to it.

bin/ircd/src/irc_server.rs

@@ -13,7 +13,7 @@ use drk::{
     Error, Result,
 };
 
-use crate::privmsg::PrivMsg;
+use crate::privmsg::{PrivMsg, SeenPrivMsgIdsPtr};
 
 /*
 NICK fifififif
@@ -34,6 +34,7 @@ PRIVMSG #dev hihi
 
 pub struct IrcServerConnection {
     write_stream: WriteHalf<Async<TcpStream>>,
+    seen_privmsg_ids: SeenPrivMsgIdsPtr,
     is_nick_init: bool,
     is_user_init: bool,
     is_registered: bool,
@@ -42,9 +43,10 @@ pub struct IrcServerConnection {
 }
 
 impl IrcServerConnection {
-    pub fn new(write_stream: WriteHalf<Async<TcpStream>>) -> Self {
+    pub fn new(write_stream: WriteHalf<Async<TcpStream>>, seen_privmsg_ids: SeenPrivMsgIdsPtr) -> Self {
         Self {
             write_stream,
+            seen_privmsg_ids,
             is_nick_init: false,
             is_user_init: false,
             is_registered: false,
@@ -95,9 +97,11 @@ impl IrcServerConnection {
                 let message = &line[substr_idx + 1..];
                 info!("Message {}: {}", channel, message);
 
+                let random_id = OsRng.next_u32();
+                self.seen_privmsg_ids.add_seen(random_id).await;
 
                 let protocol_msg = PrivMsg {
-                    id: OsRng.next_u32(),
+                    id: random_id,
                     nickname: self.nickname.clone(),
                     channel: channel.to_string(),
                     message: message.to_string(),

bin/ircd/src/main.rs

@@ -33,22 +33,25 @@ mod program_options;
 mod protocol_privmsg;
 mod irc_server;
 
-use crate::privmsg::PrivMsg;
-use crate::program_options::ProgramOptions;
-use crate::protocol_privmsg::ProtocolPrivMsg;
-use crate::irc_server::IrcServerConnection;
+use crate::{
+    privmsg::{PrivMsg, PrivMsgId, SeenPrivMsgIds, SeenPrivMsgIdsPtr},
+    program_options::ProgramOptions,
+    protocol_privmsg::ProtocolPrivMsg,
+    irc_server::IrcServerConnection,
+};
 
 async fn process(
     recvr: async_channel::Receiver<Arc<PrivMsg>>,
     stream: Async<TcpStream>,
     peer_addr: SocketAddr,
     p2p: net::P2pPtr,
+    seen_privmsg_ids: SeenPrivMsgIdsPtr,
     _executor: Arc<Executor<'_>>,
 ) -> Result<()> {
     let (reader, writer) = stream.split();
 
     let mut reader = BufReader::new(reader);
-    let mut connection = IrcServerConnection::new(writer);
+    let mut connection = IrcServerConnection::new(writer, seen_privmsg_ids);
 
     loop {
         let mut line = String::new();
@@ -70,7 +73,7 @@ async fn process(
                     warn!("Read line error. Closing stream for {}: {}", peer_addr, err);
                     return Ok(())
                 }
-                process_user_input(line, peer_addr, &mut connection, p2p.clone()).await;
+                process_user_input(line, peer_addr, &mut connection, p2p.clone()).await?;
             }
         };
     }
@@ -81,10 +84,10 @@ async fn process_user_input(
     peer_addr: SocketAddr,
     connection: &mut IrcServerConnection,
     p2p: net::P2pPtr,
-) {
+) -> Result<()> {
     if line.len() == 0 {
         warn!("Received empty line from {}. Closing connection.", peer_addr);
-        return
+        return Err(Error::ChannelStopped)
     }
     assert!(&line[(line.len() - 1)..] == "\n");
     // Remove the \n character
@@ -94,13 +97,16 @@ async fn process_user_input(
 
     if let Err(err) = connection.update(line, p2p.clone()).await {
         warn!("Connection error: {} for {}", err, peer_addr);
-        return
+        return Err(Error::ChannelStopped)
     }
+
+    Ok(())
 }
 
 async fn channel_loop(
     p2p: net::P2pPtr,
     sender: async_channel::Sender<Arc<PrivMsg>>,
+    seen_privmsg_ids: SeenPrivMsgIdsPtr,
     executor: Arc<Executor<'_>>,
 ) -> Result<()> {
     debug!("CHANNEL SUBS LOOP");
@@ -111,7 +117,7 @@ async fn channel_loop(
 
         debug!("NEWCHANNEL");
 
-        let protocol_privmsg = ProtocolPrivMsg::new(channel, sender.clone(), p2p.clone()).await;
+        let protocol_privmsg = ProtocolPrivMsg::new(channel, sender.clone(), seen_privmsg_ids.clone(), p2p.clone()).await;
         protocol_privmsg.start(executor.clone()).await;
     }
 }
@@ -142,6 +148,8 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
         identity_pass: "test".to_string(),
     };
 
+    let seen_privmsg_ids = SeenPrivMsgIds::new();
+
     let p2p = net::P2p::new(options.network_settings);
     // Performs seed session
     p2p.clone().start(executor.clone()).await?;
@@ -159,7 +167,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
     let (sender, recvr) = async_channel::unbounded();
     // for now the p2p and channel sub sessions just run forever
     // so detach them as background processes.
-    executor.spawn(channel_loop(p2p.clone(), sender, executor.clone())).detach();
+    executor.spawn(channel_loop(p2p.clone(), sender, seen_privmsg_ids.clone(), executor.clone())).detach();
 
     let ex2 = executor.clone();
     let ex3 = ex2.clone();
@@ -180,7 +188,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
 
         let p2p2 = p2p.clone();
         let ex2 = executor.clone();
-        executor.spawn(process(recvr.clone(), stream, peer_addr, p2p2, ex2)).detach();
+        executor.spawn(process(recvr.clone(), stream, peer_addr, p2p2, seen_privmsg_ids.clone(), ex2)).detach();
     }
 }
 

bin/ircd/src/privmsg.rs

@@ -1,4 +1,9 @@
-use std::io;
+use async_std::sync::Mutex;
+use std::{
+    collections::HashSet,
+    io,
+    sync::Arc,
+};
 use drk::{
     net,
     serial::{Decodable, Encodable}, Result,
@@ -42,3 +47,23 @@ impl Decodable for PrivMsg {
     }
 }
 
+pub struct SeenPrivMsgIds {
+    privmsg_ids: Mutex<HashSet<PrivMsgId>>,
+}
+
+pub type SeenPrivMsgIdsPtr = Arc<SeenPrivMsgIds>;
+
+impl SeenPrivMsgIds {
+    pub fn new() -> Arc<Self> {
+        Arc::new(Self { privmsg_ids: Mutex::new(HashSet::new()) })
+    }
+
+    pub async fn add_seen(&self, id: u32) {
+        self.privmsg_ids.lock().await.insert(id);
+    }
+
+    pub async fn is_seen(&self, id: u32) -> bool {
+        self.privmsg_ids.lock().await.contains(&id)
+    }
+}
+

bin/ircd/src/protocol_privmsg.rs

@@ -9,13 +9,13 @@ use drk::{
     net, Result,
 };
 
-use crate::privmsg::{PrivMsgId, PrivMsg};
+use crate::privmsg::{PrivMsgId, PrivMsg, SeenPrivMsgIdsPtr};
 
 pub struct ProtocolPrivMsg {
     notify_queue_sender: async_channel::Sender<Arc<PrivMsg>>,
     privmsg_sub: net::MessageSubscription<PrivMsg>,
     jobsman: net::ProtocolJobsManagerPtr,
-    privmsg_ids: Mutex<HashSet<PrivMsgId>>,
+    seen_privmsg_ids: SeenPrivMsgIdsPtr,
     p2p: net::P2pPtr,
 }
 
@@ -23,6 +23,7 @@ impl ProtocolPrivMsg {
     pub async fn new(
         channel: net::ChannelPtr,
         notify_queue_sender: async_channel::Sender<Arc<PrivMsg>>,
+        seen_privmsg_ids: SeenPrivMsgIdsPtr,
         p2p: net::P2pPtr,
     ) -> Arc<Self> {
         let message_subsytem = channel.get_message_subsystem();
@@ -37,7 +38,7 @@ impl ProtocolPrivMsg {
             notify_queue_sender,
             privmsg_sub,
             jobsman: net::ProtocolJobsManager::new("PrivMsgProtocol", channel),
-            privmsg_ids: Mutex::new(HashSet::new()),
+            seen_privmsg_ids,
             p2p,
         })
     }
@@ -61,15 +62,13 @@ impl ProtocolPrivMsg {
             );
 
             // Do we already have this message?
-            if self.privmsg_ids.lock().await.contains(&privmsg.id) {
+            if self.seen_privmsg_ids.is_seen(privmsg.id).await {
                 continue
             }
 
-            // If not then broadcast to everybody else
-
-            // First update list of privmsg ids
-            self.privmsg_ids.lock().await.insert(privmsg.id);
+            self.seen_privmsg_ids.add_seen(privmsg.id).await;
 
+            // If not then broadcast to everybody else
             let privmsg_copy = (*privmsg).clone();
             self.p2p.broadcast(privmsg_copy).await?;
 

src/net/sessions/manual_session.rs

@@ -12,6 +12,7 @@ use crate::{
         protocols::{ProtocolAddress, ProtocolPing},
         sessions::Session,
         ChannelPtr, Connector, P2p,
+        utility::sleep,
     },
     system::{StoppableTask, StoppableTaskPtr},
 };
@@ -56,8 +57,9 @@ impl ManualSession {
         executor: Arc<Executor<'_>>,
     ) -> Result<()> {
         let connector = Connector::new(self.p2p().settings());
+        let settings = self.p2p().settings();
 
-        let attempts = self.p2p().settings().manual_attempt_limit;
+        let attempts = settings.manual_attempt_limit;
         let mut remaining = attempts;
 
         loop {
@@ -94,6 +96,8 @@ impl ManualSession {
                 }
                 Err(err) => {
                     info!(target: "net", "Unable to connect to manual outbound [{}]: {}", addr, err);
+
+                    sleep(settings.connect_timeout_seconds).await;
                 }
             }
         }