rice_proto/
component.rs

1// Copyright (C) 2024 Matthew Waters <matthew@centricular.com>
2//
3// Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
4// http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
5// <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
6// option. This file may not be copied, modified, or distributed
7// except according to those terms.
8//
9// SPDX-License-Identifier: MIT OR Apache-2.0
10
11//! A [`Component`] in an ICE [`Stream`]
12
13use alloc::boxed::Box;
14use core::net::SocketAddr;
15
16use stun_proto::Instant;
17use stun_proto::agent::Transmit;
18use stun_proto::types::message::{BINDING, Message, MessageWriteVec};
19use stun_proto::types::prelude::MessageWrite;
20use turn_client_proto::api::TurnClientApi;
21use turn_client_proto::types::prelude::DelayedTransmitBuild;
22
23use crate::candidate::{CandidatePair, CandidateType, TransportType};
24
25use crate::agent::{Agent, AgentError};
26pub use crate::conncheck::SelectedPair;
27use crate::conncheck::{RequestRto, transmit_send};
28use crate::gathering::StunGatherer;
29use crate::stream::Stream;
30use crate::turn::TurnConfig;
31
32use tracing::{debug, trace};
33
34/// The component id for RTP streaming (and general data).
35pub const RTP: usize = 1;
36/// The component id for RTCP streaming (if rtcp-mux is not in use).
37pub const RTCP: usize = 2;
38
39/// A [`Component`] in an ICE [`Stream`]
40#[derive(Debug)]
41#[repr(C)]
42pub struct Component<'a> {
43    agent: &'a Agent,
44    stream_id: usize,
45    component_id: usize,
46}
47
48impl<'a> Component<'a> {
49    pub(crate) fn from_stream(agent: &'a Agent, stream_id: usize, component_id: usize) -> Self {
50        Self {
51            agent,
52            stream_id,
53            component_id,
54        }
55    }
56
57    /// The component identifier within a particular ICE [`Stream`]
58    pub fn id(&self) -> usize {
59        self.component_id
60    }
61
62    /// Retrieve the [`Stream`] for this component.
63    ///
64    /// # Examples
65    ///
66    /// ```
67    /// # use rice_proto::component::Component;
68    /// # use rice_proto::agent::Agent;
69    /// # use rice_proto::stream::Stream;
70    /// let mut agent = Agent::default();
71    /// let stream_id = agent.add_stream();
72    /// let mut stream = agent.mut_stream(stream_id).unwrap();
73    /// let component_id = stream.add_component().unwrap();
74    /// let component = stream.component(component_id).unwrap();
75    /// assert_eq!(component.stream().id(), stream_id);
76    /// ```
77    pub fn stream(&self) -> Stream<'a> {
78        self.agent.stream(self.stream_id).unwrap()
79    }
80
81    /// Retrieve the current state of a `Component`
82    ///
83    /// # Examples
84    ///
85    /// The initial state is `ComponentState::New`
86    ///
87    /// ```
88    /// # use rice_proto::component::{Component, ComponentConnectionState};
89    /// # use rice_proto::agent::Agent;
90    /// # use rice_proto::stream::Stream;
91    /// let mut agent = Agent::default();
92    /// let stream_id = agent.add_stream();
93    /// let mut stream = agent.mut_stream(stream_id).unwrap();
94    /// let component_id = stream.add_component().unwrap();
95    /// let component = stream.component(component_id).unwrap();
96    /// assert_eq!(component.state(), ComponentConnectionState::New);
97    /// ```
98    pub fn state(&self) -> ComponentConnectionState {
99        let stream = self.agent.stream_state(self.stream_id).unwrap();
100        let component = stream.component_state(self.component_id).unwrap();
101        component.state
102    }
103
104    /// The [`CandidatePair`] this component has selected to send/receive data with.  This will not
105    /// be valid until the [`Component`] has reached [`ComponentConnectionState::Connected`]
106    pub fn selected_pair(&self) -> Option<&CandidatePair> {
107        let stream = self.agent.stream_state(self.stream_id).unwrap();
108        let component = stream.component_state(self.component_id).unwrap();
109        component
110            .selected_pair
111            .as_ref()
112            .map(|pair| pair.candidate_pair())
113    }
114}
115
116/// A mutable component in an ICE [`Stream`]
117#[derive(Debug)]
118#[repr(C)]
119pub struct ComponentMut<'a> {
120    agent: &'a mut Agent,
121    stream_id: usize,
122    component_id: usize,
123}
124
125impl<'a> core::ops::Deref for ComponentMut<'a> {
126    type Target = Component<'a>;
127
128    fn deref(&self) -> &Self::Target {
129        unsafe { core::mem::transmute(self) }
130    }
131}
132
133impl<'a> ComponentMut<'a> {
134    pub(crate) fn from_stream(agent: &'a mut Agent, stream_id: usize, component_id: usize) -> Self {
135        Self {
136            agent,
137            stream_id,
138            component_id,
139        }
140    }
141
142    /// Start gathering candidates for this component.  The parent
143    /// [`Agent::poll`](crate::agent::Agent::poll) is used to progress
144    /// the gathering.
145    ///
146    /// Candidates will be generated as follows (if they succeed):
147    ///
148    /// 1. A host candidate for each `sockets[i]`. If TCP, then both an active and passive host
149    ///    candidate will be generated.
150    /// 2. For each `sockets[i]` a reflexive candidate for each `stun_server[i]` if
151    ///    different from any other candidate produced. The local address for each STUN server
152    ///    connection will be one of the entries provided in `sockets`.
153    /// 3. For each `turn_servers[i]` a TURN allocation will be attempted and a relayed candidate
154    ///    produced on success.  If you would like multiple options for relayed candidates,
155    ///    e.g. UDP, TCP, TCP/TLS, then provide each option as different entries in the provided
156    ///    slice. The `SocketAddr` for each TURN server is the local address to communicate with
157    ///    the TURN server and should be different than any value provided through `sockets`.
158    pub fn gather_candidates(
159        &mut self,
160        sockets: &[(TransportType, SocketAddr)],
161        stun_servers: &[(TransportType, SocketAddr)],
162        turn_servers: &[(SocketAddr, &TurnConfig)],
163    ) -> Result<(), AgentError> {
164        let rto = self.agent.rto.clone();
165        let stream = self.agent.mut_stream_state(self.stream_id).unwrap();
166        let component = stream.mut_component_state(self.component_id).unwrap();
167        component.gather_candidates(sockets, stun_servers, turn_servers, rto)
168    }
169
170    /// Set the pair that will be used to send/receive data.  This will override the ICE
171    /// negotiation chosen value.
172    pub fn set_selected_pair(&mut self, selected: CandidatePair) -> Result<(), AgentError> {
173        // TODO: handle TURN
174        let stream = self.agent.mut_stream_state(self.stream_id).unwrap();
175        let checklist_id = stream.checklist_id;
176        let checklist = self
177            .agent
178            .checklistset
179            .mut_list(checklist_id)
180            .ok_or(AgentError::ResourceNotFound)?;
181        let (agent_id, agent) = if let Some((agent_id, agent)) = checklist.mut_agent_for_5tuple(
182            selected.local.transport_type,
183            selected.local.base_address,
184            selected.remote.address,
185        ) {
186            (agent_id, agent)
187        } else {
188            let agent_id = checklist
189                .add_agent_for_5tuple(
190                    selected.local.transport_type,
191                    selected.local.base_address,
192                    selected.remote.address,
193                    None,
194                )
195                .0;
196            let agent = checklist.mut_agent_by_id(agent_id).unwrap();
197            (agent_id, agent)
198        };
199        if !agent.is_validated_peer(selected.remote.address) {
200            // ensure that we can receive from the provided remote address.
201            let transmit = agent
202                .send_request(
203                    Message::builder_request(BINDING, MessageWriteVec::new()).finish(),
204                    selected.remote.address,
205                    Instant::ZERO,
206                )
207                .unwrap();
208            let msg = Message::from_bytes(&transmit.data).unwrap();
209            let response = Message::builder_success(&msg, MessageWriteVec::new()).finish();
210            let response = Message::from_bytes(&response).unwrap();
211            agent.handle_stun_message(&response, selected.remote.address);
212        }
213
214        let selected_pair = SelectedPair::new(selected, agent_id, None);
215        self.set_selected_pair_with_agent(selected_pair);
216        Ok(())
217    }
218
219    pub(crate) fn set_selected_pair_with_agent(&mut self, selected: SelectedPair) {
220        let stream = self.agent.mut_stream_state(self.stream_id).unwrap();
221        let component = stream.mut_component_state(self.component_id).unwrap();
222        component.selected_pair = Some(selected);
223    }
224
225    /// Send data to the peer using the selected pair.  This will not succeed until the
226    /// [`Component`] has reached [`ComponentConnectionState::Connected`]
227    pub fn send<T: AsRef<[u8]> + core::fmt::Debug>(
228        &mut self,
229        data: T,
230        now: Instant,
231    ) -> Result<Transmit<Box<[u8]>>, AgentError> {
232        // TODO: store statistics about bytes/packets sent
233        let stream = self.agent.stream_state(self.stream_id).unwrap();
234        let checklist_id = stream.checklist_id;
235        let component = stream.component_state(self.component_id).unwrap();
236        let selected_pair = component
237            .selected_pair
238            .as_ref()
239            .ok_or(AgentError::ResourceNotFound)?;
240        let pair = selected_pair.candidate_pair();
241        let local_candidate_type = pair.local.candidate_type;
242        let local_transport = pair.local.transport_type;
243        let local_base_addr = pair.local.base_address;
244        let local_addr = pair.local.address;
245        let remote_addr = pair.remote.address;
246        let stun_agent_id = selected_pair.stun_agent_id();
247
248        let data_len = data.as_ref().len();
249
250        let checklist = self.agent.checklistset.mut_list(checklist_id).unwrap();
251        if local_candidate_type == CandidateType::Relayed {
252            let turn_client = checklist
253                .mut_turn_client_by_allocated_address(local_transport, local_base_addr)
254                .ok_or(AgentError::ResourceNotFound)?
255                .1;
256            let transmit = turn_client
257                .send_to(local_transport, remote_addr, data, now)
258                .map_err(|_| AgentError::ResourceNotFound)?
259                .unwrap();
260            trace!(
261                "sending {} bytes from {} {} through TURN server {} with allocation {local_transport} {local_base_addr} to {remote_addr}",
262                data_len, transmit.transport, transmit.from, transmit.to,
263            );
264            Ok(Transmit::new(
265                transmit.data.build().into_boxed_slice(),
266                transmit.transport,
267                transmit.from,
268                transmit.to,
269            ))
270        } else {
271            let stun_agent = checklist
272                .agent_by_id(stun_agent_id)
273                .ok_or(AgentError::ResourceNotFound)?;
274            trace!(
275                "sending {} bytes directly over {local_transport} {local_addr} -> {remote_addr}",
276                data_len
277            );
278            let transmit = stun_agent.send_data(data, remote_addr);
279            let transport = transmit.transport;
280            Ok(transmit.reinterpret_data(|data| transmit_send(transport, data.as_ref())))
281        }
282    }
283}
284
285#[derive(Debug, Default, PartialEq, Eq)]
286pub(crate) enum GatherProgress {
287    #[default]
288    New,
289    InProgress,
290    Completed,
291}
292
293/// The state of a component
294#[derive(Copy, Clone, Debug, PartialEq, Eq)]
295#[repr(u32)]
296pub enum ComponentConnectionState {
297    /// Component is in initial state and no connectivity checks are in progress.
298    New,
299    /// Connectivity checks are in progress for this candidate
300    Connecting,
301    /// A [`CandidatePair`](crate::candidate::CandidatePair`) has been selected for this component
302    Connected,
303    /// No connection could be found for this Component
304    Failed,
305}
306
307#[derive(Debug)]
308pub(crate) struct ComponentState {
309    pub(crate) id: usize,
310    state: ComponentConnectionState,
311    selected_pair: Option<SelectedPair>,
312    pub(crate) gather_state: GatherProgress,
313    pub(crate) gatherer: Option<StunGatherer>,
314}
315
316impl ComponentState {
317    pub(crate) fn new(id: usize) -> Self {
318        Self {
319            id,
320            state: ComponentConnectionState::New,
321            selected_pair: None,
322            gather_state: GatherProgress::New,
323            gatherer: None,
324        }
325    }
326
327    pub(crate) fn gather_candidates(
328        &mut self,
329        sockets: &[(TransportType, SocketAddr)],
330        stun_servers: &[(TransportType, SocketAddr)],
331        turn_servers: &[(SocketAddr, &TurnConfig)],
332        rto: Option<RequestRto>,
333    ) -> Result<(), AgentError> {
334        if self.gather_state != GatherProgress::New {
335            return Err(AgentError::AlreadyInProgress);
336        }
337
338        let mut gatherer = StunGatherer::new(self.id, sockets, stun_servers, turn_servers);
339        if let Some(rto) = rto {
340            gatherer.set_request_retransmits(rto);
341        }
342        self.gatherer = Some(gatherer);
343        self.gather_state = GatherProgress::InProgress;
344
345        Ok(())
346    }
347
348    #[tracing::instrument(name = "set_component_state", level = "debug", skip(self, state))]
349    pub(crate) fn set_state(&mut self, state: ComponentConnectionState) -> bool {
350        if self.state != state {
351            debug!(old_state = ?self.state, new_state = ?state, "setting");
352            self.state = state;
353            true
354        } else {
355            false
356        }
357    }
358}
359
360#[cfg(test)]
361mod tests {
362    use super::*;
363    use crate::agent::Agent;
364    use crate::candidate::Candidate;
365    use alloc::vec;
366
367    #[test]
368    fn initial_state_new() {
369        let _log = crate::tests::test_init_log();
370        let mut agent = Agent::builder().build();
371        let sid = agent.add_stream();
372        let mut s = agent.mut_stream(sid).unwrap();
373        let cid = s.add_component().unwrap();
374        let c = s.component(cid).unwrap();
375        assert_eq!(c.state(), ComponentConnectionState::New);
376    }
377
378    #[test]
379    fn send_recv() {
380        let _log = crate::tests::test_init_log();
381        let mut agent = Agent::builder().controlling(false).build();
382        let stream_id = agent.add_stream();
383        let mut stream = agent.mut_stream(stream_id).unwrap();
384        let send_id = stream.add_component().unwrap();
385        let local_addr = "127.0.0.1:1000".parse().unwrap();
386        let remote_addr = "127.0.0.1:2000".parse().unwrap();
387        let now = Instant::ZERO;
388
389        let local_cand = Candidate::builder(
390            send_id,
391            CandidateType::Host,
392            TransportType::Udp,
393            "0",
394            local_addr,
395        )
396        .build();
397        let remote_cand = Candidate::builder(
398            send_id,
399            CandidateType::Host,
400            TransportType::Udp,
401            "0",
402            remote_addr,
403        )
404        .build();
405        let candidate_pair = CandidatePair::new(local_cand, remote_cand);
406
407        let mut send = stream.mut_component(send_id).unwrap();
408        send.set_selected_pair(candidate_pair.clone()).unwrap();
409        assert_eq!(send.selected_pair().unwrap(), &candidate_pair);
410
411        let data = vec![3; 4];
412        let transmit = send.send(&data, now).unwrap();
413        assert_eq!(transmit.transport, TransportType::Udp);
414        assert_eq!(transmit.from, local_addr);
415        assert_eq!(transmit.to, remote_addr);
416        assert_eq!(transmit.data.as_ref(), data.as_slice());
417
418        let recved_data = vec![7; 6];
419        let ret = stream.handle_incoming_data(
420            send_id,
421            Transmit::new(
422                recved_data.clone(),
423                TransportType::Udp,
424                remote_addr,
425                local_addr,
426            ),
427            now,
428        );
429        assert_eq!(recved_data.as_slice(), ret.data.unwrap().as_ref());
430        assert!(!ret.handled);
431        assert!(!ret.have_more_data);
432
433        // Unknown remote is ignored
434        let recved_data2 = vec![9; 12];
435        let ret = stream.handle_incoming_data(
436            send_id,
437            Transmit::new(recved_data2, TransportType::Udp, local_addr, local_addr),
438            now,
439        );
440        assert!(ret.data.is_none());
441        assert!(!ret.handled);
442        assert!(!ret.have_more_data);
443    }
444}
rice_proto/component.rs

rice_proto/
component.rs
