Revert of Add TransportSession interface to prepare for WebRTC-based transport.

remoting/protocol/jingle_session.cc

 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 #include "remoting/protocol/jingle_session.h"
 
 #include "base/bind.h"
 #include "base/rand_util.h"
 #include "base/single_thread_task_runner.h"
 #include "base/stl_util.h"
 #include "base/strings/string_number_conversions.h"
 #include "base/thread_task_runner_handle.h"
 #include "base/time/time.h"
 #include "remoting/base/constants.h"
 #include "remoting/protocol/authenticator.h"
+#include "remoting/protocol/channel_authenticator.h"
+#include "remoting/protocol/channel_multiplexer.h"
 #include "remoting/protocol/content_description.h"
 #include "remoting/protocol/jingle_messages.h"
 #include "remoting/protocol/jingle_session_manager.h"
+#include "remoting/protocol/pseudotcp_channel_factory.h"
 #include "remoting/protocol/quic_channel_factory.h"
+#include "remoting/protocol/secure_channel_factory.h"
 #include "remoting/protocol/session_config.h"
+#include "remoting/protocol/stream_channel_factory.h"
 #include "remoting/signaling/iq_sender.h"
 #include "third_party/webrtc/libjingle/xmllite/xmlelement.h"
 #include "third_party/webrtc/p2p/base/candidate.h"
 
 using buzz::XmlElement;
 
 namespace remoting {
 namespace protocol {
 
 namespace {
 
+// Delay after candidate creation before sending transport-info message to
+// accumulate multiple candidates. This is an optimization to reduce number of
+// transport-info messages.
+const int kTransportInfoSendDelayMs = 20;
+
 // How long we should wait for a response from the other end. This value is used
 // for all requests except |transport-info|.
 const int kDefaultMessageTimeout = 10;
 
 // During a reconnection, it usually takes longer for the peer to respond due to
 // pending messages in the channel from the previous session.  From experiment,
 // it can take up to 20s for the session to reconnect. To make it safe, setting
 // the timeout to 30s.
 const int kSessionInitiateAndAcceptTimeout = kDefaultMessageTimeout * 3;
 
 // Timeout for the transport-info messages.
 const int kTransportInfoTimeout = 10 * 60;
 
+// Name of the multiplexed channel.
+const char kMuxChannelName[] = "mux";
+
 ErrorCode AuthRejectionReasonToErrorCode(
     Authenticator::RejectionReason reason) {
   switch (reason) {
     case Authenticator::INVALID_CREDENTIALS:
       return AUTHENTICATION_FAILED;
     case Authenticator::PROTOCOL_ERROR:
       return INCOMPATIBLE_PROTOCOL;
   }
   NOTREACHED();
   return UNKNOWN_ERROR;
 }
 
 }  // namespace
 
 JingleSession::JingleSession(JingleSessionManager* session_manager)
     : session_manager_(session_manager),
       event_handler_(nullptr),
       state_(INITIALIZING),
       error_(OK),
       weak_factory_(this) {
 }
 
 JingleSession::~JingleSession() {
+  channel_multiplexer_.reset();
   quic_channel_factory_.reset();
-  transport_session_.reset();
-
   STLDeleteContainerPointers(pending_requests_.begin(),
                              pending_requests_.end());
   STLDeleteContainerPointers(transport_info_requests_.begin(),
                              transport_info_requests_.end());
 
+  DCHECK(channels_.empty());
+
   session_manager_->SessionDestroyed(this);
 }
 
 void JingleSession::SetEventHandler(Session::EventHandler* event_handler) {
   DCHECK(CalledOnValidThread());
   DCHECK(event_handler);
   event_handler_ = event_handler;
 }
 
 ErrorCode JingleSession::error() {
   DCHECK(CalledOnValidThread());
   return error_;
 }
 
 void JingleSession::StartConnection(const std::string& peer_jid,
                                     scoped_ptr<Authenticator> authenticator) {
   DCHECK(CalledOnValidThread());
   DCHECK(authenticator.get());
   DCHECK_EQ(authenticator->state(), Authenticator::MESSAGE_READY);
 
   peer_jid_ = peer_jid;
   authenticator_ = authenticator.Pass();
 
   // Generate random session ID. There are usually not more than 1
   // concurrent session per host, so a random 64-bit integer provides
   // enough entropy. In the worst case connection will fail when two
   // clients generate the same session ID concurrently.
   session_id_ = base::Uint64ToString(base::RandGenerator(kuint64max));
 
-  transport_session_ =
-      session_manager_->transport_factory_->CreateTransportSession();
   quic_channel_factory_.reset(new QuicChannelFactory(session_id_, false));
 
   // Send session-initiate message.
   JingleMessage message(peer_jid_, JingleMessage::SESSION_INITIATE,
                         session_id_);
   message.initiator = session_manager_->signal_strategy_->GetLocalJid();
   message.description.reset(new ContentDescription(
       session_manager_->protocol_config_->Clone(),
       authenticator_->GetNextMessage(),
       quic_channel_factory_->CreateSessionInitiateConfigMessage()));
@@ skipping 26 matching lines @@
     return;
   }
 
   if (config_->is_using_quic()) {
     quic_channel_factory_.reset(new QuicChannelFactory(session_id_, true));
     if (!quic_channel_factory_->ProcessSessionInitiateConfigMessage(
             initiate_message.description->quic_config_message())) {
       CloseInternal(INCOMPATIBLE_PROTOCOL);
     }
   }
-
-  transport_session_ =
-      session_manager_->transport_factory_->CreateTransportSession();
 }
 
 void JingleSession::AcceptIncomingConnection(
     const JingleMessage& initiate_message) {
   DCHECK(config_);
 
   // Process the first authentication message.
   const buzz::XmlElement* first_auth_message =
       initiate_message.description->authenticator_message();
 
@@ skipping 49 matching lines @@
 const std::string& JingleSession::jid() {
   DCHECK(CalledOnValidThread());
   return peer_jid_;
 }
 
 const SessionConfig& JingleSession::config() {
   DCHECK(CalledOnValidThread());
   return *config_;
 }
 
-TransportSession* JingleSession::GetTransportSession() {
+StreamChannelFactory* JingleSession::GetTransportChannelFactory() {
   DCHECK(CalledOnValidThread());
-  return transport_session_.get();
+  return secure_channel_factory_.get();
+}
+
+StreamChannelFactory* JingleSession::GetMultiplexedChannelFactory() {
+  DCHECK(CalledOnValidThread());
+  if (!channel_multiplexer_.get()) {
+    channel_multiplexer_.reset(
+        new ChannelMultiplexer(GetTransportChannelFactory(), kMuxChannelName));
+  }
+  return channel_multiplexer_.get();
 }
 
 StreamChannelFactory* JingleSession::GetQuicChannelFactory() {
   DCHECK(CalledOnValidThread());
   return quic_channel_factory_.get();
 }
 
 void JingleSession::Close() {
   DCHECK(CalledOnValidThread());
 
   CloseInternal(OK);
 }
 
+void JingleSession::AddPendingRemoteTransportInfo(Transport* channel) {
+  std::list<JingleMessage::IceCredentials>::iterator credentials =
+      pending_remote_ice_credentials_.begin();
+  while (credentials != pending_remote_ice_credentials_.end()) {
+    if (credentials->channel == channel->name()) {
+      channel->SetRemoteCredentials(credentials->ufrag, credentials->password);
+      credentials = pending_remote_ice_credentials_.erase(credentials);
+    } else {
+      ++credentials;
+    }
+  }
+
+  std::list<JingleMessage::NamedCandidate>::iterator candidate =
+      pending_remote_candidates_.begin();
+  while (candidate != pending_remote_candidates_.end()) {
+    if (candidate->name == channel->name()) {
+      channel->AddRemoteCandidate(candidate->candidate);
+      candidate = pending_remote_candidates_.erase(candidate);
+    } else {
+      ++candidate;
+    }
+  }
+}
+
+void JingleSession::CreateChannel(const std::string& name,
+                                  const ChannelCreatedCallback& callback) {
+  DCHECK(!channels_[name]);
+
+  scoped_ptr<Transport> channel =
+      session_manager_->transport_factory_->CreateTransport();
+  channel->Connect(name, this, callback);
+  AddPendingRemoteTransportInfo(channel.get());
+  channels_[name] = channel.release();
+}
+
+void JingleSession::CancelChannelCreation(const std::string& name) {
+  ChannelsMap::iterator it = channels_.find(name);
+  if (it != channels_.end()) {
+    DCHECK(!it->second->is_connected());
+    delete it->second;
+    DCHECK(channels_.find(name) == channels_.end());
+  }
+}
+
+void JingleSession::OnTransportIceCredentials(Transport* transport,
+                                              const std::string& ufrag,
+                                              const std::string& password) {
+  EnsurePendingTransportInfoMessage();
+  pending_transport_info_message_->ice_credentials.push_back(
+      JingleMessage::IceCredentials(transport->name(), ufrag, password));
+}
+
+void JingleSession::OnTransportCandidate(Transport* transport,
+                                         const cricket::Candidate& candidate) {
+  EnsurePendingTransportInfoMessage();
+  pending_transport_info_message_->candidates.push_back(
+      JingleMessage::NamedCandidate(transport->name(), candidate));
+}
+
+void JingleSession::OnTransportRouteChange(Transport* transport,
+                                           const TransportRoute& route) {
+  if (event_handler_)
+    event_handler_->OnSessionRouteChange(transport->name(), route);
+}
+
+void JingleSession::OnTransportFailed(Transport* transport) {
+  CloseInternal(CHANNEL_CONNECTION_ERROR);
+}
+
+void JingleSession::OnTransportDeleted(Transport* transport) {
+  ChannelsMap::iterator it = channels_.find(transport->name());
+  DCHECK_EQ(it->second, transport);
+  channels_.erase(it);
+}
+
 void JingleSession::SendMessage(const JingleMessage& message) {
   scoped_ptr<IqRequest> request = session_manager_->iq_sender()->SendIq(
       message.ToXml(),
       base::Bind(&JingleSession::OnMessageResponse,
                  base::Unretained(this),
                  message.action));
 
   int timeout = kDefaultMessageTimeout;
   if (message.action == JingleMessage::SESSION_INITIATE ||
       message.action == JingleMessage::SESSION_ACCEPT) {
@@ skipping 35 matching lines @@
                  << " message: \"" << response->Str()
                  << "\". Terminating the session.";
 
       // TODO(sergeyu): There may be different reasons for error
       // here. Parse the response stanza to find failure reason.
       CloseInternal(PEER_IS_OFFLINE);
     }
   }
 }
 
-void JingleSession::OnOutgoingTransportInfo(
-    scoped_ptr<XmlElement> transport_info) {
-  JingleMessage message(peer_jid_, JingleMessage::TRANSPORT_INFO, session_id_);
-  message.transport_info = transport_info.Pass();
+void JingleSession::EnsurePendingTransportInfoMessage() {
+  // |transport_info_timer_| must be running iff
+  // |pending_transport_info_message_| exists.
+  DCHECK_EQ(pending_transport_info_message_ != nullptr,
+            transport_info_timer_.IsRunning());
+
+  if (!pending_transport_info_message_) {
+    pending_transport_info_message_.reset(new JingleMessage(
+        peer_jid_, JingleMessage::TRANSPORT_INFO, session_id_));
+    // Delay sending the new candidates in case we get more candidates
+    // that we can send in one message.
+    transport_info_timer_.Start(
+        FROM_HERE, base::TimeDelta::FromMilliseconds(kTransportInfoSendDelayMs),
+        this, &JingleSession::SendTransportInfo);
+  }
+}
+
+void JingleSession::SendTransportInfo() {
+  DCHECK(pending_transport_info_message_);
 
   scoped_ptr<IqRequest> request = session_manager_->iq_sender()->SendIq(
-      message.ToXml(), base::Bind(&JingleSession::OnTransportInfoResponse,
-                                  base::Unretained(this)));
+      pending_transport_info_message_->ToXml(),
+      base::Bind(&JingleSession::OnTransportInfoResponse,
+                 base::Unretained(this)));
+  pending_transport_info_message_.reset();
   if (request) {
     request->SetTimeout(base::TimeDelta::FromSeconds(kTransportInfoTimeout));
     transport_info_requests_.push_back(request.release());
   } else {
     LOG(ERROR) << "Failed to send a transport-info message";
   }
 }
 
-void JingleSession::OnTransportRouteChange(const std::string& channel_name,
-                                           const TransportRoute& route) {
-  event_handler_->OnSessionRouteChange(channel_name, route);
-}
-
-void JingleSession::OnTransportError(ErrorCode error) {
-  CloseInternal(error);
-}
-
 void JingleSession::OnTransportInfoResponse(IqRequest* request,
                                             const buzz::XmlElement* response) {
   DCHECK(!transport_info_requests_.empty());
 
   // Consider transport-info requests sent before this one lost and delete
   // corresponding IqRequest objects.
   while (transport_info_requests_.front() != request) {
     delete transport_info_requests_.front();
     transport_info_requests_.pop_front();
   }
@@ skipping 30 matching lines @@
   switch (message.action) {
     case JingleMessage::SESSION_ACCEPT:
       OnAccept(message, reply_callback);
       break;
 
     case JingleMessage::SESSION_INFO:
       OnSessionInfo(message, reply_callback);
       break;
 
     case JingleMessage::TRANSPORT_INFO:
-      if (transport_session_->ProcessTransportInfo(
-              message.transport_info.get())) {
-        reply_callback.Run(JingleMessageReply::NONE);
-      } else {
-        reply_callback.Run(JingleMessageReply::BAD_REQUEST);
-      }
+      reply_callback.Run(JingleMessageReply::NONE);
+      ProcessTransportInfo(message);
       break;
 
     case JingleMessage::SESSION_TERMINATE:
       OnTerminate(message, reply_callback);
       break;
 
     default:
       reply_callback.Run(JingleMessageReply::UNEXPECTED_REQUEST);
   }
 }
@@ skipping 53 matching lines @@
     CloseInternal(INCOMPATIBLE_PROTOCOL);
     return;
   }
 
   reply_callback.Run(JingleMessageReply::NONE);
 
   authenticator_->ProcessMessage(message.info.get(), base::Bind(
       &JingleSession::ProcessAuthenticationStep, base::Unretained(this)));
 }
 
