Changes made to TalkMediator so that its functionality is generic (non sync-specific) and can be used by other services needing XMPP notifications. This is in preparatoion of moving these set of classes to common.

BUG=None
TEST=Test Bookmarks sync. The sync unit-tests have been modified accordingly.

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

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

17 files changed, 244 insertions, 78 deletions

diff --git a/chrome/browser/sync/engine/syncapi.cc b/chrome/browser/sync/engine/syncapi.cc
index 331126d..244956e 100644
--- a/chrome/browser/sync/engine/syncapi.cc
+++ b/chrome/browser/sync/engine/syncapi.cc
@@ -46,6 +46,7 @@
 #include "chrome/browser/sync/engine/net/syncapi_server_connection_manager.h"
 #include "chrome/browser/sync/engine/syncer.h"
 #include "chrome/browser/sync/engine/syncer_thread.h"
+#include "chrome/browser/sync/notifier/listener/notification_constants.h"
 #include "chrome/browser/sync/notifier/listener/talk_mediator.h"
 #include "chrome/browser/sync/notifier/listener/talk_mediator_impl.h"
 #include "chrome/browser/sync/protocol/autofill_specifics.pb.h"
@@ -1410,6 +1411,11 @@ bool SyncManager::SyncInternal::Init(
 
   talk_mediator_.reset(new TalkMediatorImpl(notification_method,
                                             invalidate_xmpp_auth_token));
+  if (notification_method == browser_sync::NOTIFICATION_TRANSITIONAL) {
+    talk_mediator_->AddSubscribedServiceUrl(
+        browser_sync::kSyncLegacyServiceUrl);
+  }
+  talk_mediator_->AddSubscribedServiceUrl(browser_sync::kSyncServiceUrl);
   allstatus()->WatchTalkMediator(talk_mediator());
 
   BridgedGaiaAuthenticator* gaia_auth = new BridgedGaiaAuthenticator(
diff --git a/chrome/browser/sync/engine/syncer_thread.cc b/chrome/browser/sync/engine/syncer_thread.cc
index 20ba481..66c412b 100644
--- a/chrome/browser/sync/engine/syncer_thread.cc
+++ b/chrome/browser/sync/engine/syncer_thread.cc
@@ -20,6 +20,7 @@
 #include "chrome/browser/sync/engine/model_safe_worker.h"
 #include "chrome/browser/sync/engine/net/server_connection_manager.h"
 #include "chrome/browser/sync/engine/syncer.h"
+#include "chrome/browser/sync/notifier/listener/notification_constants.h"
 #include "chrome/browser/sync/notifier/listener/talk_mediator.h"
 #include "chrome/browser/sync/notifier/listener/talk_mediator_impl.h"
 #include "chrome/browser/sync/syncable/directory_manager.h"
@@ -632,6 +633,8 @@ void SyncerThread::HandleTalkMediatorEvent(const TalkMediatorEvent& event) {
       p2p_subscribed_ = false;
       break;
     case TalkMediatorEvent::NOTIFICATION_RECEIVED:
+      // TODO(sanjeevr): Check if the service url is a sync URL.
+      // An empty service URL is treated as a legacy sync notification.
       LOG(INFO) << "P2P: Updates on server, pushing syncer";
       if (NULL != vault_.syncer_) {
         NudgeSyncImpl(0, kNotification);
diff --git a/chrome/browser/sync/notifier/listener/listen_task.cc b/chrome/browser/sync/notifier/listener/listen_task.cc
index 48468a8..06f7ff2 100644
--- a/chrome/browser/sync/notifier/listener/listen_task.cc
+++ b/chrome/browser/sync/notifier/listener/listen_task.cc
@@ -40,8 +40,79 @@ int ListenTask::ProcessResponse() {
   scoped_ptr<buzz::XmlElement> response_stanza(MakeIqResult(stanza));
   SendStanza(response_stanza.get());
 
-  // Inform listeners that a notification has been received.
-  SignalUpdateAvailable();
+  // TODO(akalin): Write unittests to cover this.
+  // Extract the service URL and service-specific data from the stanza.
+  // The response stanza has the following format.
+  //  <iq from="{bare_jid}" to="{full_jid}" id="#" type="set">
+  //    <not:getAll xmlns:not="google:notifier">
+  //      <Timestamp long="#" xmlns=""/>
+  //      <Result xmlns="">
+  //        <Id>
+  //          <ServiceUrl data="{service_url}"/>
+  //          <ServiceId data="{service_id}"/>
+  //        </Id>
+  //        <Timestamp long="#"/>
+  //        <Content>
+  //          <Priority int="#"/>
+  //          <ServiceSpecificData data="{service_specific_data}"/>
+  //          <RequireSubscription bool="true"/>
+  //        </Content>
+  //        <State>
+  //          <Type int="#"/>
+  //          <Read bool="{true/false}"/>
+  //        </State>
+  //        <ClientActive bool="{true/false}"/>
+  //      </Result>
+  //    </not:getAll>
+  //  </iq> "
+  // Note that there can be multiple "Result" elements, so we need to loop
+  // through all of them.
+  bool update_signaled = false;
+  const buzz::XmlElement* get_all_element =
+      stanza->FirstNamed(buzz::QName(true, "google:notifier", "getAll"));
+  if (get_all_element) {
+    const buzz::XmlElement* result_element =
+        get_all_element->FirstNamed(
+            buzz::QName(true, buzz::STR_EMPTY, "Result"));
+    while (result_element) {
+      NotificationData notification_data;
+      const buzz::XmlElement* id_element =
+          result_element->FirstNamed(buzz::QName(true, buzz::STR_EMPTY, "Id"));
+      if (id_element) {
+        const buzz::XmlElement* service_url_element =
+            id_element->FirstNamed(
+                buzz::QName(true, buzz::STR_EMPTY, "ServiceUrl"));
+        if (service_url_element) {
+          notification_data.service_url = service_url_element->Attr(
+              buzz::QName(true, buzz::STR_EMPTY, "data"));
+        }
+      }
+      const buzz::XmlElement* content_element =
+          result_element->FirstNamed(
+              buzz::QName(true, buzz::STR_EMPTY, "Content"));
+      if (content_element) {
+        const buzz::XmlElement* service_data_element =
+            content_element->FirstNamed(
+                buzz::QName(true, buzz::STR_EMPTY, "ServiceSpecificData"));
+        if (service_data_element) {
+          notification_data.service_specific_data = service_data_element->Attr(
+              buzz::QName(true, buzz::STR_EMPTY, "data"));
+        }
+      }
+      // Inform listeners that a notification has been received.
+      SignalUpdateAvailable(notification_data);
+      update_signaled = true;
+      // Now go to the next Result element
+      result_element = result_element->NextNamed(
+          buzz::QName(true, buzz::STR_EMPTY, "Result"));
+    }
+  }
+  if (!update_signaled) {
+    LOG(WARNING) <<
+        "No getAll element or Result element found in response stanza";
+    // Signal an empty update to preserve old behavior
+    SignalUpdateAvailable(NotificationData());
+  }
   return STATE_RESPONSE;
 }
 
diff --git a/chrome/browser/sync/notifier/listener/listen_task.h b/chrome/browser/sync/notifier/listener/listen_task.h
index 8f5bcd3..cd2f40d 100644
--- a/chrome/browser/sync/notifier/listener/listen_task.h
+++ b/chrome/browser/sync/notifier/listener/listen_task.h
@@ -13,6 +13,7 @@
 #define CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_LISTEN_TASK_H_
 
 #include "chrome/browser/sync/notification_method.h"
+#include "chrome/browser/sync/notifier/listener/notification_defines.h"
 #include "talk/xmpp/xmpptask.h"
 
 namespace buzz {
@@ -33,7 +34,8 @@ class ListenTask : public buzz::XmppTask {
   virtual bool HandleStanza(const buzz::XmlElement* stanza);
 
   // Signal callback upon receipt of a notification.
-  sigslot::signal0<> SignalUpdateAvailable;
+  // SignalUpdateAvailable(const NotificationData& data);
+  sigslot::signal1<const NotificationData&> SignalUpdateAvailable;
 
  private:
   // Decide whether a notification should start a sync.  We only validate that
diff --git a/chrome/browser/sync/notifier/listener/mediator_thread.h b/chrome/browser/sync/notifier/listener/mediator_thread.h
index 12cef6d8..20ff1a0 100644
--- a/chrome/browser/sync/notifier/listener/mediator_thread.h
+++ b/chrome/browser/sync/notifier/listener/mediator_thread.h
@@ -8,8 +8,12 @@
 #ifndef CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_MEDIATOR_THREAD_H_
 #define CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_MEDIATOR_THREAD_H_
 
+#include <string>
+#include <vector>
+
 #include "base/logging.h"
 #include "chrome/browser/sync/notification_method.h"
+#include "chrome/browser/sync/notifier/listener/notification_defines.h"
 #include "talk/base/sigslot.h"
 #include "talk/xmpp/xmppclientsettings.h"
 
@@ -22,7 +26,6 @@ class MediatorThread {
     MSG_LOGGED_OUT,
     MSG_SUBSCRIPTION_SUCCESS,
     MSG_SUBSCRIPTION_FAILURE,
-    MSG_NOTIFICATION_RECEIVED,
     MSG_NOTIFICATION_SENT
   };
 
@@ -31,12 +34,15 @@ class MediatorThread {
   virtual void Login(const buzz::XmppClientSettings& settings) = 0;
   virtual void Logout() = 0;
   virtual void Start() = 0;
-  virtual void SubscribeForUpdates() = 0;
+  virtual void SubscribeForUpdates(
+      const std::vector<std::string>& subscribed_services_list) = 0;
   virtual void ListenForUpdates() = 0;
   virtual void SendNotification() = 0;
 
-  // Connect to this for messages about talk events.
+  // Connect to this for messages about talk events (except notifications).
   sigslot::signal1<MediatorMessage> SignalStateChange;
+  // Connect to this for notifications
+  sigslot::signal1<const NotificationData&> SignalNotificationReceived;
 
  protected:
   explicit MediatorThread(NotificationMethod notification_method)
diff --git a/chrome/browser/sync/notifier/listener/mediator_thread_impl.cc b/chrome/browser/sync/notifier/listener/mediator_thread_impl.cc
index 08bf2d8..b0645bf 100644
--- a/chrome/browser/sync/notifier/listener/mediator_thread_impl.cc
+++ b/chrome/browser/sync/notifier/listener/mediator_thread_impl.cc
@@ -81,8 +81,10 @@ void MediatorThreadImpl::ListenForUpdates() {
   Post(this, CMD_LISTEN_FOR_UPDATES);
 }
 
-void MediatorThreadImpl::SubscribeForUpdates() {
-  Post(this, CMD_SUBSCRIBE_FOR_UPDATES);
+void MediatorThreadImpl::SubscribeForUpdates(
+    const std::vector<std::string>& subscribed_services_list) {
+  Post(this, CMD_SUBSCRIBE_FOR_UPDATES,
+       new SubscriptionData(subscribed_services_list));
 }
 
 void MediatorThreadImpl::SendNotification() {
@@ -110,9 +112,13 @@ void MediatorThreadImpl::OnMessage(talk_base::Message* msg) {
     case CMD_SEND_NOTIFICATION:
       DoSendNotification();
       break;
-    case CMD_SUBSCRIBE_FOR_UPDATES:
-      DoSubscribeForUpdates();
+    case CMD_SUBSCRIBE_FOR_UPDATES: {
+      DCHECK(msg->pdata);
+      scoped_ptr<SubscriptionData> subscription_data(
+          reinterpret_cast<SubscriptionData*>(msg->pdata));
+      DoSubscribeForUpdates(*subscription_data);
       break;
+    }
     case CMD_PUMP_AUXILIARY_LOOPS:
       PumpAuxiliaryLoops();
       break;
@@ -199,14 +205,16 @@ void MediatorThreadImpl::DoDisconnect() {
   pump_.reset();
 }
 
-void MediatorThreadImpl::DoSubscribeForUpdates() {
+void MediatorThreadImpl::DoSubscribeForUpdates(
+    const SubscriptionData& subscription_data) {
   buzz::XmppClient* client = xmpp_client();
   // If there isn't an active xmpp client, return.
   if (!client) {
     return;
   }
   SubscribeTask* subscription =
-      new SubscribeTask(client, notification_method_);
+      new SubscribeTask(client, notification_method_,
+                        subscription_data.subscribed_services_list);
   subscription->SignalStatusUpdate.connect(
       this,
       &MediatorThreadImpl::OnSubscriptionStateChange);
@@ -239,8 +247,9 @@ void MediatorThreadImpl::DoSendNotification() {
   task->Start();
 }
 
-void MediatorThreadImpl::OnUpdateListenerMessage() {
-  SignalStateChange(MSG_NOTIFICATION_RECEIVED);
+void MediatorThreadImpl::OnUpdateListenerMessage(
+    const NotificationData& notification_data) {
+  SignalNotificationReceived(notification_data);
 }
 
 void MediatorThreadImpl::OnUpdateNotificationSent(bool success) {
diff --git a/chrome/browser/sync/notifier/listener/mediator_thread_impl.h b/chrome/browser/sync/notifier/listener/mediator_thread_impl.h
index 08b07cd..cc40308 100644
--- a/chrome/browser/sync/notifier/listener/mediator_thread_impl.h
+++ b/chrome/browser/sync/notifier/listener/mediator_thread_impl.h
@@ -20,6 +20,9 @@
 #ifndef CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_MEDIATOR_THREAD_IMPL_H_
 #define CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_MEDIATOR_THREAD_IMPL_H_
 
+#include <string>
+#include <vector>
+
 #include "base/logging.h"
 #include "base/scoped_ptr.h"
 #include "chrome/browser/sync/notifier/communicator/login.h"
@@ -63,6 +66,17 @@ struct LoginData : public talk_base::MessageData {
   buzz::XmppClientSettings user_settings;
 };
 
+// Used to pass subscription information from the mediator to the thread.
+// Use new to allocate it on the heap, the thread will delete it for you.
+struct SubscriptionData : public talk_base::MessageData {
+  explicit SubscriptionData(const std::vector<std::string>& services)
+      : subscribed_services_list(services) {
+  }
+  virtual ~SubscriptionData() {}
+
+  std::vector<std::string> subscribed_services_list;
+};
+
 class MediatorThreadImpl
     : public MediatorThread,
       public sigslot::has_slots<>,
@@ -82,7 +96,8 @@ class MediatorThreadImpl
   void Login(const buzz::XmppClientSettings& settings);
   void Logout();
   void ListenForUpdates();
-  void SubscribeForUpdates();
+  void SubscribeForUpdates(
+      const std::vector<std::string>& subscribed_services_list);
   void SendNotification();
   void LogStanzas();
 
@@ -92,14 +107,14 @@ class MediatorThreadImpl
   void OnMessage(talk_base::Message* msg);
   void DoLogin(LoginData* login_data);
   void DoDisconnect();
-  void DoSubscribeForUpdates();
+  void DoSubscribeForUpdates(const SubscriptionData& subscription_data);
   void DoListenForUpdates();
   void DoSendNotification();
   void DoStanzaLogging();
   void PumpAuxiliaryLoops();
 
   // These handle messages indicating an event happened in the outside world.
-  void OnUpdateListenerMessage();
+  void OnUpdateListenerMessage(const NotificationData& notification_data);
   void OnUpdateNotificationSent(bool success);
   void OnLoginFailureMessage(const notifier::LoginFailure& failure);
   void OnClientStateChangeMessage(notifier::Login::ConnectionState state);
diff --git a/chrome/browser/sync/notifier/listener/mediator_thread_mock.h b/chrome/browser/sync/notifier/listener/mediator_thread_mock.h
index ee5fde6..172e154 100644
--- a/chrome/browser/sync/notifier/listener/mediator_thread_mock.h
+++ b/chrome/browser/sync/notifier/listener/mediator_thread_mock.h
@@ -9,6 +9,9 @@
 #ifndef CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_MEDIATOR_THREAD_MOCK_H_
 #define CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_MEDIATOR_THREAD_MOCK_H_
 
+#include <string>
+#include <vector>
+
 #include "chrome/browser/sync/notifier/listener/mediator_thread.h"
 #include "chrome/browser/sync/notification_method.h"
 #include "talk/xmpp/xmppclientsettings.h"
@@ -44,7 +47,8 @@ class MockMediatorThread : public MediatorThread {
     start_calls++;
   }
 
-  virtual void SubscribeForUpdates() {
+  virtual void SubscribeForUpdates(
+      const std::vector<std::string>& subscribed_services_list) {
     subscribe_calls++;
   }
 
@@ -60,6 +64,9 @@ class MockMediatorThread : public MediatorThread {
   void ChangeState(MediatorThread::MediatorMessage message) {
     SignalStateChange(message);
   }
+  void Notify(const NotificationData& data) {
+    SignalNotificationReceived(data);
+  }
 
   // Intneral State
   int login_calls;
diff --git a/chrome/browser/sync/notifier/listener/notification_defines.h b/chrome/browser/sync/notifier/listener/notification_defines.h
new file mode 100644
index 0000000..5ea4d25
--- /dev/null
+++ b/chrome/browser/sync/notifier/listener/notification_defines.h
@@ -0,0 +1,16 @@
+// Copyright (c) 2010 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_LISTENER_NOTIFICATION_DEFINES_H_
+#define CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_NOTIFICATION_DEFINES_H_
+
+#include <string>
+
+struct NotificationData {
+  std::string service_url;
+  std::string service_specific_data;
+};
+
+#endif  // CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_NOTIFICATION_DEFINES_H_
+
diff --git a/chrome/browser/sync/notifier/listener/subscribe_task.cc b/chrome/browser/sync/notifier/listener/subscribe_task.cc
index e0f0bf9..de26466 100644
--- a/chrome/browser/sync/notifier/listener/subscribe_task.cc
+++ b/chrome/browser/sync/notifier/listener/subscribe_task.cc
@@ -18,10 +18,12 @@
 
 namespace browser_sync {
 
-SubscribeTask::SubscribeTask(Task* parent,
-                             NotificationMethod notification_method)
+SubscribeTask::SubscribeTask(
+    Task* parent, NotificationMethod notification_method,
+    const std::vector<std::string>& subscribed_services_list)
     : XmppTask(parent, buzz::XmppEngine::HL_SINGLE),
-      notification_method_(notification_method) {
+      notification_method_(notification_method),
+      subscribed_services_list_(subscribed_services_list) {
 }
 
 SubscribeTask::~SubscribeTask() {
@@ -38,6 +40,7 @@ int SubscribeTask::ProcessStart() {
   LOG(INFO) << "P2P: Subscription task started.";
   scoped_ptr<buzz::XmlElement> iq_stanza(
       MakeSubscriptionMessage(notification_method_,
+                              subscribed_services_list_,
                               GetClient()->jid().BareJid(), task_id()));
   LOG(INFO) << "P2P: Subscription stanza: "
             << XmlElementToString(*iq_stanza.get());
@@ -72,14 +75,16 @@ int SubscribeTask::ProcessResponse() {
 
 buzz::XmlElement* SubscribeTask::MakeSubscriptionMessage(
     NotificationMethod notification_method,
+    const std::vector<std::string>& subscribed_services_list,
     const buzz::Jid& to_jid_bare, const std::string& task_id) {
   switch (notification_method) {
     case NOTIFICATION_LEGACY:
       return MakeLegacySubscriptionMessage(to_jid_bare, task_id);
     case NOTIFICATION_TRANSITIONAL:
-      return MakeNonLegacySubscriptionMessage(true, to_jid_bare, task_id);
     case NOTIFICATION_NEW:
-      return MakeNonLegacySubscriptionMessage(false, to_jid_bare, task_id);
+      return MakeNonLegacySubscriptionMessage(subscribed_services_list,
+                                              to_jid_bare,
+                                              task_id);
   }
   NOTREACHED();
   return NULL;
@@ -124,8 +129,8 @@ buzz::XmlElement* SubscribeTask::MakeLegacySubscriptionMessage(
 // clients on at least NOTIFICATION_NEW.
 
 buzz::XmlElement* SubscribeTask::MakeNonLegacySubscriptionMessage(
-    bool is_transitional, const buzz::Jid& to_jid_bare,
-    const std::string& task_id) {
+    const std::vector<std::string>& subscribed_services_list,
+    const buzz::Jid& to_jid_bare, const std::string& task_id) {
   DCHECK(to_jid_bare.IsBare());
   static const buzz::QName kQnNotifierGetAll(
       true, kNotifierNamespace, "getAll");
@@ -146,12 +151,12 @@ buzz::XmlElement* SubscribeTask::MakeNonLegacySubscriptionMessage(
   iq->AddElement(get_all);
 
   get_all->AddElement(MakeBoolXmlElement("ClientActive", true));
-  if (is_transitional) {
+  for (std::vector<std::string>::const_iterator iter =
+        subscribed_services_list.begin();
+      iter != subscribed_services_list.end(); ++iter) {
     get_all->AddElement(
-        MakeStringXmlElement("SubscribedServiceUrl", kSyncLegacyServiceUrl));
+        MakeStringXmlElement("SubscribedServiceUrl", iter->c_str()));
   }
-  get_all->AddElement(
-      MakeStringXmlElement("SubscribedServiceUrl", kSyncServiceUrl));
   get_all->AddElement(MakeBoolXmlElement("FilterNonSubscribed", true));
 
   return iq;
diff --git a/chrome/browser/sync/notifier/listener/subscribe_task.h b/chrome/browser/sync/notifier/listener/subscribe_task.h
index dbef8d6..32a563b 100644
--- a/chrome/browser/sync/notifier/listener/subscribe_task.h
+++ b/chrome/browser/sync/notifier/listener/subscribe_task.h
@@ -10,6 +10,7 @@
 #define CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_SUBSCRIBE_TASK_H_
 
 #include <string>
+#include <vector>
 
 #include "chrome/browser/sync/notification_method.h"
 #include "talk/xmllite/xmlelement.h"
@@ -17,10 +18,12 @@
 #include "testing/gtest/include/gtest/gtest_prod.h"
 
 namespace browser_sync {
-
+// TODO(akalin): Remove NOTIFICATION_LEGACY and remove/refactor relevant code
+// in this class and any other class that uses notification_method.
 class SubscribeTask : public buzz::XmppTask {
  public:
-  SubscribeTask(Task* parent, NotificationMethod notification_method);
+  SubscribeTask(Task* parent, NotificationMethod notification_method,
+                const std::vector<std::string>& subscribed_services_list);
   virtual ~SubscribeTask();
 
   // Overridden from XmppTask.
@@ -35,16 +38,18 @@ class SubscribeTask : public buzz::XmppTask {
   // Assembles an Xmpp stanza which can be sent to subscribe to notifications.
   static buzz::XmlElement* MakeSubscriptionMessage(
       NotificationMethod notification_method,
+      const std::vector<std::string>& subscribed_services_list,
       const buzz::Jid& to_jid_bare, const std::string& task_id);
 
   static buzz::XmlElement* MakeLegacySubscriptionMessage(
       const buzz::Jid& to_jid_bare, const std::string& task_id);
 
   static buzz::XmlElement* MakeNonLegacySubscriptionMessage(
-      bool is_transitional,
+      const std::vector<std::string>& subscribed_services_list,
       const buzz::Jid& to_jid_bare, const std::string& task_id);
 
   NotificationMethod notification_method_;
+  std::vector<std::string> subscribed_services_list_;
 
   FRIEND_TEST(SubscribeTaskTest, MakeLegacySubscriptionMessage);
   FRIEND_TEST(SubscribeTaskTest, MakeNonLegacySubscriptionMessage);
diff --git a/chrome/browser/sync/notifier/listener/subscribe_task_unittest.cc b/chrome/browser/sync/notifier/listener/subscribe_task_unittest.cc
index 21bffc2..d58d6e1 100644
--- a/chrome/browser/sync/notifier/listener/subscribe_task_unittest.cc
+++ b/chrome/browser/sync/notifier/listener/subscribe_task_unittest.cc
@@ -8,6 +8,7 @@
 #include "base/scoped_ptr.h"
 #include "base/string_util.h"
 #include "chrome/browser/sync/notification_method.h"
+#include "chrome/browser/sync/notifier/listener/notification_constants.h"
 #include "chrome/browser/sync/notifier/listener/xml_element_util.h"
 #include "talk/xmpp/jid.h"
 #include "testing/gtest/include/gtest/gtest.h"
@@ -48,27 +49,14 @@ TEST_F(SubscribeTaskTest, MakeLegacySubscriptionMessage) {
 }
 
 TEST_F(SubscribeTaskTest, MakeNonLegacySubscriptionMessage) {
+  std::vector<std::string> subscribed_services_list;
+  subscribed_services_list.push_back(kSyncLegacyServiceUrl);
+  subscribed_services_list.push_back(kSyncServiceUrl);
   scoped_ptr<buzz::XmlElement> new_message(
-      SubscribeTask::MakeNonLegacySubscriptionMessage(false, to_jid_bare_,
+      SubscribeTask::MakeNonLegacySubscriptionMessage(subscribed_services_list,
+                                                      to_jid_bare_,
                                                       task_id_));
-  const std::string expected_new_xml_string =
-      StringPrintf(
-          "<cli:iq type=\"get\" to=\"%s\" id=\"%s\" "
-                  "xmlns:cli=\"jabber:client\">"
-            "<getAll xmlns=\"google:notifier\">"
-              "<ClientActive xmlns=\"\" bool=\"true\"/>"
-              "<SubscribedServiceUrl "
-                  "xmlns=\"\" data=\"http://www.google.com/chrome/sync\"/>"
-              "<FilterNonSubscribed xmlns=\"\" bool=\"true\"/>"
-            "</getAll>"
-          "</cli:iq>",
-          to_jid_bare_.Str().c_str(), task_id_.c_str());
-  EXPECT_EQ(expected_new_xml_string, XmlElementToString(*new_message));
-
-  scoped_ptr<buzz::XmlElement> transitional_message(
-      SubscribeTask::MakeNonLegacySubscriptionMessage(true, to_jid_bare_,
-                                                      task_id_));
-  const std::string expected_transitional_xml_string =
+  const std::string expected_xml_string =
       StringPrintf(
           "<cli:iq type=\"get\" to=\"%s\" id=\"%s\" "
                   "xmlns:cli=\"jabber:client\">"
@@ -81,33 +69,39 @@ TEST_F(SubscribeTaskTest, MakeNonLegacySubscriptionMessage) {
             "</getAll>"
           "</cli:iq>",
           to_jid_bare_.Str().c_str(), task_id_.c_str());
-  EXPECT_EQ(expected_transitional_xml_string,
-            XmlElementToString(*transitional_message));
+  EXPECT_EQ(expected_xml_string, XmlElementToString(*new_message));
 }
 
 TEST_F(SubscribeTaskTest, MakeSubscriptionMessage) {
+  std::vector<std::string> subscribed_services_list;
+  subscribed_services_list.push_back(kSyncLegacyServiceUrl);
+  subscribed_services_list.push_back(kSyncServiceUrl);
+
   scoped_ptr<buzz::XmlElement> expected_legacy_message(
       SubscribeTask::MakeLegacySubscriptionMessage(to_jid_bare_, task_id_));
   scoped_ptr<buzz::XmlElement> legacy_message(
       SubscribeTask::MakeSubscriptionMessage(NOTIFICATION_LEGACY,
+                                             subscribed_services_list,
                                              to_jid_bare_, task_id_));
   EXPECT_EQ(XmlElementToString(*expected_legacy_message),
             XmlElementToString(*legacy_message));
 
   scoped_ptr<buzz::XmlElement> expected_transitional_message(
-      SubscribeTask::MakeNonLegacySubscriptionMessage(true,
+      SubscribeTask::MakeNonLegacySubscriptionMessage(subscribed_services_list,
                                                       to_jid_bare_, task_id_));
   scoped_ptr<buzz::XmlElement> transitional_message(
       SubscribeTask::MakeSubscriptionMessage(NOTIFICATION_TRANSITIONAL,
+                                             subscribed_services_list,
                                              to_jid_bare_, task_id_));
   EXPECT_EQ(XmlElementToString(*expected_transitional_message),
             XmlElementToString(*transitional_message));
 
   scoped_ptr<buzz::XmlElement> expected_new_message(
-      SubscribeTask::MakeNonLegacySubscriptionMessage(false,
+      SubscribeTask::MakeNonLegacySubscriptionMessage(subscribed_services_list,
                                                       to_jid_bare_, task_id_));
   scoped_ptr<buzz::XmlElement> new_message(
       SubscribeTask::MakeSubscriptionMessage(NOTIFICATION_NEW,
+                                             subscribed_services_list,
                                              to_jid_bare_, task_id_));
   EXPECT_EQ(XmlElementToString(*expected_new_message),
             XmlElementToString(*new_message));
diff --git a/chrome/browser/sync/notifier/listener/talk_mediator.h b/chrome/browser/sync/notifier/listener/talk_mediator.h
index 0e61673..ef09e66 100644
--- a/chrome/browser/sync/notifier/listener/talk_mediator.h
+++ b/chrome/browser/sync/notifier/listener/talk_mediator.h
@@ -8,7 +8,7 @@
 // Example usage:
 //
 //   TalkMediator mediator();
-//   mediator.SetCredentials("email", "pass", false);
+//   mediator.SetAuthToken("email", "token");
 //   mediator.WatchAuthWatcher(auth_watcher_);
 //   AuthWatcher eventually sends AUTH_SUCCEEDED which triggers:
 //   mediator.Login();
@@ -20,11 +20,11 @@
 
 #include <string>
 
+#include "chrome/browser/sync/notifier/listener/notification_defines.h"
+
 namespace browser_sync {
 
 class AuthWatcher;
-class SyncerThread;
-
 struct TalkMediatorEvent {
   enum WhatHappened {
     LOGIN_SUCCEEDED,
@@ -44,6 +44,8 @@ struct TalkMediatorEvent {
   }
 
   WhatHappened what_happened;
+  // Data in the case of a NOTIFICATION_RECEIVED event
+  NotificationData notification_data;
 };
 
 typedef EventChannel<TalkMediatorEvent, Lock> TalkMediatorChannel;
@@ -66,6 +68,9 @@ class TalkMediator {
 
   // Channel by which talk mediator events are signaled.
   virtual TalkMediatorChannel* channel() const = 0;
+
+  // Add a URL to subscribe to for notifications.
+  virtual void AddSubscribedServiceUrl(const std::string& service_url) = 0;
 };
 
 }  // namespace browser_sync
diff --git a/chrome/browser/sync/notifier/listener/talk_mediator_impl.cc b/chrome/browser/sync/notifier/listener/talk_mediator_impl.cc
index a68d306..ca345a0e 100644
--- a/chrome/browser/sync/notifier/listener/talk_mediator_impl.cc
+++ b/chrome/browser/sync/notifier/listener/talk_mediator_impl.cc
@@ -66,6 +66,9 @@ void TalkMediatorImpl::TalkMediatorInitialization(bool should_connect) {
     mediator_thread_->SignalStateChange.connect(
         this,
         &TalkMediatorImpl::MediatorThreadMessageHandler);
+    mediator_thread_->SignalNotificationReceived.connect(
+        this,
+        &TalkMediatorImpl::MediatorThreadNotificationHandler);
     state_.connected = 1;
   }
   mediator_thread_->Start();
@@ -123,6 +126,9 @@ bool TalkMediatorImpl::DoLogin() {
     mediator_thread_->SignalStateChange.connect(
         this,
         &TalkMediatorImpl::MediatorThreadMessageHandler);
+    mediator_thread_->SignalNotificationReceived.connect(
+        this,
+        &TalkMediatorImpl::MediatorThreadNotificationHandler);
     state_.connected = 1;
   }
   if (state_.initialized && !state_.logging_in) {
@@ -138,6 +144,7 @@ bool TalkMediatorImpl::Logout() {
   // We do not want to be called back during logout since we may be closing.
   if (state_.connected) {
     mediator_thread_->SignalStateChange.disconnect(this);
+    mediator_thread_->SignalNotificationReceived.disconnect(this);
     state_.connected = 0;
   }
   if (state_.started) {
@@ -186,6 +193,16 @@ bool TalkMediatorImpl::SetAuthToken(const std::string& email,
   return true;
 }
 
+void TalkMediatorImpl::AddSubscribedServiceUrl(
+    const std::string& service_url) {
+  subscribed_services_list_.push_back(service_url);
+  if (state_.logged_in) {
+    LOG(INFO) << "Resubscribing for updates, a new service got added";
+    mediator_thread_->SubscribeForUpdates(subscribed_services_list_);
+  }
+}
+
+
 void TalkMediatorImpl::MediatorThreadMessageHandler(
     MediatorThread::MediatorMessage message) {
   LOG(INFO) << "P2P: MediatorThread has passed a message";
@@ -202,9 +219,6 @@ void TalkMediatorImpl::MediatorThreadMessageHandler(
     case MediatorThread::MSG_SUBSCRIPTION_FAILURE:
       OnSubscriptionFailure();
       break;
-    case MediatorThread::MSG_NOTIFICATION_RECEIVED:
-      OnNotificationReceived();
-      break;
     case MediatorThread::MSG_NOTIFICATION_SENT:
       OnNotificationSent();
       break;
@@ -214,6 +228,16 @@ void TalkMediatorImpl::MediatorThreadMessageHandler(
   }
 }
 
+void TalkMediatorImpl::MediatorThreadNotificationHandler(
+    const NotificationData& notification_data) {
+  LOG(INFO) << "P2P: Updates are available on the server.";
+  AutoLock lock(mutex_);
+  TalkMediatorEvent event = { TalkMediatorEvent::NOTIFICATION_RECEIVED };
+  event.notification_data = notification_data;
+  channel_->NotifyListeners(event);
+}
+
+
 void TalkMediatorImpl::OnLogin() {
   LOG(INFO) << "P2P: Logged in.";
   AutoLock lock(mutex_);
@@ -222,7 +246,8 @@ void TalkMediatorImpl::OnLogin() {
   // ListenForUpdates enables the ListenTask.  This is done before
   // SubscribeForUpdates.
   mediator_thread_->ListenForUpdates();
-  mediator_thread_->SubscribeForUpdates();
+  // Now subscribe for updates to all the services we are interested in
+  mediator_thread_->SubscribeForUpdates(subscribed_services_list_);
   TalkMediatorEvent event = { TalkMediatorEvent::LOGIN_SUCCEEDED };
   channel_->NotifyListeners(event);
 }
@@ -258,13 +283,6 @@ void TalkMediatorImpl::OnSubscriptionFailure() {
   channel_->NotifyListeners(event);
 }
 
-void TalkMediatorImpl::OnNotificationReceived() {
-  LOG(INFO) << "P2P: Updates are available on the server.";
-  AutoLock lock(mutex_);
-  TalkMediatorEvent event = { TalkMediatorEvent::NOTIFICATION_RECEIVED };
-  channel_->NotifyListeners(event);
-}
-
 void TalkMediatorImpl::OnNotificationSent() {
   LOG(INFO) <<
       "P2P: Peers were notified that updates are available on the server.";
diff --git a/chrome/browser/sync/notifier/listener/talk_mediator_impl.h b/chrome/browser/sync/notifier/listener/talk_mediator_impl.h
index 7de0885..7677277 100644
--- a/chrome/browser/sync/notifier/listener/talk_mediator_impl.h
+++ b/chrome/browser/sync/notifier/listener/talk_mediator_impl.h
@@ -10,6 +10,7 @@
 #define CHROME_BROWSER_SYNC_NOTIFIER_LISTENER_TALK_MEDIATOR_IMPL_H_
 
 #include <string>
+#include <vector>
 
 #include "base/lock.h"
 #include "base/scoped_ptr.h"
@@ -25,10 +26,6 @@ class EventListenerHookup;
 
 namespace browser_sync {
 
-class AuthWatcher;
-struct AuthWatcherEvent;
-class SyncerThread;
-
 class TalkMediatorImpl
     : public TalkMediator,
       public sigslot::has_slots<> {
@@ -49,6 +46,8 @@ class TalkMediatorImpl
 
   TalkMediatorChannel* channel() const;
 
+  virtual void AddSubscribedServiceUrl(const std::string& service_url);
+
  private:
   struct TalkMediatorState {
     TalkMediatorState()
@@ -74,11 +73,12 @@ class TalkMediatorImpl
   // class to push listening and subscription events to the mediator thread.
   void AuthWatcherEventHandler(const AuthWatcherEvent& auth_event);
 
-  // Callback for the mediator thread.
+  // Callbacks for the mediator thread.
   void MediatorThreadMessageHandler(MediatorThread::MediatorMessage message);
+  void MediatorThreadNotificationHandler(
+      const NotificationData& notification_data);
 
   // Responses to messages from the MediatorThread.
-  void OnNotificationReceived();
   void OnNotificationSent();
   void OnLogin();
   void OnLogout();
@@ -112,6 +112,8 @@ class TalkMediatorImpl
 
   bool invalidate_xmpp_auth_token_;
 
+  std::vector<std::string> subscribed_services_list_;
+
   FRIEND_TEST(TalkMediatorImplTest, SetAuthTokenWithBadInput);
   FRIEND_TEST(TalkMediatorImplTest, SetAuthTokenWithGoodInput);
   FRIEND_TEST(TalkMediatorImplTest, SendNotification);
diff --git a/chrome/browser/sync/notifier/listener/talk_mediator_unittest.cc b/chrome/browser/sync/notifier/listener/talk_mediator_unittest.cc
index 6beff1f..a1808bc 100644
--- a/chrome/browser/sync/notifier/listener/talk_mediator_unittest.cc
+++ b/chrome/browser/sync/notifier/listener/talk_mediator_unittest.cc
@@ -164,9 +164,10 @@ TEST_F(TalkMediatorImplTest, MediatorThreadCallbacks) {
   ASSERT_TRUE(talk1->SendNotification() == true);
   ASSERT_TRUE(mock->send_calls == 2);
 
-  // |MSG_NOTIFICATION_RECEIVED| from the MediatorThread triggers a callback
-  // of type |NOTIFICATION_RECEIVED|.
-  mock->ChangeState(MediatorThread::MSG_NOTIFICATION_RECEIVED);
+  NotificationData data;
+  data.service_url = "service_url";
+  data.service_specific_data = "service_data";
+  mock->Notify(data);
   ASSERT_TRUE(last_message_ == TalkMediatorEvent::NOTIFICATION_RECEIVED);
 
   // A |TALKMEDIATOR_DESTROYED| message is received during tear down.
diff --git a/chrome/chrome.gyp b/chrome/chrome.gyp
index 9f23d90..2074b13 100644
--- a/chrome/chrome.gyp
+++ b/chrome/chrome.gyp
@@ -810,6 +810,7 @@
         'browser/sync/notifier/listener/mediator_thread_mock.h',
         'browser/sync/notifier/listener/notification_constants.cc',
         'browser/sync/notifier/listener/notification_constants.h',
+        'browser/sync/notifier/listener/notification_defines.h',
         'browser/sync/notifier/listener/send_update_task.cc',
         'browser/sync/notifier/listener/send_update_task.h',
         'browser/sync/notifier/base/sigslotrepeater.h',