Home | History | Annotate | Download | only in webrtc 
      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
    569