diff --git a/buffet/buffet.gyp b/buffet/buffet.gyp
index a68d772..0d6fd92 100644
--- a/buffet/buffet.gyp
+++ b/buffet/buffet.gyp
@@ -3,6 +3,7 @@
     'variables': {
       'deps': [
         'dbus-1',
+        'expat',
         'libchrome-<(libbase_ver)',
         'libchromeos-<(libbase_ver)',
       ],
@@ -39,7 +40,9 @@
         'dbus_bindings/org.chromium.Buffet.Manager.xml',
         'dbus_constants.cc',
         'manager.cc',
+        'notification/xml_node.cc',
         'notification/xmpp_channel.cc',
+        'notification/xmpp_stream_parser.cc',
         'registration_status.cc',
         'storage_impls.cc',
         'states/error_codes.cc',
@@ -120,7 +123,9 @@
             'commands/schema_utils_unittest.cc',
             'commands/unittest_utils.cc',
             'device_registration_info_unittest.cc',
+            'notification/xml_node_unittest.cc',
             'notification/xmpp_channel_unittest.cc',
+            'notification/xmpp_stream_parser_unittest.cc',
             'states/state_change_queue_unittest.cc',
             'states/state_manager_unittest.cc',
             'states/state_package_unittest.cc',
diff --git a/buffet/notification/xml_node.cc b/buffet/notification/xml_node.cc
new file mode 100644
index 0000000..f244c3a
--- /dev/null
+++ b/buffet/notification/xml_node.cc
@@ -0,0 +1,122 @@
+// 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/xml_node.h"
+
+#include <base/strings/stringprintf.h>
+#include <chromeos/strings/string_utils.h>
+
+namespace buffet {
+
+XmlNode::XmlNode(const std::string& name,
+                 std::map<std::string, std::string> attributes)
+    : name_{name}, attributes_{std::move(attributes)} {}
+
+const std::string& XmlNode::name() const {
+  return name_;
+}
+
+const std::string& XmlNode::text() const {
+  return text_;
+}
+
+const std::map<std::string, std::string>& XmlNode::attributes() const {
+  return attributes_;
+}
+
+const std::vector<std::unique_ptr<XmlNode>>& XmlNode::children() const {
+  return children_;
+}
+
+bool XmlNode::GetAttribute(const std::string& name, std::string* value) const {
+  auto p = attributes_.find(name);
+  if (p == attributes_.end())
+    return false;
+
+  *value = p->second;
+  return true;
+}
+
+std::string XmlNode::GetAttributeOrEmpty(const std::string& name) const {
+  std::string value;
+  GetAttribute(name, &value);
+  return value;
+}
+
+const XmlNode* XmlNode::FindFirstChild(const std::string& name_path,
+                                       bool recursive) const {
+  return FindChildHelper(name_path, recursive, nullptr);
+}
+
+std::vector<const XmlNode*> XmlNode::FindChildren(const std::string& name_path,
+                                                  bool recursive) const {
+  std::vector<const XmlNode*> children;
+  FindChildHelper(name_path, recursive, &children);
+  return children;
+}
+
+const XmlNode* XmlNode::FindChildHelper(
+    const std::string& name_path,
+    bool recursive,
+    std::vector<const XmlNode*>* children) const {
+  std::string name;
+  std::string rest_of_path;
+  chromeos::string_utils::SplitAtFirst(name_path, "/", &name, &rest_of_path,
+                                       false);
+  for (const auto& child : children_) {
+    const XmlNode* found_node = nullptr;
+    if (child->name() == name) {
+      if (rest_of_path.empty()) {
+        found_node = child.get();
+      } else {
+        found_node = child->FindChildHelper(rest_of_path, false, children);
+      }
+    } else if (recursive) {
+      found_node = child->FindChildHelper(name_path, true, children);
+    }
+
+    if (found_node) {
+      if (!children)
+        return found_node;
+      children->push_back(found_node);
+    }
+  }
+  return nullptr;
+}
+
+void XmlNode::SetText(const std::string& text) {
+  text_ = text;
+}
+
+void XmlNode::AppendText(const std::string& text) {
+  text_ += text;
+}
+
+void XmlNode::AddChild(std::unique_ptr<XmlNode> child) {
+  child->parent_ = this;
+  children_.push_back(std::move(child));
+}
+
+std::string XmlNode::ToString() const {
+  std::string xml = base::StringPrintf("<%s", name_.c_str());
+  for (const auto& pair : attributes_) {
+    base::StringAppendF(&xml, " %s=\"%s\"", pair.first.c_str(),
+                        pair.second.c_str());
+  }
+  if (text_.empty() && children_.empty()) {
+    xml += "/>";
+  } else {
+    xml += '>';
+    if (!text_.empty()) {
+      xml += text_;
+    }
+    for (const auto& child : children_) {
+      xml += child->ToString();
+    }
+    base::StringAppendF(&xml, "</%s>", name_.c_str());
+  }
+  return xml;
+}
+
+}  // namespace buffet
diff --git a/buffet/notification/xml_node.h b/buffet/notification/xml_node.h
new file mode 100644
index 0000000..fc43650
--- /dev/null
+++ b/buffet/notification/xml_node.h
@@ -0,0 +1,124 @@
+// 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_XML_NODE_H_
+#define BUFFET_NOTIFICATION_XML_NODE_H_
+
+#include <map>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include <base/macros.h>
+
+namespace buffet {
+
+class XmlNodeTest;
+class XmppStreamParser;
+
+// XmlNode is a very simple class to represent the XML document element tree.
+// It is used in conjunction with expat XML parser to implement XmppStreamParser
+// class used to parse Xmpp data stream into individual stanzas.
+class XmlNode final {
+ public:
+  XmlNode(const std::string& name,
+          std::map<std::string, std::string> attributes);
+
+  // The node's name. E.g. in <foo bar="baz">quux</foo> this will return "foo".
+  const std::string& name() const;
+  // The node text content. E.g. in <foo bar="baz">quux</foo> this will return
+  // "quux".
+  const std::string& text() const;
+  // The node attribute map. E.g. in <foo bar="baz">quux</foo> this will return
+  // {{"bar", "baz"}}.
+  const std::map<std::string, std::string>& attributes() const;
+  // Returns the list of child nodes, if any.
+  const std::vector<std::unique_ptr<XmlNode>>& children() const;
+
+  // Retrieves the value of the given attribute specified by |name|.
+  // If the attribute doesn't exist, returns false and |value| is not modified.
+  bool GetAttribute(const std::string& name, std::string* value) const;
+  // Returns the value of the given attribute specified by |name|.
+  // Returns empty string if the attribute does not exist. This method should be
+  // used only in limited scopes such as unit tests.
+  std::string GetAttributeOrEmpty(const std::string& name) const;
+
+  // Finds a first occurrence of a child node specified by |name_path|. A name
+  // path is a "/"-separated list of node names to look for. If |recursive| is
+  // set to true, the children are recursively traversed trying to match the
+  // node names. Otherwise only first-level children of the current node are
+  // matched against the top-level name of |name_path|.
+  // This method returns a pointer to the first node that matches the path,
+  // otherwise a nullptr is returned.
+  const XmlNode* FindFirstChild(const std::string& name_path,
+                                bool recursive) const;
+
+  // Finds all the child nodes matching the |name_path|. This returns the list
+  // of pointers to the child nodes matching the criteria. If |recursive| is
+  // set to true, the children are recursively traversed trying to match the
+  // node names. Otherwise only first-level children of the current node are
+  // matched against the top-level name of |name_path|.
+  // For example, if the current node represents the <top> element of the
+  // following XML document:
+  //  <top>
+  //    <node1 id="1"><node2 id="2"><node3 id="3"/></node2></node1>
+  //    <node2 id="4"><node3 id="5"/></node2>
+  //    <node3 id="6"/>
+  //    <node2 id="7"><node4 id="8"><node3 id="9"/></node4></node2>
+  //  </top>
+  // Then recursively searching for nodes will produce the following results
+  // (only the node "id" attributes are listed in the results, for brevity):
+  //    FindChildren("node2/node3", false) -> {"5"}.
+  //    FindChildren("node2/node3", true) -> {"3", "5"}.
+  //    FindChildren("node3", false) -> {"6"}.
+  //    FindChildren("node3", true) -> {"3", "5", "6", "9"}.
+  std::vector<const XmlNode*> FindChildren(const std::string& name_path,
+                                           bool recursive) const;
+
+  // Adds a new child to the bottom of the child list of this node.
+  void AddChild(std::unique_ptr<XmlNode> child);
+
+  // Converts the node tree to XML-like string. Note that this not necessarily
+  // produces a valid XML string. It does not use any character escaping or
+  // canonicalization, which will produce invalid XML if any of the node or
+  // attribute names or values contain special characters such as ", <, >, etc.
+  // This function should be used only for logging/debugging purposes only and
+  // never to generate valid XML from the parsed node tree.
+  std::string ToString() const;
+
+ private:
+  friend class XmlNodeTest;
+  friend class XmppStreamParser;
+
+  // Sets the node's text. Used by XML parser.
+  void SetText(const std::string& text);
+  // Appends the |text| to the node's text string.
+  void AppendText(const std::string& text);
+
+  // Helper method used by FindFirstChild() and FindChildren(). Searches for
+  // child node(s) matching |name_path|.
+  // If |children| is not specified (nullptr), this function find the first
+  // matching node and returns it via return value of the function. If no match
+  // is found, this function will return nullptr.
+  // If |children| parameter is not nullptr, found nodes are added to the
+  // vector pointed to by |children| and search continues until the whole tree
+  // is inspected. In this mode, the function always returns nullptr.
+  const XmlNode* FindChildHelper(const std::string& name_path,
+                                 bool recursive,
+                                 std::vector<const XmlNode*>* children) const;
+
+
+  const XmlNode* parent_{nullptr};  // Weak pointer to the parent node, if any.
+  std::string name_;
+  std::string text_;
+  std::map<std::string, std::string> attributes_;
+  std::vector<std::unique_ptr<XmlNode>> children_;
+
+  DISALLOW_COPY_AND_ASSIGN(XmlNode);
+};
+
+}  // namespace buffet
+
+#endif  // BUFFET_NOTIFICATION_XML_NODE_H_
+
diff --git a/buffet/notification/xml_node_unittest.cc b/buffet/notification/xml_node_unittest.cc
new file mode 100644
index 0000000..8b2ed5c
--- /dev/null
+++ b/buffet/notification/xml_node_unittest.cc
@@ -0,0 +1,211 @@
+// 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/xml_node.h"
+
+#include <memory>
+
+#include <gtest/gtest.h>
+
+#include "buffet/notification/xmpp_stream_parser.h"
+
+namespace buffet {
+namespace {
+
+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};
+};
+
+}  // anonymous namespace
+
+class XmlNodeTest : public testing::Test {
+ public:
+  void SetUp() override {
+    node_.reset(new XmlNode{"test_node",
+                            {{"attr1", "val1"}, {"attr2", "val2"}}});
+  }
+
+  // Accessor helpers for private members of XmlNode.
+  static const XmlNode* GetParent(const XmlNode& node) {
+    return node.parent_;
+  }
+
+  static void SetText(XmlNode* node, const std::string& text) {
+    node->SetText(text);
+  }
+
+  static void AppendText(XmlNode* node, const std::string& text) {
+    node->AppendText(text);
+  }
+
+  void CreateNodeTree() {
+    node_ = XmlParser{}.Parse(R"(
+        <top>
+          <node1 id="1"><node2 id="2"><node3 id="3"/></node2></node1>
+          <node2 id="4"><node3 id="5"/></node2>
+          <node3 id="6"/>
+          <node2 id="7"><node4 id="8"><node3 id="9"/></node4></node2>
+        </top>
+        )");
+  }
+
+  std::unique_ptr<XmlNode> node_;
+};
+
+TEST_F(XmlNodeTest, DefaultConstruction) {
+  EXPECT_EQ("test_node", node_->name());
+  EXPECT_TRUE(node_->children().empty());
+  EXPECT_TRUE(node_->text().empty());
+}
+
+TEST_F(XmlNodeTest, SetText) {
+  SetText(node_.get(), "foobar");
+  EXPECT_EQ("foobar", node_->text());
+}
+
+TEST_F(XmlNodeTest, AppendText) {
+  SetText(node_.get(), "foobar");
+  AppendText(node_.get(), "-baz");
+  EXPECT_EQ("foobar-baz", node_->text());
+}
+
+TEST_F(XmlNodeTest, AddChild) {
+  std::unique_ptr<XmlNode> child{new XmlNode{"child", {}}};
+  node_->AddChild(std::move(child));
+  EXPECT_EQ(1u, node_->children().size());
+  EXPECT_EQ("child", node_->children().front()->name());
+  EXPECT_EQ(node_.get(), GetParent(*node_->children().front().get()));
+}
+
+TEST_F(XmlNodeTest, Attributes) {
+  const std::map<std::string, std::string> expected_attrs{
+      {"attr1", "val1"},
+      {"attr2", "val2"}
+  };
+  EXPECT_EQ(expected_attrs, node_->attributes());
+  std::string attr = "bar";
+  EXPECT_FALSE(node_->GetAttribute("foo", &attr));
+  EXPECT_EQ("bar", attr);  // Shouldn't be changed by failed GetAttribute().
+  EXPECT_TRUE(node_->GetAttribute("attr1", &attr));
+  EXPECT_EQ("val1", attr);
+  EXPECT_TRUE(node_->GetAttribute("attr2", &attr));
+  EXPECT_EQ("val2", attr);
+
+  XmlNode new_node{"node", {}};
+  EXPECT_FALSE(new_node.GetAttribute("attr1", &attr));
+}
+
+TEST_F(XmlNodeTest, FindFirstChild_SingleNode) {
+  CreateNodeTree();
+  const XmlNode* node = node_->FindFirstChild("node3", false);
+  ASSERT_NE(nullptr, node);
+  EXPECT_EQ("node3", node->name());
+  EXPECT_EQ("6", node->GetAttributeOrEmpty("id"));
+
+  node = node_->FindFirstChild("node3", true);
+  ASSERT_NE(nullptr, node);
+  EXPECT_EQ("node3", node->name());
+  EXPECT_EQ("3", node->GetAttributeOrEmpty("id"));
+
+  node = node_->FindFirstChild("foo", true);
+  ASSERT_EQ(nullptr, node);
+}
+
+TEST_F(XmlNodeTest, FindFirstChild_Path) {
+  CreateNodeTree();
+  const XmlNode* node = node_->FindFirstChild("node2/node3", false);
+  ASSERT_NE(nullptr, node);
+  EXPECT_EQ("node3", node->name());
+  EXPECT_EQ("5", node->GetAttributeOrEmpty("id"));
+
+  node = node_->FindFirstChild("node2/node3", true);
+  ASSERT_NE(nullptr, node);
+  EXPECT_EQ("node3", node->name());
+  EXPECT_EQ("3", node->GetAttributeOrEmpty("id"));
+
+  node = node_->FindFirstChild("node1/node2/node3", false);
+  ASSERT_NE(nullptr, node);
+  EXPECT_EQ("node3", node->name());
+  EXPECT_EQ("3", node->GetAttributeOrEmpty("id"));
+
+  node = node_->FindFirstChild("node1/node2/node3", true);
+  ASSERT_NE(nullptr, node);
+  EXPECT_EQ("node3", node->name());
+  EXPECT_EQ("3", node->GetAttributeOrEmpty("id"));
+
+  node = node_->FindFirstChild("foo/node3", true);
+  ASSERT_EQ(nullptr, node);
+}
+
+TEST_F(XmlNodeTest, FindChildren_SingleNode) {
+  CreateNodeTree();
+  auto children = node_->FindChildren("node3", false);
+  ASSERT_EQ(1u, children.size());
+  EXPECT_EQ("node3", children[0]->name());
+  EXPECT_EQ("6", children[0]->GetAttributeOrEmpty("id"));
+
+  children = node_->FindChildren("node3", true);
+  ASSERT_EQ(4u, children.size());
+  EXPECT_EQ("node3", children[0]->name());
+  EXPECT_EQ("3", children[0]->GetAttributeOrEmpty("id"));
+  EXPECT_EQ("node3", children[1]->name());
+  EXPECT_EQ("5", children[1]->GetAttributeOrEmpty("id"));
+  EXPECT_EQ("node3", children[2]->name());
+  EXPECT_EQ("6", children[2]->GetAttributeOrEmpty("id"));
+  EXPECT_EQ("node3", children[3]->name());
+  EXPECT_EQ("9", children[3]->GetAttributeOrEmpty("id"));
+}
+
+TEST_F(XmlNodeTest, FindChildren_Path) {
+  CreateNodeTree();
+  auto children = node_->FindChildren("node2/node3", false);
+  ASSERT_EQ(1u, children.size());
+  EXPECT_EQ("node3", children[0]->name());
+  EXPECT_EQ("5", children[0]->GetAttributeOrEmpty("id"));
+
+  children = node_->FindChildren("node2/node3", true);
+  ASSERT_EQ(2u, children.size());
+  EXPECT_EQ("node3", children[0]->name());
+  EXPECT_EQ("3", children[0]->GetAttributeOrEmpty("id"));
+  EXPECT_EQ("node3", children[1]->name());
+  EXPECT_EQ("5", children[1]->GetAttributeOrEmpty("id"));
+
+  children = node_->FindChildren("node1/node2/node3", false);
+  ASSERT_EQ(1u, children.size());
+  EXPECT_EQ("node3", children[0]->name());
+  EXPECT_EQ("3", children[0]->GetAttributeOrEmpty("id"));
+
+  children = node_->FindChildren("node1/node2/node3", true);
+  ASSERT_EQ(1u, children.size());
+  EXPECT_EQ("node3", children[0]->name());
+  EXPECT_EQ("3", children[0]->GetAttributeOrEmpty("id"));
+
+  children = node_->FindChildren("foo/bar", false);
+  ASSERT_EQ(0u, children.size());
+
+  children = node_->FindChildren("node2/baz", false);
+  ASSERT_EQ(0u, children.size());
+}
+
+}  // namespace buffet
diff --git a/buffet/notification/xmpp_channel.cc b/buffet/notification/xmpp_channel.cc
index 5590333..4567408 100644
--- a/buffet/notification/xmpp_channel.cc
+++ b/buffet/notification/xmpp_channel.cc
@@ -12,6 +12,7 @@
 #include <chromeos/streams/file_stream.h>
 
 #include "buffet/notification/notification_delegate.h"
+#include "buffet/notification/xml_node.h"
 #include "buffet/utils.h"
 
 namespace buffet {
@@ -104,112 +105,170 @@
 
 void XmppChannel::OnMessageRead(size_t size) {
   std::string msg(read_socket_data_.data(), size);
-  bool message_sent = false;
-
   VLOG(2) << "Received XMPP packet: " << msg;
+  read_pending_ = false;
+  stream_parser_.ParseData(msg);
+  WaitForMessage();
+}
+
+void XmppChannel::OnStreamStart(const std::string& node_name,
+                                std::map<std::string, std::string> attributes) {
+  VLOG(2) << "XMPP stream start: " << node_name;
+}
+
+void XmppChannel::OnStreamEnd(const std::string& node_name) {
+  VLOG(2) << "XMPP stream ended: " << node_name << ". Restarting XMPP";
+  task_runner_->PostTask(FROM_HERE,
+                         base::Bind(&XmppChannel::Restart,
+                                    weak_ptr_factory_.GetWeakPtr()));
+}
+
+void XmppChannel::OnStanza(std::unique_ptr<XmlNode> stanza) {
+  // Handle stanza asynchronously, since XmppChannel::OnStanza() is a callback
+  // from expat XML parser and some stanza could cause the XMPP stream to be
+  // reset and the parser to be re-initialized. We don't want to destroy the
+  // parser while it is performing a callback invocation.
+  task_runner_->PostTask(FROM_HERE,
+                         base::Bind(&XmppChannel::HandleStanza,
+                                    weak_ptr_factory_.GetWeakPtr(),
+                                    base::Passed(std::move(stanza))));
+}
+
+void XmppChannel::HandleStanza(std::unique_ptr<XmlNode> stanza) {
+  VLOG(2) << "XMPP stanza received: " << stanza->ToString();
 
   // TODO(nathanbullock): Need to add support for TLS (brillo:191).
   switch (state_) {
     case XmppState::kStarted:
-      if (std::string::npos != msg.find(":features") &&
-          std::string::npos != msg.find("X-GOOGLE-TOKEN")) {
-        state_ = XmppState::kAuthenticationStarted;
-        SendMessage(BuildXmppAuthenticateCommand(account_, access_token_));
-        message_sent = true;
+      if (stanza->name() == "stream:features") {
+        auto children = stanza->FindChildren("mechanisms/mechanism", false);
+        for (const auto& child : children) {
+          if (child->text() == "X-GOOGLE-TOKEN") {
+            state_ = XmppState::kAuthenticationStarted;
+            SendMessage(BuildXmppAuthenticateCommand(account_, access_token_));
+            return;
+          }
+        }
       }
       break;
     case XmppState::kAuthenticationStarted:
-      if (std::string::npos != msg.find("success")) {
+      if (stanza->name() == "success") {
         state_ = XmppState::kStreamRestartedPostAuthentication;
-        SendMessage(BuildXmppStartStreamCommand());
-        message_sent = true;
-      } else if (std::string::npos != msg.find("not-authorized")) {
-        state_ = XmppState::kAuthenticationFailed;
-        if (delegate_)
-          delegate_->OnPermanentFailure();
+        RestartXmppStream();
         return;
+      } else if (stanza->name() == "failure") {
+        if (stanza->FindFirstChild("not-authorized", false)) {
+          state_ = XmppState::kAuthenticationFailed;
+          if (delegate_)
+            delegate_->OnPermanentFailure();
+          return;
+        }
       }
       break;
     case XmppState::kStreamRestartedPostAuthentication:
-      if (std::string::npos != msg.find(":features") &&
-          std::string::npos != msg.find(":xmpp-session")) {
+      if (stanza->name() == "stream:features" &&
+          stanza->FindFirstChild("bind", false)) {
         state_ = XmppState::kBindSent;
         SendMessage(BuildXmppBindCommand());
-        message_sent = true;
+        return;
       }
       break;
     case XmppState::kBindSent:
-      if (std::string::npos != msg.find("iq") &&
-          std::string::npos != msg.find("result")) {
+      if (stanza->name() == "iq" &&
+          stanza->GetAttributeOrEmpty("type") == "result") {
         state_ = XmppState::kSessionStarted;
         SendMessage(BuildXmppStartSessionCommand());
-        message_sent = true;
+        return;
       }
       break;
     case XmppState::kSessionStarted:
-      if (std::string::npos != msg.find("iq") &&
-          std::string::npos != msg.find("result")) {
+      if (stanza->name() == "iq" &&
+          stanza->GetAttributeOrEmpty("type") == "result") {
         state_ = XmppState::kSubscribeStarted;
         SendMessage(BuildXmppSubscribeCommand(account_));
-        message_sent = true;
+        return;
       }
       break;
     case XmppState::kSubscribeStarted:
-      if (std::string::npos != msg.find("iq") &&
-          std::string::npos != msg.find("result")) {
+      if (stanza->name() == "iq" &&
+          stanza->GetAttributeOrEmpty("type") == "result") {
         state_ = XmppState::kSubscribed;
         if (delegate_)
           delegate_->OnConnected(GetName());
+        return;
       }
       break;
     default:
-      break;
+      LOG(INFO) << "Unexpected XMPP stanza ignored: " << stanza->ToString();
+      return;
   }
-  if (!message_sent)
-    WaitForMessage();
+  // Something bad happened. Close the stream and start over.
+  LOG(ERROR) << "Error condition occurred handling stanza: "
+             << stanza->ToString();
+  SendMessage("</stream:stream>");
 }
 
 void XmppChannel::SendMessage(const std::string& message) {
-  write_socket_data_ = message;
+  if (write_pending_) {
+    queued_write_data_ += message;
+    return;
+  }
+  write_socket_data_ = queued_write_data_ + message;
+  queued_write_data_.clear();
   chromeos::ErrorPtr error;
   VLOG(2) << "Sending XMPP message: " << message;
 
+  write_pending_ = true;
   bool ok = stream_->WriteAllAsync(
       write_socket_data_.data(),
       write_socket_data_.size(),
       base::Bind(&XmppChannel::OnMessageSent, weak_ptr_factory_.GetWeakPtr()),
-      base::Bind(&XmppChannel::OnError, weak_ptr_factory_.GetWeakPtr()),
+      base::Bind(&XmppChannel::OnWriteError, weak_ptr_factory_.GetWeakPtr()),
       &error);
 
   if (!ok)
-    OnError(error.get());
+    OnWriteError(error.get());
 }
 
 void XmppChannel::OnMessageSent() {
   chromeos::ErrorPtr error;
+  write_pending_ = false;
   if (!stream_->FlushBlocking(&error)) {
-    OnError(error.get());
+    OnWriteError(error.get());
     return;
   }
-  WaitForMessage();
+  if (queued_write_data_.empty()) {
+    WaitForMessage();
+  } else {
+    SendMessage(std::string{});
+  }
 }
 
 void XmppChannel::WaitForMessage() {
+  if (read_pending_)
+    return;
+
   chromeos::ErrorPtr error;
+  read_pending_ = true;
   bool ok = stream_->ReadAsync(
       read_socket_data_.data(),
       read_socket_data_.size(),
       base::Bind(&XmppChannel::OnMessageRead, weak_ptr_factory_.GetWeakPtr()),
-      base::Bind(&XmppChannel::OnError, weak_ptr_factory_.GetWeakPtr()),
+      base::Bind(&XmppChannel::OnReadError, weak_ptr_factory_.GetWeakPtr()),
       &error);
 
   if (!ok)
-    OnError(error.get());
+    OnReadError(error.get());
 }
 
-void XmppChannel::OnError(const chromeos::Error* error) {
-  Stop();
-  Start(delegate_);
+void XmppChannel::OnReadError(const chromeos::Error* error) {
+  read_pending_ = false;
+  Restart();
+}
+
+void XmppChannel::OnWriteError(const chromeos::Error* error) {
+  write_pending_ = false;
+  Restart();
 }
 
 void XmppChannel::Connect(const std::string& host, uint16_t port,
@@ -229,7 +288,7 @@
     stream_ = raw_socket_.get();
     callback.Run();
   } else {
-    VLOG(1) << "Delaying connection to XMPP server " << host << " for "
+    VLOG(2) << "Delaying connection to XMPP server " << host << " for "
             << backoff_entry_.GetTimeUntilRelease().InMilliseconds()
             << " milliseconds.";
     task_runner_->PostDelayedTask(
@@ -248,6 +307,11 @@
   // No extra parameters needed for XMPP.
 }
 
+void XmppChannel::Restart() {
+  Stop();
+  Start(delegate_);
+}
+
 void XmppChannel::Start(NotificationDelegate* delegate) {
   CHECK(state_ == XmppState::kNotStarted);
   delegate_ = delegate;
@@ -271,6 +335,14 @@
 
 void XmppChannel::OnConnected() {
   state_ = XmppState::kStarted;
+  RestartXmppStream();
+}
+
+void XmppChannel::RestartXmppStream() {
+  stream_parser_.Reset();
+  stream_->CancelPendingAsyncOperations();
+  read_pending_ = false;
+  write_pending_ = false;
   SendMessage(BuildXmppStartStreamCommand());
 }
 
diff --git a/buffet/notification/xmpp_channel.h b/buffet/notification/xmpp_channel.h
index 1fdd917..d6b1550 100644
--- a/buffet/notification/xmpp_channel.h
+++ b/buffet/notification/xmpp_channel.h
@@ -5,6 +5,7 @@
 #ifndef BUFFET_NOTIFICATION_XMPP_CHANNEL_H_
 #define BUFFET_NOTIFICATION_XMPP_CHANNEL_H_
 
+#include <map>
 #include <memory>
 #include <string>
 #include <vector>
@@ -17,10 +18,12 @@
 #include <chromeos/streams/stream.h>
 
 #include "buffet/notification/notification_channel.h"
+#include "buffet/notification/xmpp_stream_parser.h"
 
 namespace buffet {
 
-class XmppChannel : public NotificationChannel {
+class XmppChannel : public NotificationChannel,
+                    public XmppStreamParser::Delegate {
  public:
   // |account| is the robot account for buffet and |access_token|
   // it the OAuth token. Note that the OAuth token expires fairly frequently
@@ -59,13 +62,24 @@
   chromeos::Stream* stream_{nullptr};
 
  private:
+  // 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;
+
+  void HandleStanza(std::unique_ptr<XmlNode> stanza);
+  void RestartXmppStream();
+
   void SendMessage(const std::string& message);
   void WaitForMessage();
 
   void OnConnected();
   void OnMessageRead(size_t size);
   void OnMessageSent();
-  void OnError(const chromeos::Error* error);
+  void OnReadError(const chromeos::Error* error);
+  void OnWriteError(const chromeos::Error* error);
+  void Restart();
 
   // Robot account name for the device.
   std::string account_;
@@ -79,10 +93,14 @@
   std::vector<char> read_socket_data_;
   // Write buffer for outgoing message packets.
   std::string write_socket_data_;
+  std::string queued_write_data_;
 
   chromeos::BackoffEntry backoff_entry_;
   NotificationDelegate* delegate_{nullptr};
   scoped_refptr<base::TaskRunner> task_runner_;
+  XmppStreamParser stream_parser_{this};
+  bool read_pending_{false};
+  bool write_pending_{false};
 
   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 bbbc1be..7dbfd7a 100644
--- a/buffet/notification/xmpp_channel_unittest.cc
+++ b/buffet/notification/xmpp_channel_unittest.cc
@@ -39,10 +39,6 @@
     "<failure xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"><not-authorized/>"
     "</failure></stream:stream>";
 constexpr char kRestartStreamResponse[] =
-    "<stream:stream from=\"clouddevices.gserviceaccount.com\" "
-    "id=\"BE7D34E0B7589E2A\" version=\"1.0\" "
-    "xmlns:stream=\"http://etherx.jabber.org/streams\" "
-    "xmlns=\"jabber:client\">"
     "<stream:features><bind xmlns=\"urn:ietf:params:xml:ns:xmpp-bind\"/>"
     "<session xmlns=\"urn:ietf:params:xml:ns:xmpp-session\"/>"
     "</stream:features>";
@@ -131,15 +127,23 @@
     clock_.SetNow(base::Time::Now());
   }
 
-  void StartWithState(XmppChannel::XmppState state) {
+  void StartStream() {
     xmpp_client_->fake_stream_.ExpectWritePacketString({}, kStartStreamMessage);
+    xmpp_client_->fake_stream_.AddReadPacketString({}, kStartStreamResponse);
+    xmpp_client_->fake_stream_.ExpectWritePacketString({},
+                                                       kAuthenticationMessage);
     xmpp_client_->Start(nullptr);
-    RunTasks(1);
+    RunTasks(4);
+  }
+
+  void StartWithState(XmppChannel::XmppState state) {
+    StartStream();
     xmpp_client_->set_state(state);
   }
 
   void RunTasks(size_t count) {
     while (count > 0) {
+      ASSERT_FALSE(message_queue_.empty());
       base::Closure task = message_queue_.front();
       message_queue_.pop();
       task.Run();
@@ -161,11 +165,7 @@
 }
 
 TEST_F(XmppChannelTest, HandleStartedResponse) {
-  StartWithState(XmppChannel::XmppState::kStarted);
-  xmpp_client_->fake_stream_.AddReadPacketString({}, kStartStreamResponse);
-  xmpp_client_->fake_stream_.ExpectWritePacketString({},
-                                                     kAuthenticationMessage);
-  RunTasks(2);
+  StartStream();
   EXPECT_EQ(XmppChannel::XmppState::kAuthenticationStarted,
             xmpp_client_->state());
 }
@@ -175,7 +175,7 @@
   xmpp_client_->fake_stream_.AddReadPacketString(
       {}, kAuthenticationSucceededResponse);
   xmpp_client_->fake_stream_.ExpectWritePacketString({}, kStartStreamMessage);
-  RunTasks(2);
+  RunTasks(4);
   EXPECT_EQ(XmppChannel::XmppState::kStreamRestartedPostAuthentication,
             xmpp_client_->state());
 }
@@ -184,17 +184,16 @@
   StartWithState(XmppChannel::XmppState::kAuthenticationStarted);
   xmpp_client_->fake_stream_.AddReadPacketString(
       {}, kAuthenticationFailedResponse);
-  RunTasks(1);
+  RunTasks(3);
   EXPECT_EQ(XmppChannel::XmppState::kAuthenticationFailed,
             xmpp_client_->state());
-  EXPECT_TRUE(message_queue_.empty());
 }
 
 TEST_F(XmppChannelTest, HandleStreamRestartedResponse) {
   StartWithState(XmppChannel::XmppState::kStreamRestartedPostAuthentication);
   xmpp_client_->fake_stream_.AddReadPacketString({}, kRestartStreamResponse);
   xmpp_client_->fake_stream_.ExpectWritePacketString({}, kBindMessage);
-  RunTasks(2);
+  RunTasks(4);
   EXPECT_EQ(XmppChannel::XmppState::kBindSent,
             xmpp_client_->state());
 }
@@ -203,7 +202,7 @@
   StartWithState(XmppChannel::XmppState::kBindSent);
   xmpp_client_->fake_stream_.AddReadPacketString({}, kBindResponse);
   xmpp_client_->fake_stream_.ExpectWritePacketString({}, kSessionMessage);
-  RunTasks(2);
+  RunTasks(4);
   EXPECT_EQ(XmppChannel::XmppState::kSessionStarted,
             xmpp_client_->state());
 }
@@ -212,7 +211,7 @@
   StartWithState(XmppChannel::XmppState::kSessionStarted);
   xmpp_client_->fake_stream_.AddReadPacketString({}, kSessionResponse);
   xmpp_client_->fake_stream_.ExpectWritePacketString({}, kSubscribeMessage);
-  RunTasks(2);
+  RunTasks(4);
   EXPECT_EQ(XmppChannel::XmppState::kSubscribeStarted,
             xmpp_client_->state());
 }
@@ -220,7 +219,7 @@
 TEST_F(XmppChannelTest, HandleSubscribeResponse) {
   StartWithState(XmppChannel::XmppState::kSubscribeStarted);
   xmpp_client_->fake_stream_.AddReadPacketString({}, kSubscribedResponse);
-  RunTasks(1);
+  RunTasks(3);
   EXPECT_EQ(XmppChannel::XmppState::kSubscribed,
             xmpp_client_->state());
 }
diff --git a/buffet/notification/xmpp_stream_parser.cc b/buffet/notification/xmpp_stream_parser.cc
new file mode 100644
index 0000000..02329d0
--- /dev/null
+++ b/buffet/notification/xmpp_stream_parser.cc
@@ -0,0 +1,98 @@
+// 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_stream_parser.h"
+
+#include "buffet/notification/xml_node.h"
+
+namespace buffet {
+
+XmppStreamParser::XmppStreamParser(Delegate* delegate) : delegate_{delegate} {
+  parser_ = XML_ParserCreate(nullptr);
+  XML_SetUserData(parser_, this);
+  XML_SetElementHandler(parser_,
+                        &XmppStreamParser::HandleElementStart,
+                        &XmppStreamParser::HandleElementEnd);
+  XML_SetCharacterDataHandler(parser_, &XmppStreamParser::HandleCharData);
+}
+
+XmppStreamParser::~XmppStreamParser() {
+  XML_ParserFree(parser_);
+}
+
+void XmppStreamParser::ParseData(const std::string& data) {
+  XML_Parse(parser_, data.data(), data.size(), 0);
+}
+
+void XmppStreamParser::Reset() {
+  std::stack<std::unique_ptr<XmlNode>>{}.swap(node_stack_);
+  started_ = false;
+}
+
+void XmppStreamParser::HandleElementStart(void* user_data,
+                                          const XML_Char* element,
+                                          const XML_Char** attr) {
+  auto self = static_cast<XmppStreamParser*>(user_data);
+  std::map<std::string, std::string> attributes;
+  if (attr != nullptr) {
+    for (size_t n = 0; attr[n] != nullptr && attr[n + 1] != nullptr; n += 2) {
+      attributes.emplace(attr[n], attr[n + 1]);
+    }
+  }
+  self->OnOpenElement(element, std::move(attributes));
+}
+
+void XmppStreamParser::HandleElementEnd(void* user_data,
+                                        const XML_Char* element) {
+  auto self = static_cast<XmppStreamParser*>(user_data);
+  self->OnCloseElement(element);
+}
+
+void XmppStreamParser::HandleCharData(void* user_data,
+                                      const char* content,
+                                      int length) {
+  auto self = static_cast<XmppStreamParser*>(user_data);
+  self->OnCharData(std::string{content, static_cast<size_t>(length)});
+}
+
+void XmppStreamParser::OnOpenElement(
+    const std::string& node_name,
+    std::map<std::string, std::string> attributes) {
+  if (!started_) {
+    started_ = true;
+    if (delegate_)
+      delegate_->OnStreamStart(node_name, std::move(attributes));
+    return;
+  }
+  node_stack_.emplace(new XmlNode{node_name, std::move(attributes)});
+}
+
+void XmppStreamParser::OnCloseElement(const std::string& node_name) {
+  if (node_stack_.empty()) {
+    if (started_) {
+      started_ = false;
+      if (delegate_)
+        delegate_->OnStreamEnd(node_name);
+    }
+    return;
+  }
+
+  auto node = std::move(node_stack_.top());
+  node_stack_.pop();
+  if (!node_stack_.empty()) {
+    XmlNode* parent = node_stack_.top().get();
+    parent->AddChild(std::move(node));
+  } else if (delegate_) {
+    delegate_->OnStanza(std::move(node));
+  }
+}
+
+void XmppStreamParser::OnCharData(const std::string& text) {
+  if (!node_stack_.empty()) {
+    XmlNode* node = node_stack_.top().get();
+    node->AppendText(text);
+  }
+}
+
+}  // namespace buffet
diff --git a/buffet/notification/xmpp_stream_parser.h b/buffet/notification/xmpp_stream_parser.h
new file mode 100644
index 0000000..6bb39c7
--- /dev/null
+++ b/buffet/notification/xmpp_stream_parser.h
@@ -0,0 +1,86 @@
+// 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_STREAM_PARSER_H_
+#define BUFFET_NOTIFICATION_XMPP_STREAM_PARSER_H_
+
+#include <expat.h>
+
+#include <map>
+#include <memory>
+#include <stack>
+#include <string>
+
+#include <base/macros.h>
+
+namespace buffet {
+
+class XmlNode;
+
+// A simple XML stream parser. As the XML data is being read from a data source
+// (for example, a socket), XmppStreamParser::ParseData() should be called.
+// This method parses the provided XML data chunk and if it finds complete
+// XML elements, it will call internal OnOpenElement(), OnCloseElement() and
+// OnCharData() member functions. These will track the element nesting level.
+// When a top-level element starts, the parser will call Delegate::OnStreamStart
+// method. Once this happens, every complete XML element (including its children
+// if they are present) will trigger Delegate::OnStanze() callback.
+// Finally, when top-level element is closed, Delegate::OnStreamEnd() is called.
+// This class is specifically tailored to XMPP streams which look like this:
+// B:  <stream:stream to='example.com' xmlns='jabber:client' version='1.0'>
+// S:    <presence><show/></presence>
+// S:    <message to='foo'><body/></message>
+// S:    <iq to='bar'><query/></iq>
+// S:    ...
+// E:  </stream:stream>
+// Here, "B:" will trigger OnStreamStart(), "S:" will result in OnStanza() and
+// "E:" will result in OnStreamEnd().
+class XmppStreamParser final {
+ public:
+  // Delegate interface that interested parties implement to receive
+  // notifications of stream opening/closing and on new stanzas arriving.
+  class Delegate {
+   public:
+    virtual void OnStreamStart(
+        const std::string& node_name,
+        std::map<std::string, std::string> attributes) = 0;
+    virtual void OnStreamEnd(const std::string& node_name) = 0;
+    virtual void OnStanza(std::unique_ptr<XmlNode> stanza) = 0;
+  };
+
+  explicit XmppStreamParser(Delegate* delegate);
+  ~XmppStreamParser();
+
+  // Parses additional XML data received from an input stream.
+  void ParseData(const std::string& data);
+
+  // Resets the parser to expect the top-level stream node again.
+  void Reset();
+
+ private:
+  // Raw expat callbacks.
+  static void HandleElementStart(void* user_data,
+                                 const XML_Char* element,
+                                 const XML_Char** attr);
+  static void HandleElementEnd(void* user_data, const XML_Char* element);
+  static void HandleCharData(void* user_data, const char* content, int length);
+
+  // Reinterpreted callbacks from expat with some data pre-processed.
+  void OnOpenElement(const std::string& node_name,
+                     std::map<std::string, std::string> attributes);
+  void OnCloseElement(const std::string& node_name);
+  void OnCharData(const std::string& text);
+
+  Delegate* delegate_;
+  XML_Parser parser_{nullptr};
+  bool started_{false};
+  std::stack<std::unique_ptr<XmlNode>> node_stack_;
+
+  DISALLOW_COPY_AND_ASSIGN(XmppStreamParser);
+};
+
+}  // namespace buffet
+
+#endif  // BUFFET_NOTIFICATION_XMPP_STREAM_PARSER_H_
+
diff --git a/buffet/notification/xmpp_stream_parser_unittest.cc b/buffet/notification/xmpp_stream_parser_unittest.cc
new file mode 100644
index 0000000..578e6ad
--- /dev/null
+++ b/buffet/notification/xmpp_stream_parser_unittest.cc
@@ -0,0 +1,205 @@
+// 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_stream_parser.h"
+
+#include <memory>
+#include <vector>
+#include <gtest/gtest.h>
+
+#include "buffet/notification/xml_node.h"
+
+namespace buffet {
+namespace {
+// Use some real-world XMPP stream snippet to make sure all the expected
+// elements are parsed properly.
+const char kXmppStreamData[] =
+    "<stream:stream from=\"clouddevices.gserviceaccount.com\" id=\"76EEB8FDB449"
+    "5558\" version=\"1.0\" xmlns:stream=\"http://etherx.jabber.org/streams\" x"
+    "mlns=\"jabber:client\">"
+    "<stream:features><starttls xmlns=\"urn:ietf:params:xml:ns:xmpp-tls\"><requ"
+    "ired/></starttls><mechanisms xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"><m"
+    "echanism>X-OAUTH2</mechanism><mechanism>X-GOOGLE-TOKEN</mechanism></mechan"
+    "isms></stream:features>"
+    "<message from=\"cloud-devices@clouddevices.google.com/srvenc-xgbCfg9hX6tCp"
+    "xoMYsExqg==\" to=\"4783f652b387449fc52a76f9a16e616f@clouddevices.gservicea"
+    "ccount.com/5A85ED9C\"><push:push channel=\"cloud_devices\" xmlns:push=\"go"
+    "ogle:push\"><push:recipient to=\"4783f652b387449fc52a76f9a16e616f@clouddev"
+    "ices.gserviceaccount.com\"></push:recipient><push:data>eyJraW5kIjoiY2xvdWR"
+    "kZXZpY2VzI25vdGlmaWNhdGlvbiIsInR5cGUiOiJDT01NQU5EX0NSRUFURUQiLCJjb21tYW5kS"
+    "WQiOiIwNWE3MTA5MC1hZWE4LWMzNzQtOTYwNS0xZTRhY2JhNDRmM2Y4OTAzZmM3Yy01NjExLWI"
+    "5ODAtOTkyMy0yNjc2YjYwYzkxMGMiLCJkZXZpY2VJZCI6IjA1YTcxMDkwLWFlYTgtYzM3NC05N"
+    "jA1LTFlNGFjYmE0NGYzZiIsImNvbW1hbmQiOnsia2luZCI6ImNsb3VkZGV2aWNlcyNjb21tYW5"
+    "kIiwiaWQiOiIwNWE3MTA5MC1hZWE4LWMzNzQtOTYwNS0xZTRhY2JhNDRmM2Y4OTAzZmM3Yy01N"
+    "jExLWI5ODAtOTkyMy0yNjc2YjYwYzkxMGMiLCJkZXZpY2VJZCI6IjA1YTcxMDkwLWFlYTgtYzM"
+    "3NC05NjA1LTFlNGFjYmE0NGYzZiIsIm5hbWUiOiJiYXNlLl9qdW1wIiwic3RhdGUiOiJxdWV1Z"
+    "WQiLCJlcnJvciI6eyJhcmd1bWVudHMiOltdfSwiY3JlYXRpb25UaW1lTXMiOiIxNDMxNTY0NDY"
+    "4MjI3IiwiZXhwaXJhdGlvblRpbWVNcyI6IjE0MzE1NjgwNjgyMjciLCJleHBpcmF0aW9uVGltZ"
+    "W91dE1zIjoiMzYwMDAwMCJ9fQ==</push:data></push:push></message>";
+
+}  // anonymous namespace
+
+class XmppStreamParserTest : public testing::Test,
+                             public XmppStreamParser::Delegate {
+ public:
+  void SetUp() override {
+    parser_.reset(new XmppStreamParser{this});
+  }
+
+  void OnStreamStart(const std::string& node_name,
+                     std::map<std::string, std::string> attributes) override {
+    EXPECT_FALSE(stream_started_);
+    stream_started_ = true;
+    stream_start_node_name_ = node_name;
+    stream_start_node_attributes_ = std::move(attributes);
+  }
+
+  void OnStreamEnd(const std::string& node_name) override {
+    EXPECT_TRUE(stream_started_);
+    EXPECT_EQ(stream_start_node_name_, node_name);
+    stream_started_ = false;
+  }
+
+  void OnStanza(std::unique_ptr<XmlNode> stanza) override {
+    stanzas_.push_back(std::move(stanza));
+  }
+
+  void Reset() {
+    parser_.reset(new XmppStreamParser{this});
+    stream_started_ = false;
+    stream_start_node_name_.clear();
+    stream_start_node_attributes_.clear();
+    stanzas_.clear();
+  }
+
+  std::unique_ptr<XmppStreamParser> parser_;
+  bool stream_started_{false};
+  std::string stream_start_node_name_;
+  std::map<std::string, std::string> stream_start_node_attributes_;
+  std::vector<std::unique_ptr<XmlNode>> stanzas_;
+};
+
+TEST_F(XmppStreamParserTest, InitialState) {
+  EXPECT_FALSE(stream_started_);
+  EXPECT_TRUE(stream_start_node_name_.empty());
+  EXPECT_TRUE(stream_start_node_attributes_.empty());
+  EXPECT_TRUE(stanzas_.empty());
+}
+
+TEST_F(XmppStreamParserTest, FullStartElement) {
+  parser_->ParseData("<foo bar=\"baz\" quux=\"1\">");
+  EXPECT_TRUE(stream_started_);
+  EXPECT_EQ("foo", stream_start_node_name_);
+  const std::map<std::string, std::string> expected_attrs{
+      {"bar", "baz"},
+      {"quux", "1"}
+  };
+  EXPECT_EQ(expected_attrs, stream_start_node_attributes_);
+}
+
+TEST_F(XmppStreamParserTest, PartialStartElement) {
+  parser_->ParseData("<foo bar=\"baz");
+  EXPECT_FALSE(stream_started_);
+  EXPECT_TRUE(stream_start_node_name_.empty());
+  EXPECT_TRUE(stream_start_node_attributes_.empty());
+  EXPECT_TRUE(stanzas_.empty());
+  parser_->ParseData("\" quux");
+  EXPECT_FALSE(stream_started_);
+  parser_->ParseData("=\"1\">");
+  EXPECT_TRUE(stream_started_);
+  EXPECT_EQ("foo", stream_start_node_name_);
+  const std::map<std::string, std::string> expected_attrs{
+      {"bar", "baz"},
+      {"quux", "1"}
+  };
+  EXPECT_EQ(expected_attrs, stream_start_node_attributes_);
+}
+
+TEST_F(XmppStreamParserTest, VariableLengthPackets) {
+  std::string value;
+  const std::string xml_data = kXmppStreamData;
+  const std::map<std::string, std::string> expected_stream_attrs{
+      {"from", "clouddevices.gserviceaccount.com"},
+      {"id", "76EEB8FDB4495558"},
+      {"version", "1.0"},
+      {"xmlns:stream", "http://etherx.jabber.org/streams"},
+      {"xmlns", "jabber:client"}
+  };
+  // Try splitting the data into pieces from 1 character in size to the whole
+  // data block and verify that we still can parse the whole message correctly.
+  // Here |step| is the size of each individual data chunk.
+  for (size_t step = 1; step <= xml_data.size(); step++) {
+    // Feed each individual chunk to the parser and hope it can piece everything
+    // together correctly.
+    for (size_t pos = 0; pos < xml_data.size(); pos += step) {
+      parser_->ParseData(xml_data.substr(pos, step));
+    }
+    EXPECT_TRUE(stream_started_);
+    EXPECT_EQ("stream:stream", stream_start_node_name_);
+    EXPECT_EQ(expected_stream_attrs, stream_start_node_attributes_);
+    EXPECT_EQ(2u, stanzas_.size());
+
+    const XmlNode* stanza1 = stanzas_[0].get();
+    EXPECT_EQ("stream:features", stanza1->name());
+    ASSERT_EQ(2u, stanza1->children().size());
+    const XmlNode* child1 = stanza1->children()[0].get();
+    EXPECT_EQ("starttls", child1->name());
+    ASSERT_EQ(1u, child1->children().size());
+    EXPECT_EQ("required", child1->children()[0]->name());
+    const XmlNode* child2 = stanza1->children()[1].get();
+    EXPECT_EQ("mechanisms", child2->name());
+    ASSERT_EQ(2u, child2->children().size());
+    EXPECT_EQ("mechanism", child2->children()[0]->name());
+    EXPECT_EQ("X-OAUTH2", child2->children()[0]->text());
+    EXPECT_EQ("mechanism", child2->children()[1]->name());
+    EXPECT_EQ("X-GOOGLE-TOKEN", child2->children()[1]->text());
+
+    const XmlNode* stanza2 = stanzas_[1].get();
+    EXPECT_EQ("message", stanza2->name());
+    ASSERT_EQ(2u, stanza2->attributes().size());
+    EXPECT_TRUE(stanza2->GetAttribute("from", &value));
+    EXPECT_EQ("cloud-devices@clouddevices.google.com/"
+              "srvenc-xgbCfg9hX6tCpxoMYsExqg==", value);
+    EXPECT_TRUE(stanza2->GetAttribute("to", &value));
+    EXPECT_EQ("4783f652b387449fc52a76f9a16e616f@clouddevices.gserviceaccount."
+              "com/5A85ED9C", value);
+    ASSERT_EQ(1u, stanza2->children().size());
+
+    const XmlNode* child = stanza2->children().back().get();
+    EXPECT_EQ("push:push", child->name());
+    ASSERT_EQ(2u, child->attributes().size());
+    EXPECT_TRUE(child->GetAttribute("channel", &value));
+    EXPECT_EQ("cloud_devices", value);
+    EXPECT_TRUE(child->GetAttribute("xmlns:push", &value));
+    EXPECT_EQ("google:push", value);
+    ASSERT_EQ(2u, child->children().size());
+
+    child1 = child->children()[0].get();
+    EXPECT_EQ("push:recipient", child1->name());
+    ASSERT_EQ(1u, child1->attributes().size());
+    EXPECT_TRUE(child1->GetAttribute("to", &value));
+    EXPECT_EQ("4783f652b387449fc52a76f9a16e616f@clouddevices.gserviceaccount."
+              "com", value);
+    EXPECT_TRUE(child1->children().empty());
+
+    child2 = child->children()[1].get();
+    EXPECT_EQ("push:data", child2->name());
+    EXPECT_TRUE(child2->attributes().empty());
+    EXPECT_TRUE(child2->children().empty());
+    const std::string expected_data = "eyJraW5kIjoiY2xvdWRkZXZpY2VzI25vdGlmaWNh"
+        "dGlvbiIsInR5cGUiOiJDT01NQU5EX0NSRUFURUQiLCJjb21tYW5kSWQiOiIwNWE3MTA5MC"
+        "1hZWE4LWMzNzQtOTYwNS0xZTRhY2JhNDRmM2Y4OTAzZmM3Yy01NjExLWI5ODAtOTkyMy0y"
+        "Njc2YjYwYzkxMGMiLCJkZXZpY2VJZCI6IjA1YTcxMDkwLWFlYTgtYzM3NC05NjA1LTFlNG"
+        "FjYmE0NGYzZiIsImNvbW1hbmQiOnsia2luZCI6ImNsb3VkZGV2aWNlcyNjb21tYW5kIiwi"
+        "aWQiOiIwNWE3MTA5MC1hZWE4LWMzNzQtOTYwNS0xZTRhY2JhNDRmM2Y4OTAzZmM3Yy01Nj"
+        "ExLWI5ODAtOTkyMy0yNjc2YjYwYzkxMGMiLCJkZXZpY2VJZCI6IjA1YTcxMDkwLWFlYTgt"
+        "YzM3NC05NjA1LTFlNGFjYmE0NGYzZiIsIm5hbWUiOiJiYXNlLl9qdW1wIiwic3RhdGUiOi"
+        "JxdWV1ZWQiLCJlcnJvciI6eyJhcmd1bWVudHMiOltdfSwiY3JlYXRpb25UaW1lTXMiOiIx"
+        "NDMxNTY0NDY4MjI3IiwiZXhwaXJhdGlvblRpbWVNcyI6IjE0MzE1NjgwNjgyMjciLCJleH"
+        "BpcmF0aW9uVGltZW91dE1zIjoiMzYwMDAwMCJ9fQ==";
+    EXPECT_EQ(expected_data, child2->text());
+  }
+}
+
+}  // namespace buffet