TransportController refactoring

talk/session/media/channel.cc

 /*
  * libjingle
  * Copyright 2004 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,
@@ skipping 42 matching lines @@
   MSG_DATARECEIVED,
   MSG_FIRSTPACKETRECEIVED,
   MSG_STREAMCLOSEDREMOTELY,
 };
 
 // Value specified in RFC 5764.
 static const char kDtlsSrtpExporterLabel[] = "EXTRACTOR-dtls_srtp";
 
 static const int kAgcMinus10db = -10;
 
-static void SetSessionError(BaseSession* session, BaseSession::Error error,
-                            const std::string& error_desc) {
-  session->SetError(error, error_desc);
-}
-
 static void SafeSetError(const std::string& message, std::string* error_desc) {
   if (error_desc) {
     *error_desc = message;
   }
 }
 
 struct PacketMessageData : public rtc::MessageData {
   rtc::Buffer packet;
   rtc::DiffServCodePoint dscp;
 };
@@ skipping 66 matching lines @@
 }
 
 static const MediaContentDescription* GetContentDescription(
     const ContentInfo* cinfo) {
   if (cinfo == NULL)
     return NULL;
   return static_cast<const MediaContentDescription*>(cinfo->description);
 }
 
 BaseChannel::BaseChannel(rtc::Thread* thread,
-                         MediaChannel* media_channel, BaseSession* session,
-                         const std::string& content_name, bool rtcp)
+                         MediaChannel* media_channel,
+                         TransportController* transport_controller,
+                         const std::string& content_name,
+                         bool rtcp)
     : worker_thread_(thread),
-      session_(session),
+      transport_controller_(transport_controller),
       media_channel_(media_channel),
       content_name_(content_name),
       rtcp_(rtcp),
       transport_channel_(NULL),
       rtcp_transport_channel_(NULL),
       enabled_(false),
       writable_(false),
       rtp_ready_to_send_(false),
       rtcp_ready_to_send_(false),
       was_ever_writable_(false),
@@ skipping 10 matching lines @@
 BaseChannel::~BaseChannel() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   Deinit();
   StopConnectionMonitor();
   FlushRtcpMessages();  // Send any outstanding RTCP packets.
   worker_thread_->Clear(this);  // eats any outstanding messages or packets
   // We must destroy the media channel before the transport channel, otherwise
   // the media channel may try to send on the dead transport channel. NULLing
   // is not an effective strategy since the sends will come on another thread.
   delete media_channel_;
-  set_transport_channel(nullptr);
-  set_rtcp_transport_channel(nullptr);
+  if (transport_channel_) {
+    DisconnectFromTransportChannel(transport_channel_);
+    transport_controller_->DestroyTransportChannel_w(
+        transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTP);
+  }
+  if (rtcp_transport_channel_) {
+    DisconnectFromTransportChannel(rtcp_transport_channel_);
+    transport_controller_->DestroyTransportChannel_w(
+        transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTCP);
+  }
   LOG(LS_INFO) << "Destroyed channel";
 }
 
 bool BaseChannel::Init() {
-  if (!SetTransportChannels(session(), rtcp())) {
+  if (!SetTransportChannels(content_name(), rtcp())) {
     return false;
   }
 
   if (!SetDtlsSrtpCiphers(transport_channel(), false)) {
     return false;
   }
   if (rtcp() && !SetDtlsSrtpCiphers(rtcp_transport_channel(), true)) {
     return false;
   }
 
   // Both RTP and RTCP channels are set, we can call SetInterface on
   // media channel and it can set network options.
   media_channel_->SetInterface(this);
   return true;
 }
 
 void BaseChannel::Deinit() {
   media_channel_->SetInterface(NULL);
 }
 
-bool BaseChannel::SetTransportChannels(BaseSession* session, bool rtcp) {
-  return worker_thread_->Invoke<bool>(Bind(
-      &BaseChannel::SetTransportChannels_w, this, session, rtcp));
+bool BaseChannel::SetTransport(const std::string& transport_name) {
+  // Only enable an RTCP transport channel if we already had one.
+  bool rtcp = (rtcp_transport_channel() != nullptr);
+  if (!SetTransportChannels(transport_name, rtcp)) {
+    return false;
+  }
+  return true;
 }
 
-bool BaseChannel::SetTransportChannels_w(BaseSession* session, bool rtcp) {
+bool BaseChannel::SetTransportChannels(const std::string& transport_name,
+                                       bool rtcp) {
+  return worker_thread_->Invoke<bool>(
+      Bind(&BaseChannel::SetTransportChannels_w, this, transport_name, rtcp));
+}
+
+bool BaseChannel::SetTransportChannels_w(const std::string& transport_name,
+                                         bool rtcp) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
-  set_transport_channel(session->CreateChannel(
-      content_name(), cricket::ICE_CANDIDATE_COMPONENT_RTP));
-  if (!transport_channel()) {
-    return false;
+  if (transport_name != transport_name_) {
+    set_transport_channel(transport_controller_->CreateTransportChannel_w(
+        transport_name, cricket::ICE_CANDIDATE_COMPONENT_RTP));
+    if (!transport_channel()) {
+      return false;
+    }
   }
   if (rtcp) {
-    set_rtcp_transport_channel(session->CreateChannel(
-        content_name(), cricket::ICE_CANDIDATE_COMPONENT_RTCP));
-    if (!rtcp_transport_channel()) {
-      return false;
+    if (!rtcp_transport_channel_ || transport_name != transport_name_) {
+      LOG(LS_INFO) << "Create RTCP TransportChannel for " << content_name()
+                   << " on " << transport_name << " transport ";
+      set_rtcp_transport_channel(
+          transport_controller_->CreateTransportChannel_w(
+              transport_name, cricket::ICE_CANDIDATE_COMPONENT_RTCP));
+      if (!rtcp_transport_channel()) {
+        return false;
+      }
     }
   } else {
     set_rtcp_transport_channel(nullptr);
   }
+  transport_name_ = transport_name;
 
   return true;
 }
 
 void BaseChannel::set_transport_channel(TransportChannel* new_tc) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   TransportChannel* old_tc = transport_channel_;
-
-  if (old_tc == new_tc) {
-    return;
-  }
   if (old_tc) {
     DisconnectFromTransportChannel(old_tc);
-    session()->DestroyChannel(
-        content_name(), cricket::ICE_CANDIDATE_COMPONENT_RTP);
+    transport_controller_->DestroyTransportChannel_w(
+        transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTP);
   }
 
   transport_channel_ = new_tc;
 
   if (new_tc) {
     ConnectToTransportChannel(new_tc);
+    for (const auto& pair : socket_options_) {
+      new_tc->SetOption(pair.first, pair.second);
+    }
   }
+
+  // Update aggregate writable/ready-to-send state upon setting new channel
+  UpdateWritableState_w();
+  SetReadyToSend(false, new_tc && new_tc->writable());
 }
 
 void BaseChannel::set_rtcp_transport_channel(TransportChannel* new_tc) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   TransportChannel* old_tc = rtcp_transport_channel_;
-
-  if (old_tc == new_tc) {
-    return;
-  }
   if (old_tc) {
     DisconnectFromTransportChannel(old_tc);
-    session()->DestroyChannel(
-        content_name(), cricket::ICE_CANDIDATE_COMPONENT_RTCP);
+    transport_controller_->DestroyTransportChannel_w(
+        transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTCP);
   }
 
   rtcp_transport_channel_ = new_tc;
 
   if (new_tc) {
     ConnectToTransportChannel(new_tc);
+    for (const auto& pair : rtcp_socket_options_) {
+      new_tc->SetOption(pair.first, pair.second);
+    }
   }
+
+  // Update aggregate writable/ready-to-send state upon setting new channel
+  UpdateWritableState_w();
+  SetReadyToSend(true, new_tc && new_tc->writable());
 }
 
 void BaseChannel::ConnectToTransportChannel(TransportChannel* tc) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   tc->SignalWritableState.connect(this, &BaseChannel::OnWritableState);
   tc->SignalReadPacket.connect(this, &BaseChannel::OnChannelRead);
   tc->SignalReadyToSend.connect(this, &BaseChannel::OnReadyToSend);
 }
 
@@ skipping 98 matching lines @@
                            rtc::DiffServCodePoint dscp) {
   return SendPacket(true, packet, dscp);
 }
 
 int BaseChannel::SetOption(SocketType type, rtc::Socket::Option opt,
                            int value) {
   TransportChannel* channel = NULL;
   switch (type) {
     case ST_RTP:
       channel = transport_channel_;
+      socket_options_.push_back(
+          std::pair<rtc::Socket::Option, int>(opt, value));
       break;
     case ST_RTCP:
       channel = rtcp_transport_channel_;
+      rtcp_socket_options_.push_back(
+          std::pair<rtc::Socket::Option, int>(opt, value));
       break;
   }
   return channel ? channel->SetOption(opt, value) : -1;
 }
 
 void BaseChannel::OnWritableState(TransportChannel* channel) {
   ASSERT(channel == transport_channel_ || channel == rtcp_transport_channel_);
-  if (transport_channel_->writable()
-      && (!rtcp_transport_channel_ || rtcp_transport_channel_->writable())) {
-    ChannelWritable_w();
-  } else {
-    ChannelNotWritable_w();
-  }
+  UpdateWritableState_w();
 }
 
 void BaseChannel::OnChannelRead(TransportChannel* channel,
                                 const char* data, size_t len,
                                 const rtc::PacketTime& packet_time,
                                 int flags) {
   // OnChannelRead gets called from P2PSocket; now pass data to MediaEngine
   ASSERT(worker_thread_ == rtc::Thread::Current());
 
   // 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(channel, data, len);
   rtc::Buffer packet(data, len);
   HandlePacket(rtcp, &packet, packet_time);
 }
 
 void BaseChannel::OnReadyToSend(TransportChannel* channel) {
-  SetReadyToSend(channel, true);
+  ASSERT(channel == transport_channel_ || channel == rtcp_transport_channel_);
+  SetReadyToSend(channel == rtcp_transport_channel_, true);
 }
 
-void BaseChannel::SetReadyToSend(TransportChannel* channel, bool ready) {
-  ASSERT(channel == transport_channel_ || channel == rtcp_transport_channel_);
-  if (channel == transport_channel_) {
+void BaseChannel::SetReadyToSend(bool rtcp, bool ready) {
+  if (rtcp) {
+    rtcp_ready_to_send_ = ready;
+  } else {
     rtp_ready_to_send_ = ready;
   }
-  if (channel == rtcp_transport_channel_) {
-    rtcp_ready_to_send_ = ready;
-  }
 
-  if (!ready) {
+  if (rtp_ready_to_send_ &&
+      // In the case of rtcp mux |rtcp_transport_channel_| will be null.
+      (rtcp_ready_to_send_ || !rtcp_transport_channel_)) {
+    // Notify the MediaChannel when both rtp and rtcp channel can send.
+    media_channel_->OnReadyToSend(true);
+  } else {
     // Notify the MediaChannel when either rtp or rtcp channel can't send.
     media_channel_->OnReadyToSend(false);
-  } else if (rtp_ready_to_send_ &&
-             // In the case of rtcp mux |rtcp_transport_channel_| will be null.
-             (rtcp_ready_to_send_ || !rtcp_transport_channel_)) {
-    // Notify the MediaChannel when both rtp and rtcp channel can send.
-    media_channel_->OnReadyToSend(true);
   }
 }
 
 bool BaseChannel::PacketIsRtcp(const TransportChannel* channel,
                                const char* data, size_t len) {
   return (channel == rtcp_transport_channel_ ||
           rtcp_mux_filter_.DemuxRtcp(data, static_cast<int>(len)));
 }
 
 bool BaseChannel::SendPacket(bool rtcp, rtc::Buffer* packet,
@@ skipping 101 matching lines @@
     return false;
   }
 
   // Bon voyage.
   int ret =
       channel->SendPacket(packet->data<char>(), packet->size(), options,
                           (secure() && secure_dtls()) ? PF_SRTP_BYPASS : 0);
   if (ret != static_cast<int>(packet->size())) {
     if (channel->GetError() == EWOULDBLOCK) {
       LOG(LS_WARNING) << "Got EWOULDBLOCK from socket.";
-      SetReadyToSend(channel, false);
+      SetReadyToSend(rtcp, false);
     }
     return false;
   }
   return true;
 }
 
 bool BaseChannel::WantsPacket(bool rtcp, rtc::Buffer* packet) {
   // Protect ourselves against crazy data.
   if (!ValidPacket(rtcp, packet)) {
     LOG(LS_ERROR) << "Dropping incoming " << content_name_ << " "
@@ skipping 130 matching lines @@
       muted_streams_.erase(ssrc);
   }
   return ret;
 }
 
 bool BaseChannel::IsStreamMuted_w(uint32 ssrc) {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   return muted_streams_.find(ssrc) != muted_streams_.end();
 }
 
+void BaseChannel::UpdateWritableState_w() {
+  if (transport_channel_ && transport_channel_->writable() &&
+      (!rtcp_transport_channel_ || rtcp_transport_channel_->writable())) {
+    ChannelWritable_w();
+  } else {
+    ChannelNotWritable_w();
+  }
+}
+
 void BaseChannel::ChannelWritable_w() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   if (writable_)
     return;
 
-  LOG(LS_INFO) << "Channel socket writable ("
-               << transport_channel_->content_name() << ", "
-               << transport_channel_->component() << ")"
+  LOG(LS_INFO) << "Channel writable (" << content_name_ << ")"
                << (was_ever_writable_ ? "" : " for the first time");
 
   std::vector<ConnectionInfo> infos;
   transport_channel_->GetStats(&infos);
   for (std::vector<ConnectionInfo>::const_iterator it = infos.begin();
        it != infos.end(); ++it) {
     if (it->best_connection) {
       LOG(LS_INFO) << "Using " << it->local_candidate.ToSensitiveString()
                    << "->" << it->remote_candidate.ToSensitiveString();
       break;
     }
   }
 
   // If we're doing DTLS-SRTP, now is the time.
   if (!was_ever_writable_ && ShouldSetupDtlsSrtp()) {
     if (!SetupDtlsSrtp(false)) {
-      SignalDtlsSetupFailure(this, false);
+      SignalDtlsSetupFailure_w(false);
       return;
     }
 
     if (rtcp_transport_channel_) {
       if (!SetupDtlsSrtp(true)) {
-        SignalDtlsSetupFailure(this, true);
+        SignalDtlsSetupFailure_w(true);
         return;
       }
     }
   }
 
   was_ever_writable_ = true;
   writable_ = true;
   ChangeState();
 }
 
@@ skipping 22 matching lines @@
 
 bool BaseChannel::ShouldSetupDtlsSrtp() const {
   return true;
 }
 
 // This function returns true if either DTLS-SRTP is not in use
 // *or* DTLS-SRTP is successfully set up.
 bool BaseChannel::SetupDtlsSrtp(bool rtcp_channel) {
   bool ret = false;
 
-  TransportChannel *channel = rtcp_channel ?
-      rtcp_transport_channel_ : transport_channel_;
+  TransportChannel* channel =
+      rtcp_channel ? rtcp_transport_channel_ : transport_channel_;
 
   // No DTLS
   if (!channel->IsDtlsActive())
     return true;
 
   std::string selected_cipher;
 
   if (!channel->GetSrtpCipher(&selected_cipher)) {
     LOG(LS_ERROR) << "No DTLS-SRTP selected cipher";
     return false;
@@ skipping 74 matching lines @@
     dtls_keyed_ = true;
 
   return ret;
 }
 
 void BaseChannel::ChannelNotWritable_w() {
   ASSERT(worker_thread_ == rtc::Thread::Current());
   if (!writable_)
     return;
 
-  LOG(LS_INFO) << "Channel socket not writable ("
-               << transport_channel_->content_name() << ", "
-               << transport_channel_->component() << ")";
+  LOG(LS_INFO) << "Channel not writable (" << content_name_ << ")";
   writable_ = false;
   ChangeState();
 }
 
 // |dtls| will be set to true if DTLS is active for transport channel and
 // crypto is empty.
 bool BaseChannel::CheckSrtpConfig(const std::vector<CryptoParams>& cryptos,
                                   bool* dtls,
                                   std::string* error_desc) {
   *dtls = transport_channel_->IsDtlsActive();
@@ skipping 88 matching lines @@
 }
 
 void BaseChannel::ActivateRtcpMux() {
   worker_thread_->Invoke<void>(Bind(
       &BaseChannel::ActivateRtcpMux_w, this));
 }
 
 void BaseChannel::ActivateRtcpMux_w() {
   if (!rtcp_mux_filter_.IsActive()) {
     rtcp_mux_filter_.SetActive();
-    set_rtcp_transport_channel(NULL);
+    set_rtcp_transport_channel(nullptr);
   }
 }
 
 bool BaseChannel::SetRtcpMux_w(bool enable, ContentAction action,
                                ContentSource src,
                                std::string* error_desc) {
   bool ret = false;
   switch (action) {
     case CA_OFFER:
       ret = rtcp_mux_filter_.SetOffer(enable, src);
       break;
     case CA_PRANSWER:
       ret = rtcp_mux_filter_.SetProvisionalAnswer(enable, src);
       break;
     case CA_ANSWER:
       ret = rtcp_mux_filter_.SetAnswer(enable, src);
       if (ret && rtcp_mux_filter_.IsActive()) {
         // We activated RTCP mux, close down the RTCP transport.
-        set_rtcp_transport_channel(NULL);
+        LOG(LS_INFO) << "Enabling rtcp-mux for " << content_name()
+                     << " by destroying RTCP transport channel for "
+                     << transport_name();
+        set_rtcp_transport_channel(nullptr);
       }
       break;
     case CA_UPDATE:
       // No RTCP mux info.
       ret = true;
       break;
     default:
       break;
   }
   if (!ret) {
@@ skipping 249 matching lines @@
   worker_thread_->Clear(this, MSG_RTCPPACKET, &rtcp_messages);
   for (rtc::MessageList::iterator it = rtcp_messages.begin();
        it != rtcp_messages.end(); ++it) {
     worker_thread_->Send(this, MSG_RTCPPACKET, it->pdata);
   }
 }
 
 VoiceChannel::VoiceChannel(rtc::Thread* thread,
                            MediaEngineInterface* media_engine,
                            VoiceMediaChannel* media_channel,
-                           BaseSession* session,
+                           TransportController* transport_controller,
                            const std::string& content_name,
                            bool rtcp)
-    : BaseChannel(thread, media_channel, session, content_name,
+    : BaseChannel(thread,
+                  media_channel,
+                  transport_controller,
+                  content_name,
                   rtcp),
       media_engine_(media_engine),
       received_media_(false) {
 }
 
 VoiceChannel::~VoiceChannel() {
   StopAudioMonitor();
   StopMediaMonitor();
   // this can't be done in the base class, since it calls a virtual
   DisableMedia_w();
@@ skipping 371 matching lines @@
       break;
   }
 }
 
 void VoiceChannel::GetSrtpCiphers(std::vector<std::string>* ciphers) const {
   GetSupportedAudioCryptoSuites(ciphers);
 }
 
 VideoChannel::VideoChannel(rtc::Thread* thread,
                            VideoMediaChannel* media_channel,
-                           BaseSession* session,
+                           TransportController* transport_controller,
                            const std::string& content_name,
                            bool rtcp)
-    : BaseChannel(thread, media_channel, session, content_name,
+    : BaseChannel(thread,
+                  media_channel,
+                  transport_controller,
+                  content_name,
                   rtcp),
       renderer_(NULL),
       previous_we_(rtc::WE_CLOSE) {
 }
 
 bool VideoChannel::Init() {
   if (!BaseChannel::Init()) {
     return false;
   }
   media_channel()->SignalMediaError.connect(
@@ skipping 414 matching lines @@
   }
 }
 
 
 void VideoChannel::GetSrtpCiphers(std::vector<std::string>* ciphers) const {
   GetSupportedVideoCryptoSuites(ciphers);
 }
 
 DataChannel::DataChannel(rtc::Thread* thread,
                          DataMediaChannel* media_channel,
-                         BaseSession* session,
+                         TransportController* transport_controller,
                          const std::string& content_name,
                          bool rtcp)
-    : BaseChannel(thread, media_channel, session, content_name, rtcp),
+    : BaseChannel(thread,
+                  media_channel,
+                  transport_controller,
+                  content_name,
+                  rtcp),
       data_channel_type_(cricket::DCT_NONE),
       ready_to_send_data_(false) {
 }
 
 DataChannel::~DataChannel() {
   StopMediaMonitor();
   // this can't be done in the base class, since it calls a virtual
   DisableMedia_w();
 
   Deinit();
@@ skipping 333 matching lines @@
   return (data_channel_type_ == DCT_RTP);
 }
 
 void DataChannel::OnStreamClosedRemotely(uint32 sid) {
   rtc::TypedMessageData<uint32>* message =
       new rtc::TypedMessageData<uint32>(sid);
   signaling_thread()->Post(this, MSG_STREAMCLOSEDREMOTELY, message);
 }
 
 }  // namespace cricket