Chromoting protocol implementation based on P2P Transport API.

remoting/protocol/pepper_session.cc

+// Copyright (c) 2011 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/pepper_session.h"
+
+#include "base/bind.h"
+#include "base/rand_util.h"
+#include "base/stl_util.h"
+#include "base/string_number_conversions.h"
+#include "remoting/base/constants.h"
+#include "remoting/jingle_glue/iq_request.h"
+#include "remoting/protocol/content_description.h"
+#include "remoting/protocol/jingle_messages.h"
+#include "remoting/protocol/pepper_session_manager.h"
+#include "remoting/protocol/pepper_stream_channel.h"
+#include "third_party/libjingle/source/talk/p2p/base/candidate.h"
+#include "third_party/libjingle/source/talk/xmllite/xmlelement.h"
+
+using buzz::XmlElement;
+
+namespace remoting {
+namespace protocol {
+
+namespace {
+// Delay after candidate creation before sending transport-info
+// message. This is neccessary to be able to pack multiple candidates
+// into one transport-info messages. The value needs to be greater
+// than zero because ports are opened asynchronously in the browser
+// process.
+const int kTransportInfoSendDelayMs = 2;
+}  // namespace
+
+PepperSession::PepperSession(PepperSessionManager* session_manager)
+    : session_manager_(session_manager),
+      state_(INITIALIZING),
+      error_(ERROR_NO_ERROR) {
+}
+
+PepperSession::~PepperSession() {
+  control_channel_socket_.reset();
+  event_channel_socket_.reset();
+  STLDeleteContainerPairSecondPointers(channels_.begin(), channels_.end());
+  session_manager_->SessionDestroyed(this);
+}
+
+PepperSession::Error PepperSession::error() {
+  DCHECK(CalledOnValidThread());
+  return error_;
+}
+
+void PepperSession::SetStateChangeCallback(StateChangeCallback* callback) {
+  DCHECK(CalledOnValidThread());
+  state_change_callback_.reset(callback);
+}
+
+void PepperSession::StartConnection(
+    const std::string& peer_jid,
+    const std::string& peer_public_key,
+    const std::string& client_token,
+    CandidateSessionConfig* config,
+    Session::StateChangeCallback* state_change_callback) {
+  DCHECK(CalledOnValidThread());
+
+  peer_jid_ = peer_jid;
+  peer_public_key_ = peer_public_key;
+  initiator_token_ = client_token;
+  candidate_config_.reset(config);
+  state_change_callback_.reset(state_change_callback);
+
+  // 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::Int64ToString(base::RandGenerator(kint64max));
+
+  // Send session-initiate message.
+  JingleMessage message(peer_jid_, JingleMessage::SESSION_INITIATE,
+                        session_id_);
+  message.from = session_manager_->local_jid_;
+  message.description.reset(
+      new ContentDescription(candidate_config_->Clone(), initiator_token_, ""));
+  initiate_request_.reset(session_manager_->CreateIqRequest());
+  initiate_request_->set_callback(base::Bind(
+      &PepperSession::OnSessionInitiateResponse, base::Unretained(this)));
+  initiate_request_->SendIq(message.ToXml());
+
+  SetState(CONNECTING);
+}
+
+void PepperSession::OnSessionInitiateResponse(
+    const buzz::XmlElement* response) {
+  const std::string& type = response->Attr(buzz::QName("", "type"));
+  if (type != "result") {
+    LOG(ERROR) << "Received error in response to session-initiate message: \""
+               << response->Str()
+               << "\" Terminating the session.";
+
+    // TODO(sergeyu): There may be different reasons for error
+    // here. Parse the response stanza to find failure reason.
+    OnError(ERROR_PEER_IS_OFFLINE);
+  }
+}
+
+void PepperSession::OnError(Error error) {
+  error_ = error;
+  CloseInternal(true);
+}
+
+void PepperSession::CreateStreamChannel(
+      const std::string& name,
+      const StreamChannelCallback& callback) {
+  DCHECK(!channels_[name]);
+
+  PepperStreamChannel* channel = new PepperStreamChannel(this, name, callback);
+  channels_[name] = channel;
+  channel->Connect(session_manager_->pp_instance_,
+                   session_manager_->transport_config_, remote_cert_);
+}
+
+void PepperSession::CreateDatagramChannel(
+    const std::string& name,
+    const DatagramChannelCallback& callback) {
+  // TODO(sergeyu): Implement datagram channel support.
+  NOTREACHED();
+}
+
+net::Socket* PepperSession::control_channel() {
+  DCHECK(CalledOnValidThread());
+  return control_channel_socket_.get();
+}
+
+net::Socket* PepperSession::event_channel() {
+  DCHECK(CalledOnValidThread());
+  return event_channel_socket_.get();
+}
+
+const std::string& PepperSession::jid() {
+  DCHECK(CalledOnValidThread());
+  return peer_jid_;
+}
+
+const CandidateSessionConfig* PepperSession::candidate_config() {
+  DCHECK(CalledOnValidThread());
+  return candidate_config_.get();
+}
+
+const SessionConfig& PepperSession::config() {
+  DCHECK(CalledOnValidThread());
+  return config_;
+}
+
+void PepperSession::set_config(const SessionConfig& config) {
+  DCHECK(CalledOnValidThread());
+  // set_config() should never be called on the client.
+  NOTREACHED();
+}
+
+const std::string& PepperSession::initiator_token() {
+  DCHECK(CalledOnValidThread());
+  return initiator_token_;
+}
+
+void PepperSession::set_initiator_token(const std::string& initiator_token) {
+  DCHECK(CalledOnValidThread());
+  initiator_token_ = initiator_token;
+}
+
+const std::string& PepperSession::receiver_token() {
+  DCHECK(CalledOnValidThread());
+  return receiver_token_;
+}
+
+void PepperSession::set_receiver_token(const std::string& receiver_token) {
+  DCHECK(CalledOnValidThread());
+  // set_receiver_token() should not be called on the client side.
+  NOTREACHED();
+}
+
+void PepperSession::set_shared_secret(const std::string& secret) {
+  DCHECK(CalledOnValidThread());
+  shared_secret_ = secret;
+}
+
+const std::string& PepperSession::shared_secret() {
+  DCHECK(CalledOnValidThread());
+  return shared_secret_;
+}
+
+void PepperSession::Close() {
+  DCHECK(CalledOnValidThread());
+
+  if (state_ == CONNECTING || state_ == CONNECTED ||
+      state_ == CONNECTED_CHANNELS) {
+    // Send session-terminate message.
+    JingleMessage message(peer_jid_, JingleMessage::SESSION_TERMINATE,
+                          session_id_);
+    scoped_ptr<IqRequest>  terminate_request(
+        session_manager_->CreateIqRequest());
+    terminate_request->SendIq(message.ToXml());
+  }
+
+  CloseInternal(false);
+}
+
+void PepperSession::OnIncomingMessage(const JingleMessage& message,
+                                      JingleMessageReply* reply) {
+  DCHECK(CalledOnValidThread());
+
+  if (message.from != peer_jid_) {
+    // Ignore messages received from a different Jid.
+    *reply = JingleMessageReply(JingleMessageReply::INVALID_SID);
+    return;
+  }
+
+  switch (message.action) {
+    case JingleMessage::SESSION_ACCEPT:
+      OnAccept(message, reply);
+      break;
+
+    case JingleMessage::TRANSPORT_INFO:
+      ProcessTransportInfo(message);
+      break;
+
+    case JingleMessage::SESSION_REJECT:
+      OnReject(message, reply);
+      break;
+
+    case JingleMessage::SESSION_TERMINATE:
+      OnTerminate(message, reply);
+      break;
+
+    default:
+      *reply = JingleMessageReply(JingleMessageReply::UNEXPECTED_REQUEST);
+  }
+}
+
+void PepperSession::OnAccept(const JingleMessage& message,
+                             JingleMessageReply* reply) {
+  if (state_ != CONNECTING) {
+    *reply = JingleMessageReply(JingleMessageReply::UNEXPECTED_REQUEST);
+    return;
+  }
+
+  if (!InitializeConfigFromDescription(message.description.get())) {
+    OnError(ERROR_INCOMPATIBLE_PROTOCOL);
+    return;
+  }
+
+  CreateChannels();
+  SetState(CONNECTED);
+
+   // In case there is transport information in the accept message.
+  ProcessTransportInfo(message);
+}
+
+void PepperSession::ProcessTransportInfo(const JingleMessage& message) {
+  for (std::list<cricket::Candidate>::const_iterator it =
+           message.candidates.begin();
+       it != message.candidates.end(); ++it) {
+    ChannelsMap::iterator channel = channels_.find(it->name());
+    if (channel == channels_.end()) {
+      LOG(WARNING) << "Received candidate for unknown channel " << it->name();
+      continue;
+    }
+    channel->second->AddRemoveCandidate(*it);
+  }
+}
+
+void PepperSession::OnReject(const JingleMessage& message,
+                             JingleMessageReply* reply) {
+  if (state_ != CONNECTING) {
+    *reply = JingleMessageReply(JingleMessageReply::UNEXPECTED_REQUEST);
+    return;
+  }
+
+  // TODO(sergeyu): Parse exact rejection reason from reply and pass it
+  // to OnError().
+  OnError(ERROR_SESSION_REJECTED);
+}
+
+void PepperSession::OnTerminate(const JingleMessage& message,
+                                JingleMessageReply* reply) {
+  if (state_ == CONNECTING) {
+    // If we are not connected yet, then interpret terminate message
+    // as rejection.
+    OnError(ERROR_SESSION_REJECTED);
+    return;
+  }
+
+  CloseInternal(false);
+}
+
+bool PepperSession::InitializeConfigFromDescription(
+    const ContentDescription* description) {
+  DCHECK(description);
+
+  remote_cert_ = description->certificate();
+  if (remote_cert_.empty()) {
+    LOG(ERROR) << "session-accept does not specify certificate";
+    return false;
+  }
+
+  if (!description->config()->GetFinalConfig(&config_)) {
+    LOG(ERROR) << "session-accept does not specify configuration";
+    return false;
+  }
+  if (!candidate_config()->IsSupported(config_)) {
+    LOG(ERROR) << "session-accept specifies an invalid configuration";
+    return false;
+  }
+
+  return true;
+}
+
+void PepperSession::AddLocalCandidate(const cricket::Candidate& candidate) {
+  pending_candidates_.push_back(candidate);
+
+  if (!transport_infos_timer_.IsRunning()) {
+    // Delay sending the new candidates in case we get more candidates
+    // that we can send in one message.
+    transport_infos_timer_.Start(
+        FROM_HERE, base::TimeDelta::FromMilliseconds(kTransportInfoSendDelayMs),
+        this, &PepperSession::SendTransportInfo);
+  }
+}
+
+void PepperSession::OnDeleteChannel(PepperChannel* channel) {
+  ChannelsMap::iterator it = channels_.find(channel->name());
+  DCHECK_EQ(it->second, channel);
+  channels_.erase(it);
+}
+
+void PepperSession::SendTransportInfo() {
+  JingleMessage message(peer_jid_, JingleMessage::TRANSPORT_INFO, session_id_);
+  message.candidates.swap(pending_candidates_);
+  scoped_ptr<IqRequest> request(session_manager_->CreateIqRequest());
+  request->SendIq(message.ToXml());
+}
+
+void PepperSession::CreateChannels() {
+   CreateStreamChannel(
+      kControlChannelName,
+      base::Bind(&PepperSession::OnChannelConnected,
+                 base::Unretained(this), &control_channel_socket_));
+  CreateStreamChannel(
+      kEventChannelName,
+      base::Bind(&PepperSession::OnChannelConnected,
+                 base::Unretained(this), &event_channel_socket_));
+}
+
+void PepperSession::OnChannelConnected(
+    scoped_ptr<net::Socket>* socket_container,
+    net::StreamSocket* socket) {
+  if (!socket) {
+    LOG(ERROR) << "Failed to connect control or events channel. "
+               << "Terminating connection";
+    OnError(ERROR_CHANNEL_CONNECTION_FAILURE);
+    return;
+  }
+
+  socket_container->reset(socket);
+
+  if (control_channel_socket_.get() && event_channel_socket_.get())
+    SetState(CONNECTED_CHANNELS);
+}
+
+void PepperSession::CloseInternal(bool failed) {
+  DCHECK(CalledOnValidThread());
+
+  if (state_ != FAILED && state_ != CLOSED) {
+    control_channel_socket_.reset();
+    event_channel_socket_.reset();
+
+    if (failed)
+      SetState(FAILED);
+    else
+      SetState(CLOSED);
+  }
+}
+
+void PepperSession::SetState(State new_state) {
+  DCHECK(CalledOnValidThread());
+
+  if (new_state != state_) {
+    DCHECK_NE(state_, CLOSED);
+    DCHECK_NE(state_, FAILED);
+
+    state_ = new_state;
+    if (state_change_callback_.get())
+      state_change_callback_->Run(new_state);
+  }
+}
+
+}  // namespace protocol
+}  // namespace remoting