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
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
|
use super::*;
use byteorder::{BigEndian, WriteBytesExt};
use std::{
collections::HashMap,
io::{self, Write},
sync::Arc,
};
use tokio::sync::{watch, Mutex, RwLock};
pub type Ack = Option<watch::Receiver<bool>>;
#[derive(Debug)]
pub(crate) struct AckWait {
pub(crate) tx: watch::Sender<bool>,
pub(crate) rx: watch::Receiver<bool>,
pub(crate) data: Vec<u8>,
}
#[derive(Debug)]
pub(crate) struct Chan {
pub(crate) acks: HashMap<u16, AckWait>,
pub(crate) seqnum: u16,
}
#[derive(Debug)]
pub struct Sender<S: UdpSender> {
pub(crate) id: u16,
pub(crate) remote_id: RwLock<u16>,
pub(crate) chans: [Mutex<Chan>; NUM_CHANS],
udp: S,
close: watch::Sender<bool>,
}
impl<S: UdpSender> Sender<S> {
pub fn new(udp: S, close: watch::Sender<bool>, id: u16, remote_id: u16) -> Arc<Self> {
Arc::new(Self {
id,
remote_id: RwLock::new(remote_id),
udp,
close,
chans: std::array::from_fn(|_| {
Mutex::new(Chan {
acks: HashMap::new(),
seqnum: INIT_SEQNUM,
})
}),
})
}
pub async fn send_rudp(&self, pkt: Pkt<'_>) -> io::Result<Ack> {
self.send_rudp_type(PktType::Orig, pkt).await // TODO: splits
}
pub async fn send_rudp_type(&self, tp: PktType, pkt: Pkt<'_>) -> io::Result<Ack> {
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_udp(&buf).await?;
if pkt.unrel {
Ok(None)
} else {
// TODO: reliable window
let (tx, rx) = watch::channel(false);
chan.acks.insert(
seqnum,
AckWait {
tx,
rx: rx.clone(),
data: buf,
},
);
chan.seqnum = chan.seqnum.overflowing_add(1).0;
Ok(Some(rx))
}
}
pub async fn send_udp(&self, data: &[u8]) -> io::Result<()> {
if data.len() > UDP_PKT_SIZE {
panic!("splitting packets is not implemented yet");
}
self.udp.send(data).await
}
pub async fn peer_id(&self) -> u16 {
self.id
}
pub async fn is_server(&self) -> bool {
self.id == PeerID::Srv as u16
}
pub fn close(&self) {
self.close.send(true).ok();
}
}
|
