1 /*
2 * Copyright 2020 The WebRTC project authors. All Rights Reserved.
3 *
4 * Use of this source code is governed by a BSD-style license
5 * that can be found in the LICENSE file in the root of the source
6 * tree. An additional intellectual property rights grant can be found
7 * in the file PATENTS. All contributing project authors may
8 * be found in the AUTHORS file in the root of the source tree.
9 */
10
11 #include "pc/sctp_data_channel.h"
12
13 #include <limits>
14 #include <memory>
15 #include <string>
16 #include <utility>
17
18 #include "absl/cleanup/cleanup.h"
19 #include "media/sctp/sctp_transport_internal.h"
20 #include "pc/proxy.h"
21 #include "pc/sctp_utils.h"
22 #include "rtc_base/checks.h"
23 #include "rtc_base/logging.h"
24 #include "rtc_base/system/unused.h"
25 #include "rtc_base/thread.h"
26
27 namespace webrtc {
28
29 namespace {
30
31 static size_t kMaxQueuedReceivedDataBytes = 16 * 1024 * 1024;
32
33 static std::atomic<int> g_unique_id{0};
34
GenerateUniqueId()35 int GenerateUniqueId() {
36 return ++g_unique_id;
37 }
38
39 // Define proxy for DataChannelInterface.
40 BEGIN_PRIMARY_PROXY_MAP(DataChannel)
41 PROXY_PRIMARY_THREAD_DESTRUCTOR()
42 PROXY_METHOD1(void, RegisterObserver, DataChannelObserver*)
43 PROXY_METHOD0(void, UnregisterObserver)
44 BYPASS_PROXY_CONSTMETHOD0(std::string, label)
45 BYPASS_PROXY_CONSTMETHOD0(bool, reliable)
46 BYPASS_PROXY_CONSTMETHOD0(bool, ordered)
47 BYPASS_PROXY_CONSTMETHOD0(uint16_t, maxRetransmitTime)
48 BYPASS_PROXY_CONSTMETHOD0(uint16_t, maxRetransmits)
49 BYPASS_PROXY_CONSTMETHOD0(absl::optional<int>, maxRetransmitsOpt)
50 BYPASS_PROXY_CONSTMETHOD0(absl::optional<int>, maxPacketLifeTime)
51 BYPASS_PROXY_CONSTMETHOD0(std::string, protocol)
52 BYPASS_PROXY_CONSTMETHOD0(bool, negotiated)
53 // Can't bypass the proxy since the id may change.
54 PROXY_CONSTMETHOD0(int, id)
55 BYPASS_PROXY_CONSTMETHOD0(Priority, priority)
56 PROXY_CONSTMETHOD0(DataState, state)
57 PROXY_CONSTMETHOD0(RTCError, error)
58 PROXY_CONSTMETHOD0(uint32_t, messages_sent)
59 PROXY_CONSTMETHOD0(uint64_t, bytes_sent)
60 PROXY_CONSTMETHOD0(uint32_t, messages_received)
61 PROXY_CONSTMETHOD0(uint64_t, bytes_received)
62 PROXY_CONSTMETHOD0(uint64_t, buffered_amount)
63 PROXY_METHOD0(void, Close)
64 // TODO(bugs.webrtc.org/11547): Change to run on the network thread.
65 PROXY_METHOD1(bool, Send, const DataBuffer&)
66 END_PROXY_MAP(DataChannel)
67
68 } // namespace
69
InternalDataChannelInit(const DataChannelInit & base)70 InternalDataChannelInit::InternalDataChannelInit(const DataChannelInit& base)
71 : DataChannelInit(base), open_handshake_role(kOpener) {
72 // If the channel is externally negotiated, do not send the OPEN message.
73 if (base.negotiated) {
74 open_handshake_role = kNone;
75 } else {
76 // Datachannel is externally negotiated. Ignore the id value.
77 // Specified in createDataChannel, WebRTC spec section 6.1 bullet 13.
78 id = -1;
79 }
80 // Backwards compatibility: If maxRetransmits or maxRetransmitTime
81 // are negative, the feature is not enabled.
82 // Values are clamped to a 16bit range.
83 if (maxRetransmits) {
84 if (*maxRetransmits < 0) {
85 RTC_LOG(LS_ERROR)
86 << "Accepting maxRetransmits < 0 for backwards compatibility";
87 maxRetransmits = absl::nullopt;
88 } else if (*maxRetransmits > std::numeric_limits<uint16_t>::max()) {
89 maxRetransmits = std::numeric_limits<uint16_t>::max();
90 }
91 }
92
93 if (maxRetransmitTime) {
94 if (*maxRetransmitTime < 0) {
95 RTC_LOG(LS_ERROR)
96 << "Accepting maxRetransmitTime < 0 for backwards compatibility";
97 maxRetransmitTime = absl::nullopt;
98 } else if (*maxRetransmitTime > std::numeric_limits<uint16_t>::max()) {
99 maxRetransmitTime = std::numeric_limits<uint16_t>::max();
100 }
101 }
102 }
103
AllocateSid(rtc::SSLRole role,int * sid)104 bool SctpSidAllocator::AllocateSid(rtc::SSLRole role, int* sid) {
105 int potential_sid = (role == rtc::SSL_CLIENT) ? 0 : 1;
106 while (!IsSidAvailable(potential_sid)) {
107 potential_sid += 2;
108 if (potential_sid > static_cast<int>(cricket::kMaxSctpSid)) {
109 return false;
110 }
111 }
112
113 *sid = potential_sid;
114 used_sids_.insert(potential_sid);
115 return true;
116 }
117
ReserveSid(int sid)118 bool SctpSidAllocator::ReserveSid(int sid) {
119 if (!IsSidAvailable(sid)) {
120 return false;
121 }
122 used_sids_.insert(sid);
123 return true;
124 }
125
ReleaseSid(int sid)126 void SctpSidAllocator::ReleaseSid(int sid) {
127 auto it = used_sids_.find(sid);
128 if (it != used_sids_.end()) {
129 used_sids_.erase(it);
130 }
131 }
132
IsSidAvailable(int sid) const133 bool SctpSidAllocator::IsSidAvailable(int sid) const {
134 if (sid < static_cast<int>(cricket::kMinSctpSid) ||
135 sid > static_cast<int>(cricket::kMaxSctpSid)) {
136 return false;
137 }
138 return used_sids_.find(sid) == used_sids_.end();
139 }
140
Create(SctpDataChannelControllerInterface * controller,const std::string & label,const InternalDataChannelInit & config,rtc::Thread * signaling_thread,rtc::Thread * network_thread)141 rtc::scoped_refptr<SctpDataChannel> SctpDataChannel::Create(
142 SctpDataChannelControllerInterface* controller,
143 const std::string& label,
144 const InternalDataChannelInit& config,
145 rtc::Thread* signaling_thread,
146 rtc::Thread* network_thread) {
147 auto channel = rtc::make_ref_counted<SctpDataChannel>(
148 config, controller, label, signaling_thread, network_thread);
149 if (!channel->Init()) {
150 return nullptr;
151 }
152 return channel;
153 }
154
155 // static
CreateProxy(rtc::scoped_refptr<SctpDataChannel> channel)156 rtc::scoped_refptr<DataChannelInterface> SctpDataChannel::CreateProxy(
157 rtc::scoped_refptr<SctpDataChannel> channel) {
158 // TODO(bugs.webrtc.org/11547): incorporate the network thread in the proxy.
159 auto* signaling_thread = channel->signaling_thread_;
160 return DataChannelProxy::Create(signaling_thread, std::move(channel));
161 }
162
SctpDataChannel(const InternalDataChannelInit & config,SctpDataChannelControllerInterface * controller,const std::string & label,rtc::Thread * signaling_thread,rtc::Thread * network_thread)163 SctpDataChannel::SctpDataChannel(const InternalDataChannelInit& config,
164 SctpDataChannelControllerInterface* controller,
165 const std::string& label,
166 rtc::Thread* signaling_thread,
167 rtc::Thread* network_thread)
168 : signaling_thread_(signaling_thread),
169 network_thread_(network_thread),
170 internal_id_(GenerateUniqueId()),
171 label_(label),
172 config_(config),
173 observer_(nullptr),
174 controller_(controller) {
175 RTC_DCHECK_RUN_ON(signaling_thread_);
176 RTC_UNUSED(network_thread_);
177 }
178
DetachFromController()179 void SctpDataChannel::DetachFromController() {
180 RTC_DCHECK_RUN_ON(signaling_thread_);
181 controller_detached_ = true;
182 }
183
Init()184 bool SctpDataChannel::Init() {
185 RTC_DCHECK_RUN_ON(signaling_thread_);
186 if (config_.id < -1 ||
187 (config_.maxRetransmits && *config_.maxRetransmits < 0) ||
188 (config_.maxRetransmitTime && *config_.maxRetransmitTime < 0)) {
189 RTC_LOG(LS_ERROR) << "Failed to initialize the SCTP data channel due to "
190 "invalid DataChannelInit.";
191 return false;
192 }
193 if (config_.maxRetransmits && config_.maxRetransmitTime) {
194 RTC_LOG(LS_ERROR)
195 << "maxRetransmits and maxRetransmitTime should not be both set.";
196 return false;
197 }
198
199 switch (config_.open_handshake_role) {
200 case webrtc::InternalDataChannelInit::kNone: // pre-negotiated
201 handshake_state_ = kHandshakeReady;
202 break;
203 case webrtc::InternalDataChannelInit::kOpener:
204 handshake_state_ = kHandshakeShouldSendOpen;
205 break;
206 case webrtc::InternalDataChannelInit::kAcker:
207 handshake_state_ = kHandshakeShouldSendAck;
208 break;
209 }
210
211 // Try to connect to the transport in case the transport channel already
212 // exists.
213 OnTransportChannelCreated();
214
215 // Checks if the transport is ready to send because the initial channel
216 // ready signal may have been sent before the DataChannel creation.
217 // This has to be done async because the upper layer objects (e.g.
218 // Chrome glue and WebKit) are not wired up properly until after this
219 // function returns.
220 RTC_DCHECK(!controller_detached_);
221 if (controller_->ReadyToSendData()) {
222 AddRef();
223 absl::Cleanup release = [this] { Release(); };
224 rtc::Thread::Current()->PostTask([this, release = std::move(release)] {
225 RTC_DCHECK_RUN_ON(signaling_thread_);
226 if (state_ != kClosed)
227 OnTransportReady(true);
228 });
229 }
230
231 return true;
232 }
233
~SctpDataChannel()234 SctpDataChannel::~SctpDataChannel() {
235 RTC_DCHECK_RUN_ON(signaling_thread_);
236 }
237
RegisterObserver(DataChannelObserver * observer)238 void SctpDataChannel::RegisterObserver(DataChannelObserver* observer) {
239 RTC_DCHECK_RUN_ON(signaling_thread_);
240 observer_ = observer;
241 DeliverQueuedReceivedData();
242 }
243
UnregisterObserver()244 void SctpDataChannel::UnregisterObserver() {
245 RTC_DCHECK_RUN_ON(signaling_thread_);
246 observer_ = nullptr;
247 }
248
reliable() const249 bool SctpDataChannel::reliable() const {
250 // May be called on any thread.
251 return !config_.maxRetransmits && !config_.maxRetransmitTime;
252 }
253
buffered_amount() const254 uint64_t SctpDataChannel::buffered_amount() const {
255 RTC_DCHECK_RUN_ON(signaling_thread_);
256 return queued_send_data_.byte_count();
257 }
258
Close()259 void SctpDataChannel::Close() {
260 RTC_DCHECK_RUN_ON(signaling_thread_);
261 if (state_ == kClosing || state_ == kClosed)
262 return;
263 SetState(kClosing);
264 // Will send queued data before beginning the underlying closing procedure.
265 UpdateState();
266 }
267
state() const268 SctpDataChannel::DataState SctpDataChannel::state() const {
269 RTC_DCHECK_RUN_ON(signaling_thread_);
270 return state_;
271 }
272
error() const273 RTCError SctpDataChannel::error() const {
274 RTC_DCHECK_RUN_ON(signaling_thread_);
275 return error_;
276 }
277
messages_sent() const278 uint32_t SctpDataChannel::messages_sent() const {
279 RTC_DCHECK_RUN_ON(signaling_thread_);
280 return messages_sent_;
281 }
282
bytes_sent() const283 uint64_t SctpDataChannel::bytes_sent() const {
284 RTC_DCHECK_RUN_ON(signaling_thread_);
285 return bytes_sent_;
286 }
287
messages_received() const288 uint32_t SctpDataChannel::messages_received() const {
289 RTC_DCHECK_RUN_ON(signaling_thread_);
290 return messages_received_;
291 }
292
bytes_received() const293 uint64_t SctpDataChannel::bytes_received() const {
294 RTC_DCHECK_RUN_ON(signaling_thread_);
295 return bytes_received_;
296 }
297
Send(const DataBuffer & buffer)298 bool SctpDataChannel::Send(const DataBuffer& buffer) {
299 RTC_DCHECK_RUN_ON(signaling_thread_);
300 // TODO(bugs.webrtc.org/11547): Expect this method to be called on the network
301 // thread. Bring buffer management etc to the network thread and keep the
302 // operational state management on the signaling thread.
303
304 if (state_ != kOpen) {
305 return false;
306 }
307
308 // If the queue is non-empty, we're waiting for SignalReadyToSend,
309 // so just add to the end of the queue and keep waiting.
310 if (!queued_send_data_.Empty()) {
311 if (!QueueSendDataMessage(buffer)) {
312 // Queue is full
313 return false;
314 }
315 return true;
316 }
317
318 SendDataMessage(buffer, true);
319
320 // Always return true for SCTP DataChannel per the spec.
321 return true;
322 }
323
SetSctpSid(int sid)324 void SctpDataChannel::SetSctpSid(int sid) {
325 RTC_DCHECK_RUN_ON(signaling_thread_);
326 RTC_DCHECK_LT(config_.id, 0);
327 RTC_DCHECK_GE(sid, 0);
328 RTC_DCHECK_NE(handshake_state_, kHandshakeWaitingForAck);
329 RTC_DCHECK_EQ(state_, kConnecting);
330
331 if (config_.id == sid) {
332 return;
333 }
334
335 const_cast<InternalDataChannelInit&>(config_).id = sid;
336 RTC_DCHECK(!controller_detached_);
337 controller_->AddSctpDataStream(sid);
338 }
339
OnClosingProcedureStartedRemotely(int sid)340 void SctpDataChannel::OnClosingProcedureStartedRemotely(int sid) {
341 RTC_DCHECK_RUN_ON(signaling_thread_);
342 if (sid == config_.id && state_ != kClosing && state_ != kClosed) {
343 // Don't bother sending queued data since the side that initiated the
344 // closure wouldn't receive it anyway. See crbug.com/559394 for a lengthy
345 // discussion about this.
346 queued_send_data_.Clear();
347 queued_control_data_.Clear();
348 // Just need to change state to kClosing, SctpTransport will handle the
349 // rest of the closing procedure and OnClosingProcedureComplete will be
350 // called later.
351 started_closing_procedure_ = true;
352 SetState(kClosing);
353 }
354 }
355
OnClosingProcedureComplete(int sid)356 void SctpDataChannel::OnClosingProcedureComplete(int sid) {
357 RTC_DCHECK_RUN_ON(signaling_thread_);
358 if (sid == config_.id) {
359 // If the closing procedure is complete, we should have finished sending
360 // all pending data and transitioned to kClosing already.
361 RTC_DCHECK_EQ(state_, kClosing);
362 RTC_DCHECK(queued_send_data_.Empty());
363 DisconnectFromTransport();
364 SetState(kClosed);
365 }
366 }
367
OnTransportChannelCreated()368 void SctpDataChannel::OnTransportChannelCreated() {
369 RTC_DCHECK_RUN_ON(signaling_thread_);
370 if (controller_detached_) {
371 return;
372 }
373 if (!connected_to_transport_) {
374 connected_to_transport_ = controller_->ConnectDataChannel(this);
375 }
376 // The sid may have been unassigned when controller_->ConnectDataChannel was
377 // done. So always add the streams even if connected_to_transport_ is true.
378 if (config_.id >= 0) {
379 controller_->AddSctpDataStream(config_.id);
380 }
381 }
382
OnTransportChannelClosed(RTCError error)383 void SctpDataChannel::OnTransportChannelClosed(RTCError error) {
384 // The SctpTransport is unusable, which could come from multiplie reasons:
385 // - the SCTP m= section was rejected
386 // - the DTLS transport is closed
387 // - the SCTP transport is closed
388 CloseAbruptlyWithError(std::move(error));
389 }
390
GetStats() const391 DataChannelStats SctpDataChannel::GetStats() const {
392 RTC_DCHECK_RUN_ON(signaling_thread_);
393 DataChannelStats stats{internal_id_, id(), label(),
394 protocol(), state(), messages_sent(),
395 messages_received(), bytes_sent(), bytes_received()};
396 return stats;
397 }
398
OnDataReceived(const cricket::ReceiveDataParams & params,const rtc::CopyOnWriteBuffer & payload)399 void SctpDataChannel::OnDataReceived(const cricket::ReceiveDataParams& params,
400 const rtc::CopyOnWriteBuffer& payload) {
401 RTC_DCHECK_RUN_ON(signaling_thread_);
402 if (params.sid != config_.id) {
403 return;
404 }
405
406 if (params.type == DataMessageType::kControl) {
407 if (handshake_state_ != kHandshakeWaitingForAck) {
408 // Ignore it if we are not expecting an ACK message.
409 RTC_LOG(LS_WARNING)
410 << "DataChannel received unexpected CONTROL message, sid = "
411 << params.sid;
412 return;
413 }
414 if (ParseDataChannelOpenAckMessage(payload)) {
415 // We can send unordered as soon as we receive the ACK message.
416 handshake_state_ = kHandshakeReady;
417 RTC_LOG(LS_INFO) << "DataChannel received OPEN_ACK message, sid = "
418 << params.sid;
419 } else {
420 RTC_LOG(LS_WARNING)
421 << "DataChannel failed to parse OPEN_ACK message, sid = "
422 << params.sid;
423 }
424 return;
425 }
426
427 RTC_DCHECK(params.type == DataMessageType::kBinary ||
428 params.type == DataMessageType::kText);
429
430 RTC_LOG(LS_VERBOSE) << "DataChannel received DATA message, sid = "
431 << params.sid;
432 // We can send unordered as soon as we receive any DATA message since the
433 // remote side must have received the OPEN (and old clients do not send
434 // OPEN_ACK).
435 if (handshake_state_ == kHandshakeWaitingForAck) {
436 handshake_state_ = kHandshakeReady;
437 }
438
439 bool binary = (params.type == webrtc::DataMessageType::kBinary);
440 auto buffer = std::make_unique<DataBuffer>(payload, binary);
441 if (state_ == kOpen && observer_) {
442 ++messages_received_;
443 bytes_received_ += buffer->size();
444 observer_->OnMessage(*buffer.get());
445 } else {
446 if (queued_received_data_.byte_count() + payload.size() >
447 kMaxQueuedReceivedDataBytes) {
448 RTC_LOG(LS_ERROR) << "Queued received data exceeds the max buffer size.";
449
450 queued_received_data_.Clear();
451 CloseAbruptlyWithError(
452 RTCError(RTCErrorType::RESOURCE_EXHAUSTED,
453 "Queued received data exceeds the max buffer size."));
454
455 return;
456 }
457 queued_received_data_.PushBack(std::move(buffer));
458 }
459 }
460
OnTransportReady(bool writable)461 void SctpDataChannel::OnTransportReady(bool writable) {
462 RTC_DCHECK_RUN_ON(signaling_thread_);
463
464 writable_ = writable;
465 if (!writable) {
466 return;
467 }
468
469 SendQueuedControlMessages();
470 SendQueuedDataMessages();
471
472 UpdateState();
473 }
474
CloseAbruptlyWithError(RTCError error)475 void SctpDataChannel::CloseAbruptlyWithError(RTCError error) {
476 RTC_DCHECK_RUN_ON(signaling_thread_);
477
478 if (state_ == kClosed) {
479 return;
480 }
481
482 if (connected_to_transport_) {
483 DisconnectFromTransport();
484 }
485
486 // Closing abruptly means any queued data gets thrown away.
487 queued_send_data_.Clear();
488 queued_control_data_.Clear();
489
490 // Still go to "kClosing" before "kClosed", since observers may be expecting
491 // that.
492 SetState(kClosing);
493 error_ = std::move(error);
494 SetState(kClosed);
495 }
496
CloseAbruptlyWithDataChannelFailure(const std::string & message)497 void SctpDataChannel::CloseAbruptlyWithDataChannelFailure(
498 const std::string& message) {
499 RTCError error(RTCErrorType::OPERATION_ERROR_WITH_DATA, message);
500 error.set_error_detail(RTCErrorDetailType::DATA_CHANNEL_FAILURE);
501 CloseAbruptlyWithError(std::move(error));
502 }
503
UpdateState()504 void SctpDataChannel::UpdateState() {
505 RTC_DCHECK_RUN_ON(signaling_thread_);
506 // UpdateState determines what to do from a few state variables. Include
507 // all conditions required for each state transition here for
508 // clarity. OnTransportReady(true) will send any queued data and then invoke
509 // UpdateState().
510
511 switch (state_) {
512 case kConnecting: {
513 if (connected_to_transport_) {
514 if (handshake_state_ == kHandshakeShouldSendOpen) {
515 rtc::CopyOnWriteBuffer payload;
516 WriteDataChannelOpenMessage(label_, config_, &payload);
517 SendControlMessage(payload);
518 } else if (handshake_state_ == kHandshakeShouldSendAck) {
519 rtc::CopyOnWriteBuffer payload;
520 WriteDataChannelOpenAckMessage(&payload);
521 SendControlMessage(payload);
522 }
523 if (writable_ && (handshake_state_ == kHandshakeReady ||
524 handshake_state_ == kHandshakeWaitingForAck)) {
525 SetState(kOpen);
526 // If we have received buffers before the channel got writable.
527 // Deliver them now.
528 DeliverQueuedReceivedData();
529 }
530 }
531 break;
532 }
533 case kOpen: {
534 break;
535 }
536 case kClosing: {
537 // Wait for all queued data to be sent before beginning the closing
538 // procedure.
539 if (queued_send_data_.Empty() && queued_control_data_.Empty()) {
540 // For SCTP data channels, we need to wait for the closing procedure
541 // to complete; after calling RemoveSctpDataStream,
542 // OnClosingProcedureComplete will end up called asynchronously
543 // afterwards.
544 if (connected_to_transport_ && !started_closing_procedure_ &&
545 !controller_detached_ && config_.id >= 0) {
546 started_closing_procedure_ = true;
547 controller_->RemoveSctpDataStream(config_.id);
548 }
549 }
550 break;
551 }
552 case kClosed:
553 break;
554 }
555 }
556
SetState(DataState state)557 void SctpDataChannel::SetState(DataState state) {
558 RTC_DCHECK_RUN_ON(signaling_thread_);
559 if (state_ == state) {
560 return;
561 }
562
563 state_ = state;
564 if (observer_) {
565 observer_->OnStateChange();
566 }
567 if (state_ == kOpen) {
568 SignalOpened(this);
569 } else if (state_ == kClosed) {
570 SignalClosed(this);
571 }
572 }
573
DisconnectFromTransport()574 void SctpDataChannel::DisconnectFromTransport() {
575 RTC_DCHECK_RUN_ON(signaling_thread_);
576 if (!connected_to_transport_ || controller_detached_)
577 return;
578
579 controller_->DisconnectDataChannel(this);
580 connected_to_transport_ = false;
581 }
582
DeliverQueuedReceivedData()583 void SctpDataChannel::DeliverQueuedReceivedData() {
584 RTC_DCHECK_RUN_ON(signaling_thread_);
585 if (!observer_) {
586 return;
587 }
588
589 while (!queued_received_data_.Empty()) {
590 std::unique_ptr<DataBuffer> buffer = queued_received_data_.PopFront();
591 ++messages_received_;
592 bytes_received_ += buffer->size();
593 observer_->OnMessage(*buffer);
594 }
595 }
596
SendQueuedDataMessages()597 void SctpDataChannel::SendQueuedDataMessages() {
598 RTC_DCHECK_RUN_ON(signaling_thread_);
599 if (queued_send_data_.Empty()) {
600 return;
601 }
602
603 RTC_DCHECK(state_ == kOpen || state_ == kClosing);
604
605 while (!queued_send_data_.Empty()) {
606 std::unique_ptr<DataBuffer> buffer = queued_send_data_.PopFront();
607 if (!SendDataMessage(*buffer, false)) {
608 // Return the message to the front of the queue if sending is aborted.
609 queued_send_data_.PushFront(std::move(buffer));
610 break;
611 }
612 }
613 }
614
SendDataMessage(const DataBuffer & buffer,bool queue_if_blocked)615 bool SctpDataChannel::SendDataMessage(const DataBuffer& buffer,
616 bool queue_if_blocked) {
617 RTC_DCHECK_RUN_ON(signaling_thread_);
618 SendDataParams send_params;
619 if (controller_detached_) {
620 return false;
621 }
622
623 send_params.ordered = config_.ordered;
624 // Send as ordered if it is still going through OPEN/ACK signaling.
625 if (handshake_state_ != kHandshakeReady && !config_.ordered) {
626 send_params.ordered = true;
627 RTC_LOG(LS_VERBOSE)
628 << "Sending data as ordered for unordered DataChannel "
629 "because the OPEN_ACK message has not been received.";
630 }
631
632 send_params.max_rtx_count = config_.maxRetransmits;
633 send_params.max_rtx_ms = config_.maxRetransmitTime;
634 send_params.type =
635 buffer.binary ? DataMessageType::kBinary : DataMessageType::kText;
636
637 cricket::SendDataResult send_result = cricket::SDR_SUCCESS;
638 bool success =
639 controller_->SendData(config_.id, send_params, buffer.data, &send_result);
640
641 if (success) {
642 ++messages_sent_;
643 bytes_sent_ += buffer.size();
644
645 if (observer_ && buffer.size() > 0) {
646 observer_->OnBufferedAmountChange(buffer.size());
647 }
648 return true;
649 }
650
651 if (send_result == cricket::SDR_BLOCK) {
652 if (!queue_if_blocked || QueueSendDataMessage(buffer)) {
653 return false;
654 }
655 }
656 // Close the channel if the error is not SDR_BLOCK, or if queuing the
657 // message failed.
658 RTC_LOG(LS_ERROR) << "Closing the DataChannel due to a failure to send data, "
659 "send_result = "
660 << send_result;
661 CloseAbruptlyWithError(
662 RTCError(RTCErrorType::NETWORK_ERROR, "Failure to send data"));
663
664 return false;
665 }
666
QueueSendDataMessage(const DataBuffer & buffer)667 bool SctpDataChannel::QueueSendDataMessage(const DataBuffer& buffer) {
668 RTC_DCHECK_RUN_ON(signaling_thread_);
669 size_t start_buffered_amount = queued_send_data_.byte_count();
670 if (start_buffered_amount + buffer.size() >
671 DataChannelInterface::MaxSendQueueSize()) {
672 RTC_LOG(LS_ERROR) << "Can't buffer any more data for the data channel.";
673 return false;
674 }
675 queued_send_data_.PushBack(std::make_unique<DataBuffer>(buffer));
676 return true;
677 }
678
SendQueuedControlMessages()679 void SctpDataChannel::SendQueuedControlMessages() {
680 RTC_DCHECK_RUN_ON(signaling_thread_);
681 PacketQueue control_packets;
682 control_packets.Swap(&queued_control_data_);
683
684 while (!control_packets.Empty()) {
685 std::unique_ptr<DataBuffer> buf = control_packets.PopFront();
686 SendControlMessage(buf->data);
687 }
688 }
689
QueueControlMessage(const rtc::CopyOnWriteBuffer & buffer)690 void SctpDataChannel::QueueControlMessage(
691 const rtc::CopyOnWriteBuffer& buffer) {
692 RTC_DCHECK_RUN_ON(signaling_thread_);
693 queued_control_data_.PushBack(std::make_unique<DataBuffer>(buffer, true));
694 }
695
SendControlMessage(const rtc::CopyOnWriteBuffer & buffer)696 bool SctpDataChannel::SendControlMessage(const rtc::CopyOnWriteBuffer& buffer) {
697 RTC_DCHECK_RUN_ON(signaling_thread_);
698 RTC_DCHECK(writable_);
699 RTC_DCHECK_GE(config_.id, 0);
700
701 if (controller_detached_) {
702 return false;
703 }
704 bool is_open_message = handshake_state_ == kHandshakeShouldSendOpen;
705 RTC_DCHECK(!is_open_message || !config_.negotiated);
706
707 SendDataParams send_params;
708 // Send data as ordered before we receive any message from the remote peer to
709 // make sure the remote peer will not receive any data before it receives the
710 // OPEN message.
711 send_params.ordered = config_.ordered || is_open_message;
712 send_params.type = DataMessageType::kControl;
713
714 cricket::SendDataResult send_result = cricket::SDR_SUCCESS;
715 bool retval =
716 controller_->SendData(config_.id, send_params, buffer, &send_result);
717 if (retval) {
718 RTC_LOG(LS_VERBOSE) << "Sent CONTROL message on channel " << config_.id;
719
720 if (handshake_state_ == kHandshakeShouldSendAck) {
721 handshake_state_ = kHandshakeReady;
722 } else if (handshake_state_ == kHandshakeShouldSendOpen) {
723 handshake_state_ = kHandshakeWaitingForAck;
724 }
725 } else if (send_result == cricket::SDR_BLOCK) {
726 QueueControlMessage(buffer);
727 } else {
728 RTC_LOG(LS_ERROR) << "Closing the DataChannel due to a failure to send"
729 " the CONTROL message, send_result = "
730 << send_result;
731 CloseAbruptlyWithError(RTCError(RTCErrorType::NETWORK_ERROR,
732 "Failed to send a CONTROL message"));
733 }
734 return retval;
735 }
736
737 // static
ResetInternalIdAllocatorForTesting(int new_value)738 void SctpDataChannel::ResetInternalIdAllocatorForTesting(int new_value) {
739 g_unique_id = new_value;
740 }
741
DowncastProxiedDataChannelInterfaceToSctpDataChannelForTesting(DataChannelInterface * channel)742 SctpDataChannel* DowncastProxiedDataChannelInterfaceToSctpDataChannelForTesting(
743 DataChannelInterface* channel) {
744 return static_cast<SctpDataChannel*>(
745 static_cast<DataChannelProxy*>(channel)->internal());
746 }
747
748 } // namespace webrtc
749