rtc_datachannel/data_channel/
mod.rs

1//! One data channel over an SCTP stream.
2//!
3//! A [`DataChannel`](crate::data_channel::DataChannel) is opened either by the DCEP handshake (`DATA_CHANNEL_OPEN`, then
4//! `DATA_CHANNEL_ACK`) or out of band when [`DataChannelConfig::negotiated`](crate::data_channel::DataChannelConfig::negotiated) is set and both
5//! sides already agreed the stream id through signalling.
6//!
7//! The reliability the channel is opened with maps onto SCTP send parameters:
8//! [`get_reliability_params`](crate::data_channel::DataChannel::get_reliability_params) turns a
9//! [`ChannelType`](crate::message::message_channel_open::ChannelType) into the ordered flag and
10//! partial-reliability setting SCTP needs, and
11//! [`get_channel_type_and_reliability_parameter`](crate::data_channel::DataChannel::get_channel_type_and_reliability_parameter)
12//! goes the other way from the W3C `maxPacketLifeTime`/`maxRetransmits` pair.
13#[cfg(test)]
14mod data_channel_test;
15
16use crate::message::{
17    message_channel_ack::*, message_channel_close::*, message_channel_open::*,
18    message_channel_threshold::*, *,
19};
20use bytes::{Buf, BytesMut};
21use log::debug;
22use sctp::{PayloadProtocolIdentifier, ReliabilityType};
23use shared::error::{Error, Result};
24use shared::marshal::*;
25use std::collections::VecDeque;
26
27const RECEIVE_MTU: usize = 8192;
28
29/// DataChannelConfig is used to configure the data channel.
30#[derive(Eq, PartialEq, Default, Clone, Debug)]
31pub struct DataChannelConfig {
32    /// The reliability and ordering guarantees to request.
33    pub channel_type: ChannelType,
34    /// Whether the channel was negotiated out of band.
35    ///
36    /// When `true` no DCEP `DATA_CHANNEL_OPEN` is sent — both sides are assumed to have agreed
37    /// the stream id and parameters through signalling instead.
38    pub negotiated: bool,
39    /// The channel's relative priority; see the `CHANNEL_PRIORITY_*` constants.
40    pub priority: u16,
41    /// The retransmission count or message lifetime, interpreted according to
42    /// [`Self::channel_type`].
43    pub reliability_parameter: u32,
44    /// The channel label, used to distinguish channels on the same association.
45    pub label: String,
46    /// The subprotocol name, or empty if none.
47    pub protocol: String,
48}
49
50/// DataChannelMessage is used to data sent over SCTP
51#[derive(Debug, Default, Clone)]
52pub struct DataChannelMessage {
53    /// Identifies the SCTP association this message belongs to.
54    pub association_handle: usize,
55    /// The SCTP stream the message arrived on or should be sent on.
56    pub stream_id: u16,
57    /// The payload protocol identifier, which distinguishes DCEP control messages from string
58    /// and binary user data.
59    pub ppi: PayloadProtocolIdentifier,
60    /// The message bytes.
61    pub payload: BytesMut,
62
63    /// Marks a `DATA_CHANNEL_OPEN` that belongs to an out-of-band *negotiated*
64    /// channel (W3C WebRTC `RTCDataChannelInit.negotiated`). Such a message is
65    /// only used to open and configure the local SCTP stream and must not be
66    /// transmitted to the peer, which already created its own channel with the
67    /// pre-agreed stream id. Ignored for every other message.
68    pub negotiated: bool,
69}
70
71/// DataChannel represents a data channel
72#[derive(Debug, Default, Clone)]
73pub struct DataChannel {
74    config: DataChannelConfig,
75    association_handle: usize,
76    stream_id: u16,
77
78    read_outs: VecDeque<DataChannelMessage>,
79    write_outs: VecDeque<DataChannelMessage>,
80
81    // stats
82    messages_sent: usize,
83    messages_received: usize,
84    bytes_sent: usize,
85    bytes_received: usize,
86}
87
88impl DataChannel {
89    fn new(config: DataChannelConfig, association_handle: usize, stream_id: u16) -> Self {
90        Self {
91            config,
92            association_handle,
93            stream_id,
94            read_outs: VecDeque::new(),
95            write_outs: VecDeque::new(),
96            ..Default::default()
97        }
98    }
99
100    /// Dial opens a data channels over SCTP
101    pub fn dial(
102        config: DataChannelConfig,
103        association_handle: usize,
104        stream_id: u16,
105    ) -> Result<Self> {
106        let mut data_channel = DataChannel::new(config.clone(), association_handle, stream_id);
107
108        // Both in-band and out-of-band (negotiated) channels emit a
109        // DATA_CHANNEL_OPEN so the underlying SCTP stream gets opened and its
110        // reliability parameters configured (the Channel Type / Reliability
111        // Parameter mapping in RFC 8832 section 5.1). For a negotiated channel
112        // the message is flagged so the transport opens the stream locally
113        // without sending the DCEP handshake to the peer; that suppression is
114        // required by the W3C WebRTC `negotiated` semantics, not by DCEP.
115        let msg = Message::DataChannelOpen(DataChannelOpen {
116            channel_type: config.channel_type,
117            priority: config.priority,
118            reliability_parameter: config.reliability_parameter,
119            label: config.label.bytes().collect(),
120            protocol: config.protocol.bytes().collect(),
121        })
122        .marshal()?;
123
124        data_channel.write_outs.push_back(DataChannelMessage {
125            association_handle,
126            stream_id,
127            ppi: PayloadProtocolIdentifier::Dcep,
128            payload: msg,
129            negotiated: config.negotiated,
130        });
131
132        Ok(data_channel)
133    }
134
135    /// Accept is used to accept incoming data channels over SCTP
136    pub fn accept(
137        mut config: DataChannelConfig,
138        association_handle: usize,
139        stream_id: u16,
140        ppi: PayloadProtocolIdentifier,
141        buf: &[u8],
142    ) -> Result<Self> {
143        if ppi != PayloadProtocolIdentifier::Dcep {
144            return Err(Error::InvalidPayloadProtocolIdentifier(ppi as u8));
145        }
146
147        let mut read_buf = buf;
148        let msg = Message::unmarshal(&mut read_buf)?;
149
150        if let Message::DataChannelOpen(dco) = msg {
151            config.channel_type = dco.channel_type;
152            config.priority = dco.priority;
153            config.reliability_parameter = dco.reliability_parameter;
154            config.label = String::from_utf8(dco.label)?;
155            config.protocol = String::from_utf8(dco.protocol)?;
156        } else {
157            return Err(Error::InvalidMessageType(msg.message_type() as u8));
158        };
159
160        let mut data_channel = DataChannel::new(config, association_handle, stream_id);
161
162        data_channel.write_data_channel_ack()?;
163
164        Ok(data_channel)
165    }
166
167    /// MessagesSent returns the number of messages sent
168    pub fn messages_sent(&self) -> usize {
169        self.messages_sent
170    }
171
172    /// MessagesReceived returns the number of messages received
173    pub fn messages_received(&self) -> usize {
174        self.messages_received
175    }
176
177    /// BytesSent returns the number of bytes sent
178    pub fn bytes_sent(&self) -> usize {
179        self.bytes_sent
180    }
181
182    /// BytesReceived returns the number of bytes received
183    pub fn bytes_received(&self) -> usize {
184        self.bytes_received
185    }
186
187    /// association_handle returns the association handle
188    pub fn association_handle(&self) -> usize {
189        self.association_handle
190    }
191
192    /// StreamIdentifier returns the Stream identifier associated to the stream.
193    pub fn stream_identifier(&self) -> u16 {
194        self.stream_id
195    }
196
197    /// The configuration this channel was opened with.
198    pub fn config(&self) -> &DataChannelConfig {
199        &self.config
200    }
201
202    fn handle_dcep<B>(&mut self, data: &mut B) -> Result<()>
203    where
204        B: Buf,
205    {
206        let msg = Message::unmarshal(data)?;
207
208        match msg {
209            Message::DataChannelOpen(_) => {
210                // Note: DATA_CHANNEL_OPEN message is handled inside Server() method.
211                // Therefore, the message will not reach here.
212                debug!("Received DATA_CHANNEL_OPEN");
213                self.write_data_channel_ack()?;
214            }
215            Message::DataChannelAck(_) => {
216                debug!("Received DATA_CHANNEL_ACK");
217            }
218            _ => {
219                return Err(Error::InvalidMessageType(msg.message_type() as u8));
220            }
221        };
222
223        Ok(())
224    }
225
226    fn write_data_channel_ack(&mut self) -> Result<()> {
227        let ack = Message::DataChannelAck(DataChannelAck {}).marshal()?;
228        self.write_outs.push_back(DataChannelMessage {
229            association_handle: self.association_handle,
230            stream_id: self.stream_id,
231            ppi: PayloadProtocolIdentifier::Dcep,
232            payload: ack,
233            negotiated: false,
234        });
235        Ok(())
236    }
237
238    fn write_data_channel_close(&mut self) -> Result<()> {
239        let close = Message::DataChannelClose(DataChannelClose {}).marshal()?;
240        self.write_outs.push_back(DataChannelMessage {
241            association_handle: self.association_handle,
242            stream_id: self.stream_id,
243            ppi: PayloadProtocolIdentifier::Dcep,
244            payload: close,
245            negotiated: false,
246        });
247        Ok(())
248    }
249
250    fn write_data_channel_high_threshold(&mut self, threshold: u32) -> Result<()> {
251        let low_threshold =
252            Message::DataChannelThreshold(DataChannelThreshold::High(threshold)).marshal()?;
253        self.write_outs.push_back(DataChannelMessage {
254            association_handle: self.association_handle,
255            stream_id: self.stream_id,
256            ppi: PayloadProtocolIdentifier::Dcep,
257            payload: low_threshold,
258            negotiated: false,
259        });
260        Ok(())
261    }
262
263    fn write_data_channel_low_threshold(&mut self, threshold: u32) -> Result<()> {
264        let low_threshold =
265            Message::DataChannelThreshold(DataChannelThreshold::Low(threshold)).marshal()?;
266        self.write_outs.push_back(DataChannelMessage {
267            association_handle: self.association_handle,
268            stream_id: self.stream_id,
269            ppi: PayloadProtocolIdentifier::Dcep,
270            payload: low_threshold,
271            negotiated: false,
272        });
273        Ok(())
274    }
275
276    /// SetBufferedAmountHighThreshold is used to update the threshold.
277    /// See BufferedAmountHighThreshold().
278    pub fn set_buffered_amount_high_threshold(&mut self, threshold: u32) -> Result<()> {
279        self.write_data_channel_high_threshold(threshold)
280    }
281
282    /// SetBufferedAmountLowThreshold is used to update the threshold.
283    /// See BufferedAmountLowThreshold().
284    pub fn set_buffered_amount_low_threshold(&mut self, threshold: u32) -> Result<()> {
285        self.write_data_channel_low_threshold(threshold)
286    }
287
288    /*
289    /// OnBufferedAmountLow sets the callback handler which would be called when the
290    /// number of bytes of outgoing data buffered is lower than the threshold.
291    pub fn on_buffered_amount_low(&self, f: OnBufferedAmountLowFn) {
292        self.stream.on_buffered_amount_low(f)
293    }*/
294
295    /// Decomposes a [`ChannelType`] into the SCTP send parameters it implies.
296    ///
297    /// Returns whether delivery is unordered, together with the reliability type and value SCTP
298    /// needs for partial reliability.
299    pub fn get_reliability_params(channel_type: ChannelType) -> (bool, ReliabilityType) {
300        match channel_type {
301            ChannelType::Reliable => (false, ReliabilityType::Reliable),
302            ChannelType::ReliableUnordered => (true, ReliabilityType::Reliable),
303            ChannelType::PartialReliableRexmit => (false, ReliabilityType::Rexmit),
304            ChannelType::PartialReliableRexmitUnordered => (true, ReliabilityType::Rexmit),
305            ChannelType::PartialReliableTimed => (false, ReliabilityType::Timed),
306            ChannelType::PartialReliableTimedUnordered => (true, ReliabilityType::Timed),
307        }
308    }
309
310    /// Derives the [`ChannelType`] and reliability parameter from the `maxPacketLifeTime` /
311    /// `maxRetransmits` pair the W3C API exposes.
312    ///
313    /// The two are mutually exclusive; supplying neither yields a fully reliable channel.
314    pub fn get_channel_type_and_reliability_parameter(
315        ordered: bool,
316        max_retransmits: Option<u16>,
317        max_packet_life_time: Option<u16>,
318    ) -> (ChannelType, u32) {
319        let channel_type;
320        let reliability_parameter;
321
322        match (max_retransmits, max_packet_life_time) {
323            (None, None) => {
324                reliability_parameter = 0u32;
325                if ordered {
326                    channel_type = ChannelType::Reliable;
327                } else {
328                    channel_type = ChannelType::ReliableUnordered;
329                }
330            }
331
332            (Some(max_retransmits), _) => {
333                reliability_parameter = max_retransmits as u32;
334                if ordered {
335                    channel_type = ChannelType::PartialReliableRexmit;
336                } else {
337                    channel_type = ChannelType::PartialReliableRexmitUnordered;
338                }
339            }
340
341            (None, Some(max_packet_lifetime)) => {
342                reliability_parameter = max_packet_lifetime as u32;
343                if ordered {
344                    channel_type = ChannelType::PartialReliableTimed;
345                } else {
346                    channel_type = ChannelType::PartialReliableTimedUnordered;
347                }
348            }
349        }
350
351        (channel_type, reliability_parameter)
352    }
353
354    /// Builds the payload protocol identifier and payload for a user message.
355    ///
356    /// Empty messages get their own identifiers (`StringEmpty`/`BinaryEmpty`) because SCTP
357    /// cannot carry a zero-length payload.
358    pub fn get_data_channel_message(is_string: bool, data: BytesMut) -> DataChannelMessage {
359        // https://tools.ietf.org/html/draft-ietf-rtcweb-data-channel-12#section-6.6
360        // SCTP does not support the sending of empty user messages.  Therefore,
361        // if an empty message has to be sent, the appropriate PPID (WebRTC
362        // String Empty or WebRTC Binary Empty) is used and the SCTP user
363        // message of one zero byte is sent.  When receiving an SCTP user
364        // message with one of these PPIDs, the receiver MUST ignore the SCTP
365        // user message and process it as an empty message.
366        let ppi = match (is_string, data.len()) {
367            (false, 0) => PayloadProtocolIdentifier::BinaryEmpty,
368            (false, _) => PayloadProtocolIdentifier::Binary,
369            (true, 0) => PayloadProtocolIdentifier::StringEmpty,
370            (true, _) => PayloadProtocolIdentifier::String,
371        };
372
373        if data.is_empty() {
374            DataChannelMessage {
375                ppi,
376                payload: BytesMut::from(&[0][..]),
377                ..Default::default()
378            }
379        } else {
380            DataChannelMessage {
381                ppi,
382                payload: data,
383                ..Default::default()
384            }
385        }
386    }
387}
388
389impl sansio::Protocol<DataChannelMessage, DataChannelMessage, ()> for DataChannel {
390    type Rout = DataChannelMessage;
391    type Wout = DataChannelMessage;
392    type Eout = ();
393    type Error = Error;
394    type Time = ();
395
396    /// ReadDataChannel reads a packet of len(p) bytes. It returns the number of bytes read and
397    /// `true` if the data read is a string.
398    fn handle_read(&mut self, msg: DataChannelMessage) -> Result<()> {
399        self.messages_received += 1;
400        self.bytes_received += msg.payload.len();
401
402        if msg.ppi == PayloadProtocolIdentifier::Dcep {
403            let mut data_buf = &msg.payload[..];
404            self.handle_dcep(&mut data_buf)
405        } else {
406            self.read_outs.push_back(msg);
407            Ok(())
408        }
409    }
410
411    fn poll_read(&mut self) -> Option<DataChannelMessage> {
412        self.read_outs.pop_front()
413    }
414
415    /// handle_write writes len(p) bytes from p
416    fn handle_write(&mut self, mut msg: DataChannelMessage) -> Result<()> {
417        self.messages_sent += 1;
418        self.bytes_sent += msg.payload.len();
419
420        msg.association_handle = self.association_handle;
421        msg.stream_id = self.stream_id;
422        self.write_outs.push_back(msg);
423
424        Ok(())
425    }
426
427    /// Returns packets to transmit
428    fn poll_write(&mut self) -> Option<DataChannelMessage> {
429        self.write_outs.pop_front()
430    }
431
432    /// Close closes the DataChannel and the underlying SCTP stream.
433    fn close(&mut self) -> Result<()> {
434        // https://tools.ietf.org/html/draft-ietf-rtcweb-data-channel-13#section-6.7
435        // Closing of a data channel MUST be signaled by resetting the
436        // corresponding outgoing streams [RFC6525].  This means that if one
437        // side decides to close the data channel, it resets the corresponding
438        // outgoing stream.  When the peer sees that an incoming stream was
439        // reset, it also resets its corresponding outgoing stream.  Once this
440        // is completed, the data channel is closed.  Resetting a stream sets
441        // the Stream Sequence Numbers (SSNs) of the stream back to 'zero' with
442        // a corresponding notification to the application layer that the reset
443        // has been performed.  Streams are available for reuse after a reset
444        // has been performed.
445        self.write_data_channel_close()
446    }
447}
rtc_datachannel/data_channel/mod.rs

rtc_datachannel/data_channel/
mod.rs
