reparenting checks and conn owned objects

1 files changed, 21 insertions, 4 deletions

diff --git a/server/src/network.rs b/server/src/network.rs
index 5c6f173..e632e97 100644
--- a/server/src/network.rs
+++ b/server/src/network.rs
@@ -27,7 +27,7 @@ use std::{
     thread::spawn,
     time::Instant,
 };
-use weareshared::{helper::ReadWrite, packets::Packet};
+use weareshared::{helper::ReadWrite, packets::Packet, tree::PacketSink};
 
 pub struct ServerNetwork {
     conns: Arc<Mutex<HashMap<u128, (SyncSender<Arc<Vec<u8>>>, Option<SocketAddr>)>>>,
@@ -93,9 +93,10 @@ impl ServerNetwork {
                             warn!("client outbound error: {e}");
                         }
                     });
-                    if let Err(e) = handle_conn_read(conn, sock, recv_tx) {
+                    if let Err(e) = handle_conn_read(conn, sock, &recv_tx) {
                         warn!("client inbound error: {e}");
                     }
+                    recv_tx.send((conn, Packet::Disconnect)).unwrap();
                     conns.lock().unwrap().remove(&conn);
                 });
             }
@@ -192,12 +193,16 @@ impl ServerNetwork {
     }
 }
 
-fn handle_conn_read(conn: u128, sock: TcpStream, tx: Sender<(u128, Packet)>) -> Result<()> {
+fn handle_conn_read(conn: u128, sock: TcpStream, tx: &Sender<(u128, Packet)>) -> Result<()> {
     let mut sock = BufReader::new(sock);
     loop {
         let packet = Packet::read(&mut sock)?;
         debug!("{conn} <- {packet:?}");
-        tx.send((conn, packet)).unwrap();
+        if matches!(packet, Packet::Disconnect) {
+            return Ok(()); // not sending because generic disconnect sends it too
+        } else {
+            tx.send((conn, packet)).unwrap();
+        }
     }
 }
 fn handle_conn_write(sock: TcpStream, rx: Receiver<Arc<Vec<u8>>>) -> Result<()> {
@@ -208,3 +213,15 @@ fn handle_conn_write(sock: TcpStream, rx: Receiver<Arc<Vec<u8>>>) -> Result<()>
     }
     Ok(())
 }
+
+impl ServerNetwork {
+    pub fn br<'a>(&'a self) -> Broadcaster<'a> {
+        Broadcaster(self)
+    }
+}
+pub struct Broadcaster<'a>(&'a ServerNetwork);
+impl<'a> PacketSink for Broadcaster<'a> {
+    fn push(&mut self, p: Packet) {
+        self.0.broadcast(p, true);
+    }
+}