src/send.rs

   1 use crate::*;
   2 use tokio::sync::watch;
   3
   4 type AckResult = io::Result<Option<watch::Receiver<bool>>>;
   5
   6 impl<S: UdpSender> RudpSender<S> {
   7     pub async fn send(&self, pkt: Pkt<&[u8]>) -> AckResult {
   8         self.share.send(PktType::Orig, pkt).await // TODO: splits
   9     }
  10 }
  11
  12 impl<S: UdpSender> RudpShare<S> {
  13     pub async fn send(&self, tp: PktType, pkt: Pkt<&[u8]>) -> AckResult {
  14         let mut buf = Vec::with_capacity(4 + 2 + 1 + 1 + 2 + 1 + pkt.data.len());
  15         buf.write_u32::<BigEndian>(PROTO_ID)?;
  16         buf.write_u16::<BigEndian>(*self.remote_id.read().await)?;
  17         buf.write_u8(pkt.chan as u8)?;
  18
  19         let mut chan = self.chans[pkt.chan as usize].lock().await;
  20         let seqnum = chan.seqnum;
  21
  22         if !pkt.unrel {
  23             buf.write_u8(PktType::Rel as u8)?;
  24             buf.write_u16::<BigEndian>(seqnum)?;
  25         }
  26
  27         buf.write_u8(tp as u8)?;
  28         buf.write(pkt.data)?;
  29
  30         self.send_raw(&buf).await?;
  31
  32         if pkt.unrel {
  33             Ok(None)
  34         } else {
  35             // TODO: reliable window
  36             let (tx, rx) = watch::channel(false);
  37             chan.acks.insert(
  38                 seqnum,
  39                 Ack {
  40                     tx,
  41                     rx: rx.clone(),
  42                     data: buf,
  43                 },
  44             );
  45             chan.seqnum = chan.seqnum.overflowing_add(1).0;
  46
  47             Ok(Some(rx))
  48         }
  49     }
  50
  51     pub async fn send_raw(&self, data: &[u8]) -> io::Result<()> {
  52         self.udp_tx.send(data).await
  53         // TODO: reset ping timeout
  54     }
  55 }