Use channels

[mt_rudp.git] / src / send.rs

diff --git a/src/send.rs b/src/send.rs
index 20308e6758fc3dbeaceac22ccfd805cb0cbf2c1f..90bbe2d83e0b56dd0ef1b58e41297a27a4d0600a 100644 (file)
--- a/src/send.rs
+++ b/src/send.rs
@@ -1,20 +1,61 @@
-use crate::*;
-use tokio::sync::watch;
+use super::*;
+use byteorder::{BigEndian, WriteBytesExt};
+use std::{
+    collections::HashMap,
+    io::{self, Write},
+    sync::Arc,
+};
+use tokio::sync::{watch, Mutex, RwLock};
 
-type AckResult = io::Result<Option<watch::Receiver<bool>>>;
+pub type Ack = Option<watch::Receiver<bool>>;
 
-impl<S: UdpSender> RudpSender<S> {
-    pub async fn send(&self, pkt: Pkt<&[u8]>) -> AckResult {
-        self.share.send(PktType::Orig, pkt).await // TODO: splits
-    }
+#[derive(Debug)]
+pub(crate) struct AckWait {
+    pub(crate) tx: watch::Sender<bool>,
+    pub(crate) rx: watch::Receiver<bool>,
+    pub(crate) data: Vec<u8>,
+}
+
+#[derive(Debug)]
+pub(crate) struct Chan {
+    pub(crate) acks: HashMap<u16, AckWait>,
+    pub(crate) seqnum: u16,
+}
+
+#[derive(Debug)]
+pub struct Sender<S: UdpSender> {
+    pub(crate) id: u16,
+    pub(crate) remote_id: RwLock<u16>,
+    pub(crate) chans: [Mutex<Chan>; NUM_CHANS],
+    udp: S,
+    close: watch::Sender<bool>,
 }
 
-impl<S: UdpSender> RudpShare<S> {
-    pub async fn send(&self, tp: PktType, pkt: Pkt<&[u8]>) -> AckResult {
+impl<S: UdpSender> Sender<S> {
+    pub fn new(udp: S, close: watch::Sender<bool>, id: u16, remote_id: u16) -> Arc<Self> {
+        Arc::new(Self {
+            id,
+            remote_id: RwLock::new(remote_id),
+            udp,
+            close,
+            chans: std::array::from_fn(|_| {
+                Mutex::new(Chan {
+                    acks: HashMap::new(),
+                    seqnum: INIT_SEQNUM,
+                })
+            }),
+        })
+    }
+
+    pub async fn send_rudp(&self, pkt: Pkt<'_>) -> io::Result<Ack> {
+        self.send_rudp_type(PktType::Orig, pkt).await // TODO: splits
+    }
+
+    pub async fn send_rudp_type(&self, tp: PktType, pkt: Pkt<'_>) -> io::Result<Ack> {
         let mut buf = Vec::with_capacity(4 + 2 + 1 + 1 + 2 + 1 + pkt.data.len());
         buf.write_u32::<BigEndian>(PROTO_ID)?;
         buf.write_u16::<BigEndian>(*self.remote_id.read().await)?;
-        buf.write_u8(pkt.chan as u8)?;
+        buf.write_u8(pkt.chan)?;
 
         let mut chan = self.chans[pkt.chan as usize].lock().await;
         let seqnum = chan.seqnum;
@@ -25,9 +66,9 @@ impl<S: UdpSender> RudpShare<S> {
         }
 
         buf.write_u8(tp as u8)?;
-        buf.write(pkt.data)?;
+        buf.write_all(pkt.data.as_ref())?;
 
-        self.send_raw(&buf).await?;
+        self.send_udp(&buf).await?;
 
         if pkt.unrel {
             Ok(None)
@@ -36,7 +77,7 @@ impl<S: UdpSender> RudpShare<S> {
             let (tx, rx) = watch::channel(false);
             chan.acks.insert(
                 seqnum,
-                Ack {
+                AckWait {
                     tx,
                     rx: rx.clone(),
                     data: buf,
@@ -48,8 +89,23 @@ impl<S: UdpSender> RudpShare<S> {
         }
     }
 
-    pub async fn send_raw(&self, data: &[u8]) -> io::Result<()> {
-        self.udp_tx.send(data).await
-        // TODO: reset ping timeout
+    pub async fn send_udp(&self, data: &[u8]) -> io::Result<()> {
+        if data.len() > UDP_PKT_SIZE {
+            panic!("splitting packets is not implemented yet");
+        }
+
+        self.udp.send(data).await
+    }
+
+    pub async fn peer_id(&self) -> u16 {
+        self.id
+    }
+
+    pub async fn is_server(&self) -> bool {
+        self.id == PeerID::Srv as u16
+    }
+
+    pub fn close(&self) {
+        self.close.send(true).ok();
     }
 }