s2n_quic_dc/stream/
server.rs

1// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4#![allow(clippy::type_complexity)]
5
6use crate::{
7    credentials::{self, Credentials},
8    msg::recv,
9    packet,
10    stream::socket,
11};
12use s2n_codec::{DecoderBufferMut, DecoderError};
13use s2n_quic_core::varint::VarInt;
14use std::{io, net::SocketAddr};
15use tracing::trace;
16
17pub mod accept;
18pub mod application;
19pub mod handshake;
20pub mod manager;
21pub mod stats;
22pub mod tokio;
23pub mod udp;
24
25#[derive(Clone, Copy, Debug)]
26pub struct InitialPacket {
27    pub credentials: Credentials,
28    pub stream_id: packet::stream::Id,
29    pub source_queue_id: Option<VarInt>,
30    pub payload_len: usize,
31    pub is_zero_offset: bool,
32    pub is_retransmission: bool,
33    pub is_fin: bool,
34    pub is_fin_known: bool,
35}
36
37impl InitialPacket {
38    #[inline]
39    pub fn peek(recv: &mut recv::Message, tag_len: usize) -> Result<Self, DecoderError> {
40        let segment = recv
41            .peek_segments()
42            .next()
43            .ok_or(DecoderError::UnexpectedEof(1))?;
44
45        let decoder = DecoderBufferMut::new(segment);
46        // we're just going to assume that all of the packets in this datagram
47        // pertain to the same stream
48        let (packet, _remaining) = decoder.decode_parameterized(tag_len)?;
49
50        let packet::Packet::Stream(packet) = packet else {
51            return Err(DecoderError::InvariantViolation("unexpected packet type"));
52        };
53
54        let packet: InitialPacket = packet.into();
55
56        Ok(packet)
57    }
58
59    #[inline]
60    #[expect(
61        clippy::unwrap_used,
62        reason = "VarInt::ZERO is provably in range for unreliable_unidirectional"
63    )]
64    pub fn empty() -> Self {
65        Self {
66            credentials: Credentials {
67                id: credentials::Id::default(),
68                key_id: VarInt::ZERO,
69            },
70            stream_id: packet::stream::Id::unreliable_unidirectional(VarInt::ZERO).unwrap(),
71            source_queue_id: None,
72            payload_len: 0,
73            is_zero_offset: false,
74            is_retransmission: false,
75            is_fin: false,
76            is_fin_known: false,
77        }
78    }
79}
80
81impl<'a> From<packet::stream::decoder::Packet<'a>> for InitialPacket {
82    #[inline]
83    fn from(packet: packet::stream::decoder::Packet<'a>) -> Self {
84        let credentials = *packet.credentials();
85        let stream_id = *packet.stream_id();
86        let source_queue_id = packet.source_queue_id();
87        let payload_len = packet.payload().len();
88        let is_zero_offset = packet.stream_offset().as_u64() == 0;
89        let is_retransmission = packet.is_retransmission();
90        let is_fin = packet.is_fin();
91        let is_fin_known = packet.final_offset().is_some();
92        Self {
93            credentials,
94            stream_id,
95            source_queue_id,
96            is_zero_offset,
97            payload_len,
98            is_retransmission,
99            is_fin,
100            is_fin_known,
101        }
102    }
103}
104
105pub(crate) fn spawn_initial_wildcard_pair(
106    local_addr: SocketAddr,
107    socket_opts: impl Fn(SocketAddr) -> socket::Options,
108) -> io::Result<(SocketAddr, std::net::UdpSocket, std::net::TcpListener)> {
109    debug_assert_eq!(local_addr.port(), 0);
110
111    let start = std::time::Instant::now();
112    let timeout = std::time::Duration::from_secs(5);
113
114    for iteration in 0..10_000 {
115        if start.elapsed() >= timeout {
116            return Err(io::Error::new(
117                io::ErrorKind::TimedOut,
118                "could not find free port after 5 seconds",
119            ));
120        }
121
122        trace!(wildcard_search_iteration = iteration);
123        let udp_socket = socket_opts(local_addr).build_udp()?;
124        let candidate_addr = udp_socket.local_addr()?;
125        trace!(candidate = %candidate_addr);
126        match socket_opts(candidate_addr).build_tcp_listener() {
127            Ok(tcp_socket) => {
128                trace!(selected = %candidate_addr);
129                return Ok((candidate_addr, udp_socket, tcp_socket));
130            }
131            Err(err) if err.kind() == io::ErrorKind::AddrInUse => continue,
132            Err(err) => return Err(err),
133        }
134    }
135
136    Err(io::Error::new(
137        io::ErrorKind::AddrInUse,
138        "could not find free port after 10,000 attempts",
139    ))
140}
s2n_quic_dc/stream/server.rs

s2n_quic_dc/stream/
server.rs
