src/send.rs

   1 use super::*;
   2 use byteorder::{BigEndian, WriteBytesExt};
   3 use std::{
   4     io::{self, Write},
   5     sync::Arc,
   6 };
   7 use tokio::sync::watch;
   8
   9 pub type AckResult = io::Result<Option<watch::Receiver<bool>>>;
  10
  11 pub struct RudpSender<P: UdpPeer> {
  12     pub(crate) share: Arc<RudpShare<P>>,
  13 }
  14
  15 // derive(Clone) adds unwanted Clone trait bound to P parameter
  16 impl<P: UdpPeer> Clone for RudpSender<P> {
  17     fn clone(&self) -> Self {
  18         Self {
  19             share: Arc::clone(&self.share),
  20         }
  21     }
  22 }
  23
  24 impl<P: UdpPeer> RudpSender<P> {
  25     pub async fn send(&self, pkt: Pkt<'_>) -> AckResult {
  26         self.share.send(PktType::Orig, pkt).await // TODO: splits
  27     }
  28 }
  29
  30 impl<P: UdpPeer> RudpShare<P> {
  31     pub async fn send(&self, tp: PktType, pkt: Pkt<'_>) -> AckResult {
  32         let mut buf = Vec::with_capacity(4 + 2 + 1 + 1 + 2 + 1 + pkt.data.len());
  33         buf.write_u32::<BigEndian>(PROTO_ID)?;
  34         buf.write_u16::<BigEndian>(*self.remote_id.read().await)?;
  35         buf.write_u8(pkt.chan)?;
  36
  37         let mut chan = self.chans[pkt.chan as usize].lock().await;
  38         let seqnum = chan.seqnum;
  39
  40         if !pkt.unrel {
  41             buf.write_u8(PktType::Rel as u8)?;
  42             buf.write_u16::<BigEndian>(seqnum)?;
  43         }
  44
  45         buf.write_u8(tp as u8)?;
  46         buf.write_all(pkt.data.as_ref())?;
  47
  48         self.send_raw(&buf).await?;
  49
  50         if pkt.unrel {
  51             Ok(None)
  52         } else {
  53             // TODO: reliable window
  54             let (tx, rx) = watch::channel(false);
  55             chan.acks.insert(
  56                 seqnum,
  57                 Ack {
  58                     tx,
  59                     rx: rx.clone(),
  60                     data: buf,
  61                 },
  62             );
  63             chan.seqnum = chan.seqnum.overflowing_add(1).0;
  64
  65             Ok(Some(rx))
  66         }
  67     }
  68
  69     pub async fn send_raw(&self, data: &[u8]) -> io::Result<()> {
  70         if data.len() > UDP_PKT_SIZE {
  71             panic!("splitting packets is not implemented yet");
  72         }
  73
  74         self.udp_tx.send(data).await
  75     }
  76 }