Revert 66513 - sync: remove event_sys from ServerConnectionManager and SyncerThread

BUG=26339
TEST=sync_unit_tests

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

TBR=tim@chromium.org
Review URL: http://codereview.chromium.org/5228002

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

6 files changed, 95 insertions, 84 deletions

diff --git a/chrome/browser/sync/engine/net/server_connection_manager.cc b/chrome/browser/sync/engine/net/server_connection_manager.cc
index b744874..41e79f4 100644
--- a/chrome/browser/sync/engine/net/server_connection_manager.cc
+++ b/chrome/browser/sync/engine/net/server_connection_manager.cc
@@ -143,22 +143,22 @@ ServerConnectionManager::ServerConnectionManager(
       proto_sync_path_(kSyncServerSyncPath),
       get_time_path_(kSyncServerGetTimePath),
       error_count_(0),
+      channel_(new Channel(shutdown_event)),
       server_status_(HttpResponse::NONE),
       server_reachable_(false),
-      reset_count_(0) {
+      reset_count_(0),
+      terminate_all_io_(false) {
 }
 
 ServerConnectionManager::~ServerConnectionManager() {
-  FOR_EACH_OBSERVER(ServerConnectionEventListener, listeners_,
-                    OnServerConnectionEvent(shutdown_event));
+  delete channel_;
 }
 
 void ServerConnectionManager::NotifyStatusChanged() {
   ServerConnectionEvent event = { ServerConnectionEvent::STATUS_CHANGED,
                                   server_status_,
                                   server_reachable_ };
-  FOR_EACH_OBSERVER(ServerConnectionEventListener, listeners_,
-                    OnServerConnectionEvent(event));
+  channel_->NotifyListeners(event);
 }
 
 bool ServerConnectionManager::PostBufferWithCachedAuth(
@@ -253,6 +253,13 @@ bool ServerConnectionManager::CheckServerReachable() {
   return server_is_reachable;
 }
 
+void ServerConnectionManager::kill() {
+  {
+    AutoLock lock(terminate_all_io_mutex_);
+    terminate_all_io_ = true;
+  }
+}
+
 void ServerConnectionManager::ResetConnection() {
   base::subtle::NoBarrier_AtomicIncrement(&reset_count_, 1);
 }
diff --git a/chrome/browser/sync/engine/net/server_connection_manager.h b/chrome/browser/sync/engine/net/server_connection_manager.h
index 58f3985..0363993 100644
--- a/chrome/browser/sync/engine/net/server_connection_manager.h
+++ b/chrome/browser/sync/engine/net/server_connection_manager.h
@@ -10,11 +10,11 @@
 #include <string>
 
 #include "base/atomicops.h"
-#include "base/gtest_prod_util.h"
 #include "base/lock.h"
-#include "base/observer_list.h"
 #include "base/string_util.h"
 #include "chrome/browser/sync/syncable/syncable_id.h"
+#include "chrome/common/deprecated/event_sys.h"
+#include "chrome/common/deprecated/event_sys-inl.h"
 #include "chrome/common/net/http_return.h"
 
 namespace syncable {
@@ -115,19 +115,15 @@ struct ServerConnectionEvent {
     STATUS_CHANGED
   };
 
+  static inline bool IsChannelShutdownEvent(const EventType& event) {
+    return SHUTDOWN == event.what_happened;
+  }
+
   WhatHappened what_happened;
   HttpResponse::ServerConnectionCode connection_code;
   bool server_reachable;
 };
 
-class ServerConnectionEventListener {
- public:
-  // TODO(tim): Consider splitting this up to multiple callbacks.
-  virtual void OnServerConnectionEvent(const ServerConnectionEvent& event) = 0;
- protected:
-  virtual ~ServerConnectionEventListener() {}
-};
-
 class ServerConnectionManager;
 // A helper class that automatically notifies when the status changes.
 // TODO(tim): This class shouldn't be exposed outside of the implementation,
@@ -153,6 +149,7 @@ class ScopedServerStatusWatcher {
 //  one instance for every server that you need to talk to.
 class ServerConnectionManager {
  public:
+  typedef EventChannel<ServerConnectionEvent, Lock> Channel;
 
   // buffer_in - will be POSTed
   // buffer_out - string will be overwritten with response
@@ -240,6 +237,11 @@ class ServerConnectionManager {
   // Updates status and broadcasts events on change.
   bool CheckServerReachable();
 
+  // Signal the shutdown event to notify listeners.
+  virtual void kill();
+
+  inline Channel* channel() const { return channel_; }
+
   inline std::string user_agent() const { return user_agent_; }
 
   inline HttpResponse::ServerConnectionCode server_status() const {
@@ -266,6 +268,11 @@ class ServerConnectionManager {
 
   std::string GetServerHost() const;
 
+  bool terminate_all_io() const {
+    AutoLock lock(terminate_all_io_mutex_);
+    return terminate_all_io_;
+  }
+
   // Factory method to create a Post object we can use for communication with
   // the server.
   virtual Post* MakePost() {
@@ -288,14 +295,6 @@ class ServerConnectionManager {
     return auth_token_;
   }
 
-  void AddListener(ServerConnectionEventListener* listener) {
-    listeners_.AddObserver(listener);
-  }
-
-  void RemoveListener(ServerConnectionEventListener* listener) {
-    listeners_.RemoveObserver(listener);
-  }
-
  protected:
   inline std::string proto_sync_path() const {
     AutoLock lock(path_mutex_);
@@ -349,6 +348,8 @@ class ServerConnectionManager {
   Lock error_count_mutex_;  // Protects error_count_
   int error_count_;  // Tracks the number of connection errors.
 
+  Channel* const channel_;
+
   // Volatile so various threads can call server_status() without
   // synchronization.
   volatile HttpResponse::ServerConnectionCode server_status_;
@@ -357,16 +358,15 @@ class ServerConnectionManager {
   // A counter that is incremented everytime ResetAuthStatus() is called.
   volatile base::subtle::AtomicWord reset_count_;
 
-  ObserverList<ServerConnectionEventListener> listeners_;
-
  private:
-  FRIEND_TEST_ALL_PREFIXES(SyncerThreadWithSyncerTest, AuthInvalid);
   friend class Post;
   friend class ScopedServerStatusWatcher;
 
   void NotifyStatusChanged();
   void ResetConnection();
 
+  mutable Lock terminate_all_io_mutex_;
+  bool terminate_all_io_;  // When set to true, terminate all connections asap.
   DISALLOW_COPY_AND_ASSIGN(ServerConnectionManager);
 };
 
diff --git a/chrome/browser/sync/engine/syncapi.cc b/chrome/browser/sync/engine/syncapi.cc
index 1c5f9f9..bdd7be9 100644
--- a/chrome/browser/sync/engine/syncapi.cc
+++ b/chrome/browser/sync/engine/syncapi.cc
@@ -67,7 +67,6 @@ using browser_sync::ModelSafeRoutingInfo;
 using browser_sync::ModelSafeWorker;
 using browser_sync::ModelSafeWorkerRegistrar;
 using browser_sync::ServerConnectionEvent;
-using browser_sync::ServerConnectionEventListener;
 using browser_sync::SyncEngineEvent;
 using browser_sync::SyncEngineEventListener;
 using browser_sync::Syncer;
@@ -933,8 +932,7 @@ class SyncManager::SyncInternal
       public TalkMediator::Delegate,
       public sync_notifier::StateWriter,
       public browser_sync::ChannelEventHandler<syncable::DirectoryChangeEvent>,
-      public SyncEngineEventListener,
-      public ServerConnectionEventListener {
+      public SyncEngineEventListener {
   static const int kDefaultNudgeDelayMilliseconds;
   static const int kPreferencesNudgeDelayMilliseconds;
  public:
@@ -1004,7 +1002,7 @@ class SyncManager::SyncInternal
       const syncable::DirectoryChangeEvent& event);
 
   // Listens for notifications from the ServerConnectionManager
-  virtual void OnServerConnectionEvent(const ServerConnectionEvent& event);
+  void HandleServerConnectionEvent(const ServerConnectionEvent& event);
 
   // Open the directory named with username_for_share
   bool OpenDirectory();
@@ -1234,6 +1232,9 @@ class SyncManager::SyncInternal
   scoped_ptr<browser_sync::ChannelHookup<syncable::DirectoryChangeEvent> >
       dir_change_hookup_;
 
+  // Event listener hookup for the ServerConnectionManager.
+  scoped_ptr<EventListenerHookup> connection_manager_hookup_;
+
   // The sync dir_manager to which we belong.
   SyncManager* const sync_manager_;
 
@@ -1370,7 +1371,9 @@ bool SyncManager::SyncInternal::Init(
   connection_manager_.reset(new SyncAPIServerConnectionManager(
       sync_server_and_path, port, use_ssl, user_agent, post_factory));
 
-  connection_manager_->AddListener(this);
+  connection_manager_hookup_.reset(
+      NewEventListenerHookup(connection_manager()->channel(), this,
+          &SyncManager::SyncInternal::HandleServerConnectionEvent));
 
   net::NetworkChangeNotifier::AddObserver(this);
   // TODO(akalin): CheckServerReachable() can block, which may cause jank if we
@@ -1741,7 +1744,7 @@ void SyncManager::SyncInternal::Shutdown() {
 
   net::NetworkChangeNotifier::RemoveObserver(this);
 
-  connection_manager_->RemoveListener(this);
+  connection_manager_hookup_.reset();
 
   if (dir_manager()) {
     dir_manager()->FinalSaveChangesForAll();
@@ -1821,7 +1824,7 @@ void SyncManager::SyncInternal::HandleTransactionCompleteChangeEvent(
   }
 }
 
-void SyncManager::SyncInternal::OnServerConnectionEvent(
+void SyncManager::SyncInternal::HandleServerConnectionEvent(
     const ServerConnectionEvent& event) {
   allstatus_.HandleServerConnectionEvent(event);
   if (event.what_happened == ServerConnectionEvent::STATUS_CHANGED) {
@@ -1837,9 +1840,6 @@ void SyncManager::SyncInternal::OnServerConnectionEvent(
         observer_->OnAuthError(AuthError(AuthError::INVALID_GAIA_CREDENTIALS));
       }
     }
-  } else {
-    DCHECK_EQ(ServerConnectionEvent::SHUTDOWN, event.what_happened);
-    connection_manager_->RemoveListener(this);
   }
 }
 
diff --git a/chrome/browser/sync/engine/syncer_thread.cc b/chrome/browser/sync/engine/syncer_thread.cc
index dfb626a..f5c2f80 100644
--- a/chrome/browser/sync/engine/syncer_thread.cc
+++ b/chrome/browser/sync/engine/syncer_thread.cc
@@ -77,39 +77,11 @@ void SyncerThread::NudgeSyncer(
   NudgeSyncImpl(milliseconds_from_now, source, model_types);
 }
 
-// Sets |*connected| to false if it is currently true but |code| suggests that
-// the current network configuration and/or auth state cannot be used to make
-// forward progress, and user intervention (e.g changing server URL or auth
-// credentials) is likely necessary.  If |*connected| is false, set it to true
-// if |code| suggests that we just recently made healthy contact with the
-// server.
-static inline void CheckConnected(bool* connected,
-                                  HttpResponse::ServerConnectionCode code,
-                                  ConditionVariable* condvar) {
-  if (*connected) {
-    // Note, be careful when adding cases here because if the SyncerThread
-    // thinks there is no valid connection as determined by this method, it
-    // will drop out of *all* forward progress sync loops (it won't poll and it
-    // will queue up Talk notifications but not actually call SyncShare) until
-    // some external action causes a ServerConnectionManager to broadcast that
-    // a valid connection has been re-established.
-    if (HttpResponse::CONNECTION_UNAVAILABLE == code ||
-        HttpResponse::SYNC_AUTH_ERROR == code) {
-      *connected = false;
-      condvar->Broadcast();
-    }
-  } else {
-    if (HttpResponse::SERVER_CONNECTION_OK == code) {
-      *connected = true;
-      condvar->Broadcast();
-    }
-  }
-}
-
 SyncerThread::SyncerThread(sessions::SyncSessionContext* context)
     : thread_main_started_(false, false),
       thread_("SyncEngine_SyncerThread"),
       vault_field_changed_(&lock_),
+      conn_mgr_hookup_(NULL),
       syncer_short_poll_interval_seconds_(kDefaultShortPollIntervalSeconds),
       syncer_long_poll_interval_seconds_(kDefaultLongPollIntervalSeconds),
       syncer_polling_interval_(kDefaultShortPollIntervalSeconds),
@@ -118,17 +90,12 @@ SyncerThread::SyncerThread(sessions::SyncSessionContext* context)
       disable_idle_detection_(false) {
   DCHECK(context);
 
-  if (context->connection_manager()) {
-    context->connection_manager()->AddListener(this);
-    CheckConnected(&vault_.connected_,
-                   context->connection_manager()->server_status(),
-                   &vault_field_changed_);
-  }
+  if (context->connection_manager())
+    WatchConnectionManager(context->connection_manager());
 }
 
 SyncerThread::~SyncerThread() {
-  if (session_context_->connection_manager())
-    session_context_->connection_manager()->RemoveListener(this);
+  conn_mgr_hookup_.reset();
   delete vault_.syncer_;
   CHECK(!thread_.IsRunning());
 }
@@ -636,14 +603,48 @@ void SyncerThread::CreateSyncer(const std::string& dirname) {
   vault_field_changed_.Broadcast();
 }
 
-void SyncerThread::OnServerConnectionEvent(const ServerConnectionEvent& event) {
+// Sets |*connected| to false if it is currently true but |code| suggests that
+// the current network configuration and/or auth state cannot be used to make
+// forward progress, and user intervention (e.g changing server URL or auth
+// credentials) is likely necessary.  If |*connected| is false, set it to true
+// if |code| suggests that we just recently made healthy contact with the
+// server.
+static inline void CheckConnected(bool* connected,
+                                  HttpResponse::ServerConnectionCode code,
+                                  ConditionVariable* condvar) {
+  if (*connected) {
+    // Note, be careful when adding cases here because if the SyncerThread
+    // thinks there is no valid connection as determined by this method, it
+    // will drop out of *all* forward progress sync loops (it won't poll and it
+    // will queue up Talk notifications but not actually call SyncShare) until
+    // some external action causes a ServerConnectionManager to broadcast that
+    // a valid connection has been re-established.
+    if (HttpResponse::CONNECTION_UNAVAILABLE == code ||
+        HttpResponse::SYNC_AUTH_ERROR == code) {
+      *connected = false;
+      condvar->Broadcast();
+    }
+  } else {
+    if (HttpResponse::SERVER_CONNECTION_OK == code) {
+      *connected = true;
+      condvar->Broadcast();
+    }
+  }
+}
+
+void SyncerThread::WatchConnectionManager(ServerConnectionManager* conn_mgr) {
+  conn_mgr_hookup_.reset(NewEventListenerHookup(conn_mgr->channel(), this,
+                         &SyncerThread::HandleServerConnectionEvent));
+  CheckConnected(&vault_.connected_, conn_mgr->server_status(),
+                 &vault_field_changed_);
+}
+
+void SyncerThread::HandleServerConnectionEvent(
+    const ServerConnectionEvent& event) {
   if (ServerConnectionEvent::STATUS_CHANGED == event.what_happened) {
     AutoLock lock(lock_);
     CheckConnected(&vault_.connected_, event.connection_code,
                    &vault_field_changed_);
-  } else {
-    DCHECK_EQ(ServerConnectionEvent::SHUTDOWN, event.what_happened);
-    session_context_->connection_manager()->RemoveListener(this);
   }
 }
 
diff --git a/chrome/browser/sync/engine/syncer_thread.h b/chrome/browser/sync/engine/syncer_thread.h
index 570e308..a7322be 100644
--- a/chrome/browser/sync/engine/syncer_thread.h
+++ b/chrome/browser/sync/engine/syncer_thread.h
@@ -25,9 +25,11 @@
 #include "chrome/browser/sync/engine/idle_query_linux.h"
 #endif
 #include "chrome/browser/sync/engine/syncer_types.h"
-#include "chrome/browser/sync/engine/net/server_connection_manager.h"
 #include "chrome/browser/sync/sessions/sync_session.h"
 #include "chrome/browser/sync/syncable/model_type.h"
+#include "chrome/common/deprecated/event_sys-inl.h"
+
+class EventListenerHookup;
 
 namespace browser_sync {
 
@@ -38,8 +40,7 @@ class URLFactory;
 struct ServerConnectionEvent;
 
 class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
-                     public sessions::SyncSession::Delegate,
-                     public ServerConnectionEventListener {
+                     public sessions::SyncSession::Delegate {
   FRIEND_TEST_ALL_PREFIXES(SyncerThreadTest, CalculateSyncWaitTime);
   FRIEND_TEST_ALL_PREFIXES(SyncerThreadTest, CalculatePollingWaitTime);
   FRIEND_TEST_ALL_PREFIXES(SyncerThreadWithSyncerTest, Polling);
@@ -102,6 +103,8 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
   explicit SyncerThread(sessions::SyncSessionContext* context);
   virtual ~SyncerThread();
 
+  virtual void WatchConnectionManager(ServerConnectionManager* conn_mgr);
+
   // Starts a syncer thread.
   // Returns true if it creates a thread or if there's currently a thread
   // running and false otherwise.
@@ -236,8 +239,7 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
       const base::TimeDelta& new_interval);
   virtual void OnShouldStopSyncingPermanently();
 
-  // ServerConnectionEventListener implementation.
-  virtual void OnServerConnectionEvent(const ServerConnectionEvent& event);
+  void HandleServerConnectionEvent(const ServerConnectionEvent& event);
 
   void SyncMain(Syncer* syncer);
 
@@ -294,6 +296,8 @@ class SyncerThread : public base::RefCountedThreadSafe<SyncerThread>,
 
   void Notify(SyncEngineEvent::EventCause cause);
 
+  scoped_ptr<EventListenerHookup> conn_mgr_hookup_;
+
   // Modifiable versions of kDefaultLongPollIntervalSeconds which can be
   // updated by the server.
   int syncer_short_poll_interval_seconds_;
diff --git a/chrome/browser/sync/engine/syncer_thread_unittest.cc b/chrome/browser/sync/engine/syncer_thread_unittest.cc
index d49eeb0..b2238eb 100644
--- a/chrome/browser/sync/engine/syncer_thread_unittest.cc
+++ b/chrome/browser/sync/engine/syncer_thread_unittest.cc
@@ -839,8 +839,7 @@ TEST_F(SyncerThreadWithSyncerTest, AuthInvalid) {
   ServerConnectionEvent e = {ServerConnectionEvent::STATUS_CHANGED,
                              HttpResponse::SERVER_CONNECTION_OK,
                              true};
-  FOR_EACH_OBSERVER(ServerConnectionEventListener, connection()->listeners_,
-                    OnServerConnectionEvent(e));
+  connection()->channel()->NotifyListeners(e);
 
   interceptor.WaitForSyncShare(1, TimeDelta::FromSeconds(10));
   EXPECT_FALSE(interceptor.times_sync_occured().empty());