Bugfixes in JingleSession

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 <stdint.h>
 
 #include <limits>
 #include <utility>
@@ skipping 88 matching lines @@
 // This behavior leads to transient session setup failures.  For instance,
 // a <transport-info> that is sent after a <session-info> message is sometimes
 // delivered to the client out of order, causing the client to close the
 // session due to an unexpected request.
 class JingleSession::OrderedMessageQueue {
  public:
   OrderedMessageQueue() {}
   ~OrderedMessageQueue() {}
 
   // Returns the list of messages ordered by their sequential IDs.
-  std::vector<std::unique_ptr<PendingMessage>> OnIncomingMessage(
+  std::vector<PendingMessage> OnIncomingMessage(
       const std::string& id,
-      std::unique_ptr<PendingMessage>);
+      PendingMessage&& pending_message);
 
  private:
   // Implements an ordered list by using map with the |sequence_id| as the key,
   // so that |queue_| is always sorted by |sequence_id|.
-  std::map<int, std::unique_ptr<PendingMessage>> queue_;
+  std::map<int, PendingMessage> queue_;
 
   int next_incoming_ = kAny;
 
   DISALLOW_COPY_AND_ASSIGN(OrderedMessageQueue);
 };
 
-std::vector<std::unique_ptr<JingleSession::PendingMessage>>
+std::vector<JingleSession::PendingMessage>
 JingleSession::OrderedMessageQueue::OnIncomingMessage(
     const std::string& id,
-    std::unique_ptr<JingleSession::PendingMessage> message) {
-  std::vector<std::unique_ptr<JingleSession::PendingMessage>> result;
+    JingleSession::PendingMessage&& message) {
+  std::vector<JingleSession::PendingMessage> result;
   int current = GetSequentialId(id);
   // If there is no sequencing order encoded in the id, just return the
   // message.
   if (current == kInvalid) {
     result.push_back(std::move(message));
     return result;
   }
 
   if (next_incoming_ == kAny) {
     next_incoming_ = current;
@@ skipping 12 matching lines @@
     next_incoming_++;
   }
 
   if (current - next_incoming_ >= 3) {
     LOG(WARNING) << "Multiple messages are missing: expected= "
                  << next_incoming_ << " current= " << current;
   }
   return result;
 };
 
+JingleSession::PendingMessage::PendingMessage() = default;

[kelvinp, 2016/10/31 18:23:14]

Out of curiosity, why is a default empty constructor needed?

[Sergey Ulanov, 2016/10/31 18:25:02]

std::vector<> needs it.

+JingleSession::PendingMessage::PendingMessage(PendingMessage&& moved) = default;
 JingleSession::PendingMessage::PendingMessage(
     std::unique_ptr<JingleMessage> message,
     const ReplyCallback& reply_callback)
     : message(std::move(message)), reply_callback(reply_callback) {}
+JingleSession::PendingMessage::~PendingMessage() = default;
 
-JingleSession::PendingMessage::~PendingMessage() {}
+JingleSession::PendingMessage& JingleSession::PendingMessage::operator=(
+    PendingMessage&& moved) = default;
 
 JingleSession::JingleSession(JingleSessionManager* session_manager)
     : session_manager_(session_manager),
       event_handler_(nullptr),
       state_(INITIALIZING),
       error_(OK),
       message_queue_(new OrderedMessageQueue),
       weak_factory_(this) {}
 
 JingleSession::~JingleSession() {
@@ skipping 216 matching lines @@
       std::move(stanza), base::Bind(&JingleSession::OnMessageResponse,
                                     base::Unretained(this), message->action));
 
   int timeout = kDefaultMessageTimeout;
   if (message->action == JingleMessage::SESSION_INITIATE ||
       message->action == JingleMessage::SESSION_ACCEPT) {
     timeout = kSessionInitiateAndAcceptTimeout;
   }
   if (request) {
     request->SetTimeout(base::TimeDelta::FromSeconds(timeout));
-    pending_requests_.insert(std::move(request));
+    pending_requests_.push_back(std::move(request));
   } else {
     LOG(ERROR) << "Failed to send a "
                << JingleMessage::GetActionName(message->action) << " message";
   }
 }
 
 void JingleSession::OnMessageResponse(
     JingleMessage::ActionType request_type,
     IqRequest* request,
     const buzz::XmlElement* response) {
@@ skipping 31 matching lines @@
     }
   }
 }
 
 void JingleSession::OnTransportInfoResponse(IqRequest* request,
                                             const buzz::XmlElement* response) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(!transport_info_requests_.empty());
 
   // Consider transport-info requests sent before this one lost and delete
-  // corresponding IqRequest objects.
-  while (transport_info_requests_.front().get() != request) {
-    transport_info_requests_.pop_front();
-  }
-
-  // Delete the |request| itself.
-  DCHECK_EQ(request, transport_info_requests_.front().get());
-  transport_info_requests_.pop_front();
+  // all IqRequest objects in front of |request|.
+  auto request_it = std::find_if(
+      transport_info_requests_.begin(), transport_info_requests_.end(),
+      [request](const std::unique_ptr<IqRequest>& ptr) {
+        return ptr.get() == request;
+      });
+  DCHECK(request_it != transport_info_requests_.end());
+  transport_info_requests_.erase(transport_info_requests_.begin(),
+                                 request_it + 1);
 
   // Ignore transport-info timeouts.
   if (!response) {
     LOG(ERROR) << "transport-info request has timed out.";
     return;
   }
 
   const std::string& type = response->Attr(buzz::QName(std::string(), "type"));
   if (type != "result") {
     LOG(ERROR) << "Received error in response to transport-info message: \""
                << response->Str() << "\". Terminating the session.";
     Close(PEER_IS_OFFLINE);
   }
 }
 
 void JingleSession::OnIncomingMessage(const std::string& id,
                                       std::unique_ptr<JingleMessage> message,
                                       const ReplyCallback& reply_callback) {
-  std::unique_ptr<PendingMessage> item(
-      new PendingMessage(std::move(message), reply_callback));
-  std::vector<std::unique_ptr<PendingMessage>> ordered =
-      message_queue_->OnIncomingMessage(id, std::move(item));
+  std::vector<PendingMessage> ordered = message_queue_->OnIncomingMessage(
+      id, PendingMessage{std::move(message), reply_callback});
+  base::WeakPtr<JingleSession> self = weak_factory_.GetWeakPtr();
   for (auto& message : ordered) {
-    ProcessIncomingMessage(std::move(message->message),
-                           message->reply_callback);
+    ProcessIncomingMessage(std::move(message.message), message.reply_callback);
+    if (!self)
+      return;
   }
 }
 
 void JingleSession::ProcessIncomingMessage(
     std::unique_ptr<JingleMessage> message,
     const ReplyCallback& reply_callback) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   if (peer_address_ != message->from) {
     // Ignore messages received from a different Jid.
     reply_callback.Run(JingleMessageReply::INVALID_SID);
     return;
   }
 
   switch (message->action) {
     case JingleMessage::SESSION_ACCEPT:
       OnAccept(std::move(message), reply_callback);
       break;
 
     case JingleMessage::SESSION_INFO:
       OnSessionInfo(std::move(message), reply_callback);
       break;
 
     case JingleMessage::TRANSPORT_INFO:
-      if (!transport_) {
-        LOG(ERROR) << "Received unexpected transport-info message->";
-        reply_callback.Run(JingleMessageReply::NONE);
-        return;
-      }
-
-      if (!message->transport_info ||
-          !transport_->ProcessTransportInfo(message->transport_info.get())) {
-        reply_callback.Run(JingleMessageReply::BAD_REQUEST);
-        return;
-      }
-
-      reply_callback.Run(JingleMessageReply::NONE);
+      OnTransportInfo(std::move(message), reply_callback);
       break;
 
     case JingleMessage::SESSION_TERMINATE:
       OnTerminate(std::move(message), reply_callback);
       break;
 
     default:
       reply_callback.Run(JingleMessageReply::UNEXPECTED_REQUEST);
   }
 }
@@ skipping 16 matching lines @@
   }
 
   if (!InitializeConfigFromDescription(message->description.get())) {
     Close(INCOMPATIBLE_PROTOCOL);
     return;
   }
 
   SetState(ACCEPTED);
 
   DCHECK(authenticator_->state() == Authenticator::WAITING_MESSAGE);
-  authenticator_->ProcessMessage(auth_message, base::Bind(
-      &JingleSession::ProcessAuthenticationStep,base::Unretained(this)));
+  authenticator_->ProcessMessage(
+      auth_message, base::Bind(&JingleSession::ProcessAuthenticationStep,
+                               base::Unretained(this)));
 }
 
 void JingleSession::OnSessionInfo(std::unique_ptr<JingleMessage> message,
                                   const ReplyCallback& reply_callback) {
   if (!message->info.get() ||
       !Authenticator::IsAuthenticatorMessage(message->info.get())) {
     reply_callback.Run(JingleMessageReply::UNSUPPORTED_INFO);
     return;
   }
 
   if ((state_ != ACCEPTED && state_ != AUTHENTICATING) ||
       authenticator_->state() != Authenticator::WAITING_MESSAGE) {
     LOG(WARNING) << "Received unexpected authenticator message "
                  << message->info->Str();
     reply_callback.Run(JingleMessageReply::UNEXPECTED_REQUEST);
     Close(INCOMPATIBLE_PROTOCOL);
     return;
   }
 
   reply_callback.Run(JingleMessageReply::NONE);
 
-  authenticator_->ProcessMessage(message->info.get(), base::Bind(
-      &JingleSession::ProcessAuthenticationStep, base::Unretained(this)));
+  authenticator_->ProcessMessage(
+      message->info.get(), base::Bind(&JingleSession::ProcessAuthenticationStep,
+                                      base::Unretained(this)));
+}
+
+void JingleSession::OnTransportInfo(std::unique_ptr<JingleMessage> message,
+                                    const ReplyCallback& reply_callback) {
+  if (!message->transport_info) {
+    reply_callback.Run(JingleMessageReply::BAD_REQUEST);
+    return;
+  }
+
+  if (state_ == AUTHENTICATING) {
+    pending_transport_info_.push_back(
+        PendingMessage{std::move(message), reply_callback});
+  } else if (state_ == AUTHENTICATED) {
+    reply_callback.Run(
+        transport_->ProcessTransportInfo(message->transport_info.get())
+            ? JingleMessageReply::NONE
+            : JingleMessageReply::BAD_REQUEST);
+  } else {
+    LOG(ERROR) << "Received unexpected transport-info message.";
+    reply_callback.Run(JingleMessageReply::UNEXPECTED_REQUEST);
+  }
 }
 
 void JingleSession::OnTerminate(std::unique_ptr<JingleMessage> message,
                                 const ReplyCallback& reply_callback) {
   if (!is_session_active()) {
     LOG(WARNING) << "Received unexpected session-terminate message.";
     reply_callback.Run(JingleMessageReply::UNEXPECTED_REQUEST);
     return;
   }
 
@@ skipping 76 matching lines @@
 
   if (authenticator_->state() == Authenticator::MESSAGE_READY) {
     std::unique_ptr<JingleMessage> message(new JingleMessage(
         peer_address_, JingleMessage::SESSION_INFO, session_id_));
     message->info = authenticator_->GetNextMessage();
     DCHECK(message->info.get());
     SendMessage(std::move(message));
   }
   DCHECK_NE(authenticator_->state(), Authenticator::MESSAGE_READY);
 
-  // The current JingleSession object can be destroyed by event_handler of
-  // SetState(AUTHENTICATING) and cause subsequent dereferencing of the this
-  // pointer to crash.  To protect against it, we run ContinueAuthenticationStep
-  // asychronously using a weak pointer.
-  base::ThreadTaskRunnerHandle::Get()->PostTask(
-    FROM_HERE,
-    base::Bind(&JingleSession::ContinueAuthenticationStep,
-               weak_factory_.GetWeakPtr()));
+  if (authenticator_->started()) {
+    base::WeakPtr<JingleSession> self = weak_factory_.GetWeakPtr();
+    SetState(AUTHENTICATING);
+    if (!self)
+      return;
+  }
 
-  if (authenticator_->started()) {
-    SetState(AUTHENTICATING);
-  }
-}
-
-void JingleSession::ContinueAuthenticationStep() {
   if (authenticator_->state() == Authenticator::ACCEPTED) {
     OnAuthenticated();
   } else if (authenticator_->state() == Authenticator::REJECTED) {
     Close(AuthRejectionReasonToErrorCode(
         authenticator_->rejection_reason()));
   }
 }
 
 void JingleSession::OnAuthenticated() {
   transport_->Start(authenticator_.get(),
                     base::Bind(&JingleSession::SendTransportInfo,
                                weak_factory_.GetWeakPtr()));
 
+  base::WeakPtr<JingleSession> self = weak_factory_.GetWeakPtr();
+  std::vector<PendingMessage> messages_to_process;
+  std::swap(messages_to_process, pending_transport_info_);
+  for (auto& message : messages_to_process) {
+    message.reply_callback.Run(
+        transport_->ProcessTransportInfo(message.message->transport_info.get())
+            ? JingleMessageReply::NONE
+            : JingleMessageReply::BAD_REQUEST);
+    if (!self)
+      return;
+  }
+
   SetState(AUTHENTICATED);
 }
 
 void JingleSession::SetState(State new_state) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
   if (new_state != state_) {
     DCHECK_NE(state_, CLOSED);
     DCHECK_NE(state_, FAILED);
 
     state_ = new_state;
     if (event_handler_)
       event_handler_->OnSessionStateChange(new_state);
   }
 }
 
 bool JingleSession::is_session_active() {
   return state_ == CONNECTING || state_ == ACCEPTING || state_ == ACCEPTED ||
         state_ == AUTHENTICATING || state_ == AUTHENTICATED;
 }
 
 std::string JingleSession::GetNextOutgoingId() {
   return outgoing_id_prefix_ + "_" + base::IntToString(++next_outgoing_id_);
 }
 
 }  // namespace protocol
 }  // namespace remoting