char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somrearrange libs into lib and extensions into extsf4a0f5b

main
6.0 KiB159 linesraw
1use std::convert::Infallible;
2use std::io;
3use std::sync::Arc;
4
5use avril_core::Op;
6use avril_core::serde_v8::{JsBuffer, ToJsBuffer};
7use serde::{Deserialize, Serialize};
8use tokio::sync::Mutex;
9use tokio_util::sync::CancellationToken;
10
11use super::{Authority, Resource, Sockets, closed, hostname, reply, response};
12
13const MAX_DATAGRAM: usize = 65_507;
14#[derive(Clone, Deserialize, Serialize)]
15#[serde(tag = "transport", rename_all = "lowercase")]
16enum Address {
17    Udp { hostname: String, port: u16 },
18}
19impl Address {
20    fn from(value: std::net::SocketAddr) -> Self {
21        Self::Udp {
22            hostname: value.ip().to_string(),
23            port: value.port(),
24        }
25    }
26}
27pub(super) struct Datagram {
28    socket: tokio::net::UdpSocket,
29    authority: Authority,
30    receiving: Mutex<()>,
31    pub(super) canceled: CancellationToken,
32}
33impl Sockets {
34    fn datagram(&self, id: u64) -> io::Result<Arc<Datagram>> {
35        match self.resources.lock().unwrap().get(&id) {
36            Some(Resource::Datagram(value)) => Ok(value.clone()),
37            _ => Err(closed()),
38        }
39    }
40    fn listen_datagram(&self, target: Address) -> io::Result<(u64, Address)> {
41        let Address::Udp { hostname: host, port } = target;
42        let mut authority = Authority::Net {
43            caps: self.caps.clone(),
44            hostname: host.clone(),
45            port,
46        };
47        authority.check()?;
48        let socket = std::net::UdpSocket::bind((hostname(&host), port))?;
49        let local = socket.local_addr()?;
50        if let Authority::Net { port, .. } = &mut authority {
51            *port = local.port();
52        }
53        authority.check()?;
54        socket.set_nonblocking(true)?;
55        let _enter = self.runtime.enter();
56        let socket = tokio::net::UdpSocket::from_std(socket)?;
57        let id = self.id()?;
58        self.resources.lock().unwrap().insert(
59            id,
60            Resource::Datagram(Arc::new(Datagram {
61                socket,
62                authority,
63                receiving: Mutex::new(()),
64                canceled: self.canceled.child_token(),
65            })),
66        );
67        Ok((id, Address::from(local)))
68    }
69}
70pub(super) fn ops(sockets: Arc<Sockets>) -> Vec<Op> {
71    vec![
72        Op::new("op_net_datagram_listen", {
73            let sockets = sockets.clone();
74            move |target: Address| Ok::<_, Infallible>(response(sockets.listen_datagram(target)))
75        }),
76        Op::new("op_net_datagram_broadcast", {
77            let sockets = sockets.clone();
78            move |(id, enabled): (u64, bool)| {
79                let value = (|| {
80                    let socket = sockets.datagram(id)?;
81                    socket.authority.check()?;
82                    socket.socket.set_broadcast(enabled)
83                })();
84                Ok::<_, Infallible>(response(value))
85            }
86        }),
87        Op::asynchronous("op_net_datagram_receive", {
88            let sockets = sockets.clone();
89            let runtime = sockets.runtime.clone();
90            move |(id, count): (u64, usize)| {
91                let socket = sockets.datagram(id);
92                reply(runtime.spawn(async move {
93                    let socket = socket?;
94                    let work = async {
95                        let _receiving = socket.receiving.lock().await;
96                        socket.authority.check()?;
97                        let mut bytes = vec![0; count.min(65535)];
98                        let (count, remote) = socket.socket.recv_from(&mut bytes).await?;
99                        socket.authority.check()?;
100                        bytes.truncate(count);
101                        Ok((ToJsBuffer::from(bytes), Address::from(remote)))
102                    };
103                    tokio::select! {
104                        biased;
105                        _ = socket.canceled.cancelled() => Err(closed()),
106                        value = work => value,
107                    }
108                }))
109            }
110        }),
111        Op::asynchronous("op_net_datagram_send", {
112            let sockets = sockets.clone();
113            let runtime = sockets.runtime.clone();
114            move |(id, bytes, target): (u64, JsBuffer, Address)| {
115                let socket = sockets.datagram(id);
116                let bytes = if bytes.len() > MAX_DATAGRAM {
117                    Err(io::Error::new(
118                        io::ErrorKind::InvalidInput,
119                        "Datagram exceeds 65507 bytes",
120                    ))
121                } else {
122                    Ok(bytes.to_vec())
123                };
124                let Address::Udp { hostname: host, port } = target;
125                let authority = Authority::Net {
126                    caps: sockets.caps.clone(),
127                    hostname: host.clone(),
128                    port,
129                };
130                reply(runtime.spawn(async move {
131                    let socket = socket?;
132                    let bytes = bytes?;
133                    let work = async {
134                        socket.authority.check()?;
135                        authority.check()?;
136                        let ipv4 = socket.socket.local_addr()?.is_ipv4();
137                        let target = tokio::net::lookup_host((hostname(&host), port))
138                            .await?
139                            .find(|address| address.is_ipv4() == ipv4)
140                            .ok_or_else(|| {
141                                io::Error::new(
142                                    io::ErrorKind::AddrNotAvailable,
143                                    "No compatible destination address",
144                                )
145                            })?;
146                        socket.authority.check()?;
147                        authority.check()?;
148                        socket.socket.send_to(&bytes, target).await
149                    };
150                    tokio::select! {
151                        biased;
152                        _ = socket.canceled.cancelled() => Err(closed()),
153                        value = work => value,
154                    }
155                }))
156            }
157        }),
158    ]
159}