1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
use crate::data_channel::RTCDataChannelId;
use crate::data_channel::parameters::DataChannelParameters;
use crate::data_channel::state::RTCDataChannelState;
use datachannel::data_channel::DataChannelConfig;
use sansio::Protocol;
use sctp::{PayloadProtocolIdentifier, StreamId};
use shared::error::{Error, Result};
use std::time::Instant;
#[derive(Clone)]
pub(crate) struct RTCDataChannelInternal {
/// Connection-local handle. Assigned by `DataChannelRegistry::insert`, stable for the
/// channel's lifetime, and the key this channel lives under. NOT the wire value — the
/// differing widths (`usize` vs `StreamId`) are what stop the two being confused.
///
/// Carried on the value as well as the key so the close and assignment paths can read it
/// back without a reverse lookup.
pub(crate) id: RTCDataChannelId,
/// The RFC 8832 §6 SCTP stream identifier.
///
/// `None` until the SCTP connected procedure assigns it, which cannot happen before the
/// DTLS role is resolved because the role decides the parity: even for the client, odd for
/// the server. Out-of-band (`negotiated`) channels and channels accepted from the peer
/// have it from birth, since their id is already fixed by the application or the wire.
pub(crate) stream_id: Option<StreamId>,
pub(crate) label: String,
pub(crate) ordered: bool,
pub(crate) max_packet_life_time: Option<u16>,
pub(crate) max_retransmits: Option<u16>,
pub(crate) protocol: String,
pub(crate) negotiated: bool,
pub(crate) ready_state: RTCDataChannelState,
pub(crate) buffered_amount_high_threshold: u32,
pub(crate) buffered_amount_low_threshold: u32,
/// User payload bytes handed to `send()`/`send_text()` that SCTP has not yet
/// released (acknowledged or abandoned). Incremented synchronously at the app
/// send boundary and decremented on SCTP buffer-release events, so it accounts
/// for bytes still in the app→core→SCTP send pipeline — not just the SCTP
/// stream's own `buffered_amount`, which counts only post-packetization. Used
/// for synchronous send back-pressure.
pub(crate) outstanding_bytes: usize,
/// Deadline by which an in-band channel's DCEP handshake must complete.
/// Set when the channel is dialed; cleared on handshake completion or close.
pub(crate) handshake_deadline: Option<Instant>,
/// Set when the DCEP handshake times out and `OnClose` has already been
/// emitted. Prevents `SCTPStreamClosed` from emitting/counting a second close.
pub(crate) close_emitted: bool,
pub(crate) data_channel: Option<::datachannel::data_channel::DataChannel>,
}
impl Default for RTCDataChannelInternal {
fn default() -> Self {
Self {
id: 0,
stream_id: None,
label: "".to_string(),
ordered: false,
max_packet_life_time: None,
max_retransmits: None,
protocol: "".to_string(),
negotiated: false,
ready_state: RTCDataChannelState::default(),
buffered_amount_high_threshold: u32::MAX,
buffered_amount_low_threshold: 0,
outstanding_bytes: 0,
handshake_deadline: None,
close_emitted: false,
data_channel: None,
}
}
}
impl RTCDataChannelInternal {
/// Creates the DataChannel object before the networking is set up.
///
/// The handle is left at its placeholder; [`DataChannelRegistry::insert`] assigns the real
/// one. The stream id comes from `params.negotiated` — `Some` for an out-of-band channel,
/// whose id the application already fixed, and `None` for an in-band one, whose id has to
/// wait for the DTLS role.
///
/// [`DataChannelRegistry::insert`]: crate::data_channel::registry::DataChannelRegistry::insert
pub(crate) fn new(params: DataChannelParameters) -> Self {
Self {
id: 0,
stream_id: params.negotiated,
label: params.label,
protocol: params.protocol,
negotiated: params.negotiated.is_some(),
ordered: params.ordered,
max_packet_life_time: params.max_packet_life_time,
max_retransmits: params.max_retransmits,
ready_state: RTCDataChannelState::Connecting,
buffered_amount_high_threshold: u32::MAX,
buffered_amount_low_threshold: 0,
outstanding_bytes: 0,
handshake_deadline: None,
close_emitted: false,
data_channel: None,
}
}
pub(crate) fn dial(&mut self, association_handle: usize) -> Result<()> {
// A channel cannot be dialed before its stream id exists, and the stream id cannot
// exist before the DTLS role does. Making that an error rather than a fallback is the
// whole point: guessing a parity here is what issue #199 was.
let stream_id = self
.stream_id
.ok_or(Error::ErrDataChannelStreamIdNotAssigned)?;
let (channel_type, reliability_parameter) =
::datachannel::data_channel::DataChannel::get_channel_type_and_reliability_parameter(
self.ordered,
self.max_retransmits,
self.max_packet_life_time,
);
let config = ::datachannel::data_channel::DataChannelConfig {
channel_type,
priority: ::datachannel::message::message_channel_open::CHANNEL_PRIORITY_NORMAL,
reliability_parameter,
label: self.label.clone(),
protocol: self.protocol.clone(),
negotiated: self.negotiated,
};
let mut data_channel =
::datachannel::data_channel::DataChannel::dial(config, association_handle, stream_id)?;
data_channel.set_buffered_amount_low_threshold(self.buffered_amount_low_threshold)?;
data_channel.set_buffered_amount_high_threshold(self.buffered_amount_high_threshold)?;
self.data_channel = Some(data_channel);
// An in-band channel stays `Connecting` until the peer's `DATA_CHANNEL_ACK`
// is processed in `DataChannelHandler::handle_read`. An out-of-band
// `negotiated` channel has no DCEP handshake, so it is open immediately.
self.ready_state = if self.negotiated {
RTCDataChannelState::Open
} else {
RTCDataChannelState::Connecting
};
Ok(())
}
pub(crate) fn accept(
association_handle: usize,
stream_id: StreamId,
ppi: PayloadProtocolIdentifier,
buf: &[u8],
) -> Result<Self> {
let data_channel = ::datachannel::data_channel::DataChannel::accept(
DataChannelConfig::default(),
association_handle,
stream_id,
ppi,
buf,
)?;
let data_channel_config = data_channel.config();
let (unordered, _reliability_type) =
::datachannel::data_channel::DataChannel::get_reliability_params(
data_channel_config.channel_type,
);
// `negotiated: None` — the peer opened this channel in-band over DCEP, so it is not an
// out-of-band channel however well-known its stream id now is. The stream id is set
// below instead of through `negotiated`, which would also flip the `negotiated` flag.
let mut data_channel_internal = RTCDataChannelInternal::new(DataChannelParameters {
label: data_channel_config.label.clone(),
protocol: data_channel_config.protocol.clone(),
ordered: !unordered,
max_packet_life_time: None,
max_retransmits: None,
negotiated: None,
});
// Known from the wire: no deferral needed, and it must be registered so locally
// generated ids cannot collide with it.
data_channel_internal.stream_id = Some(stream_id);
data_channel_internal.data_channel = Some(data_channel);
data_channel_internal.ready_state = RTCDataChannelState::Open;
Ok(data_channel_internal)
}
pub(crate) fn close(&mut self) -> Result<()> {
if let Some(data_channel) = self.data_channel.as_mut() {
data_channel.close()?;
}
self.handshake_deadline = None;
self.ready_state = RTCDataChannelState::Closed;
Ok(())
}
}