9f8164c060
1. OnStateChange should not be fired if state is not changed. 2. RemotePeerRequestClose should be a no-op if it's already closed. TBR=pthatcher@webrtc.org BUG= Review URL: https://webrtc-codereview.appspot.com/21559004 git-svn-id: http://webrtc.googlecode.com/svn/trunk@6290 4adac7df-926f-26a2-2b94-8c16560cd09d
563 lines
17 KiB
C++
563 lines
17 KiB
C++
/*
|
|
* libjingle
|
|
* Copyright 2012, Google Inc.
|
|
*
|
|
* Redistribution and use in source and binary forms, with or without
|
|
* modification, are permitted provided that the following conditions are met:
|
|
*
|
|
* 1. Redistributions of source code must retain the above copyright notice,
|
|
* this list of conditions and the following disclaimer.
|
|
* 2. Redistributions in binary form must reproduce the above copyright notice,
|
|
* this list of conditions and the following disclaimer in the documentation
|
|
* and/or other materials provided with the distribution.
|
|
* 3. The name of the author may not be used to endorse or promote products
|
|
* derived from this software without specific prior written permission.
|
|
*
|
|
* THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR IMPLIED
|
|
* WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
|
|
* MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO
|
|
* EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
|
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
|
|
* PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
|
|
* OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
|
|
* WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
|
|
* OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
|
|
* ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
|
*/
|
|
#include "talk/app/webrtc/datachannel.h"
|
|
|
|
#include <string>
|
|
|
|
#include "talk/app/webrtc/mediastreamprovider.h"
|
|
#include "talk/app/webrtc/sctputils.h"
|
|
#include "talk/base/logging.h"
|
|
#include "talk/base/refcount.h"
|
|
|
|
namespace webrtc {
|
|
|
|
static size_t kMaxQueuedReceivedDataPackets = 100;
|
|
static size_t kMaxQueuedSendDataPackets = 100;
|
|
|
|
enum {
|
|
MSG_CHANNELREADY,
|
|
};
|
|
|
|
talk_base::scoped_refptr<DataChannel> DataChannel::Create(
|
|
DataChannelProviderInterface* provider,
|
|
cricket::DataChannelType dct,
|
|
const std::string& label,
|
|
const InternalDataChannelInit& config) {
|
|
talk_base::scoped_refptr<DataChannel> channel(
|
|
new talk_base::RefCountedObject<DataChannel>(provider, dct, label));
|
|
if (!channel->Init(config)) {
|
|
return NULL;
|
|
}
|
|
return channel;
|
|
}
|
|
|
|
DataChannel::DataChannel(
|
|
DataChannelProviderInterface* provider,
|
|
cricket::DataChannelType dct,
|
|
const std::string& label)
|
|
: label_(label),
|
|
observer_(NULL),
|
|
state_(kConnecting),
|
|
data_channel_type_(dct),
|
|
provider_(provider),
|
|
waiting_for_open_ack_(false),
|
|
was_ever_writable_(false),
|
|
connected_to_provider_(false),
|
|
send_ssrc_set_(false),
|
|
receive_ssrc_set_(false),
|
|
send_ssrc_(0),
|
|
receive_ssrc_(0) {
|
|
}
|
|
|
|
bool DataChannel::Init(const InternalDataChannelInit& config) {
|
|
if (data_channel_type_ == cricket::DCT_RTP &&
|
|
(config.reliable ||
|
|
config.id != -1 ||
|
|
config.maxRetransmits != -1 ||
|
|
config.maxRetransmitTime != -1)) {
|
|
LOG(LS_ERROR) << "Failed to initialize the RTP data channel due to "
|
|
<< "invalid DataChannelInit.";
|
|
return false;
|
|
} else if (data_channel_type_ == cricket::DCT_SCTP) {
|
|
if (config.id < -1 ||
|
|
config.maxRetransmits < -1 ||
|
|
config.maxRetransmitTime < -1) {
|
|
LOG(LS_ERROR) << "Failed to initialize the SCTP data channel due to "
|
|
<< "invalid DataChannelInit.";
|
|
return false;
|
|
}
|
|
if (config.maxRetransmits != -1 && config.maxRetransmitTime != -1) {
|
|
LOG(LS_ERROR) <<
|
|
"maxRetransmits and maxRetransmitTime should not be both set.";
|
|
return false;
|
|
}
|
|
config_ = config;
|
|
|
|
// Try to connect to the transport in case the transport channel already
|
|
// exists.
|
|
OnTransportChannelCreated();
|
|
|
|
// Checks if the transport is ready to send because the initial channel
|
|
// ready signal may have been sent before the DataChannel creation.
|
|
// This has to be done async because the upper layer objects (e.g.
|
|
// Chrome glue and WebKit) are not wired up properly until after this
|
|
// function returns.
|
|
if (provider_->ReadyToSendData()) {
|
|
talk_base::Thread::Current()->Post(this, MSG_CHANNELREADY, NULL);
|
|
}
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
DataChannel::~DataChannel() {
|
|
ClearQueuedReceivedData();
|
|
ClearQueuedSendData();
|
|
ClearQueuedControlData();
|
|
}
|
|
|
|
void DataChannel::RegisterObserver(DataChannelObserver* observer) {
|
|
observer_ = observer;
|
|
DeliverQueuedReceivedData();
|
|
}
|
|
|
|
void DataChannel::UnregisterObserver() {
|
|
observer_ = NULL;
|
|
}
|
|
|
|
bool DataChannel::reliable() const {
|
|
if (data_channel_type_ == cricket::DCT_RTP) {
|
|
return false;
|
|
} else {
|
|
return config_.maxRetransmits == -1 &&
|
|
config_.maxRetransmitTime == -1;
|
|
}
|
|
}
|
|
|
|
uint64 DataChannel::buffered_amount() const {
|
|
uint64 buffered_amount = 0;
|
|
for (std::deque<DataBuffer*>::const_iterator it = queued_send_data_.begin();
|
|
it != queued_send_data_.end();
|
|
++it) {
|
|
buffered_amount += (*it)->size();
|
|
}
|
|
return buffered_amount;
|
|
}
|
|
|
|
void DataChannel::Close() {
|
|
if (state_ == kClosed)
|
|
return;
|
|
send_ssrc_ = 0;
|
|
send_ssrc_set_ = false;
|
|
SetState(kClosing);
|
|
UpdateState();
|
|
}
|
|
|
|
bool DataChannel::Send(const DataBuffer& buffer) {
|
|
if (state_ != kOpen) {
|
|
return false;
|
|
}
|
|
// If the queue is non-empty, we're waiting for SignalReadyToSend,
|
|
// so just add to the end of the queue and keep waiting.
|
|
if (!queued_send_data_.empty()) {
|
|
if (!QueueSendData(buffer)) {
|
|
if (data_channel_type_ == cricket::DCT_RTP) {
|
|
return false;
|
|
}
|
|
Close();
|
|
}
|
|
return true;
|
|
}
|
|
|
|
cricket::SendDataResult send_result;
|
|
if (!InternalSendWithoutQueueing(buffer, &send_result)) {
|
|
if (data_channel_type_ == cricket::DCT_RTP) {
|
|
return false;
|
|
}
|
|
if (send_result != cricket::SDR_BLOCK || !QueueSendData(buffer)) {
|
|
Close();
|
|
}
|
|
}
|
|
return true;
|
|
}
|
|
|
|
void DataChannel::QueueControl(const talk_base::Buffer* buffer) {
|
|
queued_control_data_.push(buffer);
|
|
}
|
|
|
|
bool DataChannel::SendOpenMessage(const talk_base::Buffer* raw_buffer) {
|
|
ASSERT(data_channel_type_ == cricket::DCT_SCTP &&
|
|
was_ever_writable_ &&
|
|
config_.id >= 0 &&
|
|
!config_.negotiated);
|
|
|
|
talk_base::scoped_ptr<const talk_base::Buffer> buffer(raw_buffer);
|
|
|
|
cricket::SendDataParams send_params;
|
|
send_params.ssrc = config_.id;
|
|
send_params.ordered = true;
|
|
send_params.type = cricket::DMT_CONTROL;
|
|
|
|
cricket::SendDataResult send_result;
|
|
bool retval = provider_->SendData(send_params, *buffer, &send_result);
|
|
if (retval) {
|
|
LOG(LS_INFO) << "Sent OPEN message on channel " << config_.id;
|
|
// Send data as ordered before we receive any mesage from the remote peer
|
|
// to make sure the remote peer will not receive any data before it receives
|
|
// the OPEN message.
|
|
waiting_for_open_ack_ = true;
|
|
} else if (send_result == cricket::SDR_BLOCK) {
|
|
// Link is congested. Queue for later.
|
|
QueueControl(buffer.release());
|
|
} else {
|
|
LOG(LS_ERROR) << "Failed to send OPEN message with result "
|
|
<< send_result << " on channel " << config_.id;
|
|
}
|
|
return retval;
|
|
}
|
|
|
|
bool DataChannel::SendOpenAckMessage(const talk_base::Buffer* raw_buffer) {
|
|
ASSERT(data_channel_type_ == cricket::DCT_SCTP &&
|
|
was_ever_writable_ &&
|
|
config_.id >= 0);
|
|
|
|
talk_base::scoped_ptr<const talk_base::Buffer> buffer(raw_buffer);
|
|
|
|
cricket::SendDataParams send_params;
|
|
send_params.ssrc = config_.id;
|
|
send_params.ordered = config_.ordered;
|
|
send_params.type = cricket::DMT_CONTROL;
|
|
|
|
cricket::SendDataResult send_result;
|
|
bool retval = provider_->SendData(send_params, *buffer, &send_result);
|
|
if (retval) {
|
|
LOG(LS_INFO) << "Sent OPEN_ACK message on channel " << config_.id;
|
|
} else if (send_result == cricket::SDR_BLOCK) {
|
|
// Link is congested. Queue for later.
|
|
QueueControl(buffer.release());
|
|
} else {
|
|
LOG(LS_ERROR) << "Failed to send OPEN_ACK message with result "
|
|
<< send_result << " on channel " << config_.id;
|
|
}
|
|
return retval;
|
|
}
|
|
|
|
void DataChannel::SetReceiveSsrc(uint32 receive_ssrc) {
|
|
ASSERT(data_channel_type_ == cricket::DCT_RTP);
|
|
|
|
if (receive_ssrc_set_) {
|
|
return;
|
|
}
|
|
receive_ssrc_ = receive_ssrc;
|
|
receive_ssrc_set_ = true;
|
|
UpdateState();
|
|
}
|
|
|
|
// The remote peer request that this channel shall be closed.
|
|
void DataChannel::RemotePeerRequestClose() {
|
|
DoClose();
|
|
}
|
|
|
|
void DataChannel::SetSendSsrc(uint32 send_ssrc) {
|
|
ASSERT(data_channel_type_ == cricket::DCT_RTP);
|
|
if (send_ssrc_set_) {
|
|
return;
|
|
}
|
|
send_ssrc_ = send_ssrc;
|
|
send_ssrc_set_ = true;
|
|
UpdateState();
|
|
}
|
|
|
|
void DataChannel::OnMessage(talk_base::Message* msg) {
|
|
switch (msg->message_id) {
|
|
case MSG_CHANNELREADY:
|
|
OnChannelReady(true);
|
|
break;
|
|
}
|
|
}
|
|
|
|
// The underlaying data engine is closing.
|
|
// This function makes sure the DataChannel is disconnected and changes state to
|
|
// kClosed.
|
|
void DataChannel::OnDataEngineClose() {
|
|
DoClose();
|
|
}
|
|
|
|
void DataChannel::OnDataReceived(cricket::DataChannel* channel,
|
|
const cricket::ReceiveDataParams& params,
|
|
const talk_base::Buffer& payload) {
|
|
uint32 expected_ssrc =
|
|
(data_channel_type_ == cricket::DCT_RTP) ? receive_ssrc_ : config_.id;
|
|
if (params.ssrc != expected_ssrc) {
|
|
return;
|
|
}
|
|
|
|
if (params.type == cricket::DMT_CONTROL) {
|
|
ASSERT(data_channel_type_ == cricket::DCT_SCTP);
|
|
if (!waiting_for_open_ack_) {
|
|
// Ignore it if we are not expecting an ACK message.
|
|
LOG(LS_WARNING) << "DataChannel received unexpected CONTROL message, "
|
|
<< "sid = " << params.ssrc;
|
|
return;
|
|
}
|
|
if (ParseDataChannelOpenAckMessage(payload)) {
|
|
// We can send unordered as soon as we receive the ACK message.
|
|
waiting_for_open_ack_ = false;
|
|
LOG(LS_INFO) << "DataChannel received OPEN_ACK message, sid = "
|
|
<< params.ssrc;
|
|
} else {
|
|
LOG(LS_WARNING) << "DataChannel failed to parse OPEN_ACK message, sid = "
|
|
<< params.ssrc;
|
|
}
|
|
return;
|
|
}
|
|
|
|
ASSERT(params.type == cricket::DMT_BINARY ||
|
|
params.type == cricket::DMT_TEXT);
|
|
|
|
LOG(LS_VERBOSE) << "DataChannel received DATA message, sid = " << params.ssrc;
|
|
// We can send unordered as soon as we receive any DATA message since the
|
|
// remote side must have received the OPEN (and old clients do not send
|
|
// OPEN_ACK).
|
|
waiting_for_open_ack_ = false;
|
|
|
|
bool binary = (params.type == cricket::DMT_BINARY);
|
|
talk_base::scoped_ptr<DataBuffer> buffer(new DataBuffer(payload, binary));
|
|
if (was_ever_writable_ && observer_) {
|
|
observer_->OnMessage(*buffer.get());
|
|
} else {
|
|
if (queued_received_data_.size() > kMaxQueuedReceivedDataPackets) {
|
|
LOG(LS_ERROR)
|
|
<< "Queued received data exceeds the max number of packets.";
|
|
ClearQueuedReceivedData();
|
|
}
|
|
queued_received_data_.push(buffer.release());
|
|
}
|
|
}
|
|
|
|
void DataChannel::OnChannelReady(bool writable) {
|
|
if (!writable) {
|
|
return;
|
|
}
|
|
// Update the readyState and send the queued control message if the channel
|
|
// is writable for the first time; otherwise it means the channel was blocked
|
|
// for sending and now unblocked, so send the queued data now.
|
|
if (!was_ever_writable_) {
|
|
was_ever_writable_ = true;
|
|
|
|
if (data_channel_type_ == cricket::DCT_SCTP) {
|
|
if (config_.open_handshake_role == InternalDataChannelInit::kOpener) {
|
|
talk_base::Buffer* payload = new talk_base::Buffer;
|
|
WriteDataChannelOpenMessage(label_, config_, payload);
|
|
SendOpenMessage(payload);
|
|
} else if (config_.open_handshake_role ==
|
|
InternalDataChannelInit::kAcker) {
|
|
talk_base::Buffer* payload = new talk_base::Buffer;
|
|
WriteDataChannelOpenAckMessage(payload);
|
|
SendOpenAckMessage(payload);
|
|
}
|
|
}
|
|
|
|
UpdateState();
|
|
ASSERT(queued_send_data_.empty());
|
|
} else if (state_ == kOpen) {
|
|
DeliverQueuedSendData();
|
|
}
|
|
}
|
|
|
|
void DataChannel::DoClose() {
|
|
if (state_ == kClosed)
|
|
return;
|
|
|
|
receive_ssrc_set_ = false;
|
|
send_ssrc_set_ = false;
|
|
SetState(kClosing);
|
|
UpdateState();
|
|
}
|
|
|
|
void DataChannel::UpdateState() {
|
|
switch (state_) {
|
|
case kConnecting: {
|
|
if (send_ssrc_set_ == receive_ssrc_set_) {
|
|
if (data_channel_type_ == cricket::DCT_RTP && !connected_to_provider_) {
|
|
connected_to_provider_ = provider_->ConnectDataChannel(this);
|
|
}
|
|
if (was_ever_writable_) {
|
|
// TODO(jiayl): Do not transition to kOpen if we failed to send the
|
|
// OPEN message.
|
|
DeliverQueuedControlData();
|
|
SetState(kOpen);
|
|
// If we have received buffers before the channel got writable.
|
|
// Deliver them now.
|
|
DeliverQueuedReceivedData();
|
|
}
|
|
}
|
|
break;
|
|
}
|
|
case kOpen: {
|
|
break;
|
|
}
|
|
case kClosing: {
|
|
DisconnectFromTransport();
|
|
|
|
if (!send_ssrc_set_ && !receive_ssrc_set_) {
|
|
SetState(kClosed);
|
|
}
|
|
break;
|
|
}
|
|
case kClosed:
|
|
break;
|
|
}
|
|
}
|
|
|
|
void DataChannel::SetState(DataState state) {
|
|
if (state_ == state)
|
|
return;
|
|
|
|
state_ = state;
|
|
if (observer_) {
|
|
observer_->OnStateChange();
|
|
}
|
|
}
|
|
|
|
void DataChannel::DisconnectFromTransport() {
|
|
if (!connected_to_provider_)
|
|
return;
|
|
|
|
provider_->DisconnectDataChannel(this);
|
|
connected_to_provider_ = false;
|
|
|
|
if (data_channel_type_ == cricket::DCT_SCTP) {
|
|
provider_->RemoveSctpDataStream(config_.id);
|
|
}
|
|
}
|
|
|
|
void DataChannel::DeliverQueuedReceivedData() {
|
|
if (!was_ever_writable_ || !observer_) {
|
|
return;
|
|
}
|
|
|
|
while (!queued_received_data_.empty()) {
|
|
DataBuffer* buffer = queued_received_data_.front();
|
|
observer_->OnMessage(*buffer);
|
|
queued_received_data_.pop();
|
|
delete buffer;
|
|
}
|
|
}
|
|
|
|
void DataChannel::ClearQueuedReceivedData() {
|
|
while (!queued_received_data_.empty()) {
|
|
DataBuffer* buffer = queued_received_data_.front();
|
|
queued_received_data_.pop();
|
|
delete buffer;
|
|
}
|
|
}
|
|
|
|
void DataChannel::DeliverQueuedSendData() {
|
|
ASSERT(was_ever_writable_ && state_ == kOpen);
|
|
|
|
// TODO(jiayl): Sending OPEN message here contradicts with the pre-condition
|
|
// that the readyState is open. According to the standard, the channel should
|
|
// not become open before the OPEN message is sent.
|
|
DeliverQueuedControlData();
|
|
|
|
while (!queued_send_data_.empty()) {
|
|
DataBuffer* buffer = queued_send_data_.front();
|
|
cricket::SendDataResult send_result;
|
|
if (!InternalSendWithoutQueueing(*buffer, &send_result)) {
|
|
LOG(LS_WARNING) << "DeliverQueuedSendData aborted due to send_result "
|
|
<< send_result;
|
|
break;
|
|
}
|
|
queued_send_data_.pop_front();
|
|
delete buffer;
|
|
}
|
|
}
|
|
|
|
void DataChannel::ClearQueuedControlData() {
|
|
while (!queued_control_data_.empty()) {
|
|
const talk_base::Buffer *buf = queued_control_data_.front();
|
|
queued_control_data_.pop();
|
|
delete buf;
|
|
}
|
|
}
|
|
|
|
void DataChannel::DeliverQueuedControlData() {
|
|
ASSERT(was_ever_writable_);
|
|
while (!queued_control_data_.empty()) {
|
|
const talk_base::Buffer* buf = queued_control_data_.front();
|
|
queued_control_data_.pop();
|
|
if (config_.open_handshake_role == InternalDataChannelInit::kOpener) {
|
|
SendOpenMessage(buf);
|
|
} else {
|
|
ASSERT(config_.open_handshake_role == InternalDataChannelInit::kAcker);
|
|
SendOpenAckMessage(buf);
|
|
}
|
|
}
|
|
}
|
|
|
|
void DataChannel::ClearQueuedSendData() {
|
|
while (!queued_send_data_.empty()) {
|
|
DataBuffer* buffer = queued_send_data_.front();
|
|
queued_send_data_.pop_front();
|
|
delete buffer;
|
|
}
|
|
}
|
|
|
|
bool DataChannel::InternalSendWithoutQueueing(
|
|
const DataBuffer& buffer, cricket::SendDataResult* send_result) {
|
|
cricket::SendDataParams send_params;
|
|
|
|
if (data_channel_type_ == cricket::DCT_SCTP) {
|
|
send_params.ordered = config_.ordered;
|
|
// Send as ordered if it is waiting for the OPEN_ACK message.
|
|
if (waiting_for_open_ack_ && !config_.ordered) {
|
|
send_params.ordered = true;
|
|
LOG(LS_VERBOSE) << "Sending data as ordered for unordered DataChannel "
|
|
<< "because the OPEN_ACK message has not been received.";
|
|
}
|
|
|
|
send_params.max_rtx_count = config_.maxRetransmits;
|
|
send_params.max_rtx_ms = config_.maxRetransmitTime;
|
|
send_params.ssrc = config_.id;
|
|
} else {
|
|
send_params.ssrc = send_ssrc_;
|
|
}
|
|
send_params.type = buffer.binary ? cricket::DMT_BINARY : cricket::DMT_TEXT;
|
|
|
|
return provider_->SendData(send_params, buffer.data, send_result);
|
|
}
|
|
|
|
bool DataChannel::QueueSendData(const DataBuffer& buffer) {
|
|
if (queued_send_data_.size() >= kMaxQueuedSendDataPackets) {
|
|
LOG(LS_ERROR) << "Can't buffer any more data for the data channel.";
|
|
return false;
|
|
}
|
|
queued_send_data_.push_back(new DataBuffer(buffer));
|
|
return true;
|
|
}
|
|
|
|
void DataChannel::SetSctpSid(int sid) {
|
|
ASSERT(config_.id < 0 && sid >= 0 && data_channel_type_ == cricket::DCT_SCTP);
|
|
config_.id = sid;
|
|
provider_->AddSctpDataStream(sid);
|
|
}
|
|
|
|
void DataChannel::OnTransportChannelCreated() {
|
|
ASSERT(data_channel_type_ == cricket::DCT_SCTP);
|
|
if (!connected_to_provider_) {
|
|
connected_to_provider_ = provider_->ConnectDataChannel(this);
|
|
}
|
|
// The sid may have been unassigned when provider_->ConnectDataChannel was
|
|
// done. So always add the streams even if connected_to_provider_ is true.
|
|
if (config_.id >= 0) {
|
|
provider_->AddSctpDataStream(config_.id);
|
|
}
|
|
}
|
|
|
|
} // namespace webrtc
|