src/third_party/libjingle/source/talk/app/webrtc/datachannel.cc

   1 /*
   2  * libjingle
   3  * Copyright 2012, Google Inc.
   4  *
   5  * Redistribution and use in source and binary forms, with or without
   6  * modification, are permitted provided that the following conditions are met:
   7  *
   8  *  1. Redistributions of source code must retain the above copyright notice,
   9  *     this list of conditions and the following disclaimer.
  10  *  2. Redistributions in binary form must reproduce the above copyright notice,
  11  *     this list of conditions and the following disclaimer in the documentation
  12  *     and/or other materials provided with the distribution.
  13  *  3. The name of the author may not be used to endorse or promote products
  14  *     derived from this software without specific prior written permission.
  15  *
  16  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
  17  * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
  18  * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
  19  * EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  20  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
  21  * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
  22  * OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
  23  * WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
  24  * OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
  25  * ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  26  */
  27 #include "talk/app/webrtc/datachannel.h"
  28
  29 #include <string>
  30
  31 #include "talk/app/webrtc/mediastreamprovider.h"
  32 #include "talk/app/webrtc/sctputils.h"
  33 #include "webrtc/base/logging.h"
  34 #include "webrtc/base/refcount.h"
  35
  36 namespace webrtc {
  37
  38 static size_t kMaxQueuedReceivedDataBytes = 16 * 1024 * 1024;
  39 static size_t kMaxQueuedSendDataBytes = 16 * 1024 * 1024;
  40
  41 enum {
  42   MSG_CHANNELREADY,
  43 };
  44
  45 DataChannel::PacketQueue::PacketQueue() : byte_count_(0) {}
  46
  47 DataChannel::PacketQueue::~PacketQueue() {
  48   Clear();
  49 }
  50
  51 bool DataChannel::PacketQueue::Empty() const {
  52   return packets_.empty();
  53 }
  54
  55 DataBuffer* DataChannel::PacketQueue::Front() {
  56   return packets_.front();
  57 }
  58
  59 void DataChannel::PacketQueue::Pop() {
  60   if (packets_.empty()) {
  61     return;
  62   }
  63
  64   byte_count_ -= packets_.front()->size();
  65   packets_.pop_front();
  66 }
  67
  68 void DataChannel::PacketQueue::Push(DataBuffer* packet) {
  69   byte_count_ += packet->size();
  70   packets_.push_back(packet);
  71 }
  72
  73 void DataChannel::PacketQueue::Clear() {
  74   while (!packets_.empty()) {
  75     delete packets_.front();
  76     packets_.pop_front();
  77   }
  78   byte_count_ = 0;
  79 }
  80
  81 void DataChannel::PacketQueue::Swap(PacketQueue* other) {
  82   size_t other_byte_count = other->byte_count_;
  83   other->byte_count_ = byte_count_;
  84   byte_count_ = other_byte_count;
  85
  86   other->packets_.swap(packets_);
  87 }
  88
  89 rtc::scoped_refptr<DataChannel> DataChannel::Create(
  90     DataChannelProviderInterface* provider,
  91     cricket::DataChannelType dct,
  92     const std::string& label,
  93     const InternalDataChannelInit& config) {
  94   rtc::scoped_refptr<DataChannel> channel(
  95       new rtc::RefCountedObject<DataChannel>(provider, dct, label));
  96   if (!channel->Init(config)) {
  97     return NULL;
  98   }
  99   return channel;
 100 }
 101
 102 DataChannel::DataChannel(
 103     DataChannelProviderInterface* provider,
 104     cricket::DataChannelType dct,
 105     const std::string& label)
 106     : label_(label),
 107       observer_(NULL),
 108       state_(kConnecting),
 109       data_channel_type_(dct),
 110       provider_(provider),
 111       waiting_for_open_ack_(false),
 112       was_ever_writable_(false),
 113       connected_to_provider_(false),
 114       send_ssrc_set_(false),
 115       receive_ssrc_set_(false),
 116       send_ssrc_(0),
 117       receive_ssrc_(0) {
 118 }
 119
 120 bool DataChannel::Init(const InternalDataChannelInit& config) {
 121   if (data_channel_type_ == cricket::DCT_RTP &&
 122       (config.reliable ||
 123        config.id != -1 ||
 124        config.maxRetransmits != -1 ||
 125        config.maxRetransmitTime != -1)) {
 126     LOG(LS_ERROR) << "Failed to initialize the RTP data channel due to "
 127                   << "invalid DataChannelInit.";
 128     return false;
 129   } else if (data_channel_type_ == cricket::DCT_SCTP) {
 130     if (config.id < -1 ||
 131         config.maxRetransmits < -1 ||
 132         config.maxRetransmitTime < -1) {
 133       LOG(LS_ERROR) << "Failed to initialize the SCTP data channel due to "
 134                     << "invalid DataChannelInit.";
 135       return false;
 136     }
 137     if (config.maxRetransmits != -1 && config.maxRetransmitTime != -1) {
 138       LOG(LS_ERROR) <<
 139           "maxRetransmits and maxRetransmitTime should not be both set.";
 140       return false;
 141     }
 142     config_ = config;
 143
 144     // Try to connect to the transport in case the transport channel already
 145     // exists.
 146     OnTransportChannelCreated();
 147
 148     // Checks if the transport is ready to send because the initial channel
 149     // ready signal may have been sent before the DataChannel creation.
 150     // This has to be done async because the upper layer objects (e.g.
 151     // Chrome glue and WebKit) are not wired up properly until after this
 152     // function returns.
 153     if (provider_->ReadyToSendData()) {
 154       rtc::Thread::Current()->Post(this, MSG_CHANNELREADY, NULL);
 155     }
 156   }
 157
 158   return true;
 159 }
 160
 161 DataChannel::~DataChannel() {}
 162
 163 void DataChannel::RegisterObserver(DataChannelObserver* observer) {
 164   observer_ = observer;
 165   DeliverQueuedReceivedData();
 166 }
 167
 168 void DataChannel::UnregisterObserver() {
 169   observer_ = NULL;
 170 }
 171
 172 bool DataChannel::reliable() const {
 173   if (data_channel_type_ == cricket::DCT_RTP) {
 174     return false;
 175   } else {
 176     return config_.maxRetransmits == -1 &&
 177            config_.maxRetransmitTime == -1;
 178   }
 179 }
 180
 181 uint64 DataChannel::buffered_amount() const {
 182   return queued_send_data_.byte_count();
 183 }
 184
 185 void DataChannel::Close() {
 186   if (state_ == kClosed)
 187     return;
 188   send_ssrc_ = 0;
 189   send_ssrc_set_ = false;
 190   SetState(kClosing);
 191   UpdateState();
 192 }
 193
 194 bool DataChannel::Send(const DataBuffer& buffer) {
 195   if (state_ != kOpen) {
 196     return false;
 197   }
 198
 199   // TODO(jiayl): the spec is unclear about if the remote side should get the
 200   // onmessage event. We need to figure out the expected behavior and change the
 201   // code accordingly.
 202   if (buffer.size() == 0) {
 203     return true;
 204   }
 205
 206   // If the queue is non-empty, we're waiting for SignalReadyToSend,
 207   // so just add to the end of the queue and keep waiting.
 208   if (!queued_send_data_.Empty()) {
 209     // Only SCTP DataChannel queues the outgoing data when the transport is
 210     // blocked.
 211     ASSERT(data_channel_type_ == cricket::DCT_SCTP);
 212     if (!QueueSendDataMessage(buffer)) {
 213       Close();
 214     }
 215     return true;
 216   }
 217
 218   bool success = SendDataMessage(buffer);
 219   if (data_channel_type_ == cricket::DCT_RTP) {
 220     return success;
 221   }
 222
 223   // Always return true for SCTP DataChannel per the spec.
 224   return true;
 225 }
 226
 227 void DataChannel::SetReceiveSsrc(uint32 receive_ssrc) {
 228   ASSERT(data_channel_type_ == cricket::DCT_RTP);
 229
 230   if (receive_ssrc_set_) {
 231     return;
 232   }
 233   receive_ssrc_ = receive_ssrc;
 234   receive_ssrc_set_ = true;
 235   UpdateState();
 236 }
 237
 238 // The remote peer request that this channel shall be closed.
 239 void DataChannel::RemotePeerRequestClose() {
 240   DoClose();
 241 }
 242
 243 void DataChannel::SetSctpSid(int sid) {
 244   ASSERT(config_.id < 0 && sid >= 0 && data_channel_type_ == cricket::DCT_SCTP);
 245   if (config_.id == sid)
 246     return;
 247
 248   config_.id = sid;
 249   provider_->AddSctpDataStream(sid);
 250 }
 251
 252 void DataChannel::OnTransportChannelCreated() {
 253   ASSERT(data_channel_type_ == cricket::DCT_SCTP);
 254   if (!connected_to_provider_) {
 255     connected_to_provider_ = provider_->ConnectDataChannel(this);
 256   }
 257   // The sid may have been unassigned when provider_->ConnectDataChannel was
 258   // done. So always add the streams even if connected_to_provider_ is true.
 259   if (config_.id >= 0) {
 260     provider_->AddSctpDataStream(config_.id);
 261   }
 262 }
 263
 264 void DataChannel::SetSendSsrc(uint32 send_ssrc) {
 265   ASSERT(data_channel_type_ == cricket::DCT_RTP);
 266   if (send_ssrc_set_) {
 267     return;
 268   }
 269   send_ssrc_ = send_ssrc;
 270   send_ssrc_set_ = true;
 271   UpdateState();
 272 }
 273
 274 void DataChannel::OnMessage(rtc::Message* msg) {
 275   switch (msg->message_id) {
 276     case MSG_CHANNELREADY:
 277       OnChannelReady(true);
 278       break;
 279   }
 280 }
 281
 282 // The underlaying data engine is closing.
 283 // This function makes sure the DataChannel is disconnected and changes state to
 284 // kClosed.
 285 void DataChannel::OnDataEngineClose() {
 286   DoClose();
 287 }
 288
 289 void DataChannel::OnDataReceived(cricket::DataChannel* channel,
 290                                  const cricket::ReceiveDataParams& params,
 291                                  const rtc::Buffer& payload) {
 292   uint32 expected_ssrc =
 293       (data_channel_type_ == cricket::DCT_RTP) ? receive_ssrc_ : config_.id;
 294   if (params.ssrc != expected_ssrc) {
 295     return;
 296   }
 297
 298   if (params.type == cricket::DMT_CONTROL) {
 299     ASSERT(data_channel_type_ == cricket::DCT_SCTP);
 300     if (!waiting_for_open_ack_) {
 301       // Ignore it if we are not expecting an ACK message.
 302       LOG(LS_WARNING) << "DataChannel received unexpected CONTROL message, "
 303                       << "sid = " << params.ssrc;
 304       return;
 305     }
 306     if (ParseDataChannelOpenAckMessage(payload)) {
 307       // We can send unordered as soon as we receive the ACK message.
 308       waiting_for_open_ack_ = false;
 309       LOG(LS_INFO) << "DataChannel received OPEN_ACK message, sid = "
 310                    << params.ssrc;
 311     } else {
 312       LOG(LS_WARNING) << "DataChannel failed to parse OPEN_ACK message, sid = "
 313                       << params.ssrc;
 314     }
 315     return;
 316   }
 317
 318   ASSERT(params.type == cricket::DMT_BINARY ||
 319          params.type == cricket::DMT_TEXT);
 320
 321   LOG(LS_VERBOSE) << "DataChannel received DATA message, sid = " << params.ssrc;
 322   // We can send unordered as soon as we receive any DATA message since the
 323   // remote side must have received the OPEN (and old clients do not send
 324   // OPEN_ACK).
 325   waiting_for_open_ack_ = false;
 326
 327   bool binary = (params.type == cricket::DMT_BINARY);
 328   rtc::scoped_ptr<DataBuffer> buffer(new DataBuffer(payload, binary));
 329   if (was_ever_writable_ && observer_) {
 330     observer_->OnMessage(*buffer.get());
 331   } else {
 332     if (queued_received_data_.byte_count() + payload.length() >
 333             kMaxQueuedReceivedDataBytes) {
 334       LOG(LS_ERROR) << "Queued received data exceeds the max buffer size.";
 335
 336       queued_received_data_.Clear();
 337       if (data_channel_type_ != cricket::DCT_RTP) {
 338         Close();
 339       }
 340
 341       return;
 342     }
 343     queued_received_data_.Push(buffer.release());
 344   }
 345 }
 346
 347 void DataChannel::OnChannelReady(bool writable) {
 348   if (!writable) {
 349     return;
 350   }
 351   // Update the readyState and send the queued control message if the channel
 352   // is writable for the first time; otherwise it means the channel was blocked
 353   // for sending and now unblocked, so send the queued data now.
 354   if (!was_ever_writable_) {
 355     was_ever_writable_ = true;
 356
 357     if (data_channel_type_ == cricket::DCT_SCTP) {
 358       rtc::Buffer payload;
 359
 360       if (config_.open_handshake_role == InternalDataChannelInit::kOpener) {
 361         WriteDataChannelOpenMessage(label_, config_, &payload);
 362         SendControlMessage(payload);
 363       } else if (config_.open_handshake_role ==
 364                      InternalDataChannelInit::kAcker) {
 365         WriteDataChannelOpenAckMessage(&payload);
 366         SendControlMessage(payload);
 367       }
 368     }
 369
 370     UpdateState();
 371     ASSERT(queued_send_data_.Empty());
 372   } else if (state_ == kOpen) {
 373     // TODO(jiayl): Sending OPEN message here contradicts with the pre-condition
 374     // that the readyState is open. According to the standard, the channel
 375     // should not become open before the OPEN message is sent.
 376     SendQueuedControlMessages();
 377
 378     SendQueuedDataMessages();
 379   }
 380 }
 381
 382 void DataChannel::DoClose() {
 383   if (state_ == kClosed)
 384     return;
 385
 386   receive_ssrc_set_ = false;
 387   send_ssrc_set_ = false;
 388   SetState(kClosing);
 389   UpdateState();
 390 }
 391
 392 void DataChannel::UpdateState() {
 393   switch (state_) {
 394     case kConnecting: {
 395       if (send_ssrc_set_ == receive_ssrc_set_) {
 396         if (data_channel_type_ == cricket::DCT_RTP && !connected_to_provider_) {
 397           connected_to_provider_ = provider_->ConnectDataChannel(this);
 398         }
 399         if (was_ever_writable_) {
 400           // TODO(jiayl): Do not transition to kOpen if we failed to send the
 401           // OPEN message.
 402           SendQueuedControlMessages();
 403           SetState(kOpen);
 404           // If we have received buffers before the channel got writable.
 405           // Deliver them now.
 406           DeliverQueuedReceivedData();
 407         }
 408       }
 409       break;
 410     }
 411     case kOpen: {
 412       break;
 413     }
 414     case kClosing: {
 415       DisconnectFromTransport();
 416
 417       if (!send_ssrc_set_ && !receive_ssrc_set_) {
 418         SetState(kClosed);
 419       }
 420       break;
 421     }
 422     case kClosed:
 423       break;
 424   }
 425 }
 426
 427 void DataChannel::SetState(DataState state) {
 428   if (state_ == state)
 429     return;
 430
 431   state_ = state;
 432   if (observer_) {
 433     observer_->OnStateChange();
 434   }
 435 }
 436
 437 void DataChannel::DisconnectFromTransport() {
 438   if (!connected_to_provider_)
 439     return;
 440
 441   provider_->DisconnectDataChannel(this);
 442   connected_to_provider_ = false;
 443
 444   if (data_channel_type_ == cricket::DCT_SCTP) {
 445     provider_->RemoveSctpDataStream(config_.id);
 446   }
 447 }
 448
 449 void DataChannel::DeliverQueuedReceivedData() {
 450   if (!was_ever_writable_ || !observer_) {
 451     return;
 452   }
 453
 454   while (!queued_received_data_.Empty()) {
 455     rtc::scoped_ptr<DataBuffer> buffer(queued_received_data_.Front());
 456     observer_->OnMessage(*buffer);
 457     queued_received_data_.Pop();
 458   }
 459 }
 460
 461 void DataChannel::SendQueuedDataMessages() {
 462   ASSERT(was_ever_writable_ && state_ == kOpen);
 463
 464   PacketQueue packet_buffer;
 465   packet_buffer.Swap(&queued_send_data_);
 466
 467   while (!packet_buffer.Empty()) {
 468     rtc::scoped_ptr<DataBuffer> buffer(packet_buffer.Front());
 469     SendDataMessage(*buffer);
 470     packet_buffer.Pop();
 471   }
 472 }
 473
 474 bool DataChannel::SendDataMessage(const DataBuffer& buffer) {
 475   cricket::SendDataParams send_params;
 476
 477   if (data_channel_type_ == cricket::DCT_SCTP) {
 478     send_params.ordered = config_.ordered;
 479     // Send as ordered if it is waiting for the OPEN_ACK message.
 480     if (waiting_for_open_ack_ && !config_.ordered) {
 481       send_params.ordered = true;
 482       LOG(LS_VERBOSE) << "Sending data as ordered for unordered DataChannel "
 483                       << "because the OPEN_ACK message has not been received.";
 484     }
 485
 486     send_params.max_rtx_count = config_.maxRetransmits;
 487     send_params.max_rtx_ms = config_.maxRetransmitTime;
 488     send_params.ssrc = config_.id;
 489   } else {
 490     send_params.ssrc = send_ssrc_;
 491   }
 492   send_params.type = buffer.binary ? cricket::DMT_BINARY : cricket::DMT_TEXT;
 493
 494   cricket::SendDataResult send_result = cricket::SDR_SUCCESS;
 495   bool success = provider_->SendData(send_params, buffer.data, &send_result);
 496
 497   if (!success && data_channel_type_ == cricket::DCT_SCTP) {
 498     if (send_result != cricket::SDR_BLOCK || !QueueSendDataMessage(buffer)) {
 499       LOG(LS_ERROR) << "Closing the DataChannel due to a failure to send data, "
 500                     << "send_result = " << send_result;
 501       Close();
 502     }
 503   }
 504   return success;
 505 }
 506
 507 bool DataChannel::QueueSendDataMessage(const DataBuffer& buffer) {
 508   if (queued_send_data_.byte_count() >= kMaxQueuedSendDataBytes) {
 509     LOG(LS_ERROR) << "Can't buffer any more data for the data channel.";
 510     return false;
 511   }
 512   queued_send_data_.Push(new DataBuffer(buffer));
 513   return true;
 514 }
 515
 516 void DataChannel::SendQueuedControlMessages() {
 517   ASSERT(was_ever_writable_);
 518
 519   PacketQueue control_packets;
 520   control_packets.Swap(&queued_control_data_);
 521
 522   while (!control_packets.Empty()) {
 523     rtc::scoped_ptr<DataBuffer> buf(control_packets.Front());
 524     SendControlMessage(buf->data);
 525     control_packets.Pop();
 526   }
 527 }
 528
 529 void DataChannel::QueueControlMessage(const rtc::Buffer& buffer) {
 530   queued_control_data_.Push(new DataBuffer(buffer, true));
 531 }
 532
 533 bool DataChannel::SendControlMessage(const rtc::Buffer& buffer) {
 534   bool is_open_message =
 535       (config_.open_handshake_role == InternalDataChannelInit::kOpener);
 536
 537   ASSERT(data_channel_type_ == cricket::DCT_SCTP &&
 538          was_ever_writable_ &&
 539          config_.id >= 0 &&
 540          (!is_open_message || !config_.negotiated));
 541
 542   cricket::SendDataParams send_params;
 543   send_params.ssrc = config_.id;
 544   send_params.ordered = config_.ordered || is_open_message;
 545   send_params.type = cricket::DMT_CONTROL;
 546
 547   cricket::SendDataResult send_result = cricket::SDR_SUCCESS;
 548   bool retval = provider_->SendData(send_params, buffer, &send_result);
 549   if (retval) {
 550     LOG(LS_INFO) << "Sent CONTROL message on channel " << config_.id;
 551
 552     if (is_open_message) {
 553       // Send data as ordered before we receive any message from the remote peer
 554       // to make sure the remote peer will not receive any data before it
 555       // receives the OPEN message.
 556       waiting_for_open_ack_ = true;
 557     }
 558   } else if (send_result == cricket::SDR_BLOCK) {
 559     QueueControlMessage(buffer);
 560   } else {
 561     LOG(LS_ERROR) << "Closing the DataChannel due to a failure to send"
 562                   << " the CONTROL message, send_result = " << send_result;
 563     Close();
 564   }
 565   return retval;
 566 }
 567
 568 }  // namespace webrtc