Cleanups in WebrtcTransport.

This change includes several changes needed in the Webrtc-based
transport:
1. WebrtcTransportFactory no longer owns the worker thread. Instead
   the calling code must provide a thread that has been started already.
   This also allows to make WebrtcTransport::Start() synchronous.
2. The direction in which offer and answer is sent is reversed. The
   host sends offer to the client and the client replies with an answer.
3. Offer is generated only after OnRenegotiationNeeded() notification is
   received.
4. The |is_server| flag in WebrtcDataStreamAdapter is replaced with
   |outgoing|, so connections may be created in either direction.

BUG=547158

Review URL: https://codereview.chromium.org/1510333002

Cr-Commit-Position: refs/heads/master@{#364155}

5 files changed, 123 insertions, 99 deletions

diff --git a/remoting/protocol/webrtc_data_stream_adapter.cc b/remoting/protocol/webrtc_data_stream_adapter.cc
index 9c1dac1..b53800f 100644
--- a/remoting/protocol/webrtc_data_stream_adapter.cc
+++ b/remoting/protocol/webrtc_data_stream_adapter.cc
@@ -209,11 +209,11 @@ WebrtcDataStreamAdapter::~WebrtcDataStreamAdapter() {
 
 void WebrtcDataStreamAdapter::Initialize(
     rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection,
-    bool is_server) {
+    bool outgoing) {
   peer_connection_ = peer_connection;
-  is_server_ = is_server;
+  outgoing_ = outgoing;
 
- if (!is_server_) {
+ if (outgoing_) {
     for (auto& channel : pending_channels_) {
       webrtc::DataChannelInit config;
       config.reliable = true;
@@ -226,7 +226,7 @@ void WebrtcDataStreamAdapter::Initialize(
 void WebrtcDataStreamAdapter::OnIncomingDataChannel(
     webrtc::DataChannelInterface* data_channel) {
   auto it = pending_channels_.find(data_channel->label());
-  if (!is_server_ || it == pending_channels_.end()) {
+  if (outgoing_ || it == pending_channels_.end()) {
     LOG(ERROR) << "Received unexpected data channel " << data_channel->label();
     return;
   }
@@ -243,7 +243,7 @@ void WebrtcDataStreamAdapter::CreateChannel(
                              base::Unretained(this), callback));
   pending_channels_[name] = channel;
 
-  if (peer_connection_ && !is_server_) {
+  if (peer_connection_ && outgoing_) {
     webrtc::DataChannelInit config;
     config.reliable = true;
     channel->Start(peer_connection_->CreateDataChannel(name, &config));
diff --git a/remoting/protocol/webrtc_data_stream_adapter.h b/remoting/protocol/webrtc_data_stream_adapter.h
index 1200db6..fc7e42f 100644
--- a/remoting/protocol/webrtc_data_stream_adapter.h
+++ b/remoting/protocol/webrtc_data_stream_adapter.h
@@ -20,15 +20,21 @@ class PeerConnectionInterface;
 namespace remoting {
 namespace protocol {
 
+// WebrtcDataStreamAdapter is a StreamChannelFactory that creates channels that
+// send and receive data over PeerConnection data channels.
 class WebrtcDataStreamAdapter : public StreamChannelFactory {
  public:
   WebrtcDataStreamAdapter();
   ~WebrtcDataStreamAdapter() override;
 
+  // Initializes the adapter for |peer_connection|. If |outgoing| is set to true
+  // all channels will be created as outgoing. Otherwise CreateChannel() will
+  // wait for the other end to create connection.
   void Initialize(
       rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection,
-      bool is_server);
+      bool outgoing);
 
+  // Called by WebrtcTransport.
   void OnIncomingDataChannel(webrtc::DataChannelInterface* data_channel);
 
   // StreamChannelFactory interface.
@@ -44,7 +50,7 @@ class WebrtcDataStreamAdapter : public StreamChannelFactory {
                           bool connected);
 
   rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection_;
-  bool is_server_ = false;
+  bool outgoing_ = false;
 
   std::map<std::string, Channel*> pending_channels_;
 
diff --git a/remoting/protocol/webrtc_transport.cc b/remoting/protocol/webrtc_transport.cc
index 3a6ad6f..6dcbb97 100644
--- a/remoting/protocol/webrtc_transport.cc
+++ b/remoting/protocol/webrtc_transport.cc
@@ -8,6 +8,7 @@
 #include "base/single_thread_task_runner.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/task_runner_util.h"
+#include "base/thread_task_runner_handle.h"
 #include "jingle/glue/thread_wrapper.h"
 #include "third_party/libjingle/source/talk/app/webrtc/test/fakeconstraints.h"
 #include "third_party/webrtc/libjingle/xmllite/xmlelement.h"
@@ -29,15 +30,6 @@ const int kTransportInfoSendDelayMs = 20;
 // XML namespace for the transport elements.
 const char kTransportNamespace[] = "google:remoting:webrtc";
 
-rtc::Thread* InitAndGetRtcThread() {
-  jingle_glue::JingleThreadWrapper::EnsureForCurrentMessageLoop();
-
-  // TODO(sergeyu): Investigate if it's possible to avoid Send().
-  jingle_glue::JingleThreadWrapper::current()->set_send_allowed(true);
-
-  return jingle_glue::JingleThreadWrapper::current();
-}
-
 // A webrtc::CreateSessionDescriptionObserver implementation used to receive the
 // results of creating descriptions for this end of the PeerConnection.
 class CreateSessionDescriptionObserver
@@ -108,13 +100,13 @@ class SetSessionDescriptionObserver
 }  // namespace
 
 WebrtcTransport::WebrtcTransport(
+    rtc::Thread* worker_thread,
     rtc::scoped_refptr<webrtc::PortAllocatorFactoryInterface>
         port_allocator_factory,
-    TransportRole role,
-    scoped_refptr<base::SingleThreadTaskRunner> worker_task_runner)
+    TransportRole role)
     : port_allocator_factory_(port_allocator_factory),
       role_(role),
-      worker_task_runner_(worker_task_runner),
+      worker_thread_(worker_thread),
       weak_factory_(this) {}
 
 WebrtcTransport::~WebrtcTransport() {}
@@ -126,10 +118,34 @@ void WebrtcTransport::Start(EventHandler* event_handler,
   event_handler_ = event_handler;
 
   // TODO(sergeyu): Use the |authenticator| to authenticate PeerConnection.
+  jingle_glue::JingleThreadWrapper::EnsureForCurrentMessageLoop();
+
+  // TODO(sergeyu): Investigate if it's possible to avoid Send().
+  jingle_glue::JingleThreadWrapper::current()->set_send_allowed(true);
+
+  fake_audio_device_module_.reset(new webrtc::FakeAudioDeviceModule());
+
+  peer_connection_factory_ = webrtc::CreatePeerConnectionFactory(
+      worker_thread_, rtc::Thread::Current(),
+      fake_audio_device_module_.get(), nullptr, nullptr);
+
+  webrtc::PeerConnectionInterface::IceServer stun_server;
+  stun_server.urls.push_back("stun:stun.l.google.com:19302");
+  webrtc::PeerConnectionInterface::RTCConfiguration rtc_config;
+  rtc_config.servers.push_back(stun_server);
+
+  webrtc::FakeConstraints constraints;
+  constraints.AddMandatory(webrtc::MediaConstraintsInterface::kEnableDtlsSrtp,
+                           webrtc::MediaConstraintsInterface::kValueTrue);
+
+  peer_connection_ = peer_connection_factory_->CreatePeerConnection(
+      rtc_config, &constraints, port_allocator_factory_, nullptr, this);
+
+  data_stream_adapter_.Initialize(peer_connection_,
+                                  role_ == TransportRole::SERVER);
 
-  base::PostTaskAndReplyWithResult(
-      worker_task_runner_.get(), FROM_HERE, base::Bind(&InitAndGetRtcThread),
-      base::Bind(&WebrtcTransport::DoStart, weak_factory_.GetWeakPtr()));
+  if (role_ == TransportRole::SERVER)
+    RequestNegotiation();
 }
 
 bool WebrtcTransport::ProcessTransportInfo(XmlElement* transport_info) {
@@ -145,7 +161,7 @@ bool WebrtcTransport::ProcessTransportInfo(XmlElement* transport_info) {
       QName(kTransportNamespace, "session-description"));
   if (session_description) {
     webrtc::PeerConnectionInterface::SignalingState expected_state =
-        role_ == TransportRole::SERVER
+        role_ == TransportRole::CLIENT
             ? webrtc::PeerConnectionInterface::kStable
             : webrtc::PeerConnectionInterface::kHaveLocalOffer;
     if (peer_connection_->signaling_state() != expected_state) {
@@ -156,7 +172,7 @@ bool WebrtcTransport::ProcessTransportInfo(XmlElement* transport_info) {
     std::string type = session_description->Attr(QName(std::string(), "type"));
     std::string sdp = session_description->BodyText();
     if (type.empty() || sdp.empty()) {
-      LOG(ERROR) << "Incorrect session_description format.";
+      LOG(ERROR) << "Incorrect session description format.";
       return false;
     }
 
@@ -164,15 +180,16 @@ bool WebrtcTransport::ProcessTransportInfo(XmlElement* transport_info) {
     scoped_ptr<webrtc::SessionDescriptionInterface> session_description(
         webrtc::CreateSessionDescription(type, sdp, &error));
     if (!session_description) {
-      LOG(ERROR) << "Failed to parse the offer: " << error.description
-                 << " line: " << error.line;
+      LOG(ERROR) << "Failed to parse the session description: "
+                 << error.description << " line: " << error.line;
       return false;
     }
 
     peer_connection_->SetRemoteDescription(
         SetSessionDescriptionObserver::Create(
             base::Bind(&WebrtcTransport::OnRemoteDescriptionSet,
-                       weak_factory_.GetWeakPtr())),
+                       weak_factory_.GetWeakPtr(),
+                       type == webrtc::SessionDescriptionInterface::kOffer)),
         session_description.release());
   }
 
@@ -227,54 +244,6 @@ StreamChannelFactory* WebrtcTransport::GetMultiplexedChannelFactory() {
   return GetStreamChannelFactory();
 }
 
-void WebrtcTransport::DoStart(rtc::Thread* worker_thread) {
-  DCHECK(thread_checker_.CalledOnValidThread());
-
-  jingle_glue::JingleThreadWrapper::EnsureForCurrentMessageLoop();
-
-  // TODO(sergeyu): Investigate if it's possible to avoid Send().
-  jingle_glue::JingleThreadWrapper::current()->set_send_allowed(true);
-
-  fake_audio_device_module_.reset(new webrtc::FakeAudioDeviceModule());
-
-  peer_connection_factory_ = webrtc::CreatePeerConnectionFactory(
-      worker_thread, rtc::Thread::Current(),
-      fake_audio_device_module_.get(), nullptr, nullptr);
-
-  webrtc::PeerConnectionInterface::IceServer stun_server;
-  stun_server.urls.push_back("stun:stun.l.google.com:19302");
-  webrtc::PeerConnectionInterface::RTCConfiguration rtc_config;
-  rtc_config.servers.push_back(stun_server);
-
-  webrtc::FakeConstraints constraints;
-  constraints.AddMandatory(webrtc::MediaConstraintsInterface::kEnableDtlsSrtp,
-                           webrtc::MediaConstraintsInterface::kValueTrue);
-
-  peer_connection_ = peer_connection_factory_->CreatePeerConnection(
-      rtc_config, &constraints, port_allocator_factory_, nullptr, this);
-
-  data_stream_adapter_.Initialize(peer_connection_,
-                                  role_ == TransportRole::SERVER);
-
-  if (role_ == TransportRole::CLIENT) {
-    webrtc::FakeConstraints offer_config;
-    offer_config.AddMandatory(
-        webrtc::MediaConstraintsInterface::kOfferToReceiveVideo,
-        webrtc::MediaConstraintsInterface::kValueTrue);
-    offer_config.AddMandatory(
-        webrtc::MediaConstraintsInterface::kOfferToReceiveAudio,
-        webrtc::MediaConstraintsInterface::kValueFalse);
-    offer_config.AddMandatory(
-        webrtc::MediaConstraintsInterface::kEnableDtlsSrtp,
-        webrtc::MediaConstraintsInterface::kValueTrue);
-    peer_connection_->CreateOffer(
-        CreateSessionDescriptionObserver::Create(
-            base::Bind(&WebrtcTransport::OnLocalSessionDescriptionCreated,
-                       weak_factory_.GetWeakPtr())),
-        &offer_config);
-  }
-}
-
 void WebrtcTransport::OnLocalSessionDescriptionCreated(
     scoped_ptr<webrtc::SessionDescriptionInterface> description,
     const std::string& error) {
@@ -329,7 +298,8 @@ void WebrtcTransport::OnLocalDescriptionSet(bool success,
   AddPendingCandidatesIfPossible();
 }
 
-void WebrtcTransport::OnRemoteDescriptionSet(bool success,
+void WebrtcTransport::OnRemoteDescriptionSet(bool send_answer,
+                                             bool success,
                                              const std::string& error) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
@@ -343,7 +313,7 @@ void WebrtcTransport::OnRemoteDescriptionSet(bool success,
   }
 
   // Create and send answer on the server.
-  if (role_ == TransportRole::SERVER) {
+  if (send_answer) {
     peer_connection_->CreateAnswer(
         CreateSessionDescriptionObserver::Create(
             base::Bind(&WebrtcTransport::OnLocalSessionDescriptionCreated,
@@ -389,7 +359,25 @@ void WebrtcTransport::OnDataChannel(
 
 void WebrtcTransport::OnRenegotiationNeeded() {
   DCHECK(thread_checker_.CalledOnValidThread());
-  // TODO(sergeyu): Figure out what needs to happen here.
+
+  if (role_ == TransportRole::SERVER) {
+    RequestNegotiation();
+  } else {
+    // TODO(sergeyu): Is it necessary to support renegotiation initiated by the
+    // client?
+    NOTIMPLEMENTED();
+  }
+}
+
+void WebrtcTransport::RequestNegotiation() {
+  DCHECK(role_ == TransportRole::SERVER);
+
+  if (!negotiation_pending_) {
+    negotiation_pending_ = true;
+    base::ThreadTaskRunnerHandle::Get()->PostTask(
+        FROM_HERE,
+        base::Bind(&WebrtcTransport::SendOffer, weak_factory_.GetWeakPtr()));
+  }
 }
 
 void WebrtcTransport::OnIceConnectionChange(
@@ -446,6 +434,28 @@ void WebrtcTransport::EnsurePendingTransportInfoMessage() {
   }
 }
 
+void WebrtcTransport::SendOffer() {
+  DCHECK(role_ == TransportRole::SERVER);
+
+  DCHECK(negotiation_pending_);
+  negotiation_pending_ = false;
+
+  webrtc::FakeConstraints offer_config;
+  offer_config.AddMandatory(
+      webrtc::MediaConstraintsInterface::kOfferToReceiveVideo,
+      webrtc::MediaConstraintsInterface::kValueTrue);
+  offer_config.AddMandatory(
+      webrtc::MediaConstraintsInterface::kOfferToReceiveAudio,
+      webrtc::MediaConstraintsInterface::kValueFalse);
+  offer_config.AddMandatory(webrtc::MediaConstraintsInterface::kEnableDtlsSrtp,
+                            webrtc::MediaConstraintsInterface::kValueTrue);
+  peer_connection_->CreateOffer(
+      CreateSessionDescriptionObserver::Create(
+          base::Bind(&WebrtcTransport::OnLocalSessionDescriptionCreated,
+                     weak_factory_.GetWeakPtr())),
+      &offer_config);
+}
+
 void WebrtcTransport::SendTransportInfo() {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(pending_transport_info_message_);
@@ -472,23 +482,21 @@ void WebrtcTransport::AddPendingCandidatesIfPossible() {
 }
 
 WebrtcTransportFactory::WebrtcTransportFactory(
+    rtc::Thread* worker_thread,
     SignalStrategy* signal_strategy,
     rtc::scoped_refptr<webrtc::PortAllocatorFactoryInterface>
         port_allocator_factory,
     TransportRole role)
-    : signal_strategy_(signal_strategy),
+    : worker_thread_(worker_thread),
+      signal_strategy_(signal_strategy),
       port_allocator_factory_(port_allocator_factory),
-      role_(role),
-      worker_thread_("ChromotingWebrtcWorkerThread") {
-  worker_thread_.StartWithOptions(
-      base::Thread::Options(base::MessageLoop::TYPE_IO, 0));
-}
+      role_(role) {}
 
 WebrtcTransportFactory::~WebrtcTransportFactory() {}
 
 scoped_ptr<Transport> WebrtcTransportFactory::CreateTransport() {
-  return make_scoped_ptr(new WebrtcTransport(port_allocator_factory_, role_,
-                                             worker_thread_.task_runner()));
+  return make_scoped_ptr(
+      new WebrtcTransport(worker_thread_, port_allocator_factory_, role_));
 }
 
 }  // namespace protocol
diff --git a/remoting/protocol/webrtc_transport.h b/remoting/protocol/webrtc_transport.h
index 74ec30c..ae792f2 100644
--- a/remoting/protocol/webrtc_transport.h
+++ b/remoting/protocol/webrtc_transport.h
@@ -9,7 +9,6 @@
 #include "base/memory/scoped_ptr.h"
 #include "base/memory/scoped_vector.h"
 #include "base/memory/weak_ptr.h"
-#include "base/threading/thread.h"
 #include "base/threading/thread_checker.h"
 #include "base/timer/timer.h"
 #include "remoting/protocol/transport.h"
@@ -27,13 +26,19 @@ namespace protocol {
 class WebrtcTransport : public Transport,
                         public webrtc::PeerConnectionObserver {
  public:
-  WebrtcTransport(
-      rtc::scoped_refptr<webrtc::PortAllocatorFactoryInterface>
-          port_allocator_factory,
-      TransportRole role,
-      scoped_refptr<base::SingleThreadTaskRunner> worker_task_runner);
+  WebrtcTransport(rtc::Thread* worker_thread,
+                  rtc::scoped_refptr<webrtc::PortAllocatorFactoryInterface>
+                      port_allocator_factory,
+                  TransportRole role);
   ~WebrtcTransport() override;
 
+  webrtc::PeerConnectionInterface* peer_connection() {
+    return peer_connection_;
+  }
+  webrtc::PeerConnectionFactoryInterface* peer_connection_factory() {
+    return peer_connection_factory_;
+  }
+
   // Transport interface.
   void Start(EventHandler* event_handler,
              Authenticator* authenticator) override;
@@ -42,12 +47,13 @@ class WebrtcTransport : public Transport,
   StreamChannelFactory* GetMultiplexedChannelFactory() override;
 
  private:
-  void DoStart(rtc::Thread* worker_thread);
   void OnLocalSessionDescriptionCreated(
       scoped_ptr<webrtc::SessionDescriptionInterface> description,
       const std::string& error);
   void OnLocalDescriptionSet(bool success, const std::string& error);
-  void OnRemoteDescriptionSet(bool success, const std::string& error);
+  void OnRemoteDescriptionSet(bool send_answer,
+                              bool success,
+                              const std::string& error);
 
   // webrtc::PeerConnectionObserver interface.
   void OnSignalingChange(
@@ -62,6 +68,8 @@ class WebrtcTransport : public Transport,
       webrtc::PeerConnectionInterface::IceGatheringState new_state) override;
   void OnIceCandidate(const webrtc::IceCandidateInterface* candidate) override;
 
+  void RequestNegotiation();
+  void SendOffer();
   void EnsurePendingTransportInfoMessage();
   void SendTransportInfo();
   void AddPendingCandidatesIfPossible();
@@ -74,7 +82,7 @@ class WebrtcTransport : public Transport,
       port_allocator_factory_;
   TransportRole role_;
   EventHandler* event_handler_ = nullptr;
-  scoped_refptr<base::SingleThreadTaskRunner> worker_task_runner_;
+  rtc::Thread* worker_thread_;
 
   scoped_ptr<webrtc::FakeAudioDeviceModule> fake_audio_device_module_;
 
@@ -82,6 +90,8 @@ class WebrtcTransport : public Transport,
       peer_connection_factory_;
   rtc::scoped_refptr<webrtc::PeerConnectionInterface> peer_connection_;
 
+  bool negotiation_pending_ = false;
+
   scoped_ptr<buzz::XmlElement> pending_transport_info_message_;
   base::OneShotTimer transport_info_timer_;
 
@@ -100,6 +110,7 @@ class WebrtcTransport : public Transport,
 class WebrtcTransportFactory : public TransportFactory {
  public:
   WebrtcTransportFactory(
+      rtc::Thread* worker_thread,
       SignalStrategy* signal_strategy,
       rtc::scoped_refptr<webrtc::PortAllocatorFactoryInterface>
           port_allocator_factory,
@@ -110,13 +121,12 @@ class WebrtcTransportFactory : public TransportFactory {
   scoped_ptr<Transport> CreateTransport() override;
 
  private:
+  rtc::Thread* worker_thread_;
   SignalStrategy* signal_strategy_;
   rtc::scoped_refptr<webrtc::PortAllocatorFactoryInterface>
       port_allocator_factory_;
   TransportRole role_;
 
-  base::Thread worker_thread_;
-
   DISALLOW_COPY_AND_ASSIGN(WebrtcTransportFactory);
 };
 
diff --git a/remoting/protocol/webrtc_transport_unittest.cc b/remoting/protocol/webrtc_transport_unittest.cc
index d327085..1797a7e 100644
--- a/remoting/protocol/webrtc_transport_unittest.cc
+++ b/remoting/protocol/webrtc_transport_unittest.cc
@@ -90,7 +90,7 @@ class WebrtcTransportTest : public testing::Test {
     signal_strategy_.reset(new FakeSignalStrategy(kTestJid));
 
     host_transport_factory_.reset(new WebrtcTransportFactory(
-        signal_strategy_.get(),
+        jingle_glue::JingleThreadWrapper::current(), signal_strategy_.get(),
         ChromiumPortAllocatorFactory::Create(network_settings_, nullptr),
         TransportRole::SERVER));
     host_transport_ = host_transport_factory_->CreateTransport();
@@ -98,7 +98,7 @@ class WebrtcTransportTest : public testing::Test {
         FakeAuthenticator::HOST, 0, FakeAuthenticator::ACCEPT, false));
 
     client_transport_factory_.reset(new WebrtcTransportFactory(
-        signal_strategy_.get(),
+        jingle_glue::JingleThreadWrapper::current(), signal_strategy_.get(),
         ChromiumPortAllocatorFactory::Create(network_settings_, nullptr),
         TransportRole::CLIENT));
     client_transport_ = client_transport_factory_->CreateTransport();