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
@@ skipping 175 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<QueueItem> item(
+      new QueueItem(std::move(message), reply_callback));
+  std::list<std::unique_ptr<QueueItem>> ordered =
+      message_queue_.OnIncomingMessage(id, std::move(item));
+  for (auto it = ordered.begin(); it != ordered.end(); it++) {
+    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:
@@ skipping 218 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::QueueItem::QueueItem(std::unique_ptr<JingleMessage> message,
+                                    const ReplyCallback& reply_callback)
+    : message(std::move(message)), reply_callback(reply_callback) {}
+
+JingleSession::QueueItem::~QueueItem() {}
+
 }  // namespace protocol
 }  // namespace remoting