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