src/send.rs

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