Skip to main content

srt/
builder.rs

1use std::net::{IpAddr, SocketAddr, ToSocketAddrs};
2use std::{io, time::Duration};
3
4use tokio::net::UdpSocket;
5use tokio_util::udp::UdpFramed;
6
7use futures::{future::ready, Sink, Stream, StreamExt};
8
9use crate::tokio::create_bidrectional_srt;
10use crate::{
11    multiplex, pending_connection, Connection, PackChan, Packet, PacketCodec, PacketParseError,
12    SrtSocket,
13};
14use log::warn;
15
16/// Struct to build sockets.
17///
18/// This is the typical way to create instances of [`SrtSocket`], which implements both `Sink + Stream`, as they can be both receivers and senders.
19///
20/// You need to decided on a [`ConnInitMethod`] in order to create a [`SrtSocketBuilder`]. See [that documentation](ConnInitMethod) for more details.
21///
22/// # Examples:
23/// Simple:
24/// ```
25/// # use srt::SrtSocketBuilder;
26/// # use std::io;
27/// # #[tokio::main]
28/// # async fn main() -> Result<(), io::Error> {
29/// let (a, b) = futures::try_join!(
30///     SrtSocketBuilder::new_listen().local_port(3333).connect(),
31///     SrtSocketBuilder::new_connect("127.0.0.1:3333").connect(),
32/// )?;
33/// # Ok(())
34/// # }
35/// ```
36///
37/// Rendezvous example:
38///
39/// ```
40/// # use srt::{SrtSocketBuilder, ConnInitMethod};
41/// # use std::io;
42/// # #[tokio::main]
43/// # async fn main() -> Result<(), io::Error> {
44/// let (a, b) = futures::try_join!(
45///     SrtSocketBuilder::new_rendezvous("127.0.0.1:4444").local_port(5555).connect(),
46///     SrtSocketBuilder::new_rendezvous("127.0.0.1:5555").local_port(4444).connect(),
47/// )?;
48/// # Ok(())
49/// # }
50/// ```
51///
52/// # Panics:
53/// * There is no tokio runtime
54#[derive(Debug, Clone)]
55#[must_use]
56pub struct SrtSocketBuilder {
57    local_addr: SocketAddr,
58    conn_type: ConnInitMethod,
59    latency: Duration,
60    crypto: Option<(u8, String)>,
61}
62
63/// Describes how this SRT entity will connect to the other.
64#[derive(Debug, Clone, Copy, PartialEq, Eq)]
65pub enum ConnInitMethod {
66    /// Listens on the local socket, expecting there to be a [`Connect`](ConnInitMethod::Connect) instance that eventually connects to this socket.
67    /// This almost certianly menas you should use it with [`SrtSocketBuilder::local_port`],
68    /// As otherwise there is no way to know which port it will bind to.
69    Listen,
70
71    /// Connect to a listening socket. It expects the listen socket to be on the [`SocketAddr`] provided.
72    Connect(SocketAddr),
73
74    /// Connect to another [`Rendezvous`](ConnInitMethod::Rendezvous) connection. This is useful if both sides are behind a NAT. The [`SocketAddr`]
75    /// passed should be the **public** address and port of the other [`Rendezvous`](ConnInitMethod::Rendezvous) connection.
76    Rendezvous(SocketAddr),
77}
78
79impl SrtSocketBuilder {
80    /// Defaults to binding to `0.0.0.0:0` (all adaptors, OS assigned port), 50ms latency, and no encryption.
81    /// Generally easier to use [`new_listen`](SrtSocketBuilder::new_listen), [`new_connect`](SrtSocketBuilder::new_connect) or [`new_rendezvous`](SrtSocketBuilder::new_rendezvous)
82    pub fn new(conn_type: ConnInitMethod) -> Self {
83        SrtSocketBuilder {
84            local_addr: "0.0.0.0:0".parse().unwrap(),
85            conn_type,
86            latency: Duration::from_millis(50),
87            crypto: None,
88        }
89    }
90
91    pub fn new_listen() -> Self {
92        Self::new(ConnInitMethod::Listen)
93    }
94
95    /// Connects to the first address yielded by `to`
96    ///
97    /// # Panics
98    /// * `to` fails to resolve to a [`SocketAddr`]
99    pub fn new_connect(to: impl ToSocketAddrs) -> Self {
100        Self::new(ConnInitMethod::Connect(
101            to.to_socket_addrs().unwrap().next().unwrap(),
102        ))
103    }
104
105    /// Connects to the first address yielded by `to`
106    ///
107    /// # Panics
108    /// * `to` fails to resolve to a [`SocketAddr`]
109    pub fn new_rendezvous(to: impl ToSocketAddrs) -> Self {
110        Self::new(ConnInitMethod::Rendezvous(
111            to.to_socket_addrs().unwrap().next().unwrap(),
112        ))
113    }
114
115    /// Gets the [`ConnInitMethod`] of the builder.
116    ///
117    /// ```
118    /// # use srt::{SrtSocketBuilder, ConnInitMethod};
119    /// let builder = SrtSocketBuilder::new(ConnInitMethod::Listen);
120    /// assert_eq!(builder.conn_type(), &ConnInitMethod::Listen);
121    /// ```
122    #[must_use]
123    pub fn conn_type(&self) -> &ConnInitMethod {
124        &self.conn_type
125    }
126
127    /// Sets the local address of the socket. This can be used to bind to just a specific network adapter instead of the default of all adapters.
128    pub fn local_addr(mut self, local_addr: IpAddr) -> Self {
129        self.local_addr.set_ip(local_addr);
130
131        self
132    }
133
134    /// Sets the port to bind to. In general, to be used for [`ConnInitMethod::Listen`] and [`ConnInitMethod::Rendezvous`], but generally not [`ConnInitMethod::Connect`].
135    pub fn local_port(mut self, port: u16) -> Self {
136        self.local_addr.set_port(port);
137
138        self
139    }
140
141    /// Set the latency of the connection. The more latency, the more time SRT has to recover lost packets.
142    pub fn latency(mut self, latency: Duration) -> Self {
143        self.latency = latency;
144
145        self
146    }
147
148    /// Se the crypto paramters. However, this is currently unimplemented.
149    ///
150    /// # Panics:
151    /// * size is not 16, 24, or 32.
152    pub fn crypto(mut self, size: u8, passphrase: String) -> Self {
153        match size {
154            // OK
155            16 | 24 | 32 => {}
156            // NOT
157            // TODO: size validation
158            size => panic!("Invaid crypto size {}", size),
159        }
160        self.crypto = Some((size, passphrase));
161
162        self
163    }
164
165    /// Connect with a custom socket. Not typically used, see [`connect`](SrtSocketBuilder::connect) instead.
166    pub async fn connect_with_sock<T>(self, mut socket: T) -> Result<SrtSocket, io::Error>
167    where
168        T: Stream<Item = Result<(Packet, SocketAddr), PacketParseError>>
169            + Sink<(Packet, SocketAddr), Error = io::Error>
170            + Unpin
171            + Send
172            + 'static,
173    {
174        let conn = match self.conn_type {
175            ConnInitMethod::Listen => {
176                pending_connection::listen(&mut socket, rand::random(), self.latency).await?
177            }
178            ConnInitMethod::Connect(addr) => {
179                pending_connection::connect(
180                    &mut socket,
181                    addr,
182                    rand::random(),
183                    self.local_addr.ip(),
184                    self.latency,
185                    self.crypto.clone(),
186                )
187                .await?
188            }
189            ConnInitMethod::Rendezvous(remote_public) => {
190                pending_connection::rendezvous(
191                    &mut socket,
192                    rand::random(),
193                    self.local_addr.ip(),
194                    remote_public,
195                    self.latency,
196                )
197                .await?
198            }
199        };
200
201        Ok(create_bidrectional_srt(
202            socket.filter_map(|res| {
203                ready(res.map_err(|e| warn!("Error parsing packet: {}", e)).ok())
204            }),
205            conn,
206        ))
207    }
208
209    /// Connects to the remote socket. Resolves when it has been connected successfully.
210    pub async fn connect(self) -> Result<SrtSocket, io::Error> {
211        let la = self.local_addr;
212        Ok(self
213            .connect_with_sock(UdpFramed::new(UdpSocket::bind(&la).await?, PacketCodec {}))
214            .await?)
215    }
216
217    /// Build a multiplexed connection. This acts as a sort of server, allowing many connections to this one socket.
218    ///
219    /// # Panics:
220    /// If this is built with a non-listen builder
221    pub async fn build_multiplexed(
222        self,
223    ) -> Result<impl Stream<Item = Result<(Connection, PackChan), io::Error>>, io::Error> {
224        match self.conn_type {
225            ConnInitMethod::Listen => multiplex(self.local_addr, self.latency).await,
226            _ => panic!("Cannot bind multiplexed with any connection mode other than listen"),
227        }
228    }
229}