quinn/runtime/
tokio.rs

1use std::{
2    future::Future,
3    io,
4    pin::Pin,
5    sync::Arc,
6    task::{Context, Poll, ready},
7    time::Instant,
8};
9
10use tokio::{
11    io::Interest,
12    time::{Sleep, sleep_until},
13};
14
15use super::{AsyncTimer, AsyncUdpSocket, Runtime, UdpPollHelper};
16
17/// A Quinn runtime for Tokio
18#[derive(Debug)]
19pub struct TokioRuntime;
20
21impl Runtime for TokioRuntime {
22    fn new_timer(&self, t: Instant) -> Pin<Box<dyn AsyncTimer>> {
23        Box::pin(sleep_until(t.into()))
24    }
25
26    fn spawn(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
27        tokio::spawn(future);
28    }
29
30    fn wrap_udp_socket(&self, sock: std::net::UdpSocket) -> io::Result<Arc<dyn AsyncUdpSocket>> {
31        Ok(Arc::new(UdpSocket {
32            inner: udp::UdpSocketState::new((&sock).into())?,
33            io: tokio::net::UdpSocket::from_std(sock)?,
34        }))
35    }
36
37    fn now(&self) -> Instant {
38        tokio::time::Instant::now().into_std()
39    }
40}
41
42impl AsyncTimer for Sleep {
43    fn reset(self: Pin<&mut Self>, t: Instant) {
44        Self::reset(self, t.into())
45    }
46    fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<()> {
47        Future::poll(self, cx)
48    }
49}
50
51#[derive(Debug)]
52struct UdpSocket {
53    io: tokio::net::UdpSocket,
54    inner: udp::UdpSocketState,
55}
56
57impl AsyncUdpSocket for UdpSocket {
58    fn create_io_poller(self: Arc<Self>) -> Pin<Box<dyn super::UdpPoller>> {
59        Box::pin(UdpPollHelper::new(move || {
60            let socket = self.clone();
61            async move { socket.io.writable().await }
62        }))
63    }
64
65    fn try_send(&self, transmit: &udp::Transmit) -> io::Result<()> {
66        self.io.try_io(Interest::WRITABLE, || {
67            self.inner.send((&self.io).into(), transmit)
68        })
69    }
70
71    fn poll_recv(
72        &self,
73        cx: &mut Context,
74        bufs: &mut [std::io::IoSliceMut<'_>],
75        meta: &mut [udp::RecvMeta],
76    ) -> Poll<io::Result<usize>> {
77        loop {
78            ready!(self.io.poll_recv_ready(cx))?;
79            if let Ok(res) = self.io.try_io(Interest::READABLE, || {
80                self.inner.recv((&self.io).into(), bufs, meta)
81            }) {
82                return Poll::Ready(Ok(res));
83            }
84        }
85    }
86
87    fn local_addr(&self) -> io::Result<std::net::SocketAddr> {
88        self.io.local_addr()
89    }
90
91    fn may_fragment(&self) -> bool {
92        self.inner.may_fragment()
93    }
94
95    fn max_transmit_segments(&self) -> usize {
96        self.inner.max_gso_segments()
97    }
98
99    fn max_receive_segments(&self) -> usize {
100        self.inner.gro_segments()
101    }
102}
quinn/runtime/tokio.rs

quinn/runtime/
tokio.rs