+void JingleSession::ProcessTransportInfo(const JingleMessage& message) {
+  for (std::list<JingleMessage::IceCredentials>::const_iterator it =
+           message.ice_credentials.begin();
+       it != message.ice_credentials.end(); ++it) {
+    ChannelsMap::iterator channel = channels_.find(it->channel);
+    if (channel != channels_.end()) {
+      channel->second->SetRemoteCredentials(it->ufrag, it->password);
+    } else {
+      // Transport info was received before the channel was created.
+      // This could happen due to messages being reordered on the wire.
+      pending_remote_ice_credentials_.push_back(*it);
+    }
+  }
+
+  for (std::list<JingleMessage::NamedCandidate>::const_iterator it =
+           message.candidates.begin();
+       it != message.candidates.end(); ++it) {
+    ChannelsMap::iterator channel = channels_.find(it->name);
+    if (channel != channels_.end()) {
+      channel->second->AddRemoteCandidate(it->candidate);
+    } else {
+      // Transport info was received before the channel was created.
+      // This could happen due to messages being reordered on the wire.
+      pending_remote_candidates_.push_back(*it);
+    }
+  }
+}
+
 void JingleSession::OnTerminate(const JingleMessage& message,
                                 const ReplyCallback& reply_callback) {
   if (!is_session_active()) {
     LOG(WARNING) << "Received unexpected session-terminate message.";
     reply_callback.Run(JingleMessageReply::UNEXPECTED_REQUEST);
     return;
   }
 
   reply_callback.Run(JingleMessageReply::NONE);
 
@@ skipping 80 matching lines @@
 void JingleSession::ContinueAuthenticationStep() {
   if (authenticator_->state() == Authenticator::ACCEPTED) {
     OnAuthenticated();
   } else if (authenticator_->state() == Authenticator::REJECTED) {
     CloseInternal(AuthRejectionReasonToErrorCode(
         authenticator_->rejection_reason()));
   }
 }
 
 void JingleSession::OnAuthenticated() {
-  transport_session_->Start(this, authenticator_.get());
+  pseudotcp_channel_factory_.reset(new PseudoTcpChannelFactory(this));
+  secure_channel_factory_.reset(
+      new SecureChannelFactory(pseudotcp_channel_factory_.get(),
+                               authenticator_.get()));
 
-  if (quic_channel_factory_) {
-    quic_channel_factory_->Start(
-        transport_session_->GetDatagramChannelFactory(),
-        authenticator_->GetAuthKey());
-  }
+  if (quic_channel_factory_)
+    quic_channel_factory_->Start(this, authenticator_->GetAuthKey());
 
   SetState(AUTHENTICATED);
 }
 
 void JingleSession::CloseInternal(ErrorCode error) {
   DCHECK(CalledOnValidThread());
 
   if (is_session_active()) {
     // Send session-terminate message with the appropriate error code.
     JingleMessage::Reason reason;
@@ skipping 45 matching lines @@
   }
 }
 
 bool JingleSession::is_session_active() {
   return state_ == CONNECTING || state_ == ACCEPTING || state_ == CONNECTED ||
         state_ == AUTHENTICATING || state_ == AUTHENTICATED;
 }
 
 }  // namespace protocol
 }  // namespace remoting