[Sync] Delete ServerNotifierThread and refactor InvalidationNotifier

Basically replace the InvalidationNotifier/ServerNotifierThread combo
with InvalidationNotifier and NonBlockingInvalidationNotifier.

Some minor cleanup in jingle files.

This finishes steps 1 and 2 in the bug.

BUG=76764
TEST=

Review URL: http://codereview.chromium.org/6696051

git-svn-id: svn://svn.chromium.org/chrome/trunk/src@79334 0039d316-1c4b-4281-b951-d872f2087c98

16 files changed, 720 insertions, 487 deletions

diff --git a/chrome/browser/sync/notifier/invalidation_notifier.cc b/chrome/browser/sync/notifier/invalidation_notifier.cc
index 0936e1e..6c0fb87 100644
--- a/chrome/browser/sync/notifier/invalidation_notifier.cc
+++ b/chrome/browser/sync/notifier/invalidation_notifier.cc
@@ -4,95 +4,156 @@
 
 #include "chrome/browser/sync/notifier/invalidation_notifier.h"
 
+#include "base/logging.h"
 #include "chrome/browser/sync/notifier/sync_notifier_observer.h"
 #include "chrome/browser/sync/protocol/service_constants.h"
-#include "chrome/browser/sync/sessions/session_state.h"
+#include "chrome/browser/sync/syncable/model_type_payload_map.h"
+#include "jingle/notifier/base/const_communicator.h"
 #include "jingle/notifier/base/notifier_options_util.h"
-#include "jingle/notifier/listener/notification_constants.h"
+#include "jingle/notifier/communicator/connection_options.h"
+#include "net/base/host_port_pair.h"
+#include "talk/xmpp/jid.h"
+#include "talk/xmpp/xmppclientsettings.h"
 
 namespace sync_notifier {
 
 InvalidationNotifier::InvalidationNotifier(
     const notifier::NotifierOptions& notifier_options,
+    net::HostResolver* host_resolver,
+    net::CertVerifier* cert_verifier,
     const std::string& client_info)
-    : notifier_options_(notifier_options),
-      server_notifier_thread_(notifier_options, client_info, this),
-      logged_in_(false) {
-  server_notifier_thread_.AddObserver(this);
-  server_notifier_thread_.Start();
+    : state_(STOPPED),
+      notifier_options_(notifier_options),
+      host_resolver_(host_resolver),
+      cert_verifier_(cert_verifier),
+      client_info_(client_info) {
+  DCHECK_EQ(notifier::NOTIFICATION_SERVER,
+            notifier_options.notification_method);
 }
 
 InvalidationNotifier::~InvalidationNotifier() {
-  if (logged_in_) {
-    server_notifier_thread_.Logout();
-  }
-  server_notifier_thread_.RemoveObserver(this);
+  DCHECK(non_thread_safe_.CalledOnValidThread());
 }
 
 void InvalidationNotifier::AddObserver(SyncNotifierObserver* observer) {
-  observer_list_.AddObserver(observer);
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  observers_.AddObserver(observer);
 }
 
 void InvalidationNotifier::RemoveObserver(SyncNotifierObserver* observer) {
-  observer_list_.RemoveObserver(observer);
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  observers_.RemoveObserver(observer);
 }
 
 void InvalidationNotifier::SetState(const std::string& state) {
-  server_notifier_thread_.SetState(state);
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  invalidation_state_ = state;
 }
 
 void InvalidationNotifier::UpdateCredentials(
     const std::string& email, const std::string& token) {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  VLOG(1) << "Updating credentials for " << email;
   buzz::XmppClientSettings xmpp_client_settings =
       notifier::MakeXmppClientSettings(notifier_options_,
                                        email, token, SYNC_SERVICE_NAME);
-  if (logged_in_) {
-    server_notifier_thread_.UpdateXmppSettings(xmpp_client_settings);
+  if (state_ >= CONNECTING) {
+    login_->UpdateXmppSettings(xmpp_client_settings);
   } else {
-    server_notifier_thread_.Login(xmpp_client_settings);
-    logged_in_ = true;
+    notifier::ConnectionOptions options;
+    VLOG(1) << "First time updating credentials: connecting";
+    login_.reset(
+        new notifier::Login(this,
+                            xmpp_client_settings,
+                            notifier::ConnectionOptions(),
+                            host_resolver_,
+                            cert_verifier_,
+                            notifier::GetServerList(notifier_options_),
+                            notifier_options_.try_ssltcp_first,
+                            notifier_options_.auth_mechanism));
+    login_->StartConnection();
+    state_ = CONNECTING;
   }
 }
 
 void InvalidationNotifier::UpdateEnabledTypes(
     const syncable::ModelTypeSet& types) {
-  server_notifier_thread_.UpdateEnabledTypes(types);
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  enabled_types_ = types;
+  if (state_ >= STARTED) {
+    invalidation_client_.RegisterTypes(enabled_types_);
+  }
+  // If |invalidation_client_| hasn't been started yet, it will
+  // register |enabled_types_| in OnConnect().
 }
 
-void InvalidationNotifier::SendNotification() {}
+void InvalidationNotifier::SendNotification() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+}
 
-void InvalidationNotifier::WriteState(const std::string& state) {
-  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
-      StoreState(state));
+void InvalidationNotifier::OnConnect(
+    base::WeakPtr<talk_base::Task> base_task) {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  VLOG(1) << "OnConnect";
+  if (state_ >= STARTED) {
+    invalidation_client_.ChangeBaseTask(base_task);
+  } else {
+    VLOG(1) << "First time connecting: starting invalidation client";
+    // TODO(akalin): Make cache_guid() part of the client ID.  If we
+    // do so and we somehow propagate it up to the server somehow, we
+    // can make it so that we won't receive any notifications that
+    // were generated from our own changes.
+    const std::string kClientId = "invalidation_notifier";
+    invalidation_client_.Start(
+        kClientId, client_info_, invalidation_state_, this, this, base_task);
+    invalidation_state_.clear();
+    invalidation_client_.RegisterTypes(enabled_types_);
+    state_ = STARTED;
+  }
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observers_,
+                    OnNotificationStateChange(true));
 }
 
-void InvalidationNotifier::OnConnectionStateChange(bool logged_in) {
-  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
-                    OnNotificationStateChange(logged_in));
+void InvalidationNotifier::OnDisconnect() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observers_,
+                    OnNotificationStateChange(false));
 }
 
-void InvalidationNotifier::OnSubscriptionStateChange(bool subscribed) {}
+void InvalidationNotifier::OnInvalidate(
+    syncable::ModelType model_type, const std::string& payload) {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  DCHECK_GE(model_type, syncable::FIRST_REAL_MODEL_TYPE);
+  DCHECK_LT(model_type, syncable::MODEL_TYPE_COUNT);
+  VLOG(1) << "OnInvalidate: " << syncable::ModelTypeToString(model_type)
+          << " " << payload;
+  syncable::ModelTypeSet types;
+  types.insert(model_type);
+  EmitInvalidation(types, payload);
+}
 
-void InvalidationNotifier::OnIncomingNotification(
-    const notifier::Notification& notification) {
-  VLOG(1) << "Sync received server notification from "
-          << notification.channel << ": " << notification.data;
-  const std::string& model_type_list = notification.channel;
-  const std::string& notification_payload = notification.data;
+void InvalidationNotifier::OnInvalidateAll() {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  VLOG(1) << "OnInvalidateAll";
+  EmitInvalidation(enabled_types_, std::string());
+}
 
-  syncable::ModelTypeBitSet model_types;
-  if (!syncable::ModelTypeBitSetFromString(model_type_list, &model_types)) {
-    LOG(DFATAL) << "Could not extract model types from server data.";
-    model_types.set();
-  }
+void InvalidationNotifier::WriteState(const std::string& state) {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  VLOG(1) << "WriteState";
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observers_, StoreState(state));
+}
 
+void InvalidationNotifier::EmitInvalidation(
+    const syncable::ModelTypeSet& types, const std::string& payload) {
+  DCHECK(non_thread_safe_.CalledOnValidThread());
+  // TODO(akalin): Move all uses of ModelTypeBitSet for invalidations
+  // to ModelTypeSet.
   syncable::ModelTypePayloadMap type_payloads =
       syncable::ModelTypePayloadMapFromBitSet(
-          model_types, notification_payload);
-  FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
+          syncable::ModelTypeBitSetFromSet(types), payload);
+  FOR_EACH_OBSERVER(SyncNotifierObserver, observers_,
                     OnIncomingNotification(type_payloads));
 }
 
-void InvalidationNotifier::OnOutgoingNotification() {}
-
 }  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/invalidation_notifier.h b/chrome/browser/sync/notifier/invalidation_notifier.h
index 6ee27e9..1a951d7 100644
--- a/chrome/browser/sync/notifier/invalidation_notifier.h
+++ b/chrome/browser/sync/notifier/invalidation_notifier.h
@@ -2,30 +2,50 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 //
-// A notifier that communicates to sync servers for server-issued
-// notifications.
+// An implementation of SyncNotifier that wraps an invalidation
+// client.  Handles the details of connecting to XMPP and hooking it
+// up to the invalidation client.
+//
+// You probably don't want to use this directly; use
+// NonBlockingInvalidationNotifier.
 
 #ifndef CHROME_BROWSER_SYNC_NOTIFIER_INVALIDATION_NOTIFIER_H_
 #define CHROME_BROWSER_SYNC_NOTIFIER_INVALIDATION_NOTIFIER_H_
+#pragma once
 
 #include <string>
 
+#include "base/basictypes.h"
 #include "base/observer_list.h"
-#include "chrome/browser/sync/notifier/server_notifier_thread.h"
+#include "base/scoped_ptr.h"
+#include "base/threading/non_thread_safe.h"
+#include "chrome/browser/sync/notifier/chrome_invalidation_client.h"
 #include "chrome/browser/sync/notifier/state_writer.h"
 #include "chrome/browser/sync/notifier/sync_notifier.h"
 #include "chrome/browser/sync/syncable/model_type.h"
 #include "jingle/notifier/base/notifier_options.h"
+#include "jingle/notifier/communicator/login.h"
+#include "net/base/cert_verifier.h"
+
+namespace net {
+class HostResolver;
+class CertVerifier;
+}  // namespace net
 
 namespace sync_notifier {
 
 class InvalidationNotifier
     : public SyncNotifier,
-      public StateWriter,
-      public ServerNotifierThread::Observer {
+      public notifier::LoginDelegate,
+      public ChromeInvalidationClient::Listener,
+      public StateWriter {
  public:
-  InvalidationNotifier(const notifier::NotifierOptions& notifier_options,
-                       const std::string& client_info);
+  // Does not take ownership of |host_resolver| or |cert_verifier|.
+  InvalidationNotifier(
+      const notifier::NotifierOptions& notifier_options,
+      net::HostResolver* host_resolver,
+      net::CertVerifier* cert_verifier,
+      const std::string& client_info);
 
   virtual ~InvalidationNotifier();
 
@@ -38,26 +58,62 @@ class InvalidationNotifier
   virtual void UpdateEnabledTypes(const syncable::ModelTypeSet& types);
   virtual void SendNotification();
 
+  // notifier::LoginDelegate implementation.
+  virtual void OnConnect(base::WeakPtr<talk_base::Task> base_task);
+  virtual void OnDisconnect();
+
+  // ChromeInvalidationClient::Listener implementation.
+  virtual void OnInvalidate(syncable::ModelType model_type,
+                            const std::string& payload);
+  virtual void OnInvalidateAll();
+
   // StateWriter implementation.
   virtual void WriteState(const std::string& state);
 
-  // ServerNotifierThread::Observer implementation.
-  virtual void OnConnectionStateChange(bool logged_in);
-  virtual void OnSubscriptionStateChange(bool subscribed);
-  virtual void OnIncomingNotification(
-      const notifier::Notification& notification);
-  virtual void OnOutgoingNotification();
-
  private:
+  void EmitInvalidation(const syncable::ModelTypeSet& types,
+                        const std::string& payload);
+
+  base::NonThreadSafe non_thread_safe_;
+
+  // We start off in the STOPPED state.  When we get our initial
+  // credentials, we connect and move to the CONNECTING state.  When
+  // we're connected we start the invalidation client and move to the
+  // STARTED state.  We never go back to a previous state.
+  enum State {
+    STOPPED,
+    CONNECTING,
+    STARTED
+  };
+  State state_;
+
+  // Used to build parameters for |login_|.
   const notifier::NotifierOptions notifier_options_;
-  // The actual notification listener.
-  ServerNotifierThread server_notifier_thread_;
-  // Whether we called Login() on |server_notifier_thread_| yet.
-  bool logged_in_;
+  net::HostResolver* const host_resolver_;
+  net::CertVerifier* const cert_verifier_;
 
-  ObserverList<SyncNotifierObserver> observer_list_;
+  // Passed to |invalidation_client_|.
+  const std::string client_info_;
+
+  // Our observers (which must live on the same thread).
+  ObserverList<SyncNotifierObserver> observers_;
+
+  // The state to pass to |chrome_invalidation_client_|.
+  std::string invalidation_state_;
+
+  // The XMPP connection manager.
+  scoped_ptr<notifier::Login> login_;
+
+  // The invalidation client.
+  ChromeInvalidationClient invalidation_client_;
+
+  // Passed to |invalidation_client_| and also used to synthesize
+  // notifications by OnInvalidateAll().
   syncable::ModelTypeSet enabled_types_;
+
+  DISALLOW_COPY_AND_ASSIGN(InvalidationNotifier);
 };
 
-}
+}  // namespace sync_notifier
+
 #endif  // CHROME_BROWSER_SYNC_NOTIFIER_INVALIDATION_NOTIFIER_H_
diff --git a/chrome/browser/sync/notifier/invalidation_notifier_unittest.cc b/chrome/browser/sync/notifier/invalidation_notifier_unittest.cc
new file mode 100644
index 0000000..ab1fd7f
--- /dev/null
+++ b/chrome/browser/sync/notifier/invalidation_notifier_unittest.cc
@@ -0,0 +1,113 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/notifier/invalidation_notifier.h"
+
+#include "base/message_loop.h"
+#include "base/scoped_ptr.h"
+#include "chrome/browser/sync/notifier/mock_sync_notifier_observer.h"
+#include "chrome/browser/sync/syncable/model_type.h"
+#include "chrome/browser/sync/syncable/model_type_payload_map.h"
+#include "jingle/notifier/base/fake_base_task.h"
+#include "net/base/cert_verifier.h"
+#include "net/base/host_resolver.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace sync_notifier {
+
+namespace {
+
+using ::testing::InSequence;
+using ::testing::StrictMock;
+
+class InvalidationNotifierTest : public testing::Test {
+ public:
+  InvalidationNotifierTest()
+      : host_resolver_(
+          net::CreateSystemHostResolver(
+              net::HostResolver::kDefaultParallelism, NULL, NULL)),
+        invalidation_notifier_(notifier::NotifierOptions(),
+                               host_resolver_.get(),
+                               &cert_verifier_,
+                               "fake_client_info") {}
+
+ protected:
+  virtual void SetUp() {
+    invalidation_notifier_.AddObserver(&mock_observer_);
+  }
+
+  virtual void TearDown() {
+    invalidation_notifier_.RemoveObserver(&mock_observer_);
+  }
+
+  MessageLoop message_loop_;
+  scoped_ptr<net::HostResolver> host_resolver_;
+  net::CertVerifier cert_verifier_;
+  InvalidationNotifier invalidation_notifier_;
+  StrictMock<MockSyncNotifierObserver> mock_observer_;
+  notifier::FakeBaseTask fake_base_task_;
+};
+
+TEST_F(InvalidationNotifierTest, Basic) {
+  InSequence dummy;
+
+  syncable::ModelTypeSet types;
+  types.insert(syncable::BOOKMARKS);
+  types.insert(syncable::AUTOFILL);
+
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
+  EXPECT_CALL(mock_observer_, StoreState("new_fake_state"));
+  {
+    syncable::ModelTypePayloadMap type_payloads;
+    type_payloads[syncable::PREFERENCES] = "payload";
+    EXPECT_CALL(mock_observer_, OnIncomingNotification(type_payloads));
+  }
+  {
+    syncable::ModelTypePayloadMap type_payloads;
+    type_payloads[syncable::BOOKMARKS] = "";
+    type_payloads[syncable::AUTOFILL] = "";
+    EXPECT_CALL(mock_observer_, OnIncomingNotification(type_payloads));
+  }
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(false));
+
+  invalidation_notifier_.SetState("fake_state");
+  invalidation_notifier_.UpdateCredentials("foo@bar.com", "fake_token");
+  invalidation_notifier_.UpdateEnabledTypes(types);
+
+  invalidation_notifier_.OnConnect(fake_base_task_.AsWeakPtr());
+
+  invalidation_notifier_.WriteState("new_fake_state");
+
+  // Even though preferences isn't in the set of enabled types, we
+  // should still propagate the notification.
+  invalidation_notifier_.OnInvalidate(syncable::PREFERENCES, "payload");
+  invalidation_notifier_.OnInvalidateAll();
+
+  invalidation_notifier_.OnDisconnect();
+}
+
+TEST_F(InvalidationNotifierTest, UpdateEnabledTypes) {
+  InSequence dummy;
+
+  syncable::ModelTypeSet types;
+  types.insert(syncable::BOOKMARKS);
+  types.insert(syncable::AUTOFILL);
+
+  EXPECT_CALL(mock_observer_, OnNotificationStateChange(true));
+  {
+    syncable::ModelTypePayloadMap type_payloads;
+    type_payloads[syncable::BOOKMARKS] = "";
+    type_payloads[syncable::AUTOFILL] = "";
+    EXPECT_CALL(mock_observer_, OnIncomingNotification(type_payloads));
+  }
+
+  invalidation_notifier_.OnConnect(fake_base_task_.AsWeakPtr());
+  invalidation_notifier_.UpdateEnabledTypes(types);
+  invalidation_notifier_.OnInvalidateAll();
+}
+
+}  // namespace
+
+}  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/mock_sync_notifier_observer.cc b/chrome/browser/sync/notifier/mock_sync_notifier_observer.cc
new file mode 100644
index 0000000..b46ac68
--- /dev/null
+++ b/chrome/browser/sync/notifier/mock_sync_notifier_observer.cc
@@ -0,0 +1,12 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/notifier/mock_sync_notifier_observer.h"
+
+namespace sync_notifier {
+
+MockSyncNotifierObserver::MockSyncNotifierObserver() {}
+MockSyncNotifierObserver::~MockSyncNotifierObserver() {}
+
+}  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/mock_sync_notifier_observer.h b/chrome/browser/sync/notifier/mock_sync_notifier_observer.h
new file mode 100644
index 0000000..a9e141f
--- /dev/null
+++ b/chrome/browser/sync/notifier/mock_sync_notifier_observer.h
@@ -0,0 +1,29 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#ifndef CHROME_BROWSER_SYNC_NOTIFIER_MOCK_SYNC_NOTIFIER_OBSERVER_H_
+#define CHROME_BROWSER_SYNC_NOTIFIER_MOCK_SYNC_NOTIFIER_OBSERVER_H_
+#pragma once
+
+#include <string>
+
+#include "chrome/browser/sync/notifier/sync_notifier_observer.h"
+#include "testing/gmock/include/gmock/gmock.h"
+
+namespace sync_notifier {
+
+class MockSyncNotifierObserver : public SyncNotifierObserver {
+ public:
+  MockSyncNotifierObserver();
+  virtual ~MockSyncNotifierObserver();
+
+  MOCK_METHOD1(OnIncomingNotification,
+               void(const syncable::ModelTypePayloadMap&));
+  MOCK_METHOD1(OnNotificationStateChange, void(bool));
+  MOCK_METHOD1(StoreState, void(const std::string&));
+};
+
+}  // namespace sync_notifier
+
+#endif  // CHROME_BROWSER_SYNC_NOTIFIER_MOCK_SYNC_NOTIFIER_OBSERVER_H_
diff --git a/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.cc b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.cc
new file mode 100644
index 0000000..11adf97
--- /dev/null
+++ b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.cc
@@ -0,0 +1,177 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h"
+
+#include "base/logging.h"
+#include "base/message_loop.h"
+
+namespace sync_notifier {
+
+NonBlockingInvalidationNotifier::NonBlockingInvalidationNotifier(
+    const notifier::NotifierOptions& notifier_options,
+    const std::string& client_info)
+    : parent_message_loop_(MessageLoop::current()),
+      observers_(new ObserverListThreadSafe<SyncNotifierObserver>()),
+      worker_thread_("InvalidationNotifier worker thread"),
+      worker_thread_vars_(NULL) {
+  DCHECK_EQ(notifier::NOTIFICATION_SERVER,
+            notifier_options.notification_method);
+  const base::Thread::Options options(MessageLoop::TYPE_IO, 0);
+  CHECK(worker_thread_.StartWithOptions(options));
+  worker_message_loop()->PostTask(
+      FROM_HERE,
+      NewRunnableMethod(
+          this,
+          &NonBlockingInvalidationNotifier::CreateWorkerThreadVars,
+          notifier_options, client_info));
+}
+
+NonBlockingInvalidationNotifier::~NonBlockingInvalidationNotifier() {
+  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  worker_message_loop()->PostTask(
+      FROM_HERE,
+      NewRunnableMethod(
+          this,
+          &NonBlockingInvalidationNotifier::DestroyWorkerThreadVars));
+  worker_thread_.Stop();
+  CHECK(!worker_thread_vars_);
+}
+
+void NonBlockingInvalidationNotifier::AddObserver(
+    SyncNotifierObserver* observer) {
+  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  observers_->AddObserver(observer);
+}
+
+void NonBlockingInvalidationNotifier::RemoveObserver(
+    SyncNotifierObserver* observer) {
+  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  observers_->RemoveObserver(observer);
+}
+
+void NonBlockingInvalidationNotifier::SetState(const std::string& state) {
+  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  worker_message_loop()->PostTask(
+      FROM_HERE,
+      NewRunnableMethod(
+          this,
+          &NonBlockingInvalidationNotifier::SetStateOnWorkerThread,
+          state));
+}
+
+void NonBlockingInvalidationNotifier::UpdateCredentials(
+    const std::string& email, const std::string& token) {
+  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  worker_message_loop()->PostTask(
+      FROM_HERE,
+      NewRunnableMethod(
+          this,
+          &NonBlockingInvalidationNotifier::UpdateCredentialsOnWorkerThread,
+          email, token));
+}
+
+void NonBlockingInvalidationNotifier::UpdateEnabledTypes(
+    const syncable::ModelTypeSet& types) {
+  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  worker_message_loop()->PostTask(
+      FROM_HERE,
+      NewRunnableMethod(
+          this,
+          &NonBlockingInvalidationNotifier::UpdateEnabledTypesOnWorkerThread,
+          types));
+}
+
+void NonBlockingInvalidationNotifier::SendNotification() {
+  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
+  // InvalidationClient doesn't implement SendNotification(), so no
+  // need to forward on the call.
+}
+
+MessageLoop* NonBlockingInvalidationNotifier::worker_message_loop() {
+  MessageLoop* current_message_loop = MessageLoop::current();
+  DCHECK(current_message_loop);
+  MessageLoop* worker_message_loop = worker_thread_.message_loop();
+  DCHECK(worker_message_loop);
+  DCHECK(current_message_loop == parent_message_loop_ ||
+         current_message_loop == worker_message_loop);
+  return worker_message_loop;
+}
+
+void NonBlockingInvalidationNotifier::CreateWorkerThreadVars(
+    const notifier::NotifierOptions& notifier_options,
+    const std::string& client_info) {
+  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
+  worker_thread_vars_ =
+      new WorkerThreadVars(notifier_options, client_info, observers_);
+}
+
+void NonBlockingInvalidationNotifier::DestroyWorkerThreadVars() {
+  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
+  delete worker_thread_vars_;
+  worker_thread_vars_ = NULL;
+}
+
+void NonBlockingInvalidationNotifier::SetStateOnWorkerThread(
+    const std::string& state) {
+  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
+  worker_thread_vars_->invalidation_notifier.SetState(state);
+}
+
+void NonBlockingInvalidationNotifier::UpdateCredentialsOnWorkerThread(
+    const std::string& email, const std::string& token) {
+  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
+  worker_thread_vars_->invalidation_notifier.UpdateCredentials(email, token);
+}
+
+void NonBlockingInvalidationNotifier::UpdateEnabledTypesOnWorkerThread(
+    const syncable::ModelTypeSet& types) {
+  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
+  worker_thread_vars_->invalidation_notifier.UpdateEnabledTypes(types);
+}
+
+NonBlockingInvalidationNotifier::ObserverRouter::ObserverRouter(
+    const scoped_refptr<ObserverListThreadSafe<SyncNotifierObserver> >&
+        observers) : observers_(observers) {}
+
+NonBlockingInvalidationNotifier::ObserverRouter::~ObserverRouter() {}
+
+void NonBlockingInvalidationNotifier::ObserverRouter::
+    OnIncomingNotification(
+        const syncable::ModelTypePayloadMap& type_payloads) {
+  observers_->Notify(&SyncNotifierObserver::OnIncomingNotification,
+                     type_payloads);
+}
+
+void NonBlockingInvalidationNotifier::ObserverRouter::
+    OnNotificationStateChange(
+        bool notifications_enabled) {
+  observers_->Notify(&SyncNotifierObserver::OnNotificationStateChange,
+                     notifications_enabled);
+}
+
+void NonBlockingInvalidationNotifier::ObserverRouter::StoreState(
+    const std::string& state) {
+  observers_->Notify(&SyncNotifierObserver::StoreState, state);
+}
+
+NonBlockingInvalidationNotifier::WorkerThreadVars::WorkerThreadVars(
+    const notifier::NotifierOptions& notifier_options,
+    const std::string& client_info,
+    const scoped_refptr<ObserverListThreadSafe<SyncNotifierObserver> >&
+        observers)
+    : host_resolver_(
+        net::CreateSystemHostResolver(net::HostResolver::kDefaultParallelism,
+                                      NULL, NULL)),
+      invalidation_notifier(notifier_options, host_resolver_.get(),
+                            &cert_verifier_, client_info),
+      observer_router_(observers) {
+  invalidation_notifier.AddObserver(&observer_router_);
+}
+
+NonBlockingInvalidationNotifier::WorkerThreadVars::~WorkerThreadVars() {
+  invalidation_notifier.RemoveObserver(&observer_router_);
+}
+
+}  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h
new file mode 100644
index 0000000..115a43c
--- /dev/null
+++ b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h
@@ -0,0 +1,124 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+//
+// An implementation of SyncNotifier that wraps InvalidationNotifier
+// on its own thread.
+
+#ifndef CHROME_BROWSER_SYNC_NOTIFIER_NON_BLOCKING_INVALIDATION_NOTIFIER_H_
+#define CHROME_BROWSER_SYNC_NOTIFIER_NON_BLOCKING_INVALIDATION_NOTIFIER_H_
+#pragma once
+
+#include <string>
+
+#include "base/basictypes.h"
+#include "base/observer_list_threadsafe.h"
+#include "base/ref_counted.h"
+#include "base/scoped_ptr.h"
+#include "base/threading/thread.h"
+#include "chrome/browser/sync/notifier/invalidation_notifier.h"
+#include "chrome/browser/sync/notifier/sync_notifier.h"
+#include "chrome/browser/sync/notifier/sync_notifier_observer.h"
+
+class MessageLoop;
+
+namespace sync_notifier {
+
+class NonBlockingInvalidationNotifier : public SyncNotifier {
+ public:
+  NonBlockingInvalidationNotifier(
+      const notifier::NotifierOptions& notifier_options,
+      const std::string& client_info);
+
+  virtual ~NonBlockingInvalidationNotifier();
+
+  // SyncNotifier implementation.
+  virtual void AddObserver(SyncNotifierObserver* observer);
+  virtual void RemoveObserver(SyncNotifierObserver* observer);
+  virtual void SetState(const std::string& state);
+  virtual void UpdateCredentials(
+      const std::string& email, const std::string& token);
+  virtual void UpdateEnabledTypes(const syncable::ModelTypeSet& types);
+  virtual void SendNotification();
+
+ private:
+  // Utility class that routes received notifications to a given
+  // thread-safe observer list.
+  class ObserverRouter : public SyncNotifierObserver {
+   public:
+    explicit ObserverRouter(
+        const scoped_refptr<ObserverListThreadSafe<SyncNotifierObserver> >&
+            observers);
+
+    virtual ~ObserverRouter();
+
+    // SyncNotifierObserver implementation.
+    virtual void OnIncomingNotification(
+        const syncable::ModelTypePayloadMap& type_payloads);
+    virtual void OnNotificationStateChange(bool notifications_enabled);
+    virtual void StoreState(const std::string& state);
+
+   private:
+    scoped_refptr<ObserverListThreadSafe<SyncNotifierObserver> > observers_;
+  };
+
+  // The set of variables that should only be created/used on the
+  // worker thread.
+  struct WorkerThreadVars {
+    WorkerThreadVars(
+        const notifier::NotifierOptions& notifier_options,
+        const std::string& client_info,
+        const scoped_refptr<ObserverListThreadSafe<SyncNotifierObserver> >&
+            observers);
+    ~WorkerThreadVars();
+
+   private:
+    scoped_ptr<net::HostResolver> host_resolver_;
+    net::CertVerifier cert_verifier_;
+
+   public:
+    // This needs to be initialized after |host_resolver_| and
+    // |cert_verifier_|.
+    InvalidationNotifier invalidation_notifier;
+
+   private:
+    ObserverRouter observer_router_;
+
+    DISALLOW_COPY_AND_ASSIGN(WorkerThreadVars);
+  };
+
+  MessageLoop* worker_message_loop();
+
+  void CreateWorkerThreadVars(
+      const notifier::NotifierOptions& notifier_options,
+      const std::string& client_info);
+  void DestroyWorkerThreadVars();
+
+  // Equivalents of the public functions that are run on the worker
+  // thread.
+  void SetStateOnWorkerThread(const std::string& state);
+  void UpdateCredentialsOnWorkerThread(const std::string& email,
+                                       const std::string& token);
+  void UpdateEnabledTypesOnWorkerThread(const syncable::ModelTypeSet& types);
+
+  MessageLoop* parent_message_loop_;
+
+  scoped_refptr<ObserverListThreadSafe<SyncNotifierObserver> > observers_;
+
+  base::Thread worker_thread_;
+  // Created and destroyed on the worker thread.  Not a scoped_ptr as
+  // it's better to leak memory than to delete on the wrong thread.
+  // Created by CreateWorkerThreadVars() and destroyed by
+  // DestroyWorkerThreadVars().
+  WorkerThreadVars* worker_thread_vars_;
+
+  DISALLOW_COPY_AND_ASSIGN(NonBlockingInvalidationNotifier);
+};
+
+}  // namespace sync_notifier
+
+// We own our worker thread, so we don't need to be ref-counted.
+DISABLE_RUNNABLE_METHOD_REFCOUNT(
+    sync_notifier::NonBlockingInvalidationNotifier);
+
+#endif  // CHROME_BROWSER_SYNC_NOTIFIER_NON_BLOCKING_INVALIDATION_NOTIFIER_H_
diff --git a/chrome/browser/sync/notifier/non_blocking_invalidation_notifier_unittest.cc b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier_unittest.cc
new file mode 100644
index 0000000..64b78e3
--- /dev/null
+++ b/chrome/browser/sync/notifier/non_blocking_invalidation_notifier_unittest.cc
@@ -0,0 +1,59 @@
+// Copyright (c) 2011 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h"
+
+#include "base/message_loop.h"
+#include "base/scoped_ptr.h"
+#include "chrome/browser/sync/notifier/mock_sync_notifier_observer.h"
+#include "chrome/browser/sync/syncable/model_type.h"
+#include "chrome/browser/sync/syncable/model_type_payload_map.h"
+#include "jingle/notifier/base/fake_base_task.h"
+#include "testing/gmock/include/gmock/gmock.h"
+#include "testing/gtest/include/gtest/gtest.h"
+
+namespace sync_notifier {
+
+namespace {
+
+using ::testing::InSequence;
+using ::testing::StrictMock;
+
+class NonBlockingInvalidationNotifierTest : public testing::Test {
+ public:
+  NonBlockingInvalidationNotifierTest()
+      : invalidation_notifier_(notifier::NotifierOptions(),
+                               "fake_client_info") {}
+
+ protected:
+  virtual void SetUp() {
+    invalidation_notifier_.AddObserver(&mock_observer_);
+  }
+
+  virtual void TearDown() {
+    invalidation_notifier_.RemoveObserver(&mock_observer_);
+  }
+
+  MessageLoop message_loop_;
+  NonBlockingInvalidationNotifier invalidation_notifier_;
+  StrictMock<MockSyncNotifierObserver> mock_observer_;
+  notifier::FakeBaseTask fake_base_task_;
+};
+
+TEST_F(NonBlockingInvalidationNotifierTest, Basic) {
+  syncable::ModelTypeSet types;
+  types.insert(syncable::BOOKMARKS);
+  types.insert(syncable::AUTOFILL);
+
+  invalidation_notifier_.SetState("fake_state");
+  invalidation_notifier_.UpdateCredentials("foo@bar.com", "fake_token");
+  invalidation_notifier_.UpdateEnabledTypes(types);
+}
+
+// TODO(akalin): Add synchronous operations for testing to
+// NonBlockingInvalidationNotifierTest and use that to test it.
+
+}  // namespace
+
+}  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/server_notifier_thread.cc b/chrome/browser/sync/notifier/server_notifier_thread.cc
deleted file mode 100644
index 9c980275..0000000
--- a/chrome/browser/sync/notifier/server_notifier_thread.cc
+++ /dev/null
@@ -1,152 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include "chrome/browser/sync/notifier/server_notifier_thread.h"
-
-#include <string>
-#include <vector>
-
-#include "base/logging.h"
-#include "chrome/browser/sync/notifier/chrome_invalidation_client.h"
-#include "chrome/browser/sync/protocol/service_constants.h"
-#include "jingle/notifier/base/notifier_options.h"
-#include "jingle/notifier/listener/notification_defines.h"
-#include "talk/xmpp/xmppclient.h"
-#include "talk/xmpp/xmppclientsettings.h"
-
-namespace sync_notifier {
-
-ServerNotifierThread::ServerNotifierThread(
-    const notifier::NotifierOptions& notifier_options,
-    const std::string& client_info, StateWriter* state_writer)
-    : notifier::MediatorThreadImpl(notifier_options),
-      client_info_(client_info),
-      state_writers_(new ObserverListThreadSafe<StateWriter>()),
-      state_writer_(state_writer) {
-  DCHECK_EQ(notifier::NOTIFICATION_SERVER,
-            notifier_options.notification_method);
-  DCHECK(state_writer_);
-  state_writers_->AddObserver(state_writer_);
-}
-
-ServerNotifierThread::~ServerNotifierThread() {}
-
-void ServerNotifierThread::SetState(const std::string& state) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  state_ = state;
-}
-
-void ServerNotifierThread::UpdateEnabledTypes(
-    const syncable::ModelTypeSet& types) {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-
-  worker_message_loop()->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(
-          this, &ServerNotifierThread::DoUpdateEnabledTypes, types));
-
-  worker_message_loop()->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(
-          this, &ServerNotifierThread::RegisterEnabledTypes));
-}
-
-void ServerNotifierThread::Logout() {
-  DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  state_writers_->RemoveObserver(state_writer_);
-  state_writer_ = NULL;
-  worker_message_loop()->PostTask(
-      FROM_HERE,
-      NewRunnableMethod(this,
-                        &ServerNotifierThread::StopInvalidationListener));
-  MediatorThreadImpl::Logout();
-}
-
-void ServerNotifierThread::OnConnect(
-    base::WeakPtr<talk_base::Task> base_task) {
-  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  // Sets |base_task_|.
-  MediatorThreadImpl::OnConnect(base_task);
-  if (chrome_invalidation_client_.get()) {
-    // If we already have an invalidation client, simply change the
-    // base task.
-    chrome_invalidation_client_->ChangeBaseTask(base_task_);
-  } else {
-    // Otherwise, create the invalidation client.
-    chrome_invalidation_client_.reset(new ChromeInvalidationClient());
-
-    // TODO(akalin): Make cache_guid() part of the client ID.  If we do
-    // so and we somehow propagate it up to the server somehow, we can
-    // make it so that we won't receive any notifications that were
-    // generated from our own changes.
-    const std::string kClientId = "server_notifier_thread";
-    // It's okay to read/write |state_| here, since it was written
-    // before the worker thread was created and is never read/written
-    // to by the parent thread after that.
-    chrome_invalidation_client_->Start(
-        kClientId, client_info_, state_, this, this, base_task_);
-    state_.clear();
-    RegisterEnabledTypes();
-  }
-}
-
-void ServerNotifierThread::OnInvalidate(
-    syncable::ModelType model_type,
-    const std::string& payload) {
-  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  DCHECK_GE(model_type, syncable::FIRST_REAL_MODEL_TYPE);
-  DCHECK_LT(model_type, syncable::MODEL_TYPE_COUNT);
-  VLOG(1) << "OnInvalidate: " << syncable::ModelTypeToString(model_type);
-
-  syncable::ModelTypeBitSet model_types;
-  model_types[model_type] = true;
-  notifier::Notification notification;
-  notification.channel = model_types.to_string();
-  notification.data = payload;
-  observers_->Notify(&Observer::OnIncomingNotification, notification);
-}
-
-void ServerNotifierThread::OnInvalidateAll() {
-  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  VLOG(1) << "OnInvalidateAll";
-
-  syncable::ModelTypeBitSet model_types;
-  model_types.set();  // InvalidateAll, so set all datatypes to true.
-  notifier::Notification notification;
-  notification.channel = model_types.to_string();
-  notification.data = std::string();  // No payload.
-  observers_->Notify(&Observer::OnIncomingNotification, notification);
-}
-
-void ServerNotifierThread::WriteState(const std::string& state) {
-  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  VLOG(1) << "WriteState";
-  state_writers_->Notify(&StateWriter::WriteState, state);
-}
-
-void ServerNotifierThread::SendNotification(
-    const notifier::Notification& notification) {
-  NOTREACHED();
-}
-
-void ServerNotifierThread::DoUpdateEnabledTypes(
-    const syncable::ModelTypeSet& types) {
-  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  enabled_types_ = types;
-}
-
-void ServerNotifierThread::RegisterEnabledTypes() {
-  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  if (!chrome_invalidation_client_.get()) {
-    return;
-  }
-  chrome_invalidation_client_->RegisterTypes(enabled_types_);
-}
-
-void ServerNotifierThread::StopInvalidationListener() {
-  DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  chrome_invalidation_client_.reset();
-}
-
-}  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/server_notifier_thread.h b/chrome/browser/sync/notifier/server_notifier_thread.h
deleted file mode 100644
index e09811e..0000000
--- a/chrome/browser/sync/notifier/server_notifier_thread.h
+++ /dev/null
@@ -1,110 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-//
-// This class is the (hackish) way to use the XMPP parts of
-// MediatorThread for server-issued notifications.
-//
-// TODO(akalin): Separate this code from notifier::MediatorThreadImpl
-// and combine it with InvalidationNotifier.
-
-#ifndef CHROME_BROWSER_SYNC_NOTIFIER_SERVER_NOTIFIER_THREAD_H_
-#define CHROME_BROWSER_SYNC_NOTIFIER_SERVER_NOTIFIER_THREAD_H_
-#pragma once
-
-#include <string>
-#include <vector>
-
-#include "base/observer_list_threadsafe.h"
-#include "base/ref_counted.h"
-#include "base/scoped_ptr.h"
-#include "chrome/browser/sync/notifier/chrome_invalidation_client.h"
-#include "chrome/browser/sync/notifier/state_writer.h"
-#include "chrome/browser/sync/syncable/model_type.h"
-#include "jingle/notifier/listener/mediator_thread_impl.h"
-
-namespace notifier {
-struct NotifierOptions;
-}
-
-namespace sync_notifier {
-
-class ServerNotifierThread
-    : public notifier::MediatorThreadImpl,
-      public ChromeInvalidationClient::Listener,
-      public StateWriter {
- public:
-  // Does not take ownership of |state_writer| (which may not
-  // be NULL).
-  explicit ServerNotifierThread(
-      const notifier::NotifierOptions& notifier_options,
-      const std::string& client_info,
-      StateWriter* state_writer);
-
-  virtual ~ServerNotifierThread();
-
-  // Should be called exactly once before Login().
-  void SetState(const std::string& state);
-
-  // Must be called after Start() is called and before Logout() is
-  // called.
-  void UpdateEnabledTypes(const syncable::ModelTypeSet& types);
-
-  // Overridden to stop the invalidation listener.
-  virtual void Logout();
-
-  // Overridden to do actual login.
-  virtual void OnConnect(base::WeakPtr<talk_base::Task> base_task);
-
-  // ChromeInvalidationClient::Listener implementation.
-  // We pass on two pieces of information to observers through the
-  // Notification.
-  // - the model type being invalidated, through the Notification's
-  //       |channel| member.
-  // - the invalidation payload for that model type, through the
-  //       Notification's |data| member.
-  //
-  // TODO(akalin): Remove this hack once we merge with
-  // InvalidationNotifier.
-  virtual void OnInvalidate(syncable::ModelType model_type,
-                            const std::string& payload);
-  virtual void OnInvalidateAll();
-
-  // StateWriter implementation.
-  virtual void WriteState(const std::string& state);
-
- private:
-  // Made private because this function shouldn't be called.
-  virtual void SendNotification(const notifier::Notification& data);
-
-  // Posted to the worker thread by UpdateEnabledTypes().
-  void DoUpdateEnabledTypes(const syncable::ModelTypeSet& types);
-
-  // Posted to the worker thread by UpdateEnabledTypes() and also
-  // called by OnConnect().
-  void RegisterEnabledTypes();
-
-  // Posted to the worker thread by Logout().
-  void StopInvalidationListener();
-
-  const std::string client_info_;
-  // Hack to get the nice thread-safe behavior for |state_writer_|.
-  scoped_refptr<ObserverListThreadSafe<StateWriter> > state_writers_;
-  // We still need to keep |state_writer_| around to remove it from
-  // |state_writers_|.
-  StateWriter* state_writer_;
-  scoped_ptr<ChromeInvalidationClient> chrome_invalidation_client_;
-
-  // The state to pass to |chrome_invalidation_client_|.
-  std::string state_;
-
-  // The types to register.  Should only be accessed on the worker
-  // thread.
-  syncable::ModelTypeSet enabled_types_;
-};
-
-}  // namespace sync_notifier
-
-DISABLE_RUNNABLE_METHOD_REFCOUNT(sync_notifier::ServerNotifierThread);
-
-#endif  // CHROME_BROWSER_SYNC_NOTIFIER_SERVER_NOTIFIER_THREAD_H_
diff --git a/chrome/browser/sync/notifier/server_notifier_thread_unittest.cc b/chrome/browser/sync/notifier/server_notifier_thread_unittest.cc
deleted file mode 100644
index 15072bd..0000000
--- a/chrome/browser/sync/notifier/server_notifier_thread_unittest.cc
+++ /dev/null
@@ -1,155 +0,0 @@
-// Copyright (c) 2011 The Chromium Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style license that can be
-// found in the LICENSE file.
-
-#include <string>
-
-#include "base/compiler_specific.h"
-#include "base/message_loop.h"
-#include "base/scoped_ptr.h"
-#include "base/task.h"
-#include "chrome/browser/sync/notifier/server_notifier_thread.h"
-#include "chrome/browser/sync/notifier/state_writer.h"
-#include "jingle/notifier/base/fake_base_task.h"
-#include "jingle/notifier/base/notifier_options.h"
-#include "talk/xmpp/xmppclientsettings.h"
-#include "testing/gmock/include/gmock/gmock.h"
-#include "testing/gtest/include/gtest/gtest.h"
-
-namespace sync_notifier {
-
-class FakeServerNotifierThread : public ServerNotifierThread {
- public:
-   FakeServerNotifierThread()
-      : ServerNotifierThread(notifier::NotifierOptions(),
-                             "fake client info",
-                             ALLOW_THIS_IN_INITIALIZER_LIST(this)) {}
-
-  virtual ~FakeServerNotifierThread() {}
-
-  virtual void Start() {
-    DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-    ServerNotifierThread::Start();
-    worker_message_loop()->PostTask(
-        FROM_HERE,
-        NewRunnableMethod(this,
-                          &FakeServerNotifierThread::InitFakes));
-  }
-
-  virtual void Logout() {
-    DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-    worker_message_loop()->PostTask(
-        FROM_HERE,
-        NewRunnableMethod(this,
-                          &FakeServerNotifierThread::DestroyFakes));
-    ServerNotifierThread::Logout();
-  }
-
-  // We prevent the real connection attempt from happening and use
-  // SimulateConnection()/SimulateDisconnection() instead.
-  virtual void Login(const buzz::XmppClientSettings& settings) {
-    DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-  }
-
-  // We pass ourselves as the StateWriter in the constructor, so shim
-  // out WriteState() to prevent an infinite loop.
-  virtual void WriteState(const std::string& state) {
-    DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-  }
-
-  void SimulateConnect() {
-    DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-    worker_message_loop()->PostTask(
-        FROM_HERE,
-        NewRunnableMethod(this,
-                          &FakeServerNotifierThread::DoSimulateConnect));
-  }
-
-  void SimulateDisconnect() {
-    DCHECK_EQ(MessageLoop::current(), parent_message_loop_);
-    worker_message_loop()->PostTask(
-        FROM_HERE,
-        NewRunnableMethod(this,
-                          &FakeServerNotifierThread::DoSimulateDisconnect));
-  }
-
- private:
-  void InitFakes() {
-    DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-    fake_base_task_.reset(new notifier::FakeBaseTask());
-  }
-
-  void DestroyFakes() {
-    DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-    fake_base_task_.reset();
-  }
-
-  void DoSimulateConnect() {
-    DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-    OnConnect(fake_base_task_->AsWeakPtr());
-  }
-
-  void DoSimulateDisconnect() {
-    DCHECK_EQ(MessageLoop::current(), worker_message_loop());
-    OnDisconnect();
-  }
-
-  // Used only on the worker thread.
-  scoped_ptr<notifier::FakeBaseTask> fake_base_task_;
-};
-
-}  // namespace sync_notifier
-
-DISABLE_RUNNABLE_METHOD_REFCOUNT(sync_notifier::FakeServerNotifierThread);
-
-namespace sync_notifier {
-
-namespace {
-
-class ServerNotifierThreadTest : public testing::Test {
- protected:
-  MessageLoop message_loop_;
-};
-
-syncable::ModelTypeSet GetTypeSetWithAllTypes() {
-  syncable::ModelTypeSet all_types;
-
-  for (int i = syncable::FIRST_REAL_MODEL_TYPE;
-       i < syncable::MODEL_TYPE_COUNT; ++i) {
-    syncable::ModelType model_type = syncable::ModelTypeFromInt(i);
-    all_types.insert(model_type);
-  }
-
-  return all_types;
-}
-
-TEST_F(ServerNotifierThreadTest, Basic) {
-  FakeServerNotifierThread server_notifier_thread;
-
-  server_notifier_thread.Start();
-  server_notifier_thread.Login(buzz::XmppClientSettings());
-  server_notifier_thread.SimulateConnect();
-  server_notifier_thread.Logout();
-}
-
-TEST_F(ServerNotifierThreadTest, LoginLogout) {
-  FakeServerNotifierThread server_notifier_thread;
-
-  server_notifier_thread.Start();
-  server_notifier_thread.Login(buzz::XmppClientSettings());
-  server_notifier_thread.Logout();
-}
-
-TEST_F(ServerNotifierThreadTest, DisconnectLogout) {
-  FakeServerNotifierThread server_notifier_thread;
-
-  server_notifier_thread.Start();
-  server_notifier_thread.Login(buzz::XmppClientSettings());
-  server_notifier_thread.SimulateConnect();
-  server_notifier_thread.SimulateDisconnect();
-  server_notifier_thread.Logout();
-}
-
-}  // namespace
-
-}  // namespace sync_notifier
diff --git a/chrome/browser/sync/notifier/sync_notifier_factory.cc b/chrome/browser/sync/notifier/sync_notifier_factory.cc
index 6380e54..204faeb 100644
--- a/chrome/browser/sync/notifier/sync_notifier_factory.cc
+++ b/chrome/browser/sync/notifier/sync_notifier_factory.cc
@@ -9,7 +9,7 @@
 #include "base/command_line.h"
 #include "base/string_number_conversions.h"
 #include "base/string_util.h"
-#include "chrome/browser/sync/notifier/invalidation_notifier.h"
+#include "chrome/browser/sync/notifier/non_blocking_invalidation_notifier.h"
 #include "chrome/browser/sync/notifier/p2p_notifier.h"
 #include "chrome/browser/sync/notifier/sync_notifier.h"
 #include "chrome/common/chrome_switches.h"
@@ -88,7 +88,7 @@ SyncNotifier* CreateDefaultSyncNotifier(const CommandLine& command_line,
     return new P2PNotifier(notifier_options);
   }
 
-  return new InvalidationNotifier(notifier_options, client_info);
+  return new NonBlockingInvalidationNotifier(notifier_options, client_info);
 }
 }  // namespace
 
diff --git a/chrome/browser/sync/notifier/sync_notifier_observer.h b/chrome/browser/sync/notifier/sync_notifier_observer.h
index 79439ce..f2d7cda 100644
--- a/chrome/browser/sync/notifier/sync_notifier_observer.h
+++ b/chrome/browser/sync/notifier/sync_notifier_observer.h
@@ -2,9 +2,9 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
-
 #ifndef CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_OBSERVER_H_
 #define CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_OBSERVER_H_
+#pragma once
 
 #include <string>
 
@@ -25,5 +25,7 @@ class SyncNotifierObserver {
   // sync notifier implementation.
   virtual void StoreState(const std::string& state) = 0;
 };
+
 }  // namespace sync_notifier
+
 #endif  // CHROME_BROWSER_SYNC_NOTIFIER_SYNC_NOTIFIER_OBSERVER_H_
diff --git a/chrome/browser/sync/tools/sync_listen_notifications.cc b/chrome/browser/sync/tools/sync_listen_notifications.cc
index e0e1eeb..4030a12 100644
--- a/chrome/browser/sync/tools/sync_listen_notifications.cc
+++ b/chrome/browser/sync/tools/sync_listen_notifications.cc
@@ -74,8 +74,8 @@ int main(int argc, char* argv[]) {
     std::printf("Usage: %s --email=foo@bar.com --token=token\n\n"
                 "See sync_notifier_factory.cc for more switches.\n\n"
                 "Run chrome and set a breakpoint on "
-                "SyncNotifierImpl::UpdateCredentials() after\n"
-                "logging into sync to get the token to pass into this "
+                "sync_api::SyncManager::SyncInternal::UpdateCredentials() "
+                "after logging into sync to get the token to pass into this "
                 "utility.\n",
                 argv[0]);
     return -1;
diff --git a/chrome/chrome.gyp b/chrome/chrome.gyp
index 5645acd..ece8c95 100644
--- a/chrome/chrome.gyp
+++ b/chrome/chrome.gyp
@@ -794,12 +794,12 @@
         'browser/sync/notifier/invalidation_notifier.cc',
         'browser/sync/notifier/invalidation_util.cc',
         'browser/sync/notifier/invalidation_util.h',
+        'browser/sync/notifier/non_blocking_invalidation_notifier.h',
+        'browser/sync/notifier/non_blocking_invalidation_notifier.cc',
         'browser/sync/notifier/p2p_notifier.h',
         'browser/sync/notifier/p2p_notifier.cc',
         'browser/sync/notifier/registration_manager.cc',
         'browser/sync/notifier/registration_manager.h',
-        'browser/sync/notifier/server_notifier_thread.cc',
-        'browser/sync/notifier/server_notifier_thread.h',
         'browser/sync/notifier/state_writer.h',
         'browser/sync/notifier/sync_notifier.h',
         'browser/sync/notifier/sync_notifier_factory.h',
diff --git a/chrome/chrome_tests.gypi b/chrome/chrome_tests.gypi
index dc64c17..1d7a98b 100644
--- a/chrome/chrome_tests.gypi
+++ b/chrome/chrome_tests.gypi
@@ -263,6 +263,21 @@
       ],
     },
     {
+      'target_name': 'test_support_sync_notifier',
+      'type': '<(library)',
+      'dependencies': [
+        '../testing/gmock.gyp:gmock',
+        'sync_notifier',
+      ],
+      'include_dirs': [
+        '..',
+      ],
+      'sources': [
+        'browser/sync/notifier/mock_sync_notifier_observer.cc',
+        'browser/sync/notifier/mock_sync_notifier_observer.h',
+      ],
+    },
+    {
       'target_name': 'test_support_unit',
       'type': '<(library)',
       'dependencies': [
@@ -2855,8 +2870,9 @@
         'browser/sync/notifier/cache_invalidation_packet_handler_unittest.cc',
         'browser/sync/notifier/chrome_invalidation_client_unittest.cc',
         'browser/sync/notifier/chrome_system_resources_unittest.cc',
+        'browser/sync/notifier/invalidation_notifier_unittest.cc',
+        'browser/sync/notifier/non_blocking_invalidation_notifier_unittest.cc',
         'browser/sync/notifier/registration_manager_unittest.cc',
-        'browser/sync/notifier/server_notifier_thread_unittest.cc',
         'browser/sync/profile_sync_factory_mock.h',
         'browser/sync/protocol/proto_enum_conversions_unittest.cc',
         'browser/sync/protocol/proto_value_conversions_unittest.cc',
@@ -2916,6 +2932,7 @@
         'sync_notifier',
         'test_support_common',
         'test_support_sync',
+        'test_support_sync_notifier',
         'test_support_unit',
       ],
       'conditions': [