send acks

[mt_rudp.git] / src / main.rs

diff --git a/src/main.rs b/src/main.rs
index d5fa9526d89e7365dcd7d7abb0267df55ebc6cf7..1f0aca01f1a53a87b617be8cfc1cff2a0be890d6 100644 (file)
--- a/src/main.rs
+++ b/src/main.rs
@@ -10,11 +10,12 @@ use byteorder::{BigEndian, WriteBytesExt};
 pub use client::{connect, Sender as Client};
 use num_enum::TryFromPrimitive;
 use std::{
+    collections::HashMap,
     io::{self, Write},
     ops,
     sync::Arc,
 };
-use tokio::sync::{mpsc, RwLock};
+use tokio::sync::{mpsc, watch, Mutex, RwLock};
 
 pub const PROTO_ID: u32 = 0x4f457403;
 pub const UDP_PKT_SIZE: usize = 512;
@@ -68,15 +69,13 @@ pub struct Pkt<T> {
 
 pub type Error = error::Error;
 pub type InPkt = Result<Pkt<Vec<u8>>, Error>;
-
-#[derive(Debug)]
-pub struct AckChan;
+type AckChan = (watch::Sender<bool>, watch::Receiver<bool>);
 
 #[derive(Debug)]
 pub struct RudpShare<S: UdpSender> {
     pub id: u16,
     pub remote_id: RwLock<u16>,
-    pub chans: Vec<AckChan>,
+    pub ack_chans: Mutex<HashMap<u16, AckChan>>,
     udp_tx: S,
 }
 
@@ -156,7 +155,7 @@ pub fn new<S: UdpSender, R: UdpReceiver>(
         id,
         remote_id: RwLock::new(remote_id),
         udp_tx,
-        chans: (0..NUM_CHANS).map(|_| AckChan).collect(),
+        ack_chans: Mutex::new(HashMap::new()),
     });
     let recv_share = Arc::clone(&share);
 
@@ -205,5 +204,7 @@ async fn main() -> io::Result<()> {
     }
     println!("disco");
 
+    // close()ing rx is not needed because it has been consumed to the end
+
     Ok(())
 }