diff --git a/buffet/notification/xmpp_channel.cc b/buffet/notification/xmpp_channel.cc
index 3ef7975..ab657d1 100644
--- a/buffet/notification/xmpp_channel.cc
+++ b/buffet/notification/xmpp_channel.cc
@@ -44,23 +44,6 @@
   return msg;
 }
 
-std::string BuildXmppBindCommand() {
-  return "<iq type='set' id='0'>"
-      "<bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'/></iq>";
-}
-
-std::string BuildXmppStartSessionCommand() {
-  return "<iq type='set' id='1'>"
-      "<session xmlns='urn:ietf:params:xml:ns:xmpp-session'/></iq>";
-}
-
-std::string BuildXmppSubscribeCommand(const std::string& account) {
-  return "<iq type='set' to='" + account + "' "
-      "id='pushsubscribe1'><subscribe xmlns='google:push'>"
-      "<item channel='cloud_devices' from=''/>"
-      "</subscribe></iq>";
-}
-
 // Backoff policy.
 // Note: In order to ensure a minimum of 20 seconds between server errors,
 // we have a 30s +- 10s (33%) jitter initial backoff.
@@ -95,13 +78,15 @@
 
 }  // namespace
 
-XmppChannel::XmppChannel(const std::string& account,
-                         const std::string& access_token,
-                         const scoped_refptr<base::TaskRunner>& task_runner)
+XmppChannel::XmppChannel(
+    const std::string& account,
+    const std::string& access_token,
+    const scoped_refptr<base::SingleThreadTaskRunner>& task_runner)
     : account_{account},
       access_token_{access_token},
       backoff_entry_{&kDefaultBackoffPolicy},
-      task_runner_{task_runner} {
+      task_runner_{task_runner},
+      iq_stanza_handler_{new IqStanzaHandler{this, task_runner}} {
   read_socket_data_.resize(4096);
 }
 
@@ -190,32 +175,11 @@
       if (stanza->name() == "stream:features" &&
           stanza->FindFirstChild("bind", false)) {
         state_ = XmppState::kBindSent;
-        SendMessage(BuildXmppBindCommand());
-        return;
-      }
-      break;
-    case XmppState::kBindSent:
-      if (stanza->name() == "iq" &&
-          stanza->GetAttributeOrEmpty("type") == "result") {
-        state_ = XmppState::kSessionStarted;
-        SendMessage(BuildXmppStartSessionCommand());
-        return;
-      }
-      break;
-    case XmppState::kSessionStarted:
-      if (stanza->name() == "iq" &&
-          stanza->GetAttributeOrEmpty("type") == "result") {
-        state_ = XmppState::kSubscribeStarted;
-        SendMessage(BuildXmppSubscribeCommand(account_));
-        return;
-      }
-      break;
-    case XmppState::kSubscribeStarted:
-      if (stanza->name() == "iq" &&
-          stanza->GetAttributeOrEmpty("type") == "result") {
-        state_ = XmppState::kSubscribed;
-        if (delegate_)
-          delegate_->OnConnected(GetName());
+        iq_stanza_handler_->SendRequest(
+            "set", "", "", "<bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'/>",
+            base::Bind(&XmppChannel::OnBindCompleted,
+                       weak_ptr_factory_.GetWeakPtr()),
+            base::Bind(&XmppChannel::Restart, weak_ptr_factory_.GetWeakPtr()));
         return;
       }
       break;
@@ -223,6 +187,12 @@
       if (stanza->name() == "message") {
         HandleMessageStanza(std::move(stanza));
         return;
+      } else if (stanza->name() == "iq") {
+        if (!iq_stanza_handler_->HandleIqStanza(std::move(stanza))) {
+          LOG(ERROR) << "Failed to handle IQ stanza";
+          CloseStream();
+        }
+        return;
       }
       LOG(INFO) << "Unexpected XMPP stanza ignored: " << stanza->ToString();
       return;
@@ -230,9 +200,59 @@
   // Something bad happened. Close the stream and start over.
   LOG(ERROR) << "Error condition occurred handling stanza: "
              << stanza->ToString();
+  CloseStream();
+}
+
+void XmppChannel::CloseStream() {
   SendMessage("</stream:stream>");
 }
 
+void XmppChannel::OnBindCompleted(std::unique_ptr<XmlNode> reply) {
+  if (reply->GetAttributeOrEmpty("type") != "result") {
+    CloseStream();
+    return;
+  }
+  const XmlNode* jid_node = reply->FindFirstChild("bind/jid", false);
+  if (!jid_node) {
+    LOG(ERROR) << "XMPP Bind response is missing JID";
+    CloseStream();
+    return;
+  }
+
+  jid_ = jid_node->text();
+  state_ = XmppState::kSessionStarted;
+  iq_stanza_handler_->SendRequest(
+      "set", "", "", "<session xmlns='urn:ietf:params:xml:ns:xmpp-session'/>",
+      base::Bind(&XmppChannel::OnSessionEstablished,
+                 weak_ptr_factory_.GetWeakPtr()),
+      base::Bind(&XmppChannel::Restart, weak_ptr_factory_.GetWeakPtr()));
+}
+
+void XmppChannel::OnSessionEstablished(std::unique_ptr<XmlNode> reply) {
+  if (reply->GetAttributeOrEmpty("type") != "result") {
+    CloseStream();
+    return;
+  }
+  state_ = XmppState::kSubscribeStarted;
+  std::string body = "<subscribe xmlns='google:push'>"
+                     "<item channel='cloud_devices' from=''/></subscribe>";
+  iq_stanza_handler_->SendRequest(
+      "set", "", account_, body,
+      base::Bind(&XmppChannel::OnSubscribed,
+                  weak_ptr_factory_.GetWeakPtr()),
+      base::Bind(&XmppChannel::Restart, weak_ptr_factory_.GetWeakPtr()));
+}
+
+void XmppChannel::OnSubscribed(std::unique_ptr<XmlNode> reply) {
+  if (reply->GetAttributeOrEmpty("type") != "result") {
+    CloseStream();
+    return;
+  }
+  state_ = XmppState::kSubscribed;
+  if (delegate_)
+    delegate_->OnConnected(GetName());
+}
+
 void XmppChannel::HandleMessageStanza(std::unique_ptr<XmlNode> stanza) {
   const XmlNode* node = stanza->FindFirstChild("push:push/push:data", true);
   if (!node) {
diff --git a/buffet/notification/xmpp_channel.h b/buffet/notification/xmpp_channel.h
index d83df48..a36a070 100644
--- a/buffet/notification/xmpp_channel.h
+++ b/buffet/notification/xmpp_channel.h
@@ -13,24 +13,35 @@
 #include <base/callback_forward.h>
 #include <base/macros.h>
 #include <base/memory/weak_ptr.h>
-#include <base/task_runner.h>
+#include <base/single_thread_task_runner.h>
 #include <chromeos/backoff_entry.h>
 #include <chromeos/streams/stream.h>
 
 #include "buffet/notification/notification_channel.h"
+#include "buffet/notification/xmpp_iq_stanza_handler.h"
 #include "buffet/notification/xmpp_stream_parser.h"
 
 namespace buffet {
 
+// Simple interface to abstract XmppChannel's SendMessage() method.
+class XmppChannelInterface {
+ public:
+  virtual void SendMessage(const std::string& message) = 0;
+
+ protected:
+  virtual ~XmppChannelInterface() = default;
+};
+
 class XmppChannel : public NotificationChannel,
-                    public XmppStreamParser::Delegate {
+                    public XmppStreamParser::Delegate,
+                    public XmppChannelInterface {
  public:
   // |account| is the robot account for buffet and |access_token|
   // it the OAuth token. Note that the OAuth token expires fairly frequently
   // so you will need to reset the XmppClient every time this happens.
   XmppChannel(const std::string& account,
               const std::string& access_token,
-              const scoped_refptr<base::TaskRunner>& task_runner);
+              const scoped_refptr<base::SingleThreadTaskRunner>& task_runner);
   ~XmppChannel() override = default;
 
   // Overrides from NotificationChannel.
@@ -40,6 +51,8 @@
   void Start(NotificationDelegate* delegate) override;
   void Stop() override;
 
+  const std::string& jid() const { return jid_; }
+
   // Internal states for the XMPP stream.
   enum class XmppState {
     kNotStarted,
@@ -56,6 +69,8 @@
   };
 
  protected:
+  // These methods are internal helpers that can be overloaded by unit tests
+  // to help provide unit-test-specific functionality.
   virtual void Connect(const std::string& host, uint16_t port,
                        const base::Closure& callback);
 
@@ -65,12 +80,17 @@
   chromeos::Stream* stream_{nullptr};
 
  private:
+  friend class IqStanzaHandler;
+
   // Overrides from XmppStreamParser::Delegate.
   void OnStreamStart(const std::string& node_name,
                      std::map<std::string, std::string> attributes) override;
   void OnStreamEnd(const std::string& node_name) override;
   void OnStanza(std::unique_ptr<XmlNode> stanza) override;
 
+  // Overrides from XmppChannelInterface.
+  void SendMessage(const std::string& message) override;
+
   void HandleStanza(std::unique_ptr<XmlNode> stanza);
   void HandleMessageStanza(std::unique_ptr<XmlNode> stanza);
   void RestartXmppStream();
@@ -79,7 +99,6 @@
   void OnTlsHandshakeComplete(chromeos::StreamPtr tls_stream);
   void OnTlsError(const chromeos::Error* error);
 
-  void SendMessage(const std::string& message);
   void WaitForMessage();
 
   void OnConnected();
@@ -88,10 +107,19 @@
   void OnReadError(const chromeos::Error* error);
   void OnWriteError(const chromeos::Error* error);
   void Restart();
+  void CloseStream();
+
+  // XMPP connection state machine's state handlers.
+  void OnBindCompleted(std::unique_ptr<XmlNode> reply);
+  void OnSessionEstablished(std::unique_ptr<XmlNode> reply);
+  void OnSubscribed(std::unique_ptr<XmlNode> reply);
 
   // Robot account name for the device.
   std::string account_;
 
+  // Full JID of this device.
+  std::string jid_;
+
   // OAuth access token for the account. Expires fairly frequently.
   std::string access_token_;
 
@@ -110,10 +138,11 @@
 
   chromeos::BackoffEntry backoff_entry_;
   NotificationDelegate* delegate_{nullptr};
-  scoped_refptr<base::TaskRunner> task_runner_;
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
   XmppStreamParser stream_parser_{this};
   bool read_pending_{false};
   bool write_pending_{false};
+  std::unique_ptr<IqStanzaHandler> iq_stanza_handler_;
 
   base::WeakPtrFactory<XmppChannel> weak_ptr_factory_{this};
   DISALLOW_COPY_AND_ASSIGN(XmppChannel);
diff --git a/buffet/notification/xmpp_channel_unittest.cc b/buffet/notification/xmpp_channel_unittest.cc
index 6e4f4b1..f739ef7 100644
--- a/buffet/notification/xmpp_channel_unittest.cc
+++ b/buffet/notification/xmpp_channel_unittest.cc
@@ -47,18 +47,18 @@
     "<session xmlns=\"urn:ietf:params:xml:ns:xmpp-session\"/>"
     "</stream:features>";
 constexpr char kBindResponse[] =
-    "<iq id=\"0\" type=\"result\">"
+    "<iq id=\"1\" type=\"result\">"
     "<bind xmlns=\"urn:ietf:params:xml:ns:xmpp-bind\">"
     "<jid>110cc78f78d7032cc7bf2c6e14c1fa7d@clouddevices.gserviceaccount.com"
     "/19853128</jid></bind></iq>";
 constexpr char kSessionResponse[] =
-    "<iq type=\"result\" id=\"1\"/>";
+    "<iq type=\"result\" id=\"2\"/>";
 constexpr char kSubscribedResponse[] =
     "<iq to=\""
     "110cc78f78d7032cc7bf2c6e14c1fa7d@clouddevices.gserviceaccount.com/"
     "19853128\" from=\""
     "110cc78f78d7032cc7bf2c6e14c1fa7d@clouddevices.gserviceaccount.com\" "
-    "id=\"pushsubscribe1\" type=\"result\"/>";
+    "id=\"3\" type=\"result\"/>";
 constexpr char kStartStreamMessage[] =
     "<stream:stream to='clouddevices.gserviceaccount.com' "
     "xmlns:stream='http://etherx.jabber.org/streams' xml:lang='*' "
@@ -72,31 +72,37 @@
     "xmlns:auth='http://www.google.com/talk/protocol/auth'>"
     "AEFjY291bnRATmFtZQBBY2Nlc3NUb2tlbg==</auth>";
 constexpr char kBindMessage[] =
-    "<iq type='set' id='0'><bind "
+    "<iq id='1' type='set'><bind "
     "xmlns='urn:ietf:params:xml:ns:xmpp-bind'/></iq>";
 constexpr char kSessionMessage[] =
-    "<iq type='set' id='1'><session "
+    "<iq id='2' type='set'><session "
     "xmlns='urn:ietf:params:xml:ns:xmpp-session'/></iq>";
 constexpr char kSubscribeMessage[] =
-    "<iq type='set' to='Account@Name' id='pushsubscribe1'>"
+    "<iq id='3' type='set' to='Account@Name'>"
     "<subscribe xmlns='google:push'><item channel='cloud_devices' from=''/>"
     "</subscribe></iq>";
 }  // namespace
 
 // Mock-like task runner that allow the tests to inspect the calls to
 // TaskRunner::PostDelayedTask and verify the delays.
-class TestTaskRunner : public base::TaskRunner {
+class TestTaskRunner : public base::SingleThreadTaskRunner {
  public:
   MOCK_METHOD3(PostDelayedTask, bool(const tracked_objects::Location&,
                                      const base::Closure&,
                                      base::TimeDelta));
+  MOCK_METHOD3(PostNonNestableDelayedTask,
+               bool(const tracked_objects::Location&,
+                    const base::Closure&,
+                    base::TimeDelta));
+
   bool RunsTasksOnCurrentThread() const { return true; }
 };
 
 class FakeXmppChannel : public XmppChannel {
  public:
-  FakeXmppChannel(const scoped_refptr<base::TaskRunner>& task_runner,
-                  base::Clock* clock)
+  FakeXmppChannel(
+      const scoped_refptr<base::SingleThreadTaskRunner>& task_runner,
+      base::Clock* clock)
       : XmppChannel{kAccountName, kAccessToken, task_runner},
         fake_stream_{chromeos::Stream::AccessMode::READ_WRITE, task_runner,
                      clock} {}
@@ -121,6 +127,8 @@
     auto callback = [this](const tracked_objects::Location& from_here,
                            const base::Closure& task,
                            base::TimeDelta delay) -> bool {
+      if (ignore_delayed_tasks_ && delay > base::TimeDelta::FromMilliseconds(0))
+        return true;
       clock_.Advance(delay);
       message_queue_.push(task);
       return true;
@@ -155,10 +163,16 @@
       count--;
     }
   }
+
+  void SetIgnoreDelayedTasks(bool ignore) {
+    ignore_delayed_tasks_ = true;
+  }
+
   std::unique_ptr<FakeXmppChannel> xmpp_client_;
   base::SimpleTestClock clock_;
   scoped_refptr<TestTaskRunner> task_runner_;
   std::queue<base::Closure> message_queue_;
+  bool ignore_delayed_tasks_{false};
 };
 
 TEST_F(XmppChannelTest, StartStream) {
@@ -205,36 +219,31 @@
 }
 
 TEST_F(XmppChannelTest, HandleStreamRestartedResponse) {
+  SetIgnoreDelayedTasks(true);
   StartWithState(XmppChannel::XmppState::kStreamRestartedPostAuthentication);
   xmpp_client_->fake_stream_.AddReadPacketString({}, kRestartStreamResponse);
   xmpp_client_->fake_stream_.ExpectWritePacketString({}, kBindMessage);
-  RunTasks(4);
+  RunTasks(3);
   EXPECT_EQ(XmppChannel::XmppState::kBindSent,
             xmpp_client_->state());
-}
+  EXPECT_TRUE(xmpp_client_->jid().empty());
 
-TEST_F(XmppChannelTest, HandleBindResponse) {
-  StartWithState(XmppChannel::XmppState::kBindSent);
   xmpp_client_->fake_stream_.AddReadPacketString({}, kBindResponse);
   xmpp_client_->fake_stream_.ExpectWritePacketString({}, kSessionMessage);
-  RunTasks(4);
+  RunTasks(9);
   EXPECT_EQ(XmppChannel::XmppState::kSessionStarted,
             xmpp_client_->state());
-}
+  EXPECT_EQ("110cc78f78d7032cc7bf2c6e14c1fa7d@clouddevices.gserviceaccount.com"
+            "/19853128", xmpp_client_->jid());
 
-TEST_F(XmppChannelTest, HandleSessionResponse) {
-  StartWithState(XmppChannel::XmppState::kSessionStarted);
   xmpp_client_->fake_stream_.AddReadPacketString({}, kSessionResponse);
   xmpp_client_->fake_stream_.ExpectWritePacketString({}, kSubscribeMessage);
   RunTasks(4);
   EXPECT_EQ(XmppChannel::XmppState::kSubscribeStarted,
             xmpp_client_->state());
-}
 
-TEST_F(XmppChannelTest, HandleSubscribeResponse) {
-  StartWithState(XmppChannel::XmppState::kSubscribeStarted);
   xmpp_client_->fake_stream_.AddReadPacketString({}, kSubscribedResponse);
-  RunTasks(3);
+  RunTasks(5);
   EXPECT_EQ(XmppChannel::XmppState::kSubscribed,
             xmpp_client_->state());
 }
diff --git a/buffet/notification/xmpp_iq_stanza_handler.cc b/buffet/notification/xmpp_iq_stanza_handler.cc
new file mode 100644
index 0000000..e3a14c0
--- /dev/null
+++ b/buffet/notification/xmpp_iq_stanza_handler.cc
@@ -0,0 +1,128 @@
+// Copyright 2015 The Chromium OS 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 "buffet/notification/xmpp_iq_stanza_handler.h"
+
+#include <base/bind.h>
+#include <base/strings/string_number_conversions.h>
+#include <base/strings/stringprintf.h>
+
+#include "buffet/notification/xml_node.h"
+#include "buffet/notification/xmpp_channel.h"
+
+namespace buffet {
+
+namespace {
+
+// Default timeout for <iq> requests to the server. If the response hasn't been
+// received within this time interval, the request is considered as failed.
+const int kTimeoutIntervalSeconds = 30;
+
+// Builds an XML stanza that looks like this:
+//  <iq id='${id}' type='${type}' from='${from}' to='${to}'>$body</iq>
+// where 'to' and 'from' are optional attributes.
+std::string BuildIqStanza(const std::string& id,
+                          const std::string& type,
+                          const std::string& to,
+                          const std::string& from,
+                          const std::string& body) {
+  std::string to_attr;
+  if (!to.empty()) {
+    CHECK_EQ(std::string::npos, to.find_first_of("<'>"))
+        << "Destination address contains invalid XML characters";
+    base::StringAppendF(&to_attr, " to='%s'", to.c_str());
+  }
+  std::string from_attr;
+  if (!from.empty()) {
+    CHECK_EQ(std::string::npos, from.find_first_of("<'>"))
+        << "Source address contains invalid XML characters";
+    base::StringAppendF(&from_attr, " from='%s'", from.c_str());
+  }
+  return base::StringPrintf("<iq id='%s' type='%s'%s%s>%s</iq>",
+                            id.c_str(), type.c_str(), from_attr.c_str(),
+                            to_attr.c_str(), body.c_str());
+}
+
+}  // anonymous namespace
+
+IqStanzaHandler::IqStanzaHandler(
+    XmppChannelInterface* xmpp_channel,
+    const scoped_refptr<base::SingleThreadTaskRunner>& task_runner)
+    : xmpp_channel_{xmpp_channel}, task_runner_{task_runner} {}
+
+void IqStanzaHandler::SendRequest(
+    const std::string& type,
+    const std::string& from,
+    const std::string& to,
+    const std::string& body,
+    const ResponseCallback& response_callback,
+    const TimeoutCallback& timeout_callback) {
+  // Remember the response callback to call later.
+  requests_.emplace(++last_request_id_, response_callback);
+  // Schedule a time-out callback for this request.
+  task_runner_->PostDelayedTask(
+      FROM_HERE,
+      base::Bind(&IqStanzaHandler::OnTimeOut,
+                 weak_ptr_factory_.GetWeakPtr(),
+                 last_request_id_,
+                 timeout_callback),
+      base::TimeDelta::FromSeconds(kTimeoutIntervalSeconds));
+
+  std::string message = BuildIqStanza(std::to_string(last_request_id_),
+                                      type, to, from, body);
+  xmpp_channel_->SendMessage(message);
+}
+
+bool IqStanzaHandler::HandleIqStanza(std::unique_ptr<XmlNode> stanza) {
+  std::string type;
+  if (!stanza->GetAttribute("type", &type)) {
+    LOG(ERROR) << "IQ stanza missing 'type' attribute";
+    return false;
+  }
+
+  std::string id_str;
+  if (!stanza->GetAttribute("id", &id_str)) {
+    LOG(ERROR) << "IQ stanza missing 'id' attribute";
+    return false;
+  }
+
+  if (type == "result" || type == "error") {
+    // These are response stanzas from the server.
+    // Find the corresponding request.
+    RequestId id;
+    if (!base::StringToInt(id_str, &id)) {
+      LOG(ERROR) << "IQ stanza's 'id' attribute is invalid";
+      return false;
+    }
+    auto p = requests_.find(id);
+    if (p != requests_.end()) {
+      task_runner_->PostTask(
+          FROM_HERE,
+          base::Bind(p->second, base::Passed(std::move(stanza))));
+      requests_.erase(p);
+    }
+  } else {
+    // We do not support server-initiated IQ requests ("set" / "get" / "query").
+    // So just reply with "not implemented" error (and swap "to"/"from" attrs).
+    std::string error_body =
+        "<error type='modify'>"
+        "<feature-not-implemented xmlns='urn:ietf:params:xml:ns:xmpp-stanzas'/>"
+        "</error>";
+    std::string message = BuildIqStanza(id_str, "error",
+                                        stanza->GetAttributeOrEmpty("from"),
+                                        stanza->GetAttributeOrEmpty("to"),
+                                        error_body);
+    xmpp_channel_->SendMessage(message);
+  }
+  return true;
+}
+
+void IqStanzaHandler::OnTimeOut(RequestId id,
+                                const TimeoutCallback& timeout_callback) {
+  // Request has not been processed yes, so a real timeout occurred.
+  if (requests_.erase(id) > 0)
+    timeout_callback.Run();
+}
+
+}  // namespace buffet
diff --git a/buffet/notification/xmpp_iq_stanza_handler.h b/buffet/notification/xmpp_iq_stanza_handler.h
new file mode 100644
index 0000000..0eb0900
--- /dev/null
+++ b/buffet/notification/xmpp_iq_stanza_handler.h
@@ -0,0 +1,72 @@
+// Copyright 2015 The Chromium OS Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef BUFFET_NOTIFICATION_XMPP_IQ_STANZA_HANDLER_H_
+#define BUFFET_NOTIFICATION_XMPP_IQ_STANZA_HANDLER_H_
+
+#include <map>
+#include <memory>
+#include <string>
+
+#include <base/callback_forward.h>
+#include <base/macros.h>
+#include <base/memory/weak_ptr.h>
+#include <base/single_thread_task_runner.h>
+
+#include "buffet/notification/xmpp_stream_parser.h"
+
+namespace buffet {
+
+class XmppChannelInterface;
+
+class IqStanzaHandler {
+ public:
+  using ResponseCallback =
+      base::Callback<void(std::unique_ptr<XmlNode>)>;
+  using TimeoutCallback = base::Closure;
+
+  IqStanzaHandler(
+      XmppChannelInterface* xmpp_channel,
+      const scoped_refptr<base::SingleThreadTaskRunner>& task_runner);
+
+  // Sends <iq> request to the server.
+  // |type| is the IQ stanza type, one of "get", "set", "query".
+  // |to| is the target of the message. If empty string, 'to' is omitted.
+  // |body| the XML snipped to include between <iq>...</iq>
+  // |response_callback| is called with result or error XML stanza received
+  // from the server in response to the request sent.
+  // |timeout_callback| is called when the response to the request hasn't been
+  // received within the time allotted.
+  void SendRequest(const std::string& type,
+                   const std::string& from,
+                   const std::string& to,
+                   const std::string& body,
+                   const ResponseCallback& response_callback,
+                   const TimeoutCallback& timeout_callback);
+
+  // Processes an <iq> stanza is received from the server. This will match the
+  // stanza's 'id' attribute with pending request ID and if found, will
+  // call the |response_callback|, or if the request is not found, an error
+  // stanza fill be sent back to the server.
+  // Returns false if some unexpected condition occurred and the stream should
+  // be restarted.
+  bool HandleIqStanza(std::unique_ptr<XmlNode> stanza);
+
+ private:
+  using RequestId = int;
+  void OnTimeOut(RequestId id, const TimeoutCallback& timeout_callback);
+
+  XmppChannelInterface* xmpp_channel_;
+  scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
+  std::map<RequestId, ResponseCallback> requests_;
+  RequestId last_request_id_{0};
+
+  base::WeakPtrFactory<IqStanzaHandler> weak_ptr_factory_{this};
+  DISALLOW_COPY_AND_ASSIGN(IqStanzaHandler);
+};
+
+}  // namespace buffet
+
+#endif  // BUFFET_NOTIFICATION_XMPP_IQ_STANZA_HANDLER_H_
+
diff --git a/buffet/notification/xmpp_iq_stanza_handler_unittest.cc b/buffet/notification/xmpp_iq_stanza_handler_unittest.cc
new file mode 100644
index 0000000..1e8be42
--- /dev/null
+++ b/buffet/notification/xmpp_iq_stanza_handler_unittest.cc
@@ -0,0 +1,222 @@
+// Copyright 2015 The Chromium OS 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 "buffet/notification/xmpp_iq_stanza_handler.h"
+
+#include <map>
+#include <memory>
+
+#include <chromeos/bind_lambda.h>
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "buffet/notification/xml_node.h"
+#include "buffet/notification/xmpp_channel.h"
+#include "buffet/notification/xmpp_stream_parser.h"
+
+using testing::Invoke;
+using testing::Return;
+using testing::_;
+
+namespace buffet {
+namespace {
+
+// Mock-like task runner that allow the tests to inspect the calls to
+// TaskRunner::PostDelayedTask and verify the delays.
+class TestTaskRunner : public base::SingleThreadTaskRunner {
+ public:
+  TestTaskRunner() = default;
+
+  MOCK_METHOD3(PostDelayedTask, bool(const tracked_objects::Location&,
+                                     const base::Closure&,
+                                     base::TimeDelta));
+  MOCK_METHOD3(PostNonNestableDelayedTask,
+               bool(const tracked_objects::Location&,
+                    const base::Closure&,
+                    base::TimeDelta));
+
+  bool RunsTasksOnCurrentThread() const { return true; }
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(TestTaskRunner);
+};
+
+class MockXmppChannelInterface : public XmppChannelInterface {
+ public:
+  MockXmppChannelInterface() = default;
+
+  MOCK_METHOD1(SendMessage, void(const std::string&));
+
+ private:
+  DISALLOW_COPY_AND_ASSIGN(MockXmppChannelInterface);
+};
+
+// Simple class that allows to parse XML from string to XmlNode.
+class XmlParser : public XmppStreamParser::Delegate {
+ public:
+  std::unique_ptr<XmlNode> Parse(const std::string& xml) {
+    parser_.ParseData(xml);
+    return std::move(node_);
+  }
+
+ private:
+  // Overrides from XmppStreamParser::Delegate.
+  void OnStreamStart(const std::string& node_name,
+                     std::map<std::string, std::string> attributes) override {
+    node_.reset(new XmlNode{node_name, std::move(attributes)});
+  }
+
+  void OnStreamEnd(const std::string& node_name) override {}
+
+  void OnStanza(std::unique_ptr<XmlNode> stanza) override {
+    node_->AddChild(std::move(stanza));
+  }
+
+  std::unique_ptr<XmlNode> node_;
+  XmppStreamParser parser_{this};
+};
+
+class MockResponseReceiver {
+ public:
+  MOCK_METHOD2(OnResponse, void(int id, const std::string&));
+
+  IqStanzaHandler::ResponseCallback callback(int id) {
+    return base::Bind(&MockResponseReceiver::OnResponseCallback,
+                      base::Unretained(this), id);
+  }
+
+ private:
+  void OnResponseCallback(int id, std::unique_ptr<XmlNode> response) {
+    OnResponse(id, response->children().front()->name());
+  }
+};
+
+// Functor to call the |task| immediately.
+struct TaskInvoker {
+  bool operator()(const tracked_objects::Location& location,
+                  const base::Closure& task,
+                  base::TimeDelta delay) {
+    task.Run();
+    return true;
+  }
+};
+
+}  // anonymous namespace
+
+class IqStanzaHandlerTest : public testing::Test {
+ public:
+  void SetUp() override {
+    task_runner_ = new TestTaskRunner;
+    iq_stanza_handler_.reset(
+        new IqStanzaHandler{&mock_xmpp_channel_, task_runner_});
+  }
+
+  testing::StrictMock<MockXmppChannelInterface> mock_xmpp_channel_;
+  scoped_refptr<TestTaskRunner> task_runner_;
+  std::unique_ptr<IqStanzaHandler> iq_stanza_handler_;
+  MockResponseReceiver receiver_;
+  TaskInvoker task_invoker_;
+};
+
+TEST_F(IqStanzaHandlerTest, SendRequest) {
+  EXPECT_CALL(*task_runner_, PostDelayedTask(_, _, _))
+      .WillRepeatedly(Return(true));
+
+  std::string expected_msg = "<iq id='1' type='set'><body/></iq>";
+  EXPECT_CALL(mock_xmpp_channel_, SendMessage(expected_msg)).Times(1);
+  iq_stanza_handler_->SendRequest("set", "", "", "<body/>", {}, {});
+
+  expected_msg = "<iq id='2' type='get'><body/></iq>";
+  EXPECT_CALL(mock_xmpp_channel_, SendMessage(expected_msg)).Times(1);
+  iq_stanza_handler_->SendRequest("get", "", "", "<body/>", {}, {});
+
+  expected_msg = "<iq id='3' type='query' from='foo@bar'><body/></iq>";
+  EXPECT_CALL(mock_xmpp_channel_, SendMessage(expected_msg)).Times(1);
+  iq_stanza_handler_->SendRequest("query", "foo@bar", "", "<body/>", {}, {});
+
+  expected_msg = "<iq id='4' type='query' to='foo@bar'><body/></iq>";
+  EXPECT_CALL(mock_xmpp_channel_, SendMessage(expected_msg)).Times(1);
+  iq_stanza_handler_->SendRequest("query", "", "foo@bar", "<body/>", {}, {});
+
+  expected_msg = "<iq id='5' type='query' from='foo@bar' to='baz'><body/></iq>";
+  EXPECT_CALL(mock_xmpp_channel_, SendMessage(expected_msg)).Times(1);
+  iq_stanza_handler_->SendRequest("query", "foo@bar", "baz", "<body/>", {}, {});
+}
+
+TEST_F(IqStanzaHandlerTest, UnsupportedIqRequest) {
+  // Server IQ requests are not supported for now. Expect an error response.
+  std::string expected_msg =
+      "<iq id='1' type='error'><error type='modify'>"
+      "<feature-not-implemented xmlns='urn:ietf:params:xml:ns:xmpp-stanzas'/>"
+      "</error></iq>";
+  EXPECT_CALL(mock_xmpp_channel_, SendMessage(expected_msg)).Times(1);
+  auto request = XmlParser{}.Parse("<iq id='1' type='set'><foo/></iq>");
+  EXPECT_TRUE(iq_stanza_handler_->HandleIqStanza(std::move(request)));
+}
+
+TEST_F(IqStanzaHandlerTest, UnknownResponseId) {
+  // No requests with ID=100 have been previously sent.
+  auto request = XmlParser{}.Parse("<iq id='100' type='result'><foo/></iq>");
+  EXPECT_TRUE(iq_stanza_handler_->HandleIqStanza(std::move(request)));
+}
+
+TEST_F(IqStanzaHandlerTest, SequentialResponses) {
+  EXPECT_CALL(*task_runner_, PostDelayedTask(_, _, _))
+      .Times(2).WillRepeatedly(Return(true));
+
+  EXPECT_CALL(mock_xmpp_channel_, SendMessage(_)).Times(2);
+  iq_stanza_handler_->SendRequest("set", "", "", "<body/>",
+                                  receiver_.callback(1), {});
+  iq_stanza_handler_->SendRequest("get", "", "", "<body/>",
+                                  receiver_.callback(2), {});
+
+  EXPECT_CALL(*task_runner_, PostDelayedTask(_, _, _))
+      .Times(2).WillRepeatedly(Invoke(task_invoker_));
+
+  EXPECT_CALL(receiver_, OnResponse(1, "foo"));
+  auto request = XmlParser{}.Parse("<iq id='1' type='result'><foo/></iq>");
+  EXPECT_TRUE(iq_stanza_handler_->HandleIqStanza(std::move(request)));
+
+  EXPECT_CALL(receiver_, OnResponse(2, "bar"));
+  request = XmlParser{}.Parse("<iq id='2' type='result'><bar/></iq>");
+  EXPECT_TRUE(iq_stanza_handler_->HandleIqStanza(std::move(request)));
+}
+
+TEST_F(IqStanzaHandlerTest, OutOfOrderResponses) {
+  EXPECT_CALL(*task_runner_, PostDelayedTask(_, _, _))
+      .Times(2).WillRepeatedly(Return(true));
+
+  EXPECT_CALL(mock_xmpp_channel_, SendMessage(_)).Times(2);
+  iq_stanza_handler_->SendRequest("set", "", "", "<body/>",
+                                  receiver_.callback(1), {});
+  iq_stanza_handler_->SendRequest("get", "", "", "<body/>",
+                                  receiver_.callback(2), {});
+
+  EXPECT_CALL(*task_runner_, PostDelayedTask(_, _, _))
+      .Times(2).WillRepeatedly(Invoke(task_invoker_));
+
+  EXPECT_CALL(receiver_, OnResponse(2, "bar"));
+  auto request = XmlParser{}.Parse("<iq id='2' type='result'><bar/></iq>");
+  EXPECT_TRUE(iq_stanza_handler_->HandleIqStanza(std::move(request)));
+
+  EXPECT_CALL(receiver_, OnResponse(1, "foo"));
+  request = XmlParser{}.Parse("<iq id='1' type='result'><foo/></iq>");
+  EXPECT_TRUE(iq_stanza_handler_->HandleIqStanza(std::move(request)));
+}
+
+TEST_F(IqStanzaHandlerTest, RequestTimeout) {
+  EXPECT_CALL(*task_runner_, PostDelayedTask(_, _, _))
+      .WillOnce(Invoke(task_invoker_));
+
+  bool called = false;
+  auto on_timeout = [&called]() { called = true; };
+
+  EXPECT_CALL(mock_xmpp_channel_, SendMessage(_)).Times(1);
+  EXPECT_FALSE(called);
+  iq_stanza_handler_->SendRequest("set", "", "", "<body/>", {},
+                                  base::Bind(on_timeout));
+  EXPECT_TRUE(called);
+}
+
+}  // namespace buffet