| OLD | NEW |
|---|
| (Empty) |
| 1 /* | |
| 2 * Copyright 2016 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 "webrtc/api/quicdatachannel.h" | |
| 12 | |
| 13 #include "webrtc/base/bind.h" | |
| 14 #include "webrtc/base/bytebuffer.h" | |
| 15 #include "webrtc/base/copyonwritebuffer.h" | |
| 16 #include "webrtc/base/logging.h" | |
| 17 #include "webrtc/p2p/quic/quictransportchannel.h" | |
| 18 #include "webrtc/p2p/quic/reliablequicstream.h" | |
| 19 | |
| 20 namespace webrtc { | |
| 21 | |
| 22 void WriteQuicDataChannelMessageHeader(int data_channel_id, | |
| 23 uint64_t message_id, | |
| 24 rtc::CopyOnWriteBuffer* header) { | |
| 25 RTC_DCHECK(header); | |
| 26 // 64-bit varints require at most 10 bytes (7*10 == 70), and 32-bit varints | |
| 27 // require at most 5 bytes (7*5 == 35). | |
| 28 size_t max_length = 15; | |
| 29 rtc::ByteBufferWriter byte_buffer(nullptr, max_length, | |
| 30 rtc::ByteBuffer::ByteOrder::ORDER_HOST); | |
| 31 byte_buffer.WriteUVarint(data_channel_id); | |
| 32 byte_buffer.WriteUVarint(message_id); | |
| 33 header->SetData(byte_buffer.Data(), byte_buffer.Length()); | |
| 34 } | |
| 35 | |
| 36 bool ParseQuicDataMessageHeader(const char* data, | |
| 37 size_t len, | |
| 38 int* data_channel_id, | |
| 39 uint64_t* message_id, | |
| 40 size_t* bytes_read) { | |
| 41 RTC_DCHECK(data_channel_id); | |
| 42 RTC_DCHECK(message_id); | |
| 43 RTC_DCHECK(bytes_read); | |
| 44 | |
| 45 rtc::ByteBufferReader byte_buffer(data, len, rtc::ByteBuffer::ORDER_HOST); | |
| 46 uint64_t dcid; | |
| 47 if (!byte_buffer.ReadUVarint(&dcid)) { | |
| 48 LOG(LS_ERROR) << "Could not read the data channel ID"; | |
| 49 return false; | |
| 50 } | |
| 51 *data_channel_id = dcid; | |
| 52 if (!byte_buffer.ReadUVarint(message_id)) { | |
| 53 LOG(LS_ERROR) << "Could not read message ID for data channel " | |
| 54 << *data_channel_id; | |
| 55 return false; | |
| 56 } | |
| 57 size_t remaining_bytes = byte_buffer.Length(); | |
| 58 *bytes_read = len - remaining_bytes; | |
| 59 return true; | |
| 60 } | |
| 61 | |
| 62 QuicDataChannel::QuicDataChannel(rtc::Thread* signaling_thread, | |
| 63 rtc::Thread* worker_thread, | |
| 64 rtc::Thread* network_thread, | |
| 65 const std::string& label, | |
| 66 const DataChannelInit& config) | |
| 67 : signaling_thread_(signaling_thread), | |
| 68 worker_thread_(worker_thread), | |
| 69 network_thread_(network_thread), | |
| 70 id_(config.id), | |
| 71 state_(kConnecting), | |
| 72 buffered_amount_(0), | |
| 73 next_message_id_(0), | |
| 74 label_(label), | |
| 75 protocol_(config.protocol) {} | |
| 76 | |
| 77 QuicDataChannel::~QuicDataChannel() {} | |
| 78 | |
| 79 void QuicDataChannel::RegisterObserver(DataChannelObserver* observer) { | |
| 80 RTC_DCHECK(signaling_thread_->IsCurrent()); | |
| 81 observer_ = observer; | |
| 82 } | |
| 83 | |
| 84 void QuicDataChannel::UnregisterObserver() { | |
| 85 RTC_DCHECK(signaling_thread_->IsCurrent()); | |
| 86 observer_ = nullptr; | |
| 87 } | |
| 88 | |
| 89 bool QuicDataChannel::Send(const DataBuffer& buffer) { | |
| 90 RTC_DCHECK(signaling_thread_->IsCurrent()); | |
| 91 if (state_ != kOpen) { | |
| 92 LOG(LS_ERROR) << "QUIC data channel " << id_ | |
| 93 << " is not open so cannot send."; | |
| 94 return false; | |
| 95 } | |
| 96 return network_thread_->Invoke<bool>( | |
| 97 RTC_FROM_HERE, rtc::Bind(&QuicDataChannel::Send_n, this, buffer)); | |
| 98 } | |
| 99 | |
| 100 bool QuicDataChannel::Send_n(const DataBuffer& buffer) { | |
| 101 RTC_DCHECK(network_thread_->IsCurrent()); | |
| 102 | |
| 103 // Encode and send the header containing the data channel ID and message ID. | |
| 104 rtc::CopyOnWriteBuffer header; | |
| 105 WriteQuicDataChannelMessageHeader(id_, ++next_message_id_, &header); | |
| 106 RTC_DCHECK(quic_transport_channel_); | |
| 107 cricket::ReliableQuicStream* stream = | |
| 108 quic_transport_channel_->CreateQuicStream(); | |
| 109 RTC_DCHECK(stream); | |
| 110 | |
| 111 // Send the header with a FIN if the message is empty. | |
| 112 bool header_fin = (buffer.size() == 0); | |
| 113 rtc::StreamResult header_result = | |
| 114 stream->Write(header.data<char>(), header.size(), header_fin); | |
| 115 | |
| 116 if (header_result == rtc::SR_BLOCK) { | |
| 117 // The header is write blocked but we should try sending the message. Since | |
| 118 // the ReliableQuicStream queues data in order, if the header is write | |
| 119 // blocked then the message will be write blocked. Otherwise if the message | |
| 120 // is sent then the header is sent. | |
| 121 LOG(LS_INFO) << "Stream " << stream->id() | |
| 122 << " header is write blocked for QUIC data channel " << id_; | |
| 123 } else if (header_result != rtc::SR_SUCCESS) { | |
| 124 LOG(LS_ERROR) << "Stream " << stream->id() | |
| 125 << " failed to write header for QUIC data channel " << id_ | |
| 126 << ". Unexpected error " << header_result; | |
| 127 return false; | |
| 128 } | |
| 129 | |
| 130 // If the message is not empty, then send the message with a FIN. | |
| 131 bool message_fin = true; | |
| 132 rtc::StreamResult message_result = | |
| 133 header_fin ? header_result : stream->Write(buffer.data.data<char>(), | |
| 134 buffer.size(), message_fin); | |
| 135 | |
| 136 if (message_result == rtc::SR_SUCCESS) { | |
| 137 // The message is sent and we don't need this QUIC stream. | |
| 138 LOG(LS_INFO) << "Stream " << stream->id() | |
| 139 << " successfully wrote message for QUIC data channel " << id_; | |
| 140 stream->Close(); | |
| 141 return true; | |
| 142 } | |
| 143 // TODO(mikescarlett): Register the ReliableQuicStream's priority to the | |
| 144 // QuicWriteBlockedList so that the QUIC session doesn't drop messages when | |
| 145 // the QUIC transport channel becomes unwritable. | |
| 146 if (message_result == rtc::SR_BLOCK) { | |
| 147 // The QUIC stream is write blocked, so the message is queued by the QUIC | |
| 148 // session. If this is due to the QUIC not being writable, it will be sent | |
| 149 // once QUIC becomes writable again. Otherwise it may be due to exceeding | |
| 150 // the QUIC flow control limit, in which case the remote peer's QUIC session | |
| 151 // will tell the QUIC stream to send more data. | |
| 152 LOG(LS_INFO) << "Stream " << stream->id() | |
| 153 << " message is write blocked for QUIC data channel " << id_; | |
| 154 SetBufferedAmount_w(buffered_amount_ + stream->queued_data_bytes()); | |
| 155 stream->SignalQueuedBytesWritten.connect( | |
| 156 this, &QuicDataChannel::OnQueuedBytesWritten); | |
| 157 write_blocked_quic_streams_[stream->id()] = stream; | |
| 158 // The QUIC stream will be removed from |write_blocked_quic_streams_| once | |
| 159 // it closes. | |
| 160 stream->SignalClosed.connect(this, | |
| 161 &QuicDataChannel::OnWriteBlockedStreamClosed); | |
| 162 return true; | |
| 163 } | |
| 164 LOG(LS_ERROR) << "Stream " << stream->id() | |
| 165 << " failed to write message for QUIC data channel " << id_ | |
| 166 << ". Unexpected error: " << message_result; | |
| 167 return false; | |
| 168 } | |
| 169 | |
| 170 void QuicDataChannel::OnQueuedBytesWritten(net::QuicStreamId stream_id, | |
| 171 uint64_t queued_bytes_written) { | |
| 172 RTC_DCHECK(worker_thread_->IsCurrent()); | |
| 173 SetBufferedAmount_w(buffered_amount_ - queued_bytes_written); | |
| 174 const auto& kv = write_blocked_quic_streams_.find(stream_id); | |
| 175 if (kv == write_blocked_quic_streams_.end()) { | |
| 176 RTC_NOTREACHED(); | |
| 177 return; | |
| 178 } | |
| 179 cricket::ReliableQuicStream* stream = kv->second; | |
| 180 // True if the QUIC stream is done sending data. | |
| 181 if (stream->fin_sent()) { | |
| 182 LOG(LS_INFO) << "Stream " << stream->id() | |
| 183 << " successfully wrote data for QUIC data channel " << id_; | |
| 184 stream->Close(); | |
| 185 } | |
| 186 } | |
| 187 | |
| 188 void QuicDataChannel::SetBufferedAmount_w(uint64_t buffered_amount) { | |
| 189 RTC_DCHECK(worker_thread_->IsCurrent()); | |
| 190 buffered_amount_ = buffered_amount; | |
| 191 invoker_.AsyncInvoke<void>( | |
| 192 RTC_FROM_HERE, signaling_thread_, | |
| 193 rtc::Bind(&QuicDataChannel::OnBufferedAmountChange_s, this, | |
| 194 buffered_amount)); | |
| 195 } | |
| 196 | |
| 197 void QuicDataChannel::Close() { | |
| 198 RTC_DCHECK(signaling_thread_->IsCurrent()); | |
| 199 if (state_ == kClosed || state_ == kClosing) { | |
| 200 return; | |
| 201 } | |
| 202 LOG(LS_INFO) << "Closing QUIC data channel."; | |
| 203 SetState_s(kClosing); | |
| 204 worker_thread_->Invoke<void>(RTC_FROM_HERE, | |
| 205 rtc::Bind(&QuicDataChannel::Close_w, this)); | |
| 206 SetState_s(kClosed); | |
| 207 } | |
| 208 | |
| 209 void QuicDataChannel::Close_w() { | |
| 210 RTC_DCHECK(worker_thread_->IsCurrent()); | |
| 211 for (auto& kv : incoming_quic_messages_) { | |
| 212 Message& message = kv.second; | |
| 213 cricket::ReliableQuicStream* stream = message.stream; | |
| 214 stream->Close(); | |
| 215 } | |
| 216 | |
| 217 for (auto& kv : write_blocked_quic_streams_) { | |
| 218 cricket::ReliableQuicStream* stream = kv.second; | |
| 219 stream->Close(); | |
| 220 } | |
| 221 } | |
| 222 | |
| 223 bool QuicDataChannel::SetTransportChannel( | |
| 224 cricket::QuicTransportChannel* channel) { | |
| 225 RTC_DCHECK(signaling_thread_->IsCurrent()); | |
| 226 | |
| 227 if (!channel) { | |
| 228 LOG(LS_ERROR) << "|channel| is NULL. Cannot set transport channel."; | |
| 229 return false; | |
| 230 } | |
| 231 if (quic_transport_channel_) { | |
| 232 if (channel == quic_transport_channel_) { | |
| 233 LOG(LS_WARNING) << "Ignoring duplicate transport channel."; | |
| 234 return true; | |
| 235 } | |
| 236 LOG(LS_ERROR) << "|channel| does not match existing transport channel."; | |
| 237 return false; | |
| 238 } | |
| 239 | |
| 240 quic_transport_channel_ = channel; | |
| 241 LOG(LS_INFO) << "Setting QuicTransportChannel for QUIC data channel " << id_; | |
| 242 DataState data_channel_state = worker_thread_->Invoke<DataState>( | |
| 243 RTC_FROM_HERE, rtc::Bind(&QuicDataChannel::SetTransportChannel_w, this)); | |
| 244 SetState_s(data_channel_state); | |
| 245 return true; | |
| 246 } | |
| 247 | |
| 248 DataChannelInterface::DataState QuicDataChannel::SetTransportChannel_w() { | |
| 249 RTC_DCHECK(worker_thread_->IsCurrent()); | |
| 250 quic_transport_channel_->SignalReadyToSend.connect( | |
| 251 this, &QuicDataChannel::OnReadyToSend); | |
| 252 quic_transport_channel_->SignalClosed.connect( | |
| 253 this, &QuicDataChannel::OnConnectionClosed); | |
| 254 if (quic_transport_channel_->writable()) { | |
| 255 return kOpen; | |
| 256 } | |
| 257 return kConnecting; | |
| 258 } | |
| 259 | |
| 260 void QuicDataChannel::OnIncomingMessage(Message&& message) { | |
| 261 RTC_DCHECK(network_thread_->IsCurrent()); | |
| 262 RTC_DCHECK(message.stream); | |
| 263 if (!observer_) { | |
| 264 LOG(LS_WARNING) << "QUIC data channel " << id_ | |
| 265 << " received a message but has no observer."; | |
| 266 message.stream->Close(); | |
| 267 return; | |
| 268 } | |
| 269 // A FIN is received if the message fits into a single QUIC stream frame and | |
| 270 // the remote peer is done sending. | |
| 271 if (message.stream->fin_received()) { | |
| 272 LOG(LS_INFO) << "Stream " << message.stream->id() | |
| 273 << " has finished receiving data for QUIC data channel " | |
| 274 << id_; | |
| 275 DataBuffer final_message(message.buffer, false); | |
| 276 invoker_.AsyncInvoke<void>(RTC_FROM_HERE, signaling_thread_, | |
| 277 rtc::Bind(&QuicDataChannel::OnMessage_s, this, | |
| 278 std::move(final_message))); | |
| 279 message.stream->Close(); | |
| 280 return; | |
| 281 } | |
| 282 // Otherwise the message is divided across multiple QUIC stream frames, so | |
| 283 // queue the data. OnDataReceived() will be called each time the remaining | |
| 284 // QUIC stream frames arrive. | |
| 285 LOG(LS_INFO) << "QUIC data channel " << id_ | |
| 286 << " is queuing incoming data for stream " | |
| 287 << message.stream->id(); | |
| 288 incoming_quic_messages_[message.stream->id()] = std::move(message); | |
| 289 message.stream->SignalDataReceived.connect(this, | |
| 290 &QuicDataChannel::OnDataReceived); | |
| 291 // The QUIC stream will be removed from |incoming_quic_messages_| once it | |
| 292 // closes. | |
| 293 message.stream->SignalClosed.connect( | |
| 294 this, &QuicDataChannel::OnIncomingQueuedStreamClosed); | |
| 295 } | |
| 296 | |
| 297 void QuicDataChannel::OnDataReceived(net::QuicStreamId stream_id, | |
| 298 const char* data, | |
| 299 size_t len) { | |
| 300 RTC_DCHECK(network_thread_->IsCurrent()); | |
| 301 RTC_DCHECK(data); | |
| 302 const auto& kv = incoming_quic_messages_.find(stream_id); | |
| 303 if (kv == incoming_quic_messages_.end()) { | |
| 304 RTC_NOTREACHED(); | |
| 305 return; | |
| 306 } | |
| 307 Message& message = kv->second; | |
| 308 cricket::ReliableQuicStream* stream = message.stream; | |
| 309 rtc::CopyOnWriteBuffer& received_data = message.buffer; | |
| 310 // If the QUIC stream has not received a FIN, then the remote peer is not | |
| 311 // finished sending data. | |
| 312 if (!stream->fin_received()) { | |
| 313 received_data.AppendData(data, len); | |
| 314 return; | |
| 315 } | |
| 316 // Otherwise we are done receiving and can provide the data channel observer | |
| 317 // with the message. | |
| 318 LOG(LS_INFO) << "Stream " << stream_id | |
| 319 << " has finished receiving data for QUIC data channel " << id_; | |
| 320 received_data.AppendData(data, len); | |
| 321 DataBuffer final_message(std::move(received_data), false); | |
| 322 invoker_.AsyncInvoke<void>( | |
| 323 RTC_FROM_HERE, signaling_thread_, | |
| 324 rtc::Bind(&QuicDataChannel::OnMessage_s, this, std::move(final_message))); | |
| 325 // Once the stream is closed, OnDataReceived will not fire for the stream. | |
| 326 stream->Close(); | |
| 327 } | |
| 328 | |
| 329 void QuicDataChannel::OnReadyToSend(cricket::TransportChannel* channel) { | |
| 330 RTC_DCHECK(network_thread_->IsCurrent()); | |
| 331 RTC_DCHECK(channel == quic_transport_channel_); | |
| 332 LOG(LS_INFO) << "QuicTransportChannel is ready to send"; | |
| 333 invoker_.AsyncInvoke<void>( | |
| 334 RTC_FROM_HERE, signaling_thread_, | |
| 335 rtc::Bind(&QuicDataChannel::SetState_s, this, kOpen)); | |
| 336 } | |
| 337 | |
| 338 void QuicDataChannel::OnWriteBlockedStreamClosed(net::QuicStreamId stream_id, | |
| 339 int error) { | |
| 340 RTC_DCHECK(worker_thread_->IsCurrent()); | |
| 341 LOG(LS_VERBOSE) << "Write blocked stream " << stream_id << " is closed."; | |
| 342 write_blocked_quic_streams_.erase(stream_id); | |
| 343 } | |
| 344 | |
| 345 void QuicDataChannel::OnIncomingQueuedStreamClosed(net::QuicStreamId stream_id, | |
| 346 int error) { | |
| 347 RTC_DCHECK(network_thread_->IsCurrent()); | |
| 348 LOG(LS_VERBOSE) << "Incoming queued stream " << stream_id << " is closed."; | |
| 349 incoming_quic_messages_.erase(stream_id); | |
| 350 } | |
| 351 | |
| 352 void QuicDataChannel::OnConnectionClosed() { | |
| 353 RTC_DCHECK(worker_thread_->IsCurrent()); | |
| 354 invoker_.AsyncInvoke<void>(RTC_FROM_HERE, signaling_thread_, | |
| 355 rtc::Bind(&QuicDataChannel::Close, this)); | |
| 356 } | |
| 357 | |
| 358 void QuicDataChannel::OnMessage_s(const DataBuffer& received_data) { | |
| 359 RTC_DCHECK(signaling_thread_->IsCurrent()); | |
| 360 if (observer_) { | |
| 361 observer_->OnMessage(received_data); | |
| 362 } | |
| 363 } | |
| 364 | |
| 365 void QuicDataChannel::SetState_s(DataState state) { | |
| 366 RTC_DCHECK(signaling_thread_->IsCurrent()); | |
| 367 if (state_ == state || state_ == kClosed) { | |
| 368 return; | |
| 369 } | |
| 370 if (state_ == kClosing && state != kClosed) { | |
| 371 return; | |
| 372 } | |
| 373 LOG(LS_INFO) << "Setting state to " << state << " for QUIC data channel " | |
| 374 << id_; | |
| 375 state_ = state; | |
| 376 if (observer_) { | |
| 377 observer_->OnStateChange(); | |
| 378 } | |
| 379 } | |
| 380 | |
| 381 void QuicDataChannel::OnBufferedAmountChange_s(uint64_t buffered_amount) { | |
| 382 RTC_DCHECK(signaling_thread_->IsCurrent()); | |
| 383 if (observer_) { | |
| 384 observer_->OnBufferedAmountChange(buffered_amount); | |
| 385 } | |
| 386 } | |
| 387 | |
| 388 size_t QuicDataChannel::GetNumWriteBlockedStreams() const { | |
| 389 return write_blocked_quic_streams_.size(); | |
| 390 } | |
| 391 | |
| 392 size_t QuicDataChannel::GetNumIncomingStreams() const { | |
| 393 return incoming_quic_messages_.size(); | |
| 394 } | |
| 395 | |
| 396 } // namespace webrtc | |
| OLD | NEW |
|---|
Chromium Code Reviews
Issue 2514883002:
Create //webrtc/api:libjingle_peerconnection_api + refactorings. (Closed)
