From a4c3c05c69a4f9de76e41980594d180ec9ffe03b Mon Sep 17 00:00:00 2001 From: Lizzy Fleckenstein Date: Sun, 26 Feb 2023 03:02:06 +0100 Subject: Implement sending splits --- src/send.rs | 64 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 59 insertions(+), 5 deletions(-) (limited to 'src/send.rs') diff --git a/src/send.rs b/src/send.rs index 90bbe2d..3681eb9 100644 --- a/src/send.rs +++ b/src/send.rs @@ -1,6 +1,7 @@ use super::*; use byteorder::{BigEndian, WriteBytesExt}; use std::{ + borrow::Cow, collections::HashMap, io::{self, Write}, sync::Arc, @@ -20,6 +21,7 @@ pub(crate) struct AckWait { pub(crate) struct Chan { pub(crate) acks: HashMap, pub(crate) seqnum: u16, + pub(crate) splits_seqnum: u16, } #[derive(Debug)] @@ -42,17 +44,58 @@ impl Sender { Mutex::new(Chan { acks: HashMap::new(), seqnum: INIT_SEQNUM, + splits_seqnum: INIT_SEQNUM, }) }), }) } pub async fn send_rudp(&self, pkt: Pkt<'_>) -> io::Result { - self.send_rudp_type(PktType::Orig, pkt).await // TODO: splits + if pkt.size() > UDP_PKT_SIZE { + let chunks = pkt + .data + .chunks(UDP_PKT_SIZE - (pkt.header_size() + 1 + 2 + 2 + 2)); + let num_chunks: u16 = chunks + .len() + .try_into() + .map_err(|_| io::Error::new(io::ErrorKind::Other, "too many chunks"))?; + + let seqnum = { + let mut chan = self.chans[pkt.chan as usize].lock().await; + let sn = chan.splits_seqnum; + chan.splits_seqnum = chan.splits_seqnum.overflowing_add(1).0; + + sn + }; + + for (i, ch) in chunks.enumerate() { + self.send_rudp_type( + PktType::Orig, + Some((seqnum, num_chunks, i as u16)), + Pkt { + unrel: pkt.unrel, + chan: pkt.chan, + data: Cow::Borrowed(ch), + }, + ) + .await?; + } + + Ok(None) // TODO: ack + } else { + self.send_rudp_type(PktType::Orig, None, pkt).await + } } - pub async fn send_rudp_type(&self, tp: PktType, pkt: Pkt<'_>) -> io::Result { - let mut buf = Vec::with_capacity(4 + 2 + 1 + 1 + 2 + 1 + pkt.data.len()); + pub async fn send_rudp_type( + &self, + tp: PktType, + chunk: Option<(u16, u16, u16)>, + pkt: Pkt<'_>, + ) -> io::Result { + let mut buf = + Vec::with_capacity(pkt.size() + if chunk.is_some() { 1 + 2 + 2 + 2 } else { 0 }); + buf.write_u32::(PROTO_ID)?; buf.write_u16::(*self.remote_id.read().await)?; buf.write_u8(pkt.chan)?; @@ -65,7 +108,15 @@ impl Sender { buf.write_u16::(seqnum)?; } - buf.write_u8(tp as u8)?; + if let Some((seqnum, count, index)) = chunk { + buf.write_u8(PktType::Split as u8)?; + buf.write_u16::(seqnum)?; + buf.write_u16::(count)?; + buf.write_u16::(index)?; + } else { + buf.write_u8(tp as u8)?; + } + buf.write_all(pkt.data.as_ref())?; self.send_udp(&buf).await?; @@ -91,7 +142,10 @@ impl Sender { pub async fn send_udp(&self, data: &[u8]) -> io::Result<()> { if data.len() > UDP_PKT_SIZE { - panic!("splitting packets is not implemented yet"); + panic!( + "attempted to send a packet with len {} > {UDP_PKT_SIZE}", + data.len() + ); } self.udp.send(data).await -- cgit v1.2.3