Process incoming IQs in the same order that they were sent.

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>
 
 #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/threading/thread_task_runner_handle.h"
 #include "base/time/time.h"
 #include "remoting/base/constants.h"
 #include "remoting/protocol/authenticator.h"
 #include "remoting/protocol/content_description.h"
 #include "remoting/protocol/jingle_messages.h"
 #include "remoting/protocol/jingle_session_manager.h"
 #include "remoting/protocol/session_config.h"
 #include "remoting/protocol/transport.h"
 #include "remoting/signaling/iq_sender.h"
 #include "third_party/webrtc/libjingle/xmllite/xmlelement.h"
+#include "third_party/webrtc/libjingle/xmpp/constants.h"
 #include "third_party/webrtc/p2p/base/candidate.h"
 
 using buzz::XmlElement;
 
 namespace remoting {
 namespace protocol {
 
 namespace {
 
 // 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;
 
+// Special value for an invalid sequential ID for an incoming IQ.
+const int kInvalid = -1;
+
+// Special value indicating that any sequential ID is valid for the next
+// incoming IQ.
+const int kAny = -1;
+
 ErrorCode AuthRejectionReasonToErrorCode(
     Authenticator::RejectionReason reason) {
   switch (reason) {
     case Authenticator::INVALID_CREDENTIALS:
       return AUTHENTICATION_FAILED;
     case Authenticator::PROTOCOL_ERROR:
       return INCOMPATIBLE_PROTOCOL;
     case Authenticator::INVALID_ACCOUNT:
       return INVALID_ACCOUNT;
     case Authenticator::REJECTED_BY_USER:
       return SESSION_REJECTED;
   }
   NOTREACHED();
   return UNKNOWN_ERROR;
 }
 
+// Extracts a sequential id from the id attribute of the IQ stanza.
+int GetSequentialId(std::string id) {

[Sergey Ulanov, 2016/10/19 21:23:38]

please use const reference for string parameters

[kelvinp, 2016/10/21 00:26:02]

Done.

+  int result = kInvalid;

[Sergey Ulanov, 2016/10/19 21:23:38]

move this below, just before StringToInt() call. See
https://google.github.io/styleguide/cppguide.html#Local_Variables

[kelvinp, 2016/10/21 00:26:02]

Done.

+  std::vector<std::string> tokens =
+      SplitString(id, "_", base::TRIM_WHITESPACE, base::SPLIT_WANT_NONEMPTY);
+  // Legacy endpoints does not encode the IQ ordering in the ID attribute
+  if (tokens.size() != 2) {
+    return kInvalid;
+  }
+
+  if (!base::StringToInt(tokens[1].c_str(), &result)) {
+    return kInvalid;
+  }
+  return result;
+};

[Sergey Ulanov, 2016/10/19 21:23:38]

nit: don't need ;

[kelvinp, 2016/10/21 00:26:02]

Done.

+
 }  // namespace
 
+// A Queue that sorts incoming messages and returns them in the ascending order
+// of sequence ids. The sequence id can be extracted from the ID attribute of
+// an IQ stanza, which have the following format <opaque_string>_<sequence_id>.
+//
+// Background:
+// The chromoting signaling channel does not guarantee that the incoming IQs are
+// delivered in the order that it is sent.
+//
+// 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()
+      : next_incoming_(kAny),
+        next_outgoing_(0),

[Sergey Ulanov, 2016/10/19 21:23:38]

Move these initializers to where they are defined.

[kelvinp, 2016/10/21 00:26:03]

Done.

+        outgoing_prefix_(base::Uint64ToString(
+            base::RandGenerator(std::numeric_limits<uint64_t>::max()))){};

[Sergey Ulanov, 2016/10/19 21:23:38]

base::RandUint64()

[Sergey Ulanov, 2016/10/19 21:23:38]

don't need semicolon

[kelvinp, 2016/10/21 00:26:02]

Done.

[kelvinp, 2016/10/21 00:26:03]

Done.

+
+  ~OrderedMessageQueue() {}
+
+  // Returns the list of messages ordered by their sequential IDs.
+  std::list<std::unique_ptr<PendingMessage>> OnIncomingMessage(

[Sergey Ulanov, 2016/10/19 21:23:38]

Better to use std::vector<> instead of std::list<> for the result.

[kelvinp, 2016/10/21 00:26:03]

Done.

+      std::string id,

[Sergey Ulanov, 2016/10/19 21:23:38]

const reference

[kelvinp, 2016/10/21 00:26:02]

Done.

+      std::unique_ptr<PendingMessage>);
+  std::string GetNextOutgoingId();
+
+ private:
+  // Implements an ordered list by using map with the |sequence_id| as the key,

[Sergey Ulanov, 2016/10/19 21:23:39]

I don't think you really need a map. A heap stored in std::vector<> would be
more appropriate (see std::push_heap and std::pop_heap), or it can be sorted
vector (there will never be more than a handful of messages, so message
insertion consts are small).

[kelvinp, 2016/10/21 00:26:02]

Looking at std::push_heap() this would require us to write a comparator and
store the id in the structure.  Alternatively, we can use a priority_queue which
called make_heap for us.
Given we only have a handful of messages, it won't make a big difference either
way we go.  I will leave the code as is as it is simpler.  WDYT?

+  // so that |queue_| is always sorted by |sequence_id|.
+  std::map<int, std::unique_ptr<PendingMessage>> queue_;
+
+  int next_incoming_;

[Sergey Ulanov, 2016/10/19 21:23:38]

 = kAny

[kelvinp, 2016/10/21 00:26:02]

Done.

+  int next_outgoing_;

[Sergey Ulanov, 2016/10/19 21:23:38]

 = 0

[kelvinp, 2016/10/21 00:26:02]

Done.

+
+  // This prefix is necessary to disambiguate between the ID's sent from the
+  // client and the ID's sent from the host.
+  std::string outgoing_prefix_;
+
+  DISALLOW_COPY_AND_ASSIGN(OrderedMessageQueue);
+};
+
+std::list<std::unique_ptr<JingleSession::PendingMessage>>
+JingleSession::OrderedMessageQueue::OnIncomingMessage(
+    std::string id,
+    std::unique_ptr<JingleSession::PendingMessage> message) {
+  std::list<std::unique_ptr<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;
+  }
+
+  DCHECK(current >= next_incoming_)
+      << "Duplicate sequence id: current= " << current

[Sergey Ulanov, 2016/10/19 21:23:38]

Messages in DCHECK() are very rarely useful, but they increase code complexity,
so I'd prefer removing them

[kelvinp, 2016/10/21 00:26:02]

Done.

+      << " expected= " << next_incoming_;
+  DCHECK(queue_.find(current) == queue_.end())
+      << "Duplicate sequence id: current = " << current;
+
+  queue_.insert(std::pair<int, std::unique_ptr<JingleSession::PendingMessage>>(

[Sergey Ulanov, 2016/10/19 21:23:38]

std::make_pair?

[kelvinp, 2016/10/21 00:26:03]

Done.

+      current, std::move(message)));
+
+  auto it = queue_.begin();
+  while (it != queue_.end() && it->first == next_incoming_) {
+    result.push_back(std::move(it->second));
+    it = queue_.erase(it);
+    next_incoming_++;
+  }
+
+  if (current - next_incoming_ >= 3) {
+    LOG(WARNING) << "Multiple messages are missing: expected= "
+                 << next_incoming_ << " current= " << current;
+  }
+  return result;
+};
+
+std::string JingleSession::OrderedMessageQueue::GetNextOutgoingId() {

[Sergey Ulanov, 2016/10/19 21:23:38]

This doesn't look related to queuing incoming messages. Move it to
JingleSession?

[kelvinp, 2016/10/21 00:26:02]

Done.

+  return outgoing_prefix_ + "_" + base::IntToString(++next_outgoing_);
+}
+
 JingleSession::JingleSession(JingleSessionManager* session_manager)
     : session_manager_(session_manager),
       event_handler_(nullptr),
       state_(INITIALIZING),
       error_(OK),
-      weak_factory_(this) {
-}
+      message_queue_(new OrderedMessageQueue),
+      weak_factory_(this) {}
 
 JingleSession::~JingleSession() {
   session_manager_->SessionDestroyed(this);
 }
 
 void JingleSession::SetEventHandler(Session::EventHandler* event_handler) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK(event_handler);
   event_handler_ = event_handler;
 }
@@ skipping 128 matching lines @@
 
 void JingleSession::SendTransportInfo(
     std::unique_ptr<buzz::XmlElement> transport_info) {
   DCHECK(thread_checker_.CalledOnValidThread());
   DCHECK_EQ(state_, AUTHENTICATED);
 
   std::unique_ptr<JingleMessage> message(new JingleMessage(
       peer_address_, JingleMessage::TRANSPORT_INFO, session_id_));
   message->transport_info = std::move(transport_info);
 
+  std::unique_ptr<buzz::XmlElement> stanza = message->ToXml();
+  stanza->AddAttr(buzz::QN_ID, message_queue_->GetNextOutgoingId());
+
   auto request = session_manager_->iq_sender()->SendIq(
-      message->ToXml(), base::Bind(&JingleSession::OnTransportInfoResponse,
-                                   base::Unretained(this)));
+      std::move(stanza), base::Bind(&JingleSession::OnTransportInfoResponse,
+                                    base::Unretained(this)));
   if (request) {
     request->SetTimeout(base::TimeDelta::FromSeconds(kTransportInfoTimeout));
     transport_info_requests_.push_back(std::move(request));
   } else {
     LOG(ERROR) << "Failed to send a transport-info message";
   }
 }
 
 void JingleSession::Close(protocol::ErrorCode error) {
   DCHECK(thread_checker_.CalledOnValidThread());
@@ skipping 40 matching lines @@
       SetState(FAILED);
     } else {
       SetState(CLOSED);
     }
   }
 }
 
 void JingleSession::SendMessage(std::unique_ptr<JingleMessage> message) {
   DCHECK(thread_checker_.CalledOnValidThread());
 
+  std::unique_ptr<buzz::XmlElement> stanza = message->ToXml();
+  stanza->AddAttr(buzz::QN_ID, message_queue_->GetNextOutgoingId());
+
   auto request = session_manager_->iq_sender()->SendIq(
-      message->ToXml(), base::Bind(&JingleSession::OnMessageResponse,
-                                   base::Unretained(this), message->action));
+      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));
   } else {
@@ skipping 63 matching lines @@
   }
 
   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(std::unique_ptr<JingleMessage> message,
+void JingleSession::OnIncomingMessage(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::list<std::unique_ptr<PendingMessage>> ordered =
+      message_queue_->OnIncomingMessage(id, std::move(item));
+  for (auto it = ordered.begin(); it != ordered.end(); it++) {

[Sergey Ulanov, 2016/10/19 21:23:38]

for (auto& message : ordered)

Also for iterators (++it is faster than it++)

[kelvinp, 2016/10/21 00:26:02]

Done.

+    OnIncomingMessageInOrder(std::move(it->get()->message),
+                             it->get()->reply_callback);
+  }
+}
+
+void JingleSession::OnIncomingMessageInOrder(
+    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())) {
+          !transport_->ProcessTransportInfo(message->transport_info.get())) {
         reply_callback.Run(JingleMessageReply::BAD_REQUEST);
         return;
       }
 
       reply_callback.Run(JingleMessageReply::NONE);
       break;
 
     case JingleMessage::SESSION_TERMINATE:
       OnTerminate(std::move(message), reply_callback);
       break;
@@ skipping 191 matching lines @@
     if (event_handler_)
       event_handler_->OnSessionStateChange(new_state);
   }
 }
 
 bool JingleSession::is_session_active() {
   return state_ == CONNECTING || state_ == ACCEPTING || state_ == ACCEPTED ||
         state_ == AUTHENTICATING || state_ == AUTHENTICATED;
 }
 
+JingleSession::PendingMessage::PendingMessage(

[Sergey Ulanov, 2016/10/19 21:23:38]

nit: put this above JingleSession::JingleSession() constructor.

[kelvinp, 2016/10/21 00:26:02]

Done.

+    std::unique_ptr<JingleMessage> message,
+    const ReplyCallback& reply_callback)
+    : message(std::move(message)), reply_callback(reply_callback) {}
+
+JingleSession::PendingMessage::~PendingMessage() {}
+
 }  // namespace protocol
 }  // namespace remoting