Move ready to send logic from BaseChannel to RtpTransport.

webrtc/pc/channel.cc

 /*
  *  Copyright 2004 The WebRTC project authors. All Rights Reserved.
  *
  *  Use of this source code is governed by a BSD-style license
  *  that can be found in the LICENSE file in the root of the source
  *  tree. An additional intellectual property rights grant can be found
  *  in the file PATENTS.  All contributing project authors may
  *  be found in the AUTHORS file in the root of the source tree.
  */
 
@@ skipping 150 matching lines @@
                          rtc::Thread* network_thread,
                          rtc::Thread* signaling_thread,
                          MediaChannel* media_channel,
                          const std::string& content_name,
                          bool rtcp_mux_required,
                          bool srtp_required)
     : worker_thread_(worker_thread),
       network_thread_(network_thread),
       signaling_thread_(signaling_thread),
       content_name_(content_name),
+      rtcp_mux_required_(rtcp_mux_required),
       rtp_transport_(rtcp_mux_required),
       srtp_required_(srtp_required),
       media_channel_(media_channel),
       selected_candidate_pair_(nullptr) {
   RTC_DCHECK(worker_thread_ == rtc::Thread::Current());
 #if defined(ENABLE_EXTERNAL_AUTH)
   srtp_filter_.EnableExternalAuth();
 #endif
+  rtp_transport_.SignalReadyToSend.connect(
+      this, &BaseChannel::OnTransportReadyToSend);
   LOG(LS_INFO) << "Created channel for " << content_name;
 }
 
 BaseChannel::~BaseChannel() {
   TRACE_EVENT0("webrtc", "BaseChannel::~BaseChannel");
   RTC_DCHECK(worker_thread_ == rtc::Thread::Current());
   Deinit();
   StopConnectionMonitor();
   // Eats any outstanding messages or packets.
   worker_thread_->Clear(&invoker_);
@@ skipping 54 matching lines @@
                   rtcp_packet_transport);
 
   if (rtp_dtls_transport_ &&
       !SetDtlsSrtpCryptoSuites_n(rtp_dtls_transport_, false)) {
     return false;
   }
   if (rtcp_dtls_transport_ &&
       !SetDtlsSrtpCryptoSuites_n(rtcp_dtls_transport_, true)) {
     return false;
   }
-  if (rtp_transport_.rtcp_mux_required()) {
+  if (rtcp_mux_required_) {
     rtcp_mux_filter_.SetActive();
   }
   return true;
 }
 
 void BaseChannel::Deinit() {
   RTC_DCHECK(worker_thread_->IsCurrent());
   media_channel_->SetInterface(NULL);
   // Packets arrive on the network thread, processing packets calls virtual
   // functions, so need to stop this process in Deinit that is called in
@@ skipping 72 matching lines @@
     SetTransport_n(true, rtcp_dtls_transport, rtcp_packet_transport);
   }
 
   LOG(LS_INFO) << "Setting RTP Transport for " << content_name() << " on "
                << debug_name << " transport " << rtp_packet_transport;
   SetTransport_n(false, rtp_dtls_transport, rtp_packet_transport);
 
   // Update aggregate writable/ready-to-send state between RTP and RTCP upon
   // setting new transport channels.
   UpdateWritableState_n();
-  // We can only update ready-to-send after updating writability.
-  //
-  // On setting a new channel, assume it's ready to send if it's writable,
-  // because we have no way of knowing otherwise (the channel doesn't give us
-  // "was last send successful?").
-  //
-  // This won't always be accurate (the last SendPacket call from another
-  // BaseChannel could have resulted in an error), but even so, we'll just
-  // encounter the error again and update "ready to send" accordingly.
-  SetTransportChannelReadyToSend(
-      false, rtp_packet_transport && rtp_packet_transport->writable());
-  SetTransportChannelReadyToSend(
-      true, rtcp_packet_transport && rtcp_packet_transport->writable());
 }
 
 void BaseChannel::SetTransport_n(
     bool rtcp,
     DtlsTransportInternal* new_dtls_transport,
     rtc::PacketTransportInternal* new_packet_transport) {
   RTC_DCHECK(network_thread_->IsCurrent());
   DtlsTransportInternal*& old_dtls_transport =
       rtcp ? rtcp_dtls_transport_ : rtp_dtls_transport_;
   rtc::PacketTransportInternal* old_packet_transport =
       rtcp ? rtp_transport_.rtcp_packet_transport()
            : rtp_transport_.rtp_packet_transport();
 
   if (!old_packet_transport && !new_packet_transport) {
     // Nothing to do.
     return;
   }
 
   RTC_DCHECK(old_packet_transport != new_packet_transport);
   if (old_dtls_transport) {
     DisconnectFromDtlsTransport(old_dtls_transport);
   } else if (old_packet_transport) {
     DisconnectFromPacketTransport(old_packet_transport);
   }
 
   if (rtcp) {
-    rtp_transport_.set_rtcp_packet_transport(new_packet_transport);
+    rtp_transport_.SetRtcpPacketTransport(new_packet_transport);
   } else {
-    rtp_transport_.set_rtp_packet_transport(new_packet_transport);
+    rtp_transport_.SetRtpPacketTransport(new_packet_transport);
   }
   old_dtls_transport = new_dtls_transport;
 
   // If there's no new transport, we're done after disconnecting from old one.
   if (!new_packet_transport) {
     return;
   }
 
   if (rtcp && new_dtls_transport) {
     RTC_CHECK(!(ShouldSetupDtlsSrtp_n() && srtp_filter_.IsActive()))
         << "Setting RTCP for DTLS/SRTP after SrtpFilter is active "
         << "should never happen.";
   }
+
   if (new_dtls_transport) {
     ConnectToDtlsTransport(new_dtls_transport);
   } else {
     ConnectToPacketTransport(new_packet_transport);
   }
   auto& socket_options = rtcp ? rtcp_socket_options_ : socket_options_;
   for (const auto& pair : socket_options) {
     new_packet_transport->SetOption(pair.first, pair.second);
   }
 }
 
 void BaseChannel::ConnectToDtlsTransport(DtlsTransportInternal* transport) {
   RTC_DCHECK(network_thread_->IsCurrent());
 
+  // TODO(zstein): de-dup with ConnectToPacketTransport
   transport->SignalWritableState.connect(this, &BaseChannel::OnWritableState);
   transport->SignalReadPacket.connect(this, &BaseChannel::OnPacketRead);
-  transport->SignalReadyToSend.connect(this, &BaseChannel::OnReadyToSend);
   transport->SignalDtlsState.connect(this, &BaseChannel::OnDtlsState);
   transport->SignalSentPacket.connect(this, &BaseChannel::SignalSentPacket_n);
   transport->ice_transport()->SignalSelectedCandidatePairChanged.connect(
       this, &BaseChannel::OnSelectedCandidatePairChanged);
 }
 
 void BaseChannel::DisconnectFromDtlsTransport(
     DtlsTransportInternal* transport) {
   RTC_DCHECK(network_thread_->IsCurrent());
   OnSelectedCandidatePairChanged(transport->ice_transport(), nullptr, -1,
                                  false);
 
   transport->SignalWritableState.disconnect(this);
   transport->SignalReadPacket.disconnect(this);
-  transport->SignalReadyToSend.disconnect(this);
   transport->SignalDtlsState.disconnect(this);
   transport->SignalSentPacket.disconnect(this);
   transport->ice_transport()->SignalSelectedCandidatePairChanged.disconnect(
       this);
 }
 
 void BaseChannel::ConnectToPacketTransport(
     rtc::PacketTransportInternal* transport) {
   RTC_DCHECK_RUN_ON(network_thread_);
   transport->SignalWritableState.connect(this, &BaseChannel::OnWritableState);
   transport->SignalReadPacket.connect(this, &BaseChannel::OnPacketRead);
-  transport->SignalReadyToSend.connect(this, &BaseChannel::OnReadyToSend);
   transport->SignalSentPacket.connect(this, &BaseChannel::SignalSentPacket_n);
 }
 
 void BaseChannel::DisconnectFromPacketTransport(
     rtc::PacketTransportInternal* transport) {
   RTC_DCHECK_RUN_ON(network_thread_);
   transport->SignalWritableState.disconnect(this);
   transport->SignalReadPacket.disconnect(this);
-  transport->SignalReadyToSend.disconnect(this);
   transport->SignalSentPacket.disconnect(this);
 }
 
 bool BaseChannel::Enable(bool enable) {
   worker_thread_->Invoke<void>(
       RTC_FROM_HERE,
       Bind(enable ? &BaseChannel::EnableMedia_w : &BaseChannel::DisableMedia_w,
            this));
   return true;
 }
@@ skipping 59 matching lines @@
   RTC_DCHECK(network_thread_->IsCurrent());
   if (!rtp_dtls_transport_) {
     return false;
   }
   return rtp_dtls_transport_->ice_transport()->GetStats(infos);
 }
 
 bool BaseChannel::NeedsRtcpTransport() {
   // If this BaseChannel doesn't require RTCP mux and we haven't fully
   // negotiated RTCP mux, we need an RTCP transport.
-  return !rtp_transport_.rtcp_mux_required() &&
-         !rtcp_mux_filter_.IsFullyActive();
+  return !rtcp_mux_required_ && !rtcp_mux_filter_.IsFullyActive();
 }
 
 bool BaseChannel::IsReadyToReceiveMedia_w() const {
   // Receive data if we are enabled and have local content,
   return enabled() && IsReceiveContentDirection(local_content_direction_);
 }
 
 bool BaseChannel::IsReadyToSendMedia_w() const {
   // Need to access some state updated on the network thread.
   return network_thread_->Invoke<bool>(
@@ skipping 66 matching lines @@
   // OnPacketRead gets called from P2PSocket; now pass data to MediaEngine
   RTC_DCHECK(network_thread_->IsCurrent());
 
   // When using RTCP multiplexing we might get RTCP packets on the RTP
   // transport. We feed RTP traffic into the demuxer to determine if it is RTCP.
   bool rtcp = PacketIsRtcp(transport, data, len);
   rtc::CopyOnWriteBuffer packet(data, len);
   HandlePacket(rtcp, &packet, packet_time);
 }
 
-void BaseChannel::OnReadyToSend(rtc::PacketTransportInternal* transport) {
-  RTC_DCHECK(transport == rtp_transport_.rtp_packet_transport() ||
-             transport == rtp_transport_.rtcp_packet_transport());
-  SetTransportChannelReadyToSend(
-      transport == rtp_transport_.rtcp_packet_transport(), true);
-}
-
 void BaseChannel::OnDtlsState(DtlsTransportInternal* transport,
                               DtlsTransportState state) {
   if (!ShouldSetupDtlsSrtp_n()) {
     return;
   }
 
   // Reset the srtp filter if it's not the CONNECTED state. For the CONNECTED
   // state, setting up DTLS-SRTP context is deferred to ChannelWritable_w to
   // cover other scenarios like the whole transport is writable (not just this
   // TransportChannel) or when TransportChannel is attached after DTLS is
@@ skipping 23 matching lines @@
         last_sent_packet_id);
 
     UpdateTransportOverhead();
   }
   invoker_.AsyncInvoke<void>(
       RTC_FROM_HERE, worker_thread_,
       Bind(&MediaChannel::OnNetworkRouteChanged, media_channel_, transport_name,
            network_route));
 }
 
-void BaseChannel::SetTransportChannelReadyToSend(bool rtcp, bool ready) {
-  RTC_DCHECK(network_thread_->IsCurrent());
-  if (rtcp) {
-    rtcp_ready_to_send_ = ready;
-  } else {
-    rtp_ready_to_send_ = ready;
-  }
-
-  bool ready_to_send =
-      (rtp_ready_to_send_ &&
-       // In the case of rtcp mux |rtcp_packet_transport_| will be null.
-       (rtcp_ready_to_send_ || !rtp_transport_.rtcp_packet_transport()));
-
+void BaseChannel::OnTransportReadyToSend(bool ready) {
   invoker_.AsyncInvoke<void>(
       RTC_FROM_HERE, worker_thread_,
-      Bind(&MediaChannel::OnReadyToSend, media_channel_, ready_to_send));
+      Bind(&MediaChannel::OnReadyToSend, media_channel_, ready));
 }
 
 bool BaseChannel::PacketIsRtcp(const rtc::PacketTransportInternal* transport,
                                const char* data,
                                size_t len) {
   return (transport == rtp_transport_.rtcp_packet_transport() ||
           rtcp_mux_filter_.DemuxRtcp(data, static_cast<int>(len)));
 }
 
 bool BaseChannel::SendPacket(bool rtcp,
@@ skipping 14 matching lines @@
     data->options = options;
     network_thread_->Post(RTC_FROM_HERE, this, message_id, data);
     return true;
   }
   TRACE_EVENT0("webrtc", "BaseChannel::SendPacket");
 
   // Now that we are on the correct thread, ensure we have a place to send this
   // packet before doing anything. (We might get RTCP packets that we don't
   // intend to send.) If we've negotiated RTCP mux, send RTCP over the RTP
   // transport.
-  rtc::PacketTransportInternal* transport =
-      (!rtcp || rtcp_mux_filter_.IsActive())
-          ? rtp_transport_.rtp_packet_transport()
-          : rtp_transport_.rtcp_packet_transport();
-  if (!transport || !transport->writable()) {
+  const bool send_on_rtcp = rtcp && !rtcp_mux_filter_.IsActive();

[Taylor Brandstetter, 2017/04/19 05:56:33]

This could be moved into RtpTransport, if rtcp_mux_filter_.IsActive() always
matched "rtcp_mux_enabled()"

[Zach Stein, 2017/04/20 19:59:10]

Done.

+  if (!rtp_transport_.IsWritable(send_on_rtcp)) {
     return false;
   }
 
   // Protect ourselves against crazy data.
   if (!ValidPacket(rtcp, packet)) {
     LOG(LS_ERROR) << "Dropping outgoing " << content_name_ << " "
                   << PacketType(rtcp)
                   << " packet: wrong size=" << packet->size();
     return false;
   }
@@ skipping 75 matching lines @@
     // However, there shouldn't be any RTP packets sent before SRTP is set up
     // (and SetSend(true) is called).
     LOG(LS_ERROR) << "Can't send outgoing RTP packet when SRTP is inactive"
                   << " and crypto is required";
     RTC_NOTREACHED();
     return false;
   }
 
   // Bon voyage.
   int flags = (secure() && secure_dtls()) ? PF_SRTP_BYPASS : PF_NORMAL;
-  int ret = transport->SendPacket(packet->data<char>(), packet->size(),
-                                  updated_options, flags);
-  if (ret != static_cast<int>(packet->size())) {
-    if (transport->GetError() == ENOTCONN) {
-      LOG(LS_WARNING) << "Got ENOTCONN from transport.";
-      SetTransportChannelReadyToSend(rtcp, false);
-    }
-    return false;
-  }
-  return true;
+  return rtp_transport_.SendPacket(send_on_rtcp, packet, updated_options,
+                                   flags);
 }
 
 bool BaseChannel::WantsPacket(bool rtcp, const rtc::CopyOnWriteBuffer* packet) {
   // Protect ourselves against crazy data.
   if (!ValidPacket(rtcp, packet)) {
     LOG(LS_ERROR) << "Dropping incoming " << content_name_ << " "
                   << PacketType(rtcp)
                   << " packet: wrong size=" << packet->size();
     return false;
   }
@@ skipping 411 matching lines @@
   }
   return true;
 }
 
 bool BaseChannel::SetRtcpMux_n(bool enable,
                                ContentAction action,
                                ContentSource src,
                                std::string* error_desc) {
   // Provide a more specific error message for the RTCP mux "require" policy
   // case.
-  if (rtp_transport_.rtcp_mux_required() && !enable) {
+  if (rtcp_mux_required_ && !enable) {
     SafeSetError(
         "rtcpMuxPolicy is 'require', but media description does not "
         "contain 'a=rtcp-mux'.",
         error_desc);
     return false;
   }
   bool ret = false;
   switch (action) {
     case CA_OFFER:
       ret = rtcp_mux_filter_.SetOffer(enable, src);
+      // TODO(zstein): rtp_transport_.set_rtcp_mux_enabled here?

[Taylor Brandstetter, 2017/04/19 05:56:33]

I think you could do
"rtp_transport_.set_rtcp_mux_enabled(rtcp_mux_filter_.IsActive());" after the
switch statement, that seems simplest.

[Zach Stein, 2017/04/20 19:59:10]

I think that is just what I want - thanks!

       break;
     case CA_PRANSWER:
       // This may activate RTCP muxing, but we don't yet destroy the transport
       // because the final answer may deactivate it.
       ret = rtcp_mux_filter_.SetProvisionalAnswer(enable, src);
+      // TODO(zstein): rtp_transport_.set_rtcp_mux_enabled here?
       break;
     case CA_ANSWER:
       ret = rtcp_mux_filter_.SetAnswer(enable, src);
       if (ret && rtcp_mux_filter_.IsActive()) {
         // We permanently activated RTCP muxing; signal that we no longer need
         // the RTCP transport.
         std::string debug_name =
             transport_name_.empty()
                 ? rtp_transport_.rtp_packet_transport()->debug_name()
                 : transport_name_;
         ;
         LOG(LS_INFO) << "Enabling rtcp-mux for " << content_name()
                      << "; no longer need RTCP transport for " << debug_name;
         if (rtp_transport_.rtcp_packet_transport()) {
           SetTransport_n(true, nullptr, nullptr);
           SignalRtcpMuxFullyActive(transport_name_);
         }
         UpdateWritableState_n();
-        SetTransportChannelReadyToSend(true, false);
+        // TODO(zstein): Maybe rtcp_mux_filter should signal these changes.

[Taylor Brandstetter, 2017/04/19 05:56:33]

Since its state will always change synchronously, I don't think that's necessary

[Zach Stein, 2017/04/20 19:59:10]

Acknowledged.

+        rtp_transport_.set_rtcp_mux_enabled(true);
       }
       break;
     case CA_UPDATE:
       // No RTCP mux info.
       ret = true;
       break;
     default:
       break;
   }
   if (!ret) {
@@ skipping 1179 matching lines @@
   signaling_thread()->Post(RTC_FROM_HERE, this, MSG_READYTOSENDDATA,
                            new DataChannelReadyToSendMessageData(writable));
 }
 
 void RtpDataChannel::GetSrtpCryptoSuites_n(
     std::vector<int>* crypto_suites) const {
   GetSupportedDataCryptoSuites(crypto_options(), crypto_suites);
 }
 
 }  // namespace cricket