diff --git a/buffet/buffet.gyp b/buffet/buffet.gyp
index 83cebba..7a466d7 100644
--- a/buffet/buffet.gyp
+++ b/buffet/buffet.gyp
@@ -39,6 +39,7 @@
         'dbus_bindings/org.chromium.Buffet.Manager.xml',
         'dbus_constants.cc',
         'manager.cc',
+        'notification/xmpp_channel.cc',
         'registration_status.cc',
         'storage_impls.cc',
         'states/error_codes.cc',
@@ -46,8 +47,6 @@
         'states/state_manager.cc',
         'states/state_package.cc',
         'utils.cc',
-        'xmpp/xmpp_client.cc',
-        'xmpp/xmpp_connection.cc',
       ],
       'includes': ['../common-mk/generate-dbus-adaptors.gypi'],
       'actions': [
@@ -120,10 +119,10 @@
             'commands/schema_utils_unittest.cc',
             'commands/unittest_utils.cc',
             'device_registration_info_unittest.cc',
+            'notification/xmpp_channel_unittest.cc',
             'states/state_change_queue_unittest.cc',
             'states/state_manager_unittest.cc',
             'states/state_package_unittest.cc',
-            'xmpp/xmpp_client_unittest.cc',
           ],
         },
       ],
diff --git a/buffet/device_registration_info.cc b/buffet/device_registration_info.cc
index 2fd2bfb..e9f806b 100644
--- a/buffet/device_registration_info.cc
+++ b/buffet/device_registration_info.cc
@@ -27,6 +27,7 @@
 #include "buffet/commands/command_manager.h"
 #include "buffet/commands/schema_constants.h"
 #include "buffet/device_registration_storage_keys.h"
+#include "buffet/notification/xmpp_channel.h"
 #include "buffet/org.chromium.Buffet.Manager.h"
 #include "buffet/states/state_manager.h"
 #include "buffet/utils.h"
@@ -133,14 +134,14 @@
     std::unique_ptr<BuffetConfig> config,
     const std::shared_ptr<chromeos::http::Transport>& transport,
     const std::shared_ptr<StorageInterface>& state_store,
-    bool xmpp_enabled,
+    bool notifications_enabled,
     org::chromium::Buffet::ManagerAdaptor* manager)
     : transport_{transport},
       storage_{state_store},
       command_manager_{command_manager},
       state_manager_{state_manager},
       config_{std::move(config)},
-      xmpp_enabled_{xmpp_enabled},
+      notifications_enabled_{notifications_enabled},
       manager_{manager} {
   OnConfigChanged();
   command_manager_->AddOnCommandDefChanged(
@@ -364,14 +365,14 @@
   LOG(INFO) << "Access token is refreshed for additional " << expires_in
             << " seconds.";
 
-  StartXmpp();
+  StartNotificationChannel();
 
   return true;
 }
 
-void DeviceRegistrationInfo::StartXmpp() {
-  if (!xmpp_enabled_) {
-    LOG(WARNING) << "XMPP support disabled by flag.";
+void DeviceRegistrationInfo::StartNotificationChannel() {
+  if (!notifications_enabled_) {
+    LOG(WARNING) << "Notification support disabled by flag.";
     return;
   }
   // If no MessageLoop assume we're in unittests.
@@ -380,37 +381,13 @@
     return;
   }
 
-  if (!fd_watcher_.StopWatchingFileDescriptor()) {
-    LOG(WARNING) << "Failed to stop the previous watcher";
-    return;
-  }
-
-  std::unique_ptr<XmppConnection> connection(new XmppConnection());
-  if (!connection->Initialize()) {
-    LOG(WARNING) << "Failed to connect to XMPP server";
-    return;
-  }
-  xmpp_client_.reset(new XmppClient(device_robot_account_, access_token_,
-                                    std::move(connection)));
-  if (!base::MessageLoopForIO::current()->WatchFileDescriptor(
-          xmpp_client_->GetFileDescriptor(), true /* persistent */,
-          base::MessageLoopForIO::WATCH_READ, &fd_watcher_, this)) {
-    LOG(WARNING) << "Failed to watch XMPP file descriptor";
-    return;
-  }
-
-  xmpp_client_->StartStream();
-}
-
-void DeviceRegistrationInfo::OnFileCanReadWithoutBlocking(int fd) {
-  if (!xmpp_client_ || xmpp_client_->GetFileDescriptor() != fd)
-    return;
-  if (!xmpp_client_->Read()) {
-    // Authentication failed or the socket was closed.
-    if (!fd_watcher_.StopWatchingFileDescriptor())
-      LOG(WARNING) << "Failed to stop the watcher";
-    return;
-  }
+  // TODO(avakulenko): Move this into a notification channel factory and out of
+  // this class completely. Also to be added the secondary (poll) notification
+  // channel.
+  primary_notification_channel_.reset(
+      new XmppChannel{device_robot_account_, access_token_,
+                      base::MessageLoop::current()->task_runner()});
+  primary_notification_channel_->Start(this);
 }
 
 std::unique_ptr<base::DictionaryValue>
@@ -437,7 +414,18 @@
     resource->SetString("location", config_->location());
   resource->SetString("modelManifestId", config_->model_id());
   resource->SetString("deviceKind", config_->device_kind());
-  resource->SetString("channel.supportedType", "xmpp");
+  std::unique_ptr<base::DictionaryValue> channel{new base::DictionaryValue};
+  if (primary_notification_channel_) {
+    channel->SetString("supportedType",
+                       primary_notification_channel_->GetName());
+    primary_notification_channel_->AddChannelParameters(channel.get());
+  } else {
+    // TODO(avakulenko): Currently GCD server doesn't support changing supported
+    // channel, so here we cannot use "pull" as supported channel type until
+    // this is fixed. See b/20895223
+    channel->SetString("supportedType", "xmpp");
+  }
+  resource->Set("channel", channel.release());
   resource->Set("commandDefs", commands.release());
   resource->Set("state", state.release());
 
@@ -587,7 +575,7 @@
                              base::TimeDelta::FromSeconds(expires_in);
 
   Save();
-  StartXmpp();
+  StartNotificationChannel();
 
   // We're going to respond with our success immediately and we'll StartDevice
   // shortly after.
@@ -1067,4 +1055,19 @@
                        base::Bind(&IgnoreCloudError));
 }
 
+void DeviceRegistrationInfo::OnConnected(const std::string& channel_name) {
+  LOG(INFO) << "Notification channel successfully established over "
+            << channel_name;
+  // TODO(avakulenko): Notify GCD server of changed supported channel.
+}
+
+void DeviceRegistrationInfo::OnDisconnected() {
+  LOG(INFO) << "Notification channel disconnected";
+  // TODO(avakulenko): Notify GCD server of changed supported channel.
+}
+
+void DeviceRegistrationInfo::OnPermanentFailure() {
+  LOG(ERROR) << "Failed to establish notification channel.";
+}
+
 }  // namespace buffet
diff --git a/buffet/device_registration_info.h b/buffet/device_registration_info.h
index 40f4f94..ab72c87 100644
--- a/buffet/device_registration_info.h
+++ b/buffet/device_registration_info.h
@@ -12,7 +12,6 @@
 
 #include <base/macros.h>
 #include <base/memory/weak_ptr.h>
-#include <base/message_loop/message_loop.h>
 #include <base/time/time.h>
 #include <base/timer/timer.h>
 #include <chromeos/data_encoding.h>
@@ -21,9 +20,10 @@
 
 #include "buffet/buffet_config.h"
 #include "buffet/commands/command_manager.h"
+#include "buffet/notification/notification_channel.h"
+#include "buffet/notification/notification_delegate.h"
 #include "buffet/registration_status.h"
 #include "buffet/storage_interface.h"
-#include "buffet/xmpp/xmpp_client.h"
 
 namespace org {
 namespace chromium {
@@ -50,7 +50,7 @@
 extern const char kErrorDomainGCDServer[];
 
 // The DeviceRegistrationInfo class represents device registration information.
-class DeviceRegistrationInfo : public base::MessageLoopForIO::Watcher {
+class DeviceRegistrationInfo : public NotificationDelegate {
  public:
   // This is a helper class for unit testing.
   class TestHelper;
@@ -61,16 +61,10 @@
       std::unique_ptr<BuffetConfig> config,
       const std::shared_ptr<chromeos::http::Transport>& transport,
       const std::shared_ptr<StorageInterface>& state_store,
-      bool xmpp_enabled,
+      bool notifications_enabled,
       org::chromium::Buffet::ManagerAdaptor* manager);
 
-  ~DeviceRegistrationInfo() override;
-
-  void OnFileCanReadWithoutBlocking(int fd) override;
-
-  void OnFileCanWriteWithoutBlocking(int fd) override {
-    LOG(FATAL) << "No write watcher is configured";
-  }
+  virtual ~DeviceRegistrationInfo();
 
   // Returns our current best known registration status.
   RegistrationStatus GetRegistrationStatus() const {
@@ -183,9 +177,9 @@
   std::unique_ptr<base::DictionaryValue> ParseOAuthResponse(
       chromeos::http::Response* response, chromeos::ErrorPtr* error);
 
-  // This attempts to open the XMPP channel. The XMPP channel needs to be
+  // This attempts to open a notification channel. The channel needs to be
   // restarted anytime the access_token is refreshed.
-  void StartXmpp();
+  void StartNotificationChannel();
 
   using CloudRequestCallback =
       base::Callback<void(const base::DictionaryValue&)>;
@@ -243,6 +237,11 @@
   // Callback called when command definitions are changed to re-publish new CDD.
   void OnCommandDefsChanged();
 
+  // Overrides from NotificationDelegate
+  void OnConnected(const std::string& channel_name) override;
+  void OnDisconnected() override;
+  void OnPermanentFailure() override;
+
   // Data that is cached here, persisted in the state store.
   std::string refresh_token_;
   std::string device_id_;
@@ -263,9 +262,8 @@
 
   std::unique_ptr<BuffetConfig> config_;
 
-  const bool xmpp_enabled_;
-  std::unique_ptr<XmppClient> xmpp_client_;
-  base::MessageLoopForIO::FileDescriptorWatcher fd_watcher_;
+  const bool notifications_enabled_;
+  std::unique_ptr<NotificationChannel> primary_notification_channel_;
 
   // Tracks our current registration status.
   RegistrationStatus registration_status_{RegistrationStatus::kUnconfigured};
diff --git a/buffet/notification/notification_channel.h b/buffet/notification/notification_channel.h
new file mode 100644
index 0000000..ab33dd2
--- /dev/null
+++ b/buffet/notification/notification_channel.h
@@ -0,0 +1,31 @@
+// 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_NOTIFICATION_CHANNEL_H_
+#define BUFFET_NOTIFICATION_NOTIFICATION_CHANNEL_H_
+
+#include <string>
+
+namespace base {
+class DictionaryValue;
+}  // namespace base
+
+namespace buffet {
+
+class NotificationDelegate;
+
+class NotificationChannel {
+ public:
+  virtual ~NotificationChannel() = default;
+
+  virtual std::string GetName() const = 0;
+  virtual void AddChannelParameters(base::DictionaryValue* channel_json) = 0;
+
+  virtual void Start(NotificationDelegate* delegate) = 0;
+  virtual void Stop() = 0;
+};
+
+}  // namespace buffet
+
+#endif  // BUFFET_NOTIFICATION_NOTIFICATION_CHANNEL_H_
diff --git a/buffet/notification/notification_delegate.h b/buffet/notification/notification_delegate.h
new file mode 100644
index 0000000..06b30d4
--- /dev/null
+++ b/buffet/notification/notification_delegate.h
@@ -0,0 +1,24 @@
+// 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_NOTIFICATION_DELEGATE_H_
+#define BUFFET_NOTIFICATION_NOTIFICATION_DELEGATE_H_
+
+#include <string>
+
+namespace buffet {
+
+class NotificationDelegate {
+ public:
+  virtual void OnConnected(const std::string& channel_name) = 0;
+  virtual void OnDisconnected() = 0;
+  virtual void OnPermanentFailure() = 0;
+
+ protected:
+  virtual ~NotificationDelegate() = default;
+};
+
+}  // namespace buffet
+
+#endif  // BUFFET_NOTIFICATION_NOTIFICATION_DELEGATE_H_
diff --git a/buffet/notification/xmpp_channel.cc b/buffet/notification/xmpp_channel.cc
new file mode 100644
index 0000000..5590333
--- /dev/null
+++ b/buffet/notification/xmpp_channel.cc
@@ -0,0 +1,277 @@
+// 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_channel.h"
+
+#include <string>
+
+#include <base/bind.h>
+#include <chromeos/backoff_entry.h>
+#include <chromeos/data_encoding.h>
+#include <chromeos/streams/file_stream.h>
+
+#include "buffet/notification/notification_delegate.h"
+#include "buffet/utils.h"
+
+namespace buffet {
+
+namespace {
+
+std::string BuildXmppStartStreamCommand() {
+  return "<stream:stream to='clouddevices.gserviceaccount.com' "
+      "xmlns:stream='http://etherx.jabber.org/streams' "
+      "xml:lang='*' version='1.0' xmlns='jabber:client'>";
+}
+
+std::string BuildXmppAuthenticateCommand(
+    const std::string& account, const std::string& token) {
+  chromeos::Blob credentials;
+  credentials.push_back(0);
+  credentials.insert(credentials.end(), account.begin(), account.end());
+  credentials.push_back(0);
+  credentials.insert(credentials.end(), token.begin(), token.end());
+  std::string msg = "<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' "
+      "mechanism='X-OAUTH2' auth:service='oauth2' "
+      "auth:allow-non-google-login='true' "
+      "auth:client-uses-full-bind-result='true' "
+      "xmlns:auth='http://www.google.com/talk/protocol/auth'>" +
+      chromeos::data_encoding::Base64Encode(credentials) +
+      "</auth>";
+  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.
+const chromeos::BackoffEntry::Policy kDefaultBackoffPolicy = {
+  // Number of initial errors (in sequence) to ignore before applying
+  // exponential back-off rules.
+  0,
+
+  // Initial delay for exponential back-off in ms.
+  30 * 1000,  // 30 seconds.
+
+  // Factor by which the waiting time will be multiplied.
+  2,
+
+  // Fuzzing percentage. ex: 10% will spread requests randomly
+  // between 90%-100% of the calculated time.
+  0.33,  // 33%.
+
+  // Maximum amount of time we are willing to delay our request in ms.
+  10 * 60 * 1000,  // 10 minutes.
+
+  // Time to keep an entry from being discarded even when it
+  // has no significant state, -1 to never discard.
+  -1,
+
+  // Don't use initial delay unless the last request was an error.
+  false,
+};
+
+const char kDefaultXmppHost[] = "talk.google.com";
+const uint16_t kDefaultXmppPort = 5222;
+
+}  // namespace
+
+XmppChannel::XmppChannel(const std::string& account,
+                         const std::string& access_token,
+                         const scoped_refptr<base::TaskRunner>& task_runner)
+    : account_{account},
+      access_token_{access_token},
+      backoff_entry_{&kDefaultBackoffPolicy},
+      task_runner_{task_runner} {
+  read_socket_data_.resize(4096);
+}
+
+void XmppChannel::OnMessageRead(size_t size) {
+  std::string msg(read_socket_data_.data(), size);
+  bool message_sent = false;
+
+  VLOG(2) << "Received XMPP packet: " << msg;
+
+  // 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;
+      }
+      break;
+    case XmppState::kAuthenticationStarted:
+      if (std::string::npos != msg.find("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();
+        return;
+      }
+      break;
+    case XmppState::kStreamRestartedPostAuthentication:
+      if (std::string::npos != msg.find(":features") &&
+          std::string::npos != msg.find(":xmpp-session")) {
+        state_ = XmppState::kBindSent;
+        SendMessage(BuildXmppBindCommand());
+        message_sent = true;
+      }
+      break;
+    case XmppState::kBindSent:
+      if (std::string::npos != msg.find("iq") &&
+          std::string::npos != msg.find("result")) {
+        state_ = XmppState::kSessionStarted;
+        SendMessage(BuildXmppStartSessionCommand());
+        message_sent = true;
+      }
+      break;
+    case XmppState::kSessionStarted:
+      if (std::string::npos != msg.find("iq") &&
+          std::string::npos != msg.find("result")) {
+        state_ = XmppState::kSubscribeStarted;
+        SendMessage(BuildXmppSubscribeCommand(account_));
+        message_sent = true;
+      }
+      break;
+    case XmppState::kSubscribeStarted:
+      if (std::string::npos != msg.find("iq") &&
+          std::string::npos != msg.find("result")) {
+        state_ = XmppState::kSubscribed;
+        if (delegate_)
+          delegate_->OnConnected(GetName());
+      }
+      break;
+    default:
+      break;
+  }
+  if (!message_sent)
+    WaitForMessage();
+}
+
+void XmppChannel::SendMessage(const std::string& message) {
+  write_socket_data_ = message;
+  chromeos::ErrorPtr error;
+  VLOG(2) << "Sending XMPP message: " << message;
+
+  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()),
+      &error);
+
+  if (!ok)
+    OnError(error.get());
+}
+
+void XmppChannel::OnMessageSent() {
+  chromeos::ErrorPtr error;
+  if (!stream_->FlushBlocking(&error)) {
+    OnError(error.get());
+    return;
+  }
+  WaitForMessage();
+}
+
+void XmppChannel::WaitForMessage() {
+  chromeos::ErrorPtr error;
+  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()),
+      &error);
+
+  if (!ok)
+    OnError(error.get());
+}
+
+void XmppChannel::OnError(const chromeos::Error* error) {
+  Stop();
+  Start(delegate_);
+}
+
+void XmppChannel::Connect(const std::string& host, uint16_t port,
+                          const base::Closure& callback) {
+  int socket_fd = ConnectSocket(host, port);
+  if (socket_fd >= 0) {
+    raw_socket_ =
+      chromeos::FileStream::FromFileDescriptor(socket_fd, true, nullptr);
+    if (!raw_socket_) {
+      close(socket_fd);
+      socket_fd = -1;
+    }
+  }
+
+  backoff_entry_.InformOfRequest(raw_socket_ != nullptr);
+  if (raw_socket_) {
+    stream_ = raw_socket_.get();
+    callback.Run();
+  } else {
+    VLOG(1) << "Delaying connection to XMPP server " << host << " for "
+            << backoff_entry_.GetTimeUntilRelease().InMilliseconds()
+            << " milliseconds.";
+    task_runner_->PostDelayedTask(
+        FROM_HERE,
+        base::Bind(&XmppChannel::Connect, weak_ptr_factory_.GetWeakPtr(),
+                    host, port, callback),
+        backoff_entry_.GetTimeUntilRelease());
+  }
+}
+
+std::string XmppChannel::GetName() const {
+  return "xmpp";
+}
+
+void XmppChannel::AddChannelParameters(base::DictionaryValue* channel_json) {
+  // No extra parameters needed for XMPP.
+}
+
+void XmppChannel::Start(NotificationDelegate* delegate) {
+  CHECK(state_ == XmppState::kNotStarted);
+  delegate_ = delegate;
+  Connect(kDefaultXmppHost, kDefaultXmppPort,
+          base::Bind(&XmppChannel::OnConnected,
+                     weak_ptr_factory_.GetWeakPtr()));
+}
+
+void XmppChannel::Stop() {
+  if (state_ == XmppState::kSubscribed && delegate_)
+    delegate_->OnDisconnected();
+
+  weak_ptr_factory_.InvalidateWeakPtrs();
+  if (raw_socket_) {
+    raw_socket_->CloseBlocking(nullptr);
+    raw_socket_.reset();
+  }
+  stream_ = nullptr;
+  state_ = XmppState::kNotStarted;
+}
+
+void XmppChannel::OnConnected() {
+  state_ = XmppState::kStarted;
+  SendMessage(BuildXmppStartStreamCommand());
+}
+
+}  // namespace buffet
diff --git a/buffet/notification/xmpp_channel.h b/buffet/notification/xmpp_channel.h
new file mode 100644
index 0000000..1fdd917
--- /dev/null
+++ b/buffet/notification/xmpp_channel.h
@@ -0,0 +1,94 @@
+// 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_CHANNEL_H_
+#define BUFFET_NOTIFICATION_XMPP_CHANNEL_H_
+
+#include <memory>
+#include <string>
+#include <vector>
+
+#include <base/callback_forward.h>
+#include <base/macros.h>
+#include <base/memory/weak_ptr.h>
+#include <base/task_runner.h>
+#include <chromeos/backoff_entry.h>
+#include <chromeos/streams/stream.h>
+
+#include "buffet/notification/notification_channel.h"
+
+namespace buffet {
+
+class XmppChannel : public NotificationChannel {
+ 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);
+  virtual ~XmppChannel() = default;
+
+  // Overrides from NotificationChannel.
+  std::string GetName() const override;
+  void AddChannelParameters(base::DictionaryValue* channel_json) override;
+  void Start(NotificationDelegate* delegate) override;
+  void Stop() override;
+
+  // Internal states for the XMPP stream.
+  enum class XmppState {
+    kNotStarted,
+    kStarted,
+    kAuthenticationStarted,
+    kAuthenticationFailed,
+    kStreamRestartedPostAuthentication,
+    kBindSent,
+    kSessionStarted,
+    kSubscribeStarted,
+    kSubscribed,
+  };
+
+ protected:
+  virtual void Connect(const std::string& host, uint16_t port,
+                       const base::Closure& callback);
+
+  XmppState state_{XmppState::kNotStarted};
+
+  // The connection socket stream to the XMPP server.
+  chromeos::Stream* stream_{nullptr};
+
+ private:
+  void SendMessage(const std::string& message);
+  void WaitForMessage();
+
+  void OnConnected();
+  void OnMessageRead(size_t size);
+  void OnMessageSent();
+  void OnError(const chromeos::Error* error);
+
+  // Robot account name for the device.
+  std::string account_;
+
+  // OAuth access token for the account. Expires fairly frequently.
+  std::string access_token_;
+
+  chromeos::StreamPtr raw_socket_;
+
+  // Read buffer for incoming message packets.
+  std::vector<char> read_socket_data_;
+  // Write buffer for outgoing message packets.
+  std::string write_socket_data_;
+
+  chromeos::BackoffEntry backoff_entry_;
+  NotificationDelegate* delegate_{nullptr};
+  scoped_refptr<base::TaskRunner> task_runner_;
+
+  base::WeakPtrFactory<XmppChannel> weak_ptr_factory_{this};
+  DISALLOW_COPY_AND_ASSIGN(XmppChannel);
+};
+
+}  // namespace buffet
+
+#endif  // BUFFET_NOTIFICATION_XMPP_CHANNEL_H_
+
diff --git a/buffet/notification/xmpp_channel_unittest.cc b/buffet/notification/xmpp_channel_unittest.cc
new file mode 100644
index 0000000..bbbc1be
--- /dev/null
+++ b/buffet/notification/xmpp_channel_unittest.cc
@@ -0,0 +1,228 @@
+// 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_channel.h"
+
+#include <queue>
+
+#include <base/test/simple_test_clock.h>
+#include <chromeos/bind_lambda.h>
+#include <chromeos/streams/fake_stream.h>
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+namespace buffet {
+
+using ::testing::DoAll;
+using ::testing::Return;
+using ::testing::SetArgPointee;
+using ::testing::_;
+
+namespace {
+
+constexpr char kAccountName[] = "Account@Name";
+constexpr char kAccessToken[] = "AccessToken";
+
+constexpr char kStartStreamResponse[] =
+    "<stream:stream from=\"clouddevices.gserviceaccount.com\" "
+    "id=\"0CCF520913ABA04B\" version=\"1.0\" "
+    "xmlns:stream=\"http://etherx.jabber.org/streams\" "
+    "xmlns=\"jabber:client\">"
+    "<stream:features><starttls xmlns=\"urn:ietf:params:xml:ns:xmpp-tls\">"
+    "<required/></starttls><mechanisms "
+    "xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"><mechanism>X-OAUTH2</mechanism>"
+    "<mechanism>X-GOOGLE-TOKEN</mechanism></mechanisms></stream:features>";
+constexpr char kAuthenticationSucceededResponse[] =
+    "<success xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"/>";
+constexpr char kAuthenticationFailedResponse[] =
+    "<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>";
+constexpr char kBindResponse[] =
+    "<iq id=\"0\" 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\"/>";
+constexpr char kSubscribedResponse[] =
+    "<iq to=\""
+    "110cc78f78d7032cc7bf2c6e14c1fa7d@clouddevices.gserviceaccount.com/"
+    "19853128\" from=\""
+    "110cc78f78d7032cc7bf2c6e14c1fa7d@clouddevices.gserviceaccount.com\" "
+    "id=\"pushsubscribe1\" type=\"result\"/>";
+constexpr char kStartStreamMessage[] =
+    "<stream:stream to='clouddevices.gserviceaccount.com' "
+    "xmlns:stream='http://etherx.jabber.org/streams' xml:lang='*' "
+    "version='1.0' xmlns='jabber:client'>";
+constexpr char kAuthenticationMessage[] =
+    "<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='X-OAUTH2' "
+    "auth:service='oauth2' auth:allow-non-google-login='true' "
+    "auth:client-uses-full-bind-result='true' "
+    "xmlns:auth='http://www.google.com/talk/protocol/auth'>"
+    "AEFjY291bnRATmFtZQBBY2Nlc3NUb2tlbg==</auth>";
+constexpr char kBindMessage[] =
+    "<iq type='set' id='0'><bind "
+    "xmlns='urn:ietf:params:xml:ns:xmpp-bind'/></iq>";
+constexpr char kSessionMessage[] =
+    "<iq type='set' id='1'><session "
+    "xmlns='urn:ietf:params:xml:ns:xmpp-session'/></iq>";
+constexpr char kSubscribeMessage[] =
+    "<iq type='set' to='Account@Name' id='pushsubscribe1'>"
+    "<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 {
+ public:
+  MOCK_METHOD3(PostDelayedTask, 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)
+      : XmppChannel{kAccountName, kAccessToken, task_runner},
+        fake_stream_{chromeos::Stream::AccessMode::READ_WRITE, task_runner,
+                     clock} {}
+
+  XmppState state() const { return state_; }
+  void set_state(XmppState state) { state_ = state; }
+
+  void Connect(const std::string& host, uint16_t port,
+               const base::Closure& callback) override {
+    stream_ = &fake_stream_;
+    callback.Run();
+  }
+
+  chromeos::FakeStream fake_stream_;
+};
+
+class XmppChannelTest : public ::testing::Test {
+ protected:
+  void SetUp() override {
+    task_runner_ = new TestTaskRunner;
+
+    auto callback = [this](const tracked_objects::Location& from_here,
+                           const base::Closure& task,
+                           base::TimeDelta delay) -> bool {
+      clock_.Advance(delay);
+      message_queue_.push(task);
+      return true;
+    };
+
+    EXPECT_CALL(*task_runner_, PostDelayedTask(_, _, _))
+        .WillRepeatedly(testing::Invoke(callback));
+
+    xmpp_client_.reset(new FakeXmppChannel{task_runner_, &clock_});
+    clock_.SetNow(base::Time::Now());
+  }
+
+  void StartWithState(XmppChannel::XmppState state) {
+    xmpp_client_->fake_stream_.ExpectWritePacketString({}, kStartStreamMessage);
+    xmpp_client_->Start(nullptr);
+    RunTasks(1);
+    xmpp_client_->set_state(state);
+  }
+
+  void RunTasks(size_t count) {
+    while (count > 0) {
+      base::Closure task = message_queue_.front();
+      message_queue_.pop();
+      task.Run();
+      count--;
+    }
+  }
+  std::unique_ptr<FakeXmppChannel> xmpp_client_;
+  base::SimpleTestClock clock_;
+  scoped_refptr<TestTaskRunner> task_runner_;
+  std::queue<base::Closure> message_queue_;
+};
+
+TEST_F(XmppChannelTest, StartStream) {
+  EXPECT_EQ(XmppChannel::XmppState::kNotStarted, xmpp_client_->state());
+  xmpp_client_->fake_stream_.ExpectWritePacketString({}, kStartStreamMessage);
+  xmpp_client_->Start(nullptr);
+  RunTasks(1);
+  EXPECT_EQ(XmppChannel::XmppState::kStarted, xmpp_client_->state());
+}
+
+TEST_F(XmppChannelTest, HandleStartedResponse) {
+  StartWithState(XmppChannel::XmppState::kStarted);
+  xmpp_client_->fake_stream_.AddReadPacketString({}, kStartStreamResponse);
+  xmpp_client_->fake_stream_.ExpectWritePacketString({},
+                                                     kAuthenticationMessage);
+  RunTasks(2);
+  EXPECT_EQ(XmppChannel::XmppState::kAuthenticationStarted,
+            xmpp_client_->state());
+}
+
+TEST_F(XmppChannelTest, HandleAuthenticationSucceededResponse) {
+  StartWithState(XmppChannel::XmppState::kAuthenticationStarted);
+  xmpp_client_->fake_stream_.AddReadPacketString(
+      {}, kAuthenticationSucceededResponse);
+  xmpp_client_->fake_stream_.ExpectWritePacketString({}, kStartStreamMessage);
+  RunTasks(2);
+  EXPECT_EQ(XmppChannel::XmppState::kStreamRestartedPostAuthentication,
+            xmpp_client_->state());
+}
+
+TEST_F(XmppChannelTest, HandleAuthenticationFailedResponse) {
+  StartWithState(XmppChannel::XmppState::kAuthenticationStarted);
+  xmpp_client_->fake_stream_.AddReadPacketString(
+      {}, kAuthenticationFailedResponse);
+  RunTasks(1);
+  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);
+  EXPECT_EQ(XmppChannel::XmppState::kBindSent,
+            xmpp_client_->state());
+}
+
+TEST_F(XmppChannelTest, HandleBindResponse) {
+  StartWithState(XmppChannel::XmppState::kBindSent);
+  xmpp_client_->fake_stream_.AddReadPacketString({}, kBindResponse);
+  xmpp_client_->fake_stream_.ExpectWritePacketString({}, kSessionMessage);
+  RunTasks(2);
+  EXPECT_EQ(XmppChannel::XmppState::kSessionStarted,
+            xmpp_client_->state());
+}
+
+TEST_F(XmppChannelTest, HandleSessionResponse) {
+  StartWithState(XmppChannel::XmppState::kSessionStarted);
+  xmpp_client_->fake_stream_.AddReadPacketString({}, kSessionResponse);
+  xmpp_client_->fake_stream_.ExpectWritePacketString({}, kSubscribeMessage);
+  RunTasks(2);
+  EXPECT_EQ(XmppChannel::XmppState::kSubscribeStarted,
+            xmpp_client_->state());
+}
+
+TEST_F(XmppChannelTest, HandleSubscribeResponse) {
+  StartWithState(XmppChannel::XmppState::kSubscribeStarted);
+  xmpp_client_->fake_stream_.AddReadPacketString({}, kSubscribedResponse);
+  RunTasks(1);
+  EXPECT_EQ(XmppChannel::XmppState::kSubscribed,
+            xmpp_client_->state());
+}
+
+}  // namespace buffet
diff --git a/buffet/utils.cc b/buffet/utils.cc
index 5a9c154..7e50d0f 100644
--- a/buffet/utils.cc
+++ b/buffet/utils.cc
@@ -5,7 +5,11 @@
 #include "buffet/utils.h"
 
 #include <map>
+#include <netdb.h>
 #include <string>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
 
 #include <base/files/file_util.h>
 #include <base/json/json_reader.h>
@@ -75,4 +79,28 @@
   return result;
 }
 
+int ConnectSocket(const std::string& host, uint16_t port) {
+  std::string service = std::to_string(port);
+  addrinfo hints = {0, AF_UNSPEC, SOCK_STREAM};
+  addrinfo* result = nullptr;
+  if (getaddrinfo(host.c_str(), service.c_str(), &hints, &result))
+    return -1;
+
+  int socket_fd = -1;
+  for (const addrinfo* info = result; info != nullptr; info = info->ai_next) {
+    socket_fd = socket(info->ai_family, info->ai_socktype, info->ai_protocol);
+    if (socket_fd < 0)
+      continue;
+
+    if (connect(socket_fd, info->ai_addr, info->ai_addrlen) == 0)
+      break;  // Success.
+
+    close(socket_fd);
+    socket_fd = -1;
+  }
+
+  freeaddrinfo(result);
+  return socket_fd;
+}
+
 }  // namespace buffet
diff --git a/buffet/utils.h b/buffet/utils.h
index 32738d2..0325171 100644
--- a/buffet/utils.h
+++ b/buffet/utils.h
@@ -37,6 +37,11 @@
 std::unique_ptr<const base::DictionaryValue> LoadJsonDict(
     const std::string& json_string, chromeos::ErrorPtr* error);
 
+// Synchronously resolves the |host| and connects a socket to the resolved
+// address/port.
+// Returns the connected socket file descriptor or -1 if failed.
+int ConnectSocket(const std::string& host, uint16_t port);
+
 }  // namespace buffet
 
 #endif  // BUFFET_UTILS_H_
diff --git a/buffet/xmpp/xmpp_client.cc b/buffet/xmpp/xmpp_client.cc
deleted file mode 100644
index f98c6bb..0000000
--- a/buffet/xmpp/xmpp_client.cc
+++ /dev/null
@@ -1,128 +0,0 @@
-// 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 "xmpp/xmpp_client.h"
-
-#include <arpa/inet.h>
-#include <netdb.h>
-#include <netinet/in.h>
-#include <unistd.h>
-
-#include <string>
-
-#include <base/files/file_util.h>
-#include <chromeos/data_encoding.h>
-#include <chromeos/syslog_logging.h>
-
-namespace buffet {
-
-namespace {
-
-std::string BuildXmppStartStreamCommand() {
-  return "<stream:stream to='clouddevices.gserviceaccount.com' "
-      "xmlns:stream='http://etherx.jabber.org/streams' "
-      "xml:lang='*' version='1.0' xmlns='jabber:client'>";
-}
-
-std::string BuildXmppAuthenticateCommand(
-    const std::string& account, const std::string& token) {
-  chromeos::Blob credentials;
-  credentials.push_back(0);
-  credentials.insert(credentials.end(), account.begin(), account.end());
-  credentials.push_back(0);
-  credentials.insert(credentials.end(), token.begin(), token.end());
-  std::string msg = "<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' "
-      "mechanism='X-OAUTH2' auth:service='oauth2' "
-      "auth:allow-non-google-login='true' "
-      "auth:client-uses-full-bind-result='true' "
-      "xmlns:auth='http://www.google.com/talk/protocol/auth'>" +
-      chromeos::data_encoding::Base64Encode(credentials) +
-      "</auth>";
-  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>";
-}
-
-}  // namespace
-
-bool XmppClient::Read() {
-  std::string msg;
-  if (!connection_->Read(&msg) || msg.size() <= 0) {
-    LOG(ERROR) << "Failed to read from stream. The socket was probably closed";
-    return false;
-  }
-
-  // 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;
-        connection_->Write(BuildXmppAuthenticateCommand(
-            account_, access_token_));
-      }
-      break;
-    case XmppState::kAuthenticationStarted:
-      if (std::string::npos != msg.find("success")) {
-        state_ = XmppState::kStreamRestartedPostAuthentication;
-        connection_->Write(BuildXmppStartStreamCommand());
-      } else if (std::string::npos != msg.find("not-authorized")) {
-        state_ = XmppState::kAuthenticationFailed;
-        return false;
-      }
-      break;
-    case XmppState::kStreamRestartedPostAuthentication:
-      if (std::string::npos != msg.find(":features") &&
-          std::string::npos != msg.find(":xmpp-session")) {
-        state_ = XmppState::kBindSent;
-        connection_->Write(BuildXmppBindCommand());
-      }
-      break;
-    case XmppState::kBindSent:
-      if (std::string::npos != msg.find("iq") &&
-          std::string::npos != msg.find("result")) {
-        state_ = XmppState::kSessionStarted;
-        connection_->Write(BuildXmppStartSessionCommand());
-      }
-      break;
-    case XmppState::kSessionStarted:
-      if (std::string::npos != msg.find("iq") &&
-          std::string::npos != msg.find("result")) {
-        state_ = XmppState::kSubscribeStarted;
-        connection_->Write(BuildXmppSubscribeCommand(account_));
-      }
-      break;
-    case XmppState::kSubscribeStarted:
-      if (std::string::npos != msg.find("iq") &&
-          std::string::npos != msg.find("result")) {
-        state_ = XmppState::kSubscribed;
-      }
-      break;
-    default:
-      break;
-  }
-  return true;
-}
-
-void XmppClient::StartStream() {
-  state_ = XmppState::kStarted;
-  connection_->Write(BuildXmppStartStreamCommand());
-}
-
-}  // namespace buffet
diff --git a/buffet/xmpp/xmpp_client.h b/buffet/xmpp/xmpp_client.h
deleted file mode 100644
index 6d847ca..0000000
--- a/buffet/xmpp/xmpp_client.h
+++ /dev/null
@@ -1,71 +0,0 @@
-// 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_XMPP_XMPP_CLIENT_H_
-#define BUFFET_XMPP_XMPP_CLIENT_H_
-
-#include <memory>
-#include <string>
-
-#include <base/macros.h>
-
-#include "buffet/xmpp/xmpp_connection.h"
-
-namespace buffet {
-
-class XmppClient final {
- 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.
-  XmppClient(const std::string& account,
-             const std::string& access_token,
-             std::unique_ptr<XmppConnection> connection)
-      : account_{account},
-        access_token_{access_token},
-        connection_{std::move(connection)} {}
-
-  int GetFileDescriptor() const {
-    return connection_->GetFileDescriptor();
-  }
-
-  // Needs to be called when new data is available from the connection.
-  bool Read();
-
-  // Start talking to the XMPP server (authenticate, etc.)
-  void StartStream();
-
-  // Internal states for the XMPP stream.
-  enum class XmppState {
-    kNotStarted,
-    kStarted,
-    kAuthenticationStarted,
-    kAuthenticationFailed,
-    kStreamRestartedPostAuthentication,
-    kBindSent,
-    kSessionStarted,
-    kSubscribeStarted,
-    kSubscribed,
-  };
-
- private:
-  // Robot account name for the device.
-  std::string account_;
-
-  // OAuth access token for the account. Expires fairly frequently.
-  std::string access_token_;
-
-  // The connection to the XMPP server.
-  std::unique_ptr<XmppConnection> connection_;
-
-  XmppState state_{XmppState::kNotStarted};
-
-  friend class TestHelper;
-  DISALLOW_COPY_AND_ASSIGN(XmppClient);
-};
-
-}  // namespace buffet
-
-#endif  // BUFFET_XMPP_XMPP_CLIENT_H_
-
diff --git a/buffet/xmpp/xmpp_client_unittest.cc b/buffet/xmpp/xmpp_client_unittest.cc
deleted file mode 100644
index 26dfacf..0000000
--- a/buffet/xmpp/xmpp_client_unittest.cc
+++ /dev/null
@@ -1,226 +0,0 @@
-// 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/device_registration_info.h"
-
-#include <base/values.h>
-#include <chromeos/key_value_store.h>
-#include <chromeos/http/curl_api.h>
-#include <gmock/gmock.h>
-#include <gtest/gtest.h>
-
-#include "buffet/xmpp/xmpp_client.h"
-#include "buffet/xmpp/xmpp_connection.h"
-
-namespace buffet {
-
-using ::testing::DoAll;
-using ::testing::Return;
-using ::testing::SetArgPointee;
-using ::testing::_;
-
-namespace {
-
-constexpr char kAccountName[] = "Account@Name";
-constexpr char kAccessToken[] = "AccessToken";
-
-constexpr char kStartStreamResponse[] =
-    "<stream:stream from=\"clouddevices.gserviceaccount.com\" "
-    "id=\"0CCF520913ABA04B\" version=\"1.0\" "
-    "xmlns:stream=\"http://etherx.jabber.org/streams\" "
-    "xmlns=\"jabber:client\">"
-    "<stream:features><starttls xmlns=\"urn:ietf:params:xml:ns:xmpp-tls\">"
-    "<required/></starttls><mechanisms "
-    "xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"><mechanism>X-OAUTH2</mechanism>"
-    "<mechanism>X-GOOGLE-TOKEN</mechanism></mechanisms></stream:features>";
-constexpr char kAuthenticationSucceededResponse[] =
-    "<success xmlns=\"urn:ietf:params:xml:ns:xmpp-sasl\"/>";
-constexpr char kAuthenticationFailedResponse[] =
-    "<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>";
-constexpr char kBindResponse[] =
-    "<iq id=\"0\" 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\"/>";
-constexpr char kSubscribedResponse[] =
-    "<iq to=\""
-    "110cc78f78d7032cc7bf2c6e14c1fa7d@clouddevices.gserviceaccount.com/"
-    "19853128\" from=\""
-    "110cc78f78d7032cc7bf2c6e14c1fa7d@clouddevices.gserviceaccount.com\" "
-    "id=\"pushsubscribe1\" type=\"result\"/>";
-
-constexpr char kStartStreamMessage[] =
-    "<stream:stream to='clouddevices.gserviceaccount.com' "
-    "xmlns:stream='http://etherx.jabber.org/streams' xml:lang='*' "
-    "version='1.0' xmlns='jabber:client'>";
-constexpr char kAuthenticationMessage[] =
-    "<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='X-OAUTH2' "
-    "auth:service='oauth2' auth:allow-non-google-login='true' "
-    "auth:client-uses-full-bind-result='true' "
-    "xmlns:auth='http://www.google.com/talk/protocol/auth'>"
-    "AEFjY291bnRATmFtZQBBY2Nlc3NUb2tlbg==</auth>";
-constexpr char kBindMessage[] =
-    "<iq type='set' id='0'><bind "
-    "xmlns='urn:ietf:params:xml:ns:xmpp-bind'/></iq>";
-constexpr char kSessionMessage[] =
-    "<iq type='set' id='1'><session "
-    "xmlns='urn:ietf:params:xml:ns:xmpp-session'/></iq>";
-constexpr char kSubscribeMessage[] =
-    "<iq type='set' to='Account@Name' id='pushsubscribe1'>"
-    "<subscribe xmlns='google:push'><item channel='cloud_devices' from=''/>"
-    "</subscribe></iq>";
-
-class MockXmppConnection : public XmppConnection {
- public:
-  MockXmppConnection() = default;
-
-  MOCK_CONST_METHOD1(Read, bool(std::string* msg));
-  MOCK_CONST_METHOD1(Write, bool(const std::string& msg));
-
- private:
-  DISALLOW_COPY_AND_ASSIGN(MockXmppConnection);
-};
-
-}  // namespace
-
-class TestHelper {
- public:
-  static void SetState(XmppClient* client, XmppClient::XmppState state) {
-    client->state_ = state;
-  }
-
-  static XmppClient::XmppState GetState(const XmppClient& client) {
-    return client.state_;
-  }
-
-  static XmppConnection* GetConnection(const XmppClient& client) {
-    return client.connection_.get();
-  }
-};
-
-class XmppClientTest : public ::testing::Test {
- protected:
-  void SetUp() override {
-    std::unique_ptr<XmppConnection> connection(new MockXmppConnection());
-    xmpp_client_.reset(new XmppClient(kAccountName, kAccessToken,
-                                      std::move(connection)));
-    connection_ = static_cast<MockXmppConnection*>(
-        TestHelper::GetConnection(*xmpp_client_));
-  }
-
-  std::unique_ptr<XmppClient> xmpp_client_;
-  MockXmppConnection* connection_;  // xmpp_client_ owns this.
-};
-
-TEST_F(XmppClientTest, StartStream) {
-  EXPECT_EQ(TestHelper::GetState(*xmpp_client_),
-            XmppClient::XmppState::kNotStarted);
-  EXPECT_CALL(*connection_, Write(kStartStreamMessage))
-      .WillOnce(Return(true));
-  xmpp_client_->StartStream();
-  EXPECT_EQ(TestHelper::GetState(*xmpp_client_),
-            XmppClient::XmppState::kStarted);
-}
-
-TEST_F(XmppClientTest, HandleStartedResponse) {
-  TestHelper::SetState(xmpp_client_.get(),
-                       XmppClient::XmppState::kStarted);
-  EXPECT_CALL(*connection_, Read(_))
-      .WillOnce(DoAll(SetArgPointee<0>(kStartStreamResponse), Return(true)));
-  EXPECT_CALL(*connection_, Write(kAuthenticationMessage))
-      .WillOnce(Return(true));
-  xmpp_client_->Read();
-  EXPECT_EQ(TestHelper::GetState(*xmpp_client_),
-            XmppClient::XmppState::kAuthenticationStarted);
-}
-
-TEST_F(XmppClientTest, HandleAuthenticationSucceededResponse) {
-  TestHelper::SetState(
-      xmpp_client_.get(),
-      XmppClient::XmppState::kAuthenticationStarted);
-  EXPECT_CALL(*connection_, Read(_))
-      .WillOnce(DoAll(SetArgPointee<0>(kAuthenticationSucceededResponse),
-                      Return(true)));
-  EXPECT_CALL(*connection_, Write(kStartStreamMessage))
-      .WillOnce(Return(true));
-  xmpp_client_->Read();
-  EXPECT_EQ(TestHelper::GetState(*xmpp_client_),
-            XmppClient::XmppState::kStreamRestartedPostAuthentication);
-}
-
-TEST_F(XmppClientTest, HandleAuthenticationFailedResponse) {
-  TestHelper::SetState(
-      xmpp_client_.get(),
-      XmppClient::XmppState::kAuthenticationStarted);
-  EXPECT_CALL(*connection_, Read(_))
-      .WillOnce(DoAll(SetArgPointee<0>(kAuthenticationFailedResponse),
-                      Return(true)));
-  EXPECT_CALL(*connection_, Write(_))
-      .Times(0);
-  xmpp_client_->Read();
-  EXPECT_EQ(TestHelper::GetState(*xmpp_client_),
-            XmppClient::XmppState::kAuthenticationFailed);
-}
-
-TEST_F(XmppClientTest, HandleStreamRestartedResponse) {
-  TestHelper::SetState(
-      xmpp_client_.get(),
-      XmppClient::XmppState::kStreamRestartedPostAuthentication);
-  EXPECT_CALL(*connection_, Read(_))
-      .WillOnce(DoAll(SetArgPointee<0>(kRestartStreamResponse), Return(true)));
-  EXPECT_CALL(*connection_, Write(kBindMessage))
-      .WillOnce(Return(true));
-  xmpp_client_->Read();
-  EXPECT_EQ(TestHelper::GetState(*xmpp_client_),
-            XmppClient::XmppState::kBindSent);
-}
-
-TEST_F(XmppClientTest, HandleBindResponse) {
-  TestHelper::SetState(xmpp_client_.get(),
-                       XmppClient::XmppState::kBindSent);
-  EXPECT_CALL(*connection_, Read(_))
-      .WillOnce(DoAll(SetArgPointee<0>(kBindResponse), Return(true)));
-  EXPECT_CALL(*connection_, Write(kSessionMessage))
-      .WillOnce(Return(true));
-  xmpp_client_->Read();
-  EXPECT_EQ(TestHelper::GetState(*xmpp_client_),
-            XmppClient::XmppState::kSessionStarted);
-}
-
-TEST_F(XmppClientTest, HandleSessionResponse) {
-  TestHelper::SetState(xmpp_client_.get(),
-                       XmppClient::XmppState::kSessionStarted);
-  EXPECT_CALL(*connection_, Read(_))
-      .WillOnce(DoAll(SetArgPointee<0>(kSessionResponse), Return(true)));
-  EXPECT_CALL(*connection_, Write(kSubscribeMessage))
-      .WillOnce(Return(true));
-  xmpp_client_->Read();
-  EXPECT_EQ(TestHelper::GetState(*xmpp_client_),
-            XmppClient::XmppState::kSubscribeStarted);
-}
-
-TEST_F(XmppClientTest, HandleSubscribeResponse) {
-  TestHelper::SetState(xmpp_client_.get(),
-                       XmppClient::XmppState::kSubscribeStarted);
-  EXPECT_CALL(*connection_, Read(_))
-      .WillOnce(DoAll(SetArgPointee<0>(kSubscribedResponse), Return(true)));
-  EXPECT_CALL(*connection_, Write(_))
-      .Times(0);
-  xmpp_client_->Read();
-  EXPECT_EQ(TestHelper::GetState(*xmpp_client_),
-            XmppClient::XmppState::kSubscribed);
-}
-
-}  // namespace buffet
diff --git a/buffet/xmpp/xmpp_connection.cc b/buffet/xmpp/xmpp_connection.cc
deleted file mode 100644
index 38e99e3..0000000
--- a/buffet/xmpp/xmpp_connection.cc
+++ /dev/null
@@ -1,72 +0,0 @@
-// 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/xmpp/xmpp_connection.h"
-
-#include <arpa/inet.h>
-#include <netdb.h>
-#include <netinet/in.h>
-#include <unistd.h>
-
-#include <string>
-
-#include <base/files/file_util.h>
-#include <chromeos/syslog_logging.h>
-
-namespace buffet {
-
-XmppConnection::~XmppConnection() {
-  if (fd_ > 0) {
-    close(fd_);
-  }
-}
-
-bool XmppConnection::Initialize() {
-  LOG(INFO) << "Opening XMPP connection";
-
-  fd_ = socket(AF_INET, SOCK_STREAM, 0);
-
-  // TODO(nathanbullock): Use gethostbyname_r.
-  struct hostent* server = gethostbyname("talk.google.com");
-  if (server == NULL) {
-    LOG(WARNING) << "Failed to find host talk.google.com";
-    return false;
-  }
-
-  sockaddr_in serv_addr;
-  memset(&serv_addr, 0, sizeof(serv_addr));
-  serv_addr.sin_family = AF_INET;
-  bcopy(server->h_addr, &(serv_addr.sin_addr), server->h_length);
-  serv_addr.sin_port = htons(5222);
-
-  if (connect(fd_, (struct sockaddr*)&serv_addr, sizeof(sockaddr)) < 0) {
-    LOG(WARNING) << "Failed to connect to talk.google.com:5222";
-    return false;
-  }
-
-  return true;
-}
-
-bool XmppConnection::Read(std::string* msg) const {
-  char buffer[4096];  // This should be large enough for our purposes.
-  ssize_t bytes = HANDLE_EINTR(read(fd_, buffer, sizeof(buffer)));
-  if (bytes < 0) {
-    LOG(WARNING) << "Failure reading";
-    return false;
-  }
-  *msg = std::string{buffer, static_cast<size_t>(bytes)};
-  LOG(INFO) << "Read: (" << msg->size() << ")" << *msg;
-  return true;
-}
-
-bool XmppConnection::Write(const std::string& msg) const {
-  LOG(INFO) << "Write: (" << msg.size() << ")" << msg;
-  if (!base::WriteFileDescriptor(fd_, msg.c_str(), msg.size())) {
-    LOG(WARNING) << "Failure writing";
-    return false;
-  }
-  return true;
-}
-
-}  // namespace buffet
diff --git a/buffet/xmpp/xmpp_connection.h b/buffet/xmpp/xmpp_connection.h
deleted file mode 100644
index 741f4c8..0000000
--- a/buffet/xmpp/xmpp_connection.h
+++ /dev/null
@@ -1,40 +0,0 @@
-// 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_XMPP_XMPP_CONNECTION_H_
-#define BUFFET_XMPP_XMPP_CONNECTION_H_
-
-#include <string>
-
-#include <base/macros.h>
-
-namespace buffet {
-
-class XmppConnection {
- public:
-  XmppConnection() {}
-  virtual ~XmppConnection();
-
-  // Initialize the XMPP client. (Connects to talk.google.com:5222).
-  virtual bool Initialize();
-
-  int GetFileDescriptor() const { return fd_; }
-
-  // Needs to be called when new data is available from the connection.
-  virtual bool Read(std::string* msg) const;
-
-  // Start talking to the XMPP server (authenticate, etc.)
-  virtual bool Write(const std::string& msg) const;
-
- private:
-  // The file descriptor for the connection.
-  int fd_{-1};
-
-  DISALLOW_COPY_AND_ASSIGN(XmppConnection);
-};
-
-}  // namespace buffet
-
-#endif  // BUFFET_XMPP_XMPP_CONNECTION_H_
-