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 16 matching lines @@
   // 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);
   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));
+  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));
+    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) {
+    if (old_tc) {
+      // Need to make sure ref count is decremented
+      transport_controller_->DestroyTransportChannel_w(
+          transport_name_, cricket::ICE_CANDIDATE_COMPONENT_RTP);
+    }

[pthatcher1, 2015/08/10 20:40:17]

This doesn't seem right.  If we set_transport_channel(transport_channel()), it
will decrement the ref count?  Then you could keep doing that until it explodes.

[Taylor Brandstetter, 2015/08/11 01:20:07]

I did this because "CreateTransportChannel_w" above will increment the ref
count. But probably a better solution would be to not call
CreateTransportChannel_w in the first place unless necessary. I added that logic
to SetTransportChannels_w.

[pthatcher1, 2015/08/18 22:32:46]

I think it's right to decrement the pointer of the old_tc if old_tc != new_tc. 
But if they are the same and we return early, we shouldn't change the ref count.

     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);
+    }
+
+    OnWritableState(new_tc);
+    SetReadyToSend(new_tc, new_tc->writable());

[pthatcher1, 2015/08/10 20:40:17]

Can you put in a nice comment about why we need to set the socket options and
update the writeable and "ready to send" states when we get a new transport
channel?

[Taylor Brandstetter, 2015/08/11 01:20:07]

Done.

   }
 }
 
 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);
+    }
+
+    OnWritableState(new_tc);
+    SetReadyToSend(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 99 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();
@@ skipping 326 matching lines @@
     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 185 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),

[pthatcher1, 2015/08/10 20:40:17]

Do we not need to set the media engine any more?

[Taylor Brandstetter, 2015/08/11 01:20:07]

Good catch... I must have deleted that line on accident when handling a merge
conflict.

       received_media_(false) {
 }
 
 VoiceChannel::~VoiceChannel() {
   StopAudioMonitor();
   StopMediaMonitor();
   // this can't be done in the base class, since it calls a virtual
   DisableMedia_w();
   Deinit();
 }
@@ skipping 369 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