msquic_async/
connection.rs

1use crate::buffer::WriteBuffer;
2use crate::stream::{ReadStream, StartError as StreamStartError, Stream, StreamType};
3
4#[cfg(feature = "msquic-2-5")]
5use msquic_v2_5 as msquic;
6#[cfg(feature = "msquic-seera")]
7use seera_msquic as msquic;
8
9use std::collections::VecDeque;
10use std::fs::File;
11use std::future::Future;
12use std::io::{Seek, SeekFrom, Write};
13use std::net::SocketAddr;
14use std::ops::Deref;
15use std::pin::Pin;
16use std::sync::{Arc, Mutex};
17use std::task::{Context, Poll, Waker};
18
19use bytes::Bytes;
20use libc::c_void;
21use msquic::ffi::QUIC_TLS_SECRETS__bindgen_ty_1;
22use thiserror::Error;
23use tracing::{info, trace};
24
25#[derive(Clone)]
26pub struct Connection(Arc<ConnectionInstance>);
27
28impl Connection {
29    /// Create a new connection.
30    ///
31    /// The connection is not started until `start` is called.
32    pub fn new(registration: &msquic::Registration) -> Result<Self, ConnectionError> {
33        let inner = Arc::new(ConnectionInner::new(ConnectionState::Open, None, None));
34        let inner_in_ev = inner.clone();
35        let msquic_conn = msquic::Connection::open(registration, move |conn_ref, ev| {
36            inner_in_ev.callback_handler_impl(conn_ref, ev)
37        })
38        .map_err(ConnectionError::OtherError)?;
39        let instance = Arc::new(ConnectionInstance { inner, msquic_conn });
40        trace!(
41            "ConnectionInstance({:p}, Inner: {:p}) Open by local",
42            instance,
43            instance.inner
44        );
45        Ok(Self(instance))
46    }
47
48    pub(crate) fn from_raw(
49        #[cfg(feature = "msquic-2-5")] handle: msquic::ffi::HQUIC,
50        #[cfg(not(feature = "msquic-2-5"))] msquic_conn: msquic::Connection,
51        tls_secrets: Option<Box<msquic::ffi::QUIC_TLS_SECRETS>>,
52        sslkeylog_file: Option<File>,
53    ) -> Self {
54        #[cfg(feature = "msquic-2-5")]
55        let msquic_conn = unsafe { msquic::Connection::from_raw(handle) };
56        let inner = Arc::new(ConnectionInner::new(
57            ConnectionState::Connected,
58            tls_secrets,
59            sslkeylog_file,
60        ));
61        let inner_in_ev = inner.clone();
62        msquic_conn.set_callback_handler(move |conn_ref, ev| {
63            inner_in_ev.callback_handler_impl(conn_ref, ev)
64        });
65        let instance = Arc::new(ConnectionInstance { inner, msquic_conn });
66        trace!(
67            "ConnectionInstance({:p}, Inner: {:p}) Open by peer",
68            instance,
69            instance.inner
70        );
71        Self(instance)
72    }
73
74    /// Start the connection.
75    pub fn start<'a>(
76        &'a self,
77        configuration: &'a msquic::Configuration,
78        host: &'a str,
79        port: u16,
80    ) -> ConnectionStart<'a> {
81        ConnectionStart {
82            conn: self,
83            configuration,
84            host,
85            port,
86        }
87    }
88
89    /// Poll to start the connection.
90    fn poll_start(
91        &self,
92        cx: &mut Context<'_>,
93        configuration: &msquic::Configuration,
94        host: &str,
95        port: u16,
96    ) -> Poll<Result<(), StartError>> {
97        let mut exclusive = self.0.exclusive.lock().unwrap();
98        match exclusive.state {
99            ConnectionState::Open => {
100                self.0
101                    .msquic_conn
102                    .start(configuration, host, port)
103                    .map_err(StartError::OtherError)?;
104                exclusive.state = ConnectionState::Connecting;
105            }
106            ConnectionState::Connecting => {}
107            ConnectionState::Connected => return Poll::Ready(Ok(())),
108            ConnectionState::Shutdown | ConnectionState::ShutdownComplete => {
109                return Poll::Ready(Err(StartError::ConnectionLost(
110                    exclusive.error.as_ref().expect("error").clone(),
111                )));
112            }
113        }
114        exclusive.start_waiters.push(cx.waker().clone());
115        Poll::Pending
116    }
117
118    /// Open a new outbound stream.
119    pub fn open_outbound_stream(
120        &self,
121        stream_type: StreamType,
122        fail_on_blocked: bool,
123    ) -> OpenOutboundStream<'_> {
124        OpenOutboundStream {
125            conn: &self.0,
126            stream_type: Some(stream_type),
127            stream: None,
128            fail_on_blocked,
129        }
130    }
131
132    /// Accept an inbound bidilectional stream.
133    pub fn accept_inbound_stream(&self) -> AcceptInboundStream<'_> {
134        AcceptInboundStream { conn: self }
135    }
136
137    /// Poll to accept an inbound bidilectional stream.
138    pub fn poll_accept_inbound_stream(
139        &self,
140        cx: &mut Context<'_>,
141    ) -> Poll<Result<Stream, StreamStartError>> {
142        let mut exclusive = self.0.exclusive.lock().unwrap();
143        match exclusive.state {
144            ConnectionState::Open => {
145                return Poll::Ready(Err(StreamStartError::ConnectionNotStarted));
146            }
147            ConnectionState::Connecting => {
148                exclusive.start_waiters.push(cx.waker().clone());
149                return Poll::Pending;
150            }
151            ConnectionState::Connected => {}
152            ConnectionState::Shutdown | ConnectionState::ShutdownComplete => {
153                return Poll::Ready(Err(StreamStartError::ConnectionLost(
154                    exclusive.error.as_ref().expect("error").clone(),
155                )));
156            }
157        }
158
159        if !exclusive.inbound_streams.is_empty() {
160            return Poll::Ready(Ok(exclusive.inbound_streams.pop_front().unwrap()));
161        }
162        exclusive.inbound_stream_waiters.push(cx.waker().clone());
163        Poll::Pending
164    }
165
166    /// Accept an inbound unidirectional stream.
167    pub fn accept_inbound_uni_stream(&self) -> AcceptInboundUniStream<'_> {
168        AcceptInboundUniStream { conn: self }
169    }
170
171    /// Poll to accept an inbound unidirectional stream.
172    pub fn poll_accept_inbound_uni_stream(
173        &self,
174        cx: &mut Context<'_>,
175    ) -> Poll<Result<ReadStream, StreamStartError>> {
176        let mut exclusive = self.0.exclusive.lock().unwrap();
177        match exclusive.state {
178            ConnectionState::Open => {
179                return Poll::Ready(Err(StreamStartError::ConnectionNotStarted));
180            }
181            ConnectionState::Connecting => {
182                exclusive.start_waiters.push(cx.waker().clone());
183                return Poll::Pending;
184            }
185            ConnectionState::Connected => {}
186            ConnectionState::Shutdown | ConnectionState::ShutdownComplete => {
187                return Poll::Ready(Err(StreamStartError::ConnectionLost(
188                    exclusive.error.as_ref().expect("error").clone(),
189                )));
190            }
191        }
192
193        if !exclusive.inbound_uni_streams.is_empty() {
194            return Poll::Ready(Ok(exclusive.inbound_uni_streams.pop_front().unwrap()));
195        }
196        exclusive
197            .inbound_uni_stream_waiters
198            .push(cx.waker().clone());
199        Poll::Pending
200    }
201
202    /// Poll to receive a datagram.
203    pub fn poll_receive_datagram(
204        &self,
205        cx: &mut Context<'_>,
206    ) -> Poll<Result<Bytes, DgramReceiveError>> {
207        let mut exclusive = self.0.exclusive.lock().unwrap();
208        match exclusive.state {
209            ConnectionState::Open => {
210                return Poll::Ready(Err(DgramReceiveError::ConnectionNotStarted));
211            }
212            ConnectionState::Connecting => {
213                exclusive.start_waiters.push(cx.waker().clone());
214                return Poll::Pending;
215            }
216            ConnectionState::Connected => {}
217            ConnectionState::Shutdown | ConnectionState::ShutdownComplete => {
218                return Poll::Ready(Err(DgramReceiveError::ConnectionLost(
219                    exclusive.error.as_ref().expect("error").clone(),
220                )));
221            }
222        }
223
224        if let Some(buf) = exclusive.recv_buffers.pop_front() {
225            Poll::Ready(Ok(buf))
226        } else {
227            exclusive.recv_waiters.push(cx.waker().clone());
228            Poll::Pending
229        }
230    }
231
232    /// Poll to send a datagram.
233    pub fn poll_send_datagram(
234        &self,
235        cx: &mut Context<'_>,
236        buf: &Bytes,
237    ) -> Poll<Result<(), DgramSendError>> {
238        let mut exclusive = self.0.exclusive.lock().unwrap();
239        match exclusive.state {
240            ConnectionState::Open => {
241                return Poll::Ready(Err(DgramSendError::ConnectionNotStarted));
242            }
243            ConnectionState::Connecting => {
244                exclusive.start_waiters.push(cx.waker().clone());
245                return Poll::Pending;
246            }
247            ConnectionState::Connected => {}
248            ConnectionState::Shutdown | ConnectionState::ShutdownComplete => {
249                return Poll::Ready(Err(DgramSendError::ConnectionLost(
250                    exclusive.error.as_ref().expect("error").clone(),
251                )));
252            }
253        }
254
255        let mut write_buf = exclusive.write_pool.pop().unwrap_or(WriteBuffer::new());
256        let _ = write_buf.put_zerocopy(buf);
257        let buffers = unsafe {
258            let (data, len) = write_buf.get_buffers();
259            std::slice::from_raw_parts(data, len)
260        };
261        let res = unsafe {
262            self.0.msquic_conn.datagram_send(
263                buffers,
264                msquic::SendFlags::NONE,
265                write_buf.into_raw() as *const _,
266            )
267        }
268        .map_err(DgramSendError::OtherError);
269        Poll::Ready(res)
270    }
271
272    /// Send a datagram.
273    pub fn send_datagram(&self, buf: &Bytes) -> Result<(), DgramSendError> {
274        let mut exclusive = self.0.exclusive.lock().unwrap();
275        match exclusive.state {
276            ConnectionState::Open => {
277                return Err(DgramSendError::ConnectionNotStarted);
278            }
279            ConnectionState::Connecting => {
280                return Err(DgramSendError::ConnectionNotStarted);
281            }
282            ConnectionState::Connected => {}
283            ConnectionState::Shutdown | ConnectionState::ShutdownComplete => {
284                return Err(DgramSendError::ConnectionLost(
285                    exclusive.error.as_ref().expect("error").clone(),
286                ));
287            }
288        }
289
290        if !exclusive.dgram_send_enabled {
291            return Err(DgramSendError::Denied);
292        }
293        if buf.len() > exclusive.dgram_max_send_length as usize {
294            return Err(DgramSendError::TooBig);
295        }
296
297        let mut write_buf = exclusive.write_pool.pop().unwrap_or(WriteBuffer::new());
298        let _ = write_buf.put_zerocopy(buf);
299        let buffers = unsafe {
300            let (data, len) = write_buf.get_buffers();
301            std::slice::from_raw_parts(data, len)
302        };
303        unsafe {
304            self.0.msquic_conn.datagram_send(
305                buffers,
306                msquic::SendFlags::NONE,
307                write_buf.into_raw() as *const _,
308            )
309        }
310        .map_err(DgramSendError::OtherError)?;
311        Ok(())
312    }
313
314    /// Poll to shutdown the connection.
315    pub fn poll_shutdown(
316        &self,
317        cx: &mut Context<'_>,
318        error_code: u64,
319    ) -> Poll<Result<(), ShutdownError>> {
320        let mut exclusive = self.0.exclusive.lock().unwrap();
321        match exclusive.state {
322            ConnectionState::Open => {
323                return Poll::Ready(Err(ShutdownError::ConnectionNotStarted));
324            }
325            ConnectionState::Connecting => {
326                exclusive.start_waiters.push(cx.waker().clone());
327                return Poll::Pending;
328            }
329            ConnectionState::Connected => {
330                self.0
331                    .msquic_conn
332                    .shutdown(msquic::ConnectionShutdownFlags::NONE, error_code);
333                exclusive.state = ConnectionState::Shutdown;
334                exclusive.error = Some(ConnectionError::ShutdownByLocal);
335            }
336            ConnectionState::Shutdown => {}
337            ConnectionState::ShutdownComplete => {
338                if let Some(ConnectionError::ShutdownByLocal) = &exclusive.error {
339                    return Poll::Ready(Ok(()));
340                } else {
341                    return Poll::Ready(Err(ShutdownError::ConnectionLost(
342                        exclusive.error.as_ref().expect("error").clone(),
343                    )));
344                }
345            }
346        }
347
348        exclusive.shutdown_waiters.push(cx.waker().clone());
349        Poll::Pending
350    }
351
352    /// Shutdown the connection.
353    pub fn shutdown(&self, error_code: u64) -> Result<(), ShutdownError> {
354        let mut exclusive = self.0.exclusive.lock().unwrap();
355        match exclusive.state {
356            ConnectionState::Open | ConnectionState::Connecting => {
357                return Err(ShutdownError::ConnectionNotStarted);
358            }
359            ConnectionState::Connected => {
360                self.0
361                    .msquic_conn
362                    .shutdown(msquic::ConnectionShutdownFlags::NONE, error_code);
363                exclusive.state = ConnectionState::Shutdown;
364                exclusive.error = Some(ConnectionError::ShutdownByLocal);
365            }
366            _ => {}
367        }
368        Ok(())
369    }
370
371    /// Get the local address of the connection.
372    pub fn get_local_addr(&self) -> Result<SocketAddr, ConnectionError> {
373        self.0
374            .msquic_conn
375            .get_local_addr()
376            .map(|addr| addr.as_socket().expect("socket addr"))
377            .map_err(ConnectionError::OtherError)
378    }
379
380    /// Get the remote address of the connection.
381    pub fn get_remote_addr(&self) -> Result<SocketAddr, ConnectionError> {
382        self.0
383            .msquic_conn
384            .get_remote_addr()
385            .map(|addr| addr.as_socket().expect("socket addr"))
386            .map_err(ConnectionError::OtherError)
387    }
388
389    /// Set whether to share the UDP binding.
390    pub fn set_share_binding(&self, share: bool) -> Result<(), ConnectionError> {
391        let share: u8 = if share { 1 } else { 0 };
392        unsafe {
393            msquic::Api::set_param(
394                self.0.msquic_conn.as_raw(),
395                msquic::ffi::QUIC_PARAM_CONN_SHARE_UDP_BINDING,
396                std::mem::size_of::<u8>() as u32,
397                &share as *const _ as *const _,
398            )
399        }
400        .map_err(ConnectionError::OtherError)
401    }
402
403    /// Add a new path to the connection.
404    #[cfg(feature = "msquic-seera")]
405    pub fn add_path(
406        &self,
407        local_addr: SocketAddr,
408        remote_addr: SocketAddr,
409    ) -> Result<(), ConnectionError> {
410        unsafe {
411            msquic::Api::set_param(
412                self.0.msquic_conn.as_raw(),
413                msquic::ffi::QUIC_PARAM_CONN_ADD_PATH,
414                std::mem::size_of::<msquic::ffi::QUIC_PATH_PARAM>() as u32,
415                &msquic::ffi::QUIC_PATH_PARAM {
416                    LocalAddress: &mut msquic::Addr::from(local_addr) as *mut _ as *mut _,
417                    RemoteAddress: &mut msquic::Addr::from(remote_addr) as *mut _ as *mut _,
418                } as *const _ as *const _,
419            )
420        }
421        .map_err(ConnectionError::OtherError)
422    }
423
424    /// Activate a path for the connection.
425    #[cfg(feature = "msquic-seera")]
426    pub fn activate_path(
427        &self,
428        local_addr: SocketAddr,
429        remote_addr: SocketAddr,
430    ) -> Result<(), ConnectionError> {
431        unsafe {
432            msquic::Api::set_param(
433                self.0.msquic_conn.as_raw(),
434                msquic::ffi::QUIC_PARAM_CONN_ACTIVATE_PATH,
435                std::mem::size_of::<msquic::ffi::QUIC_PATH_PARAM>() as u32,
436                &msquic::ffi::QUIC_PATH_PARAM {
437                    LocalAddress: &mut msquic::Addr::from(local_addr) as *mut _ as *mut _,
438                    RemoteAddress: &mut msquic::Addr::from(remote_addr) as *mut _ as *mut _,
439                } as *const _ as *const _,
440            )
441        }
442        .map_err(ConnectionError::OtherError)
443    }
444
445    /// Remove a path from the connection.
446    #[cfg(feature = "msquic-seera")]
447    pub fn remove_path(
448        &self,
449        local_addr: SocketAddr,
450        remote_addr: SocketAddr,
451    ) -> Result<(), ConnectionError> {
452        unsafe {
453            msquic::Api::set_param(
454                self.0.msquic_conn.as_raw(),
455                msquic::ffi::QUIC_PARAM_CONN_REMOVE_PATH,
456                std::mem::size_of::<msquic::ffi::QUIC_PATH_PARAM>() as u32,
457                &msquic::ffi::QUIC_PATH_PARAM {
458                    LocalAddress: &mut msquic::Addr::from(local_addr) as *mut _ as *mut _,
459                    RemoteAddress: &mut msquic::Addr::from(remote_addr) as *mut _ as *mut _,
460                } as *const _ as *const _,
461            )
462        }
463        .map_err(ConnectionError::OtherError)
464    }
465
466    /// Add a bound address to the connection.
467    #[cfg(feature = "msquic-seera")]
468    pub fn add_bound_addr(&self, addr: SocketAddr) -> Result<(), ConnectionError> {
469        unsafe {
470            msquic::Api::set_param(
471                self.0.msquic_conn.as_raw(),
472                msquic::ffi::QUIC_PARAM_CONN_ADD_BOUND_ADDRESS,
473                std::mem::size_of::<msquic::Addr>() as u32,
474                &msquic::Addr::from(addr) as *const _ as *const _,
475            )
476        }
477        .map_err(ConnectionError::OtherError)
478    }
479
480    /// Add an observed address to the connection.
481    #[cfg(feature = "msquic-seera")]
482    pub fn add_observed_addr(
483        &self,
484        addr: SocketAddr,
485        observed_addr: SocketAddr,
486    ) -> Result<(), ConnectionError> {
487        unsafe {
488            msquic::Api::set_param(
489                self.0.msquic_conn.as_raw(),
490                msquic::ffi::QUIC_PARAM_CONN_ADD_OBSERVED_ADDRESS,
491                std::mem::size_of::<msquic::ffi::QUIC_ADD_OBSERVED_ADDRESS>() as u32,
492                &msquic::ffi::QUIC_ADD_OBSERVED_ADDRESS {
493                    LocalAddress: &mut msquic::Addr::from(addr) as *mut _ as *mut _,
494                    ObservedAddress: &mut msquic::Addr::from(observed_addr) as *mut _ as *mut _,
495                } as *const _ as *const _,
496            )
497        }
498        .map_err(ConnectionError::OtherError)
499    }
500
501    /// Remove a bound address from the connection.
502    #[cfg(feature = "msquic-seera")]
503    pub fn remove_bound_addr(&self, addr: SocketAddr) -> Result<(), ConnectionError> {
504        unsafe {
505            msquic::Api::set_param(
506                self.0.msquic_conn.as_raw(),
507                msquic::ffi::QUIC_PARAM_CONN_REMOVE_BOUND_ADDRESS,
508                std::mem::size_of::<msquic::Addr>() as u32,
509                &msquic::Addr::from(addr) as *const _ as *const _,
510            )
511        }
512        .map_err(ConnectionError::OtherError)
513    }
514
515    /// Add a candidate address to the connection.
516    #[cfg(feature = "msquic-seera")]
517    pub fn add_candidate_addr(
518        &self,
519        host_addr: SocketAddr,
520        observed_addr: SocketAddr,
521    ) -> Result<(), ConnectionError> {
522        unsafe {
523            msquic::Api::set_param(
524                self.0.msquic_conn.as_raw(),
525                msquic::ffi::QUIC_PARAM_CONN_ADD_CANDIDATE_ADDRESS,
526                std::mem::size_of::<msquic::ffi::QUIC_CANDIDATE_ADDRESS>() as u32,
527                &msquic::ffi::QUIC_CANDIDATE_ADDRESS {
528                    HostAddress: &mut msquic::Addr::from(host_addr) as *mut _ as *mut _,
529                    ObservedAddress: &mut msquic::Addr::from(observed_addr) as *mut _ as *mut _,
530                } as *const _ as *const _,
531            )
532        }
533        .map_err(ConnectionError::OtherError)
534    }
535
536    /// Remove a candidate address from the connection.
537    #[cfg(feature = "msquic-seera")]
538    pub fn remove_candidate_addr(
539        &self,
540        host_addr: SocketAddr,
541        observed_addr: SocketAddr,
542    ) -> Result<(), ConnectionError> {
543        unsafe {
544            msquic::Api::set_param(
545                self.0.msquic_conn.as_raw(),
546                msquic::ffi::QUIC_PARAM_CONN_REMOVE_CANDIDATE_ADDRESS,
547                std::mem::size_of::<msquic::ffi::QUIC_CANDIDATE_ADDRESS>() as u32,
548                &msquic::ffi::QUIC_CANDIDATE_ADDRESS {
549                    HostAddress: &mut msquic::Addr::from(host_addr) as *mut _ as *mut _,
550                    ObservedAddress: &mut msquic::Addr::from(observed_addr) as *mut _ as *mut _,
551                } as *const _ as *const _,
552            )
553        }
554        .map_err(ConnectionError::OtherError)
555    }
556
557    /// Poll to receive events on the connection.
558    pub fn poll_event(&self, cx: &mut Context<'_>) -> Poll<Result<ConnectionEvent, EventError>> {
559        let mut exclusive = self.0.exclusive.lock().unwrap();
560        match exclusive.state {
561            ConnectionState::Open => {
562                return Poll::Ready(Err(EventError::ConnectionNotStarted));
563            }
564            ConnectionState::Connecting => {
565                exclusive.start_waiters.push(cx.waker().clone());
566                return Poll::Pending;
567            }
568            ConnectionState::Connected | ConnectionState::Shutdown => {}
569            ConnectionState::ShutdownComplete => {
570                return Poll::Ready(Err(EventError::ConnectionLost(
571                    exclusive.error.as_ref().expect("error").clone(),
572                )));
573            }
574        }
575
576        if exclusive.events.is_empty() {
577            exclusive.event_waiters.push(cx.waker().clone());
578            Poll::Pending
579        } else {
580            Poll::Ready(Ok(exclusive.events.pop_front().unwrap()))
581        }
582    }
583
584    /// Set the SSL key log file for the connection.
585    pub fn set_sslkeylog_file(&self, file: File) -> Result<(), ConnectionError> {
586        let mut exclusive = self.0.exclusive.lock().unwrap();
587        if exclusive.sslkeylog_file.is_some() {
588            return Err(ConnectionError::SslKeyLogFileAlreadySet);
589        }
590        if exclusive.tls_secrets.is_none() {
591            exclusive.tls_secrets = Some(Box::new(msquic::ffi::QUIC_TLS_SECRETS {
592                SecretLength: 0,
593                ClientRandom: [0; 32],
594                IsSet: QUIC_TLS_SECRETS__bindgen_ty_1 {
595                    _bitfield_align_1: [0; 0],
596                    _bitfield_1: QUIC_TLS_SECRETS__bindgen_ty_1::new_bitfield_1(
597                        0u8, 0u8, 0u8, 0u8, 0u8, 0u8,
598                    ),
599                },
600                ClientEarlyTrafficSecret: [0; 64],
601                ClientHandshakeTrafficSecret: [0; 64],
602                ServerHandshakeTrafficSecret: [0; 64],
603                ClientTrafficSecret0: [0; 64],
604                ServerTrafficSecret0: [0; 64],
605            }));
606            unsafe {
607                msquic::Api::set_param(
608                    self.0.msquic_conn.as_raw(),
609                    msquic::ffi::QUIC_PARAM_CONN_TLS_SECRETS,
610                    std::mem::size_of::<msquic::ffi::QUIC_TLS_SECRETS>() as u32,
611                    exclusive.tls_secrets.as_ref().unwrap().as_ref() as *const _ as *const _,
612                )
613            }
614            .map_err(ConnectionError::OtherError)?;
615        }
616        exclusive.sslkeylog_file = Some(file);
617        Ok(())
618    }
619}
620
621struct ConnectionInstance {
622    inner: Arc<ConnectionInner>,
623    msquic_conn: msquic::Connection,
624}
625
626impl Deref for ConnectionInstance {
627    type Target = ConnectionInner;
628
629    fn deref(&self) -> &Self::Target {
630        &self.inner
631    }
632}
633
634impl Drop for ConnectionInstance {
635    fn drop(&mut self) {
636        trace!("ConnectionInstance({:p}) dropping", self);
637    }
638}
639
640struct ConnectionInner {
641    exclusive: Mutex<ConnectionInnerExclusive>,
642}
643
644struct ConnectionInnerExclusive {
645    state: ConnectionState,
646    error: Option<ConnectionError>,
647    start_waiters: Vec<Waker>,
648    inbound_stream_waiters: Vec<Waker>,
649    inbound_uni_stream_waiters: Vec<Waker>,
650    inbound_streams: VecDeque<crate::stream::Stream>,
651    inbound_uni_streams: VecDeque<crate::stream::ReadStream>,
652    recv_buffers: VecDeque<Bytes>,
653    recv_waiters: Vec<Waker>,
654    write_pool: Vec<WriteBuffer>,
655    dgram_send_enabled: bool,
656    dgram_max_send_length: u16,
657    shutdown_waiters: Vec<Waker>,
658    events: VecDeque<ConnectionEvent>,
659    event_waiters: Vec<Waker>,
660    sslkeylog_file: Option<File>,
661    tls_secrets: Option<Box<msquic::ffi::QUIC_TLS_SECRETS>>,
662}
663
664impl ConnectionInner {
665    fn new(
666        state: ConnectionState,
667        tls_secrets: Option<Box<msquic::ffi::QUIC_TLS_SECRETS>>,
668        sslkeylog_file: Option<File>,
669    ) -> Self {
670        Self {
671            exclusive: Mutex::new(ConnectionInnerExclusive {
672                state,
673                error: None,
674                start_waiters: Vec::new(),
675                inbound_stream_waiters: Vec::new(),
676                inbound_uni_stream_waiters: Vec::new(),
677                inbound_streams: VecDeque::new(),
678                inbound_uni_streams: VecDeque::new(),
679                recv_buffers: VecDeque::new(),
680                recv_waiters: Vec::new(),
681                write_pool: Vec::new(),
682                dgram_send_enabled: false,
683                dgram_max_send_length: 0,
684                shutdown_waiters: Vec::new(),
685                events: VecDeque::new(),
686                event_waiters: Vec::new(),
687                sslkeylog_file,
688                tls_secrets,
689            }),
690        }
691    }
692
693    fn handle_event_connected(
694        &self,
695        _session_resumed: bool,
696        _negotiated_alpn: &[u8],
697    ) -> Result<(), msquic::Status> {
698        trace!("ConnectionInner({:p}) Connected", self);
699
700        let mut exclusive = self.exclusive.lock().unwrap();
701        match (
702            exclusive.tls_secrets.take(),
703            exclusive.sslkeylog_file.take(),
704        ) {
705            (Some(tls_secrets), Some(mut file)) => {
706                info!("ConnectionInner({:p}) Writing TLS secrets to file", self);
707                let client_random = if tls_secrets.IsSet.ClientRandom() != 0 {
708                    hex::encode(tls_secrets.ClientRandom)
709                } else {
710                    String::new()
711                };
712
713                let _ = file.seek(SeekFrom::End(0));
714
715                if tls_secrets.IsSet.ClientEarlyTrafficSecret() != 0 {
716                    let _ = writeln!(
717                        file,
718                        "CLIENT_EARLY_TRAFFIC_SECRET {} {}",
719                        client_random,
720                        hex::encode(
721                            &tls_secrets.ClientEarlyTrafficSecret
722                                [0..tls_secrets.SecretLength as usize]
723                        )
724                    );
725                }
726
727                if tls_secrets.IsSet.ClientHandshakeTrafficSecret() != 0 {
728                    let _ = writeln!(
729                        file,
730                        "CLIENT_HANDSHAKE_TRAFFIC_SECRET {} {}",
731                        client_random,
732                        hex::encode(
733                            &tls_secrets.ClientHandshakeTrafficSecret
734                                [0..tls_secrets.SecretLength as usize]
735                        )
736                    );
737                }
738
739                if tls_secrets.IsSet.ServerHandshakeTrafficSecret() != 0 {
740                    let _ = writeln!(
741                        file,
742                        "SERVER_HANDSHAKE_TRAFFIC_SECRET {} {}",
743                        client_random,
744                        hex::encode(
745                            &tls_secrets.ServerHandshakeTrafficSecret
746                                [0..tls_secrets.SecretLength as usize]
747                        )
748                    );
749                }
750
751                if tls_secrets.IsSet.ClientTrafficSecret0() != 0 {
752                    let _ = writeln!(
753                        file,
754                        "CLIENT_TRAFFIC_SECRET_0 {} {}",
755                        client_random,
756                        hex::encode(
757                            &tls_secrets.ClientTrafficSecret0[0..tls_secrets.SecretLength as usize]
758                        )
759                    );
760                }
761
762                if tls_secrets.IsSet.ServerTrafficSecret0() != 0 {
763                    let _ = writeln!(
764                        file,
765                        "SERVER_TRAFFIC_SECRET_0 {} {}",
766                        client_random,
767                        hex::encode(
768                            &tls_secrets.ServerTrafficSecret0[0..tls_secrets.SecretLength as usize]
769                        )
770                    );
771                }
772                exclusive.tls_secrets = Some(tls_secrets);
773            }
774            _ => { /* do nothing */ }
775        }
776        exclusive.state = ConnectionState::Connected;
777        exclusive
778            .start_waiters
779            .drain(..)
780            .for_each(|waker| waker.wake());
781        Ok(())
782    }
783
784    fn handle_event_shutdown_initiated_by_transport(
785        &self,
786        status: msquic::Status,
787        error_code: u64,
788    ) -> Result<(), msquic::Status> {
789        trace!(
790            "ConnectionInner({:p}) Transport shutdown {:?}",
791            self,
792            status
793        );
794
795        let mut exclusive = self.exclusive.lock().unwrap();
796        exclusive.state = ConnectionState::Shutdown;
797        exclusive.error = Some(ConnectionError::ShutdownByTransport(status, error_code));
798        exclusive
799            .start_waiters
800            .drain(..)
801            .for_each(|waker| waker.wake());
802        exclusive
803            .inbound_stream_waiters
804            .drain(..)
805            .for_each(|waker| waker.wake());
806        exclusive
807            .recv_waiters
808            .drain(..)
809            .for_each(|waker| waker.wake());
810        Ok(())
811    }
812
813    fn handle_event_shutdown_initiated_by_peer(
814        &self,
815        error_code: u64,
816    ) -> Result<(), msquic::Status> {
817        trace!("ConnectionInner({:p}) App shutdown {}", self, error_code);
818
819        let mut exclusive = self.exclusive.lock().unwrap();
820        exclusive.state = ConnectionState::Shutdown;
821        exclusive.error = Some(ConnectionError::ShutdownByPeer(error_code));
822        exclusive
823            .start_waiters
824            .drain(..)
825            .for_each(|waker| waker.wake());
826        exclusive
827            .inbound_stream_waiters
828            .drain(..)
829            .for_each(|waker| waker.wake());
830        exclusive
831            .recv_waiters
832            .drain(..)
833            .for_each(|waker| waker.wake());
834        Ok(())
835    }
836
837    fn handle_event_shutdown_complete(
838        &self,
839        handshake_completed: bool,
840        peer_acknowledged_shutdown: bool,
841        app_close_in_progress: bool,
842    ) -> Result<(), msquic::Status> {
843        trace!("ConnectionInner({:p}) Shutdown complete: handshake_completed={}, peer_acknowledged_shutdown={}, app_close_in_progress={}",
844            self, handshake_completed, peer_acknowledged_shutdown, app_close_in_progress
845        );
846
847        {
848            let mut exclusive = self.exclusive.lock().unwrap();
849            exclusive.state = ConnectionState::ShutdownComplete;
850            if exclusive.error.is_none() {
851                exclusive.error = Some(ConnectionError::ShutdownByLocal);
852            }
853            exclusive
854                .start_waiters
855                .drain(..)
856                .for_each(|waker| waker.wake());
857            exclusive
858                .inbound_stream_waiters
859                .drain(..)
860                .for_each(|waker| waker.wake());
861            exclusive
862                .recv_waiters
863                .drain(..)
864                .for_each(|waker| waker.wake());
865            exclusive
866                .shutdown_waiters
867                .drain(..)
868                .for_each(|waker| waker.wake());
869            exclusive
870                .event_waiters
871                .drain(..)
872                .for_each(|waker| waker.wake());
873        }
874        Ok(())
875    }
876
877    fn handle_event_peer_stream_started(
878        &self,
879        stream: msquic::StreamRef,
880        flags: msquic::StreamOpenFlags,
881    ) -> Result<(), msquic::Status> {
882        let stream_type = if (flags & msquic::StreamOpenFlags::UNIDIRECTIONAL)
883            == msquic::StreamOpenFlags::UNIDIRECTIONAL
884        {
885            StreamType::Unidirectional
886        } else {
887            StreamType::Bidirectional
888        };
889        trace!(
890            "ConnectionInner({:p}) Peer stream started {:?}",
891            self,
892            stream_type
893        );
894
895        let stream = Stream::from_raw(unsafe { stream.as_raw() }, stream_type);
896        if (flags & msquic::StreamOpenFlags::UNIDIRECTIONAL)
897            == msquic::StreamOpenFlags::UNIDIRECTIONAL
898        {
899            if let (Some(read_stream), None) = stream.split() {
900                let mut exclusive = self.exclusive.lock().unwrap();
901                exclusive.inbound_uni_streams.push_back(read_stream);
902                exclusive
903                    .inbound_uni_stream_waiters
904                    .drain(..)
905                    .for_each(|waker| waker.wake());
906            } else {
907                unreachable!();
908            }
909        } else {
910            {
911                let mut exclusive = self.exclusive.lock().unwrap();
912                exclusive.inbound_streams.push_back(stream);
913                exclusive
914                    .inbound_stream_waiters
915                    .drain(..)
916                    .for_each(|waker| waker.wake());
917            }
918        }
919
920        Ok(())
921    }
922
923    fn handle_event_streams_available(
924        &self,
925        bidirectional_count: u16,
926        unidirectional_count: u16,
927    ) -> Result<(), msquic::Status> {
928        trace!(
929            "ConnectionInner({:p}) Streams available bidirectional_count:{} unidirectional_count:{}",
930            self,
931            bidirectional_count,
932            unidirectional_count
933        );
934        Ok(())
935    }
936
937    fn handle_event_datagram_state_changed(
938        &self,
939        send_enabled: bool,
940        max_send_length: u16,
941    ) -> Result<(), msquic::Status> {
942        trace!(
943            "ConnectionInner({:p}) Datagram state changed send_enabled:{} max_send_length:{}",
944            self,
945            send_enabled,
946            max_send_length
947        );
948        let mut exclusive = self.exclusive.lock().unwrap();
949        exclusive.dgram_send_enabled = send_enabled;
950        exclusive.dgram_max_send_length = max_send_length;
951        Ok(())
952    }
953
954    fn handle_event_datagram_received(
955        &self,
956        buffer: &msquic::BufferRef,
957        _flags: msquic::ReceiveFlags,
958    ) -> Result<(), msquic::Status> {
959        trace!("ConnectionInner({:p}) Datagram received", self);
960        let buf = Bytes::copy_from_slice(buffer.as_bytes());
961        {
962            let mut exclusive = self.exclusive.lock().unwrap();
963            exclusive.recv_buffers.push_back(buf);
964            exclusive
965                .recv_waiters
966                .drain(..)
967                .for_each(|waker| waker.wake());
968        }
969        Ok(())
970    }
971
972    fn handle_event_datagram_send_state_changed(
973        &self,
974        client_context: *const c_void,
975        state: msquic::DatagramSendState,
976    ) -> Result<(), msquic::Status> {
977        trace!(
978            "ConnectionInner({:p}) Datagram send state changed state:{:?}",
979            self,
980            state
981        );
982        match state {
983            msquic::DatagramSendState::Sent | msquic::DatagramSendState::Canceled => {
984                let mut write_buf = unsafe { WriteBuffer::from_raw(client_context) };
985                let mut exclusive = self.exclusive.lock().unwrap();
986                write_buf.reset();
987                exclusive.write_pool.push(write_buf);
988            }
989            _ => {}
990        }
991        Ok(())
992    }
993
994    #[cfg(feature = "msquic-seera")]
995    fn handle_event_notify_observed_address(
996        &self,
997        local_address: &msquic::Addr,
998        observed_address: &msquic::Addr,
999    ) -> Result<(), msquic::Status> {
1000        let local_address = local_address.as_socket().expect("socket addr");
1001        let observed_address = observed_address.as_socket().expect("socket addr");
1002        trace!(
1003            "ConnectionInner({:p}) Notify observed address local_address:{} observed_address:{}",
1004            self,
1005            local_address,
1006            observed_address
1007        );
1008        let mut exclusive = self.exclusive.lock().unwrap();
1009        exclusive
1010            .events
1011            .push_back(ConnectionEvent::NotifyObservedAddress {
1012                local_address,
1013                observed_address,
1014            });
1015        exclusive
1016            .event_waiters
1017            .drain(..)
1018            .for_each(|waker| waker.wake());
1019        Ok(())
1020    }
1021
1022    #[cfg(feature = "msquic-seera")]
1023    fn handle_event_notify_remote_address_added(
1024        &self,
1025        address: &msquic::Addr,
1026        sequence_number: u64,
1027    ) -> Result<(), msquic::Status> {
1028        let address = address.as_socket().expect("socket addr");
1029        trace!(
1030            "ConnectionInner({:p}) Notify remote address added address:{} sequence_number:{}",
1031            self,
1032            address,
1033            sequence_number
1034        );
1035        let mut exclusive = self.exclusive.lock().unwrap();
1036        exclusive
1037            .events
1038            .push_back(ConnectionEvent::NotifyRemoteAddressAdded {
1039                address,
1040                sequence_number,
1041            });
1042        exclusive
1043            .event_waiters
1044            .drain(..)
1045            .for_each(|waker| waker.wake());
1046        Ok(())
1047    }
1048
1049    #[cfg(feature = "msquic-seera")]
1050    fn handle_event_path_validated(
1051        &self,
1052        local_address: &msquic::Addr,
1053        remote_address: &msquic::Addr,
1054    ) -> Result<(), msquic::Status> {
1055        let local_address = local_address.as_socket().expect("socket addr");
1056        let remote_address = remote_address.as_socket().expect("socket addr");
1057        trace!(
1058            "ConnectionInner({:p}) path validated local_address:{} remote_address:{}",
1059            self,
1060            local_address,
1061            remote_address
1062        );
1063        let mut exclusive = self.exclusive.lock().unwrap();
1064        exclusive.events.push_back(ConnectionEvent::PathValidated {
1065            local_address,
1066            remote_address,
1067        });
1068        exclusive
1069            .event_waiters
1070            .drain(..)
1071            .for_each(|waker| waker.wake());
1072        Ok(())
1073    }
1074
1075    #[cfg(feature = "msquic-seera")]
1076    fn handle_event_notify_remote_address_removed(
1077        &self,
1078        sequence_number: u64,
1079    ) -> Result<(), msquic::Status> {
1080        trace!(
1081            "ConnectionInner({:p}) Notify remote address removed sequence_number:{}",
1082            self,
1083            sequence_number
1084        );
1085        let mut exclusive = self.exclusive.lock().unwrap();
1086        exclusive
1087            .events
1088            .push_back(ConnectionEvent::NotifyRemoteAddressRemoved { sequence_number });
1089        exclusive
1090            .event_waiters
1091            .drain(..)
1092            .for_each(|waker| waker.wake());
1093        Ok(())
1094    }
1095
1096    fn callback_handler_impl(
1097        &self,
1098        _connection: msquic::ConnectionRef,
1099        ev: msquic::ConnectionEvent,
1100    ) -> Result<(), msquic::Status> {
1101        match ev {
1102            msquic::ConnectionEvent::Connected {
1103                session_resumed,
1104                negotiated_alpn,
1105            } => self.handle_event_connected(session_resumed, negotiated_alpn),
1106            msquic::ConnectionEvent::ShutdownInitiatedByTransport { status, error_code } => {
1107                self.handle_event_shutdown_initiated_by_transport(status, error_code)
1108            }
1109            msquic::ConnectionEvent::ShutdownInitiatedByPeer { error_code } => {
1110                self.handle_event_shutdown_initiated_by_peer(error_code)
1111            }
1112            msquic::ConnectionEvent::ShutdownComplete {
1113                handshake_completed,
1114                peer_acknowledged_shutdown,
1115                app_close_in_progress,
1116            } => self.handle_event_shutdown_complete(
1117                handshake_completed,
1118                peer_acknowledged_shutdown,
1119                app_close_in_progress,
1120            ),
1121            msquic::ConnectionEvent::PeerStreamStarted { stream, flags } => {
1122                self.handle_event_peer_stream_started(stream, flags)
1123            }
1124            msquic::ConnectionEvent::StreamsAvailable {
1125                bidirectional_count,
1126                unidirectional_count,
1127            } => self.handle_event_streams_available(bidirectional_count, unidirectional_count),
1128            msquic::ConnectionEvent::DatagramStateChanged {
1129                send_enabled,
1130                max_send_length,
1131            } => self.handle_event_datagram_state_changed(send_enabled, max_send_length),
1132            msquic::ConnectionEvent::DatagramReceived { buffer, flags } => {
1133                self.handle_event_datagram_received(buffer, flags)
1134            }
1135            msquic::ConnectionEvent::DatagramSendStateChanged {
1136                client_context,
1137                state,
1138            } => self.handle_event_datagram_send_state_changed(client_context, state),
1139            #[cfg(feature = "msquic-seera")]
1140            msquic::ConnectionEvent::NotifyObservedAddress {
1141                local_address,
1142                observed_address,
1143            } => self.handle_event_notify_observed_address(local_address, observed_address),
1144            #[cfg(feature = "msquic-seera")]
1145            msquic::ConnectionEvent::NotifyRemoteAddressAdded {
1146                address,
1147                sequence_number,
1148            } => self.handle_event_notify_remote_address_added(address, sequence_number),
1149            #[cfg(feature = "msquic-seera")]
1150            msquic::ConnectionEvent::PathValidated {
1151                local_address,
1152                remote_address,
1153            } => self.handle_event_path_validated(local_address, remote_address),
1154            #[cfg(feature = "msquic-seera")]
1155            msquic::ConnectionEvent::NotifyRemoteAddressRemoved { sequence_number } => {
1156                self.handle_event_notify_remote_address_removed(sequence_number)
1157            }
1158            _ => {
1159                trace!("ConnectionInner({:p}) Other callback", self);
1160                Ok(())
1161            }
1162        }
1163    }
1164}
1165impl Drop for ConnectionInner {
1166    fn drop(&mut self) {
1167        trace!("ConnectionInner({:p}) dropping", self);
1168    }
1169}
1170
1171#[derive(Debug, PartialEq)]
1172enum ConnectionState {
1173    Open,
1174    Connecting,
1175    Connected,
1176    Shutdown,
1177    ShutdownComplete,
1178}
1179
1180/// Events that can occur on a connection.
1181#[derive(Clone, Debug, PartialEq, Eq)]
1182pub enum ConnectionEvent {
1183    /// A new observed address has been detected.
1184    NotifyObservedAddress {
1185        local_address: SocketAddr,
1186        observed_address: SocketAddr,
1187    },
1188    /// A new remote address has been added.
1189    NotifyRemoteAddressAdded {
1190        address: SocketAddr,
1191        sequence_number: u64,
1192    },
1193    /// A path has been validated.
1194    PathValidated {
1195        local_address: SocketAddr,
1196        remote_address: SocketAddr,
1197    },
1198    /// A remote address has been removed.
1199    NotifyRemoteAddressRemoved { sequence_number: u64 },
1200}
1201
1202/// Errors that can occur when managing a connection.
1203#[derive(Debug, Error, Clone)]
1204pub enum ConnectionError {
1205    #[error("connection shutdown by transport: status {0:?}, error 0x{1:x}")]
1206    ShutdownByTransport(msquic::Status, u64),
1207    #[error("connection shutdown by peer: error 0x{0:x}")]
1208    ShutdownByPeer(u64),
1209    #[error("connection shutdown by local")]
1210    ShutdownByLocal,
1211    #[error("connection closed")]
1212    ConnectionClosed,
1213    #[error("SSL key log file already set")]
1214    SslKeyLogFileAlreadySet,
1215    #[error("other error: status {0:?}")]
1216    OtherError(msquic::Status),
1217}
1218
1219/// Errors that can occur when receiving a datagram.
1220#[derive(Debug, Error, Clone)]
1221pub enum DgramReceiveError {
1222    #[error("connection not started yet")]
1223    ConnectionNotStarted,
1224    #[error("connection lost")]
1225    ConnectionLost(#[from] ConnectionError),
1226    #[error("other error: status {0:?}")]
1227    OtherError(msquic::Status),
1228}
1229
1230/// Errors that can occur when sending a datagram.
1231#[derive(Debug, Error, Clone)]
1232pub enum DgramSendError {
1233    #[error("connection not started yet")]
1234    ConnectionNotStarted,
1235    #[error("not allowed for sending dgram")]
1236    Denied,
1237    #[error("exceeded maximum data size for sending dgram")]
1238    TooBig,
1239    #[error("connection lost")]
1240    ConnectionLost(#[from] ConnectionError),
1241    #[error("other error: status {0:?}")]
1242    OtherError(msquic::Status),
1243}
1244
1245/// Errors that can occur when starting a connection.
1246#[derive(Debug, Error, Clone)]
1247pub enum StartError {
1248    #[error("connection lost")]
1249    ConnectionLost(#[from] ConnectionError),
1250    #[error("other error: status {0:?}")]
1251    OtherError(msquic::Status),
1252}
1253
1254/// Errors that can occur when shutdowning a connection.
1255#[derive(Debug, Error, Clone)]
1256pub enum ShutdownError {
1257    #[error("connection not started yet")]
1258    ConnectionNotStarted,
1259    #[error("connection lost")]
1260    ConnectionLost(#[from] ConnectionError),
1261    #[error("other error: status {0:?}")]
1262    OtherError(msquic::Status),
1263}
1264
1265/// Errors that can occur when receiving events on a connection.
1266#[derive(Debug, Error, Clone)]
1267pub enum EventError {
1268    #[error("connection not started yet")]
1269    ConnectionNotStarted,
1270    #[error("connection lost")]
1271    ConnectionLost(#[from] ConnectionError),
1272    #[error("other error: status {0:?}")]
1273    OtherError(msquic::Status),
1274}
1275
1276/// Future produced by [`Connection::start()`].
1277pub struct ConnectionStart<'a> {
1278    conn: &'a Connection,
1279    configuration: &'a msquic::Configuration,
1280    host: &'a str,
1281    port: u16,
1282}
1283
1284impl Future for ConnectionStart<'_> {
1285    type Output = Result<(), StartError>;
1286
1287    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
1288        self.conn
1289            .poll_start(cx, self.configuration, self.host, self.port)
1290    }
1291}
1292
1293/// Future produced by [`Connection::open_outbound_stream()`].
1294pub struct OpenOutboundStream<'a> {
1295    conn: &'a ConnectionInstance,
1296    stream_type: Option<crate::stream::StreamType>,
1297    stream: Option<crate::stream::Stream>,
1298    fail_on_blocked: bool,
1299}
1300
1301impl Future for OpenOutboundStream<'_> {
1302    type Output = Result<crate::stream::Stream, StreamStartError>;
1303
1304    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
1305        let this = self.get_mut();
1306        let OpenOutboundStream {
1307            conn,
1308            ref mut stream_type,
1309            ref mut stream,
1310            fail_on_blocked: fail_blocked,
1311            ..
1312        } = *this;
1313
1314        let mut exclusive = conn.inner.exclusive.lock().unwrap();
1315        match exclusive.state {
1316            ConnectionState::Open => {
1317                return Poll::Ready(Err(StreamStartError::ConnectionNotStarted));
1318            }
1319            ConnectionState::Connecting => {
1320                exclusive.start_waiters.push(cx.waker().clone());
1321                return Poll::Pending;
1322            }
1323            ConnectionState::Connected => {}
1324            ConnectionState::Shutdown | ConnectionState::ShutdownComplete => {
1325                return Poll::Ready(Err(StreamStartError::ConnectionLost(
1326                    exclusive.error.as_ref().expect("error").clone(),
1327                )));
1328            }
1329        }
1330        if stream.is_none() {
1331            match Stream::open(&conn.msquic_conn, stream_type.take().unwrap()) {
1332                Ok(new_stream) => {
1333                    *stream = Some(new_stream);
1334                }
1335                Err(e) => return Poll::Ready(Err(e)),
1336            }
1337        }
1338        stream
1339            .as_mut()
1340            .unwrap()
1341            .poll_start(cx, fail_blocked)
1342            .map(|res| res.map(|_| stream.take().unwrap()))
1343    }
1344}
1345
1346/// Future produced by [`Connection::accept_inbound_stream()`].
1347pub struct AcceptInboundStream<'a> {
1348    conn: &'a Connection,
1349}
1350
1351impl Future for AcceptInboundStream<'_> {
1352    type Output = Result<Stream, StreamStartError>;
1353
1354    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
1355        self.conn.poll_accept_inbound_stream(cx)
1356    }
1357}
1358
1359/// Future produced by [`Connection::accept_inbound_uni_stream()`].
1360pub struct AcceptInboundUniStream<'a> {
1361    conn: &'a Connection,
1362}
1363
1364impl Future for AcceptInboundUniStream<'_> {
1365    type Output = Result<ReadStream, StreamStartError>;
1366
1367    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
1368        self.conn.poll_accept_inbound_uni_stream(cx)
1369    }
1370}
msquic_async/connection.rs

msquic_async/
connection.rs
