msquic_async/
listener.rs

1use crate::connection::Connection;
2
3#[cfg(feature = "msquic-2-5")]
4use msquic_v2_5 as msquic;
5#[cfg(feature = "msquic-seera")]
6use seera_msquic as msquic;
7
8use std::collections::VecDeque;
9use std::future::Future;
10use std::net::SocketAddr;
11use std::sync::{Arc, Mutex};
12use std::task::{Context, Poll, Waker};
13
14use thiserror::Error;
15use tracing::{error, info, trace};
16
17/// Listener for incoming connections.
18pub struct Listener {
19    inner: Arc<ListenerInner>,
20    msquic_listener: msquic::Listener,
21}
22
23impl Listener {
24    /// Create a new listener.
25    pub fn new(
26        registration: &msquic::Registration,
27        configuration: msquic::Configuration,
28    ) -> Result<Self, ListenError> {
29        let inner = Arc::new(ListenerInner::new(configuration));
30        let inner_in_ev = inner.clone();
31        let msquic_listener = msquic::Listener::open(registration, move |_, ev| match ev {
32            msquic::ListenerEvent::NewConnection { info, connection } => {
33                inner_in_ev.handle_event_new_connection(info, connection)
34            }
35            msquic::ListenerEvent::StopComplete {
36                app_close_in_progress,
37            } => inner_in_ev.handle_event_stop_complete(app_close_in_progress),
38        })
39        .map_err(ListenError::OtherError)?;
40        trace!("Listener({:p}) new", inner);
41        Ok(Self {
42            inner,
43            msquic_listener,
44        })
45    }
46
47    /// Start the listener.
48    pub fn start<T: AsRef<[msquic::BufferRef]>>(
49        &self,
50        alpn: &T,
51        local_address: Option<SocketAddr>,
52    ) -> Result<(), ListenError> {
53        let mut exclusive = self.inner.exclusive.lock().unwrap();
54        match exclusive.state {
55            ListenerState::Open | ListenerState::ShutdownComplete => {}
56            ListenerState::StartComplete | ListenerState::Shutdown => {
57                return Err(ListenError::AlreadyStarted);
58            }
59        }
60        let local_address: Option<msquic::Addr> = local_address.map(|x| x.into());
61        self.msquic_listener
62            .start(alpn.as_ref(), local_address.as_ref())
63            .map_err(ListenError::OtherError)?;
64        exclusive.state = ListenerState::StartComplete;
65        Ok(())
66    }
67
68    /// Start the listener on a specified port.
69    pub fn start_on_port<T: AsRef<[msquic::BufferRef]>>(
70        &self,
71        alpn: &T,
72        local_port: Option<u16>,
73    ) -> Result<(), ListenError> {
74        let mut exclusive = self.inner.exclusive.lock().unwrap();
75        match exclusive.state {
76            ListenerState::Open | ListenerState::ShutdownComplete => {}
77            ListenerState::StartComplete | ListenerState::Shutdown => {
78                return Err(ListenError::AlreadyStarted);
79            }
80        }
81        let local_address: Option<msquic::Addr> = local_port.map(|x| {
82            let mut addr = msquic::Addr::from(SocketAddr::from(([0, 0, 0, 0], x)));
83            addr.ipv4.sin_family = msquic::ffi::QUIC_ADDRESS_FAMILY_UNSPEC as u16;
84            addr
85        });
86        self.msquic_listener
87            .start(alpn.as_ref(), local_address.as_ref())
88            .map_err(ListenError::OtherError)?;
89        exclusive.state = ListenerState::StartComplete;
90        Ok(())
91    }
92
93    /// Accept a new connection.
94    pub fn accept(&self) -> Accept<'_> {
95        Accept(self)
96    }
97
98    /// Poll to accept a new connection.
99    pub fn poll_accept(&self, cx: &mut Context<'_>) -> Poll<Result<Connection, ListenError>> {
100        trace!("Listener({:p}) poll_accept", self);
101        let mut exclusive = self.inner.exclusive.lock().unwrap();
102
103        if !exclusive.new_connections.is_empty() {
104            return Poll::Ready(Ok(exclusive.new_connections.pop_front().unwrap()));
105        }
106
107        match exclusive.state {
108            ListenerState::Open => {
109                return Poll::Ready(Err(ListenError::NotStarted));
110            }
111            ListenerState::StartComplete | ListenerState::Shutdown => {}
112            ListenerState::ShutdownComplete => {
113                return Poll::Ready(Err(ListenError::Finished));
114            }
115        }
116        exclusive.new_connection_waiters.push(cx.waker().clone());
117        Poll::Pending
118    }
119
120    /// Stop the listener.
121    pub fn stop(&self) -> Stop<'_> {
122        Stop(self)
123    }
124
125    /// Poll to stop the listener.
126    pub fn poll_stop(&self, cx: &mut Context<'_>) -> Poll<Result<(), ListenError>> {
127        trace!("Listener({:p}) poll_stop", self);
128        let mut call_stop = false;
129        {
130            let mut exclusive = self.inner.exclusive.lock().unwrap();
131
132            match exclusive.state {
133                ListenerState::Open => {
134                    return Poll::Ready(Err(ListenError::NotStarted));
135                }
136                ListenerState::StartComplete => {
137                    call_stop = true;
138                    exclusive.state = ListenerState::Shutdown;
139                }
140                ListenerState::Shutdown => {}
141                ListenerState::ShutdownComplete => {
142                    return Poll::Ready(Ok(()));
143                }
144            }
145            exclusive.shutdown_complete_waiters.push(cx.waker().clone());
146        }
147        if call_stop {
148            self.msquic_listener.stop();
149        }
150        Poll::Pending
151    }
152
153    /// Get the local address the listener is bound to.
154    pub fn local_addr(&self) -> Result<SocketAddr, ListenError> {
155        self.msquic_listener
156            .get_local_addr()
157            .map(|addr| addr.as_socket().expect("not a socket address"))
158            .map_err(|_| ListenError::Failed)
159    }
160
161    /// Get the port the listener is bound to.
162    pub fn local_port(&self) -> Result<u16, ListenError> {
163        self.msquic_listener
164            .get_local_addr()
165            .map(|addr| addr.port())
166            .map_err(|_| ListenError::Failed)
167    }
168
169    /// Set the SSL key log file for new connections.
170    pub fn set_sslkeylog_file(&self, file: std::fs::File) -> Result<(), ListenError> {
171        let mut exclusive = self.inner.exclusive.lock().unwrap();
172        if exclusive.sslkeylog_file.is_some() {
173            return Err(ListenError::SslKeyLogFileAlreadySet);
174        }
175        exclusive.sslkeylog_file = Some(file);
176        Ok(())
177    }
178}
179
180impl Drop for Listener {
181    fn drop(&mut self) {
182        trace!("Listener(Inner: {:p}) dropping", self.inner);
183    }
184}
185
186struct ListenerInner {
187    exclusive: Mutex<ListenerInnerExclusive>,
188    shared: ListenerInnerShared,
189}
190
191struct ListenerInnerExclusive {
192    state: ListenerState,
193    new_connections: VecDeque<Connection>,
194    new_connection_waiters: Vec<Waker>,
195    shutdown_complete_waiters: Vec<Waker>,
196    sslkeylog_file: Option<std::fs::File>,
197}
198unsafe impl Sync for ListenerInnerExclusive {}
199unsafe impl Send for ListenerInnerExclusive {}
200
201struct ListenerInnerShared {
202    configuration: msquic::Configuration,
203}
204unsafe impl Sync for ListenerInnerShared {}
205unsafe impl Send for ListenerInnerShared {}
206
207#[derive(Debug, Clone, PartialEq)]
208enum ListenerState {
209    Open,
210    StartComplete,
211    Shutdown,
212    ShutdownComplete,
213}
214
215impl ListenerInner {
216    fn new(configuration: msquic::Configuration) -> Self {
217        Self {
218            exclusive: Mutex::new(ListenerInnerExclusive {
219                state: ListenerState::Open,
220                new_connections: VecDeque::new(),
221                new_connection_waiters: Vec::new(),
222                shutdown_complete_waiters: Vec::new(),
223                sslkeylog_file: None,
224            }),
225            shared: ListenerInnerShared { configuration },
226        }
227    }
228
229    fn handle_event_new_connection(
230        &self,
231        _info: msquic::NewConnectionInfo<'_>,
232        #[cfg(feature = "msquic-2-5")] connection: msquic::ConnectionRef,
233        #[cfg(not(feature = "msquic-2-5"))] connection: msquic::Connection,
234    ) -> Result<(), msquic::Status> {
235        trace!("Listener({:p}) New connection", self);
236
237        let mut exclusive = self.exclusive.lock().unwrap();
238
239        let (sslkeylog_file, tls_secrets) = if let Some(file) = exclusive.sslkeylog_file.as_ref() {
240            let sslkeylog_file = match file.try_clone() {
241                Ok(f) => {
242                    info!(
243                        "Listener({:p}) SSL key log file set for new connection",
244                        self
245                    );
246                    Some(f)
247                }
248                Err(e) => {
249                    error!(
250                        "Listener({:p}) Failed to clone SSL key log file: {}",
251                        self, e
252                    );
253                    None
254                }
255            };
256
257            if sslkeylog_file.is_none() {
258                (None, None)
259            } else {
260                // Create a QUIC_TLS_SECRETS structure with zeroed fields
261                let tls_secrets = Box::new(msquic::ffi::QUIC_TLS_SECRETS {
262                    SecretLength: 0,
263                    ClientRandom: [0; 32],
264                    IsSet: msquic::ffi::QUIC_TLS_SECRETS__bindgen_ty_1 {
265                        _bitfield_align_1: [0; 0],
266                        _bitfield_1: msquic::ffi::QUIC_TLS_SECRETS__bindgen_ty_1::new_bitfield_1(
267                            0u8, 0u8, 0u8, 0u8, 0u8, 0u8,
268                        ),
269                    },
270                    ClientEarlyTrafficSecret: [0; 64],
271                    ClientHandshakeTrafficSecret: [0; 64],
272                    ServerHandshakeTrafficSecret: [0; 64],
273                    ClientTrafficSecret0: [0; 64],
274                    ServerTrafficSecret0: [0; 64],
275                });
276                unsafe {
277                    msquic::Api::set_param(
278                        connection.as_raw(),
279                        msquic::ffi::QUIC_PARAM_CONN_TLS_SECRETS,
280                        std::mem::size_of::<msquic::ffi::QUIC_TLS_SECRETS>() as u32,
281                        tls_secrets.as_ref() as *const _ as *const _,
282                    )
283                }?;
284                (sslkeylog_file, Some(tls_secrets))
285            }
286        } else {
287            (None, None)
288        };
289        connection.set_configuration(&self.shared.configuration)?;
290        #[cfg(feature = "msquic-2-5")]
291        let new_conn =
292            Connection::from_raw(unsafe { connection.as_raw() }, tls_secrets, sslkeylog_file);
293        #[cfg(not(feature = "msquic-2-5"))]
294        let new_conn = Connection::from_raw(connection, tls_secrets, sslkeylog_file);
295
296        exclusive.new_connections.push_back(new_conn);
297        exclusive
298            .new_connection_waiters
299            .drain(..)
300            .for_each(|waker| waker.wake());
301        Ok(())
302    }
303
304    fn handle_event_stop_complete(
305        &self,
306        app_close_in_progress: bool,
307    ) -> Result<(), msquic::Status> {
308        trace!(
309            "Listener({:p}) Stop complete: app_close_in_progress={}",
310            self,
311            app_close_in_progress
312        );
313        {
314            let mut exclusive = self.exclusive.lock().unwrap();
315            exclusive.state = ListenerState::ShutdownComplete;
316
317            exclusive
318                .new_connection_waiters
319                .drain(..)
320                .for_each(|waker| waker.wake());
321
322            exclusive
323                .shutdown_complete_waiters
324                .drain(..)
325                .for_each(|waker| waker.wake());
326            trace!(
327                "Listener({:p}) new_connections's len={}",
328                self,
329                exclusive.new_connections.len()
330            );
331        }
332        // unsafe {
333        //     Arc::from_raw(self as *const _);
334        // }
335        Ok(())
336    }
337}
338
339impl Drop for ListenerInner {
340    fn drop(&mut self) {
341        trace!("ListenerInner({:p}) dropping", self);
342    }
343}
344
345/// Future generated by `[Listener::accept()]`.
346pub struct Accept<'a>(&'a Listener);
347
348impl Future for Accept<'_> {
349    type Output = Result<Connection, ListenError>;
350
351    fn poll(self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
352        self.0.poll_accept(cx)
353    }
354}
355
356/// Future generated by `[Listener::stop()]`.
357pub struct Stop<'a>(&'a Listener);
358
359impl Future for Stop<'_> {
360    type Output = Result<(), ListenError>;
361
362    fn poll(self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
363        self.0.poll_stop(cx)
364    }
365}
366
367#[derive(Debug, Error, Clone)]
368pub enum ListenError {
369    #[error("Not started yet")]
370    NotStarted,
371    #[error("already started")]
372    AlreadyStarted,
373    #[error("finished")]
374    Finished,
375    #[error("failed")]
376    Failed,
377    #[error("SSL key log file already set")]
378    SslKeyLogFileAlreadySet,
379    #[error("other error: status {0:?}")]
380    OtherError(msquic::Status),
381}
msquic_async/listener.rs

msquic_async/
listener.rs
