[Sync] Make P2PNotifier behave more like InvalidationNotifier

chrome/browser/sync/notifier/p2p_notifier.cc

 // 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/p2p_notifier.h"
 
+#include <algorithm>
+
+#include "base/logging.h"
 #include "base/message_loop_proxy.h"
+#include "base/json/json_reader.h"
+#include "base/json/json_writer.h"
+#include "base/values.h"
 #include "chrome/browser/sync/notifier/sync_notifier_observer.h"
 #include "chrome/browser/sync/protocol/service_constants.h"
 #include "chrome/browser/sync/syncable/model_type_payload_map.h"
-#include "jingle/notifier/listener/mediator_thread_impl.h"
-#include "jingle/notifier/listener/talk_mediator_impl.h"
 
 namespace sync_notifier {
 
 namespace {
+
 const char kSyncNotificationChannel[] = "http://www.google.com/chrome/sync";
-const char kSyncNotificationData[] = "sync-ping-p2p";
+
+const char kNotifySelf[] = "notifySelf";
+const char kNotifyOthers[] = "notifyOthers";
+const char kNotifyAll[] = "notifyAll";
+
+const char kSenderIdKey[] = "senderId";
+const char kNotificationTypeKey[] = "notificationType";
+const char kChangedTypesKey[] = "changedTypes";
+
 }  // namespace
 
-P2PNotifier::P2PNotifier(
-    const notifier::NotifierOptions& notifier_options)
-    : talk_mediator_(
-        new notifier::TalkMediatorImpl(
-            new notifier::MediatorThreadImpl(notifier_options),
-            notifier_options)),
+std::string P2PNotificationTargetToString(P2PNotificationTarget target) {
+  switch (target) {
+    case NOTIFY_SELF:
+      return kNotifySelf;
+    case NOTIFY_OTHERS:
+      return kNotifyOthers;
+    case NOTIFY_ALL:
+      return kNotifyAll;
+    default:
+      NOTREACHED();
+      return "";
+  }
+}
+
+P2PNotificationTarget P2PNotificationTargetFromString(
+    const std::string& target_str) {
+  if (target_str == kNotifySelf) {
+    return NOTIFY_SELF;
+  }
+  if (target_str == kNotifyOthers) {
+    return NOTIFY_OTHERS;
+  }
+  if (target_str == kNotifyAll) {
+    return NOTIFY_ALL;
+  }
+  LOG(WARNING) << "Could not parse " << target_str;
+  return NOTIFY_SELF;
+}
+
+P2PNotificationData::P2PNotificationData() : target_(NOTIFY_SELF) {}
+
+P2PNotificationData::P2PNotificationData(
+    const std::string& sender_id,
+    P2PNotificationTarget target,
+    const syncable::ModelTypeSet& changed_types)
+    : sender_id_(sender_id),
+      target_(target),
+      changed_types_(changed_types) {}
+
+P2PNotificationData::~P2PNotificationData() {}
+
+bool P2PNotificationData::IsTargeted(const std::string& id) const {
+  switch (target_) {
+    case NOTIFY_SELF:
+      return sender_id_ == id;
+    case NOTIFY_OTHERS:
+      return sender_id_ != id;
+    case NOTIFY_ALL:
+      return true;
+    default:
+      NOTREACHED();
+      return false;
+  }
+}
+
+const syncable::ModelTypeSet& P2PNotificationData::GetChangedTypes() const {
+  return changed_types_;
+}
+
+bool P2PNotificationData::Equals(const P2PNotificationData& other) const {
+  return
+      (sender_id_ == other.sender_id_) &&
+      (target_ == other.target_) &&
+      (changed_types_ == other.changed_types_);
+}
+
+std::string P2PNotificationData::ToString() const {
+  scoped_ptr<DictionaryValue> dict(new DictionaryValue());
+  dict->SetString(kSenderIdKey, sender_id_);
+  dict->SetString(kNotificationTypeKey,
+                  P2PNotificationTargetToString(target_));
+  dict->Set(kChangedTypesKey, syncable::ModelTypeSetToValue(changed_types_));
+  std::string json;
+  base::JSONWriter::Write(dict.get(), false /* pretty_print */, &json);
+  return json;
+}
+
+bool P2PNotificationData::ResetFromString(const std::string& str) {
+  scoped_ptr<Value> data_value(
+      base::JSONReader::Read(str, false /* allow_trailing_comma */));
+  if (!data_value.get()) {
+    LOG(WARNING) << "Could not parse " << str;
+    return false;
+  }
+  if (!data_value->IsType(Value::TYPE_DICTIONARY)) {
+    LOG(WARNING) << "Could not parse " << str << " as a dictionary";
+    return false;
+  }
+  // TODO(akalin): Use Values::AsDictionary() when it becomes
+  // available.
+  DictionaryValue* data_dict =
+      static_cast<DictionaryValue*>(data_value.get());
+  if (!data_dict->GetString(kSenderIdKey, &sender_id_)) {
+    LOG(WARNING) << "Could not find string value for " << kSenderIdKey;
+  }
+  std::string target_str;
+  if (!data_dict->GetString(kNotificationTypeKey, &target_str)) {
+    LOG(WARNING) << "Could not find string value for "
+                 << kNotificationTypeKey;
+  }
+  target_ = P2PNotificationTargetFromString(target_str);
+  ListValue* changed_types_list = NULL;
+  if (!data_dict->GetList(kChangedTypesKey, &changed_types_list)) {
+    LOG(WARNING) << "Could not find list value for "
+                 << kChangedTypesKey;
+    return false;
+  }
+  changed_types_ = syncable::ModelTypeSetFromValue(*changed_types_list);
+  return true;
+}
+
+P2PNotifier::P2PNotifier(notifier::TalkMediator* talk_mediator)
+    : talk_mediator_(talk_mediator),
       logged_in_(false),
       notifications_enabled_(false),
       parent_message_loop_proxy_(
           base::MessageLoopProxy::current()) {
   talk_mediator_->SetDelegate(this);
 }
 
 P2PNotifier::~P2PNotifier() {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
 }
@@ skipping 12 matching lines @@
   observer_list_.RemoveObserver(observer);
 
   // Logout after the last observer is removed.
   if (observer_list_.size() == 0) {
    talk_mediator_->Logout();
   }
 }
 
 void P2PNotifier::SetUniqueId(const std::string& unique_id) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  unique_id_ = unique_id;
 }
 
 void P2PNotifier::SetState(const std::string& state) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
 }
 
 void P2PNotifier::UpdateCredentials(
     const std::string& email, const std::string& token) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
   // If already logged in, the new credentials will take effect on the
@@ skipping 10 matching lines @@
     // There may be some subtle issues around case sensitivity of the
     // from field, but it doesn't matter too much since this is only
     // used in p2p mode (which is only used in testing).
     subscription.from = email;
     talk_mediator_->AddSubscription(subscription);
 
     logged_in_ = true;
   }
 }
 
-void P2PNotifier::UpdateEnabledTypes(const syncable::ModelTypeSet& types) {
+void P2PNotifier::UpdateEnabledTypes(
+    const syncable::ModelTypeSet& enabled_types) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  enabled_types_ = types;
-  MaybeEmitNotification();
+  syncable::ModelTypeSet new_enabled_types;
+  std::set_difference(enabled_types.begin(), enabled_types.end(),
+                      enabled_types_.begin(), enabled_types_.end(),
+                      std::inserter(new_enabled_types,
+                                    new_enabled_types.end()));
+  enabled_types_ = enabled_types;
+  const P2PNotificationData notification_data(
+      unique_id_, NOTIFY_SELF, new_enabled_types);
+  SendNotificationData(notification_data);
 }
 
-void P2PNotifier::SendNotification() {
+void P2PNotifier::SendNotification(
+    const syncable::ModelTypeSet& changed_types) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  VLOG(1) << "Sending XMPP notification...";
-  notifier::Notification notification;
-  notification.channel = kSyncNotificationChannel;
-  notification.data = kSyncNotificationData;
-  talk_mediator_->SendNotification(notification);
+  const P2PNotificationData notification_data(
+      unique_id_, NOTIFY_OTHERS, changed_types);
+  SendNotificationData(notification_data);
 }
 
 void P2PNotifier::OnNotificationStateChange(bool notifications_enabled) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  bool disabled_to_enabled = notifications_enabled && !notifications_enabled_;
   notifications_enabled_ = notifications_enabled;
   FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
       OnNotificationStateChange(notifications_enabled_));
-  MaybeEmitNotification();
+  if (disabled_to_enabled) {
+    const P2PNotificationData notification_data(
+        unique_id_, NOTIFY_SELF, enabled_types_);
+    SendNotificationData(notification_data);
+  }
 }
 
 void P2PNotifier::OnIncomingNotification(
     const notifier::Notification& notification) {
   DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
-  VLOG(1) << "Sync received P2P notification.";
-  if (notification.channel != kSyncNotificationChannel) {
-    LOG(WARNING) << "Notification from unexpected source: "
-                 << notification.channel;
-  }
-  MaybeEmitNotification();
-}
-
-void P2PNotifier::OnOutgoingNotification() {}
-
-void P2PNotifier::MaybeEmitNotification() {
+  VLOG(1) << "Received notification " << notification.ToString();
   if (!logged_in_) {
     VLOG(1) << "Not logged in yet -- not emitting notification";
     return;
   }
   if (!notifications_enabled_) {
     VLOG(1) << "Notifications not enabled -- not emitting notification";
     return;
   }
-  if (enabled_types_.empty()) {
-    VLOG(1) << "No enabled types -- not emitting notification";
+  if (notification.channel != kSyncNotificationChannel) {
+    LOG(WARNING) << "Notification from unexpected source "
+                 << notification.channel;
+  }
+  P2PNotificationData notification_data;
+  if (!notification_data.ResetFromString(notification.data)) {
+    LOG(WARNING) << "Could not parse notification data from "
+                 << notification.data;
+    notification_data =
+        P2PNotificationData(unique_id_, NOTIFY_ALL, enabled_types_);
+  }
+  if (!notification_data.IsTargeted(unique_id_)) {
+    VLOG(1) << "Not a target of the notification -- "
+            << "not emitting notification";
     return;
   }
-  syncable::ModelTypePayloadMap type_payloads =
+  if (notification_data.GetChangedTypes().empty()) {
+    VLOG(1) << "No changed types -- not emitting notification";
+    return;
+  }
+  const syncable::ModelTypePayloadMap& type_payloads =
       syncable::ModelTypePayloadMapFromBitSet(
-          syncable::ModelTypeBitSetFromSet(enabled_types_), std::string());
+          syncable::ModelTypeBitSetFromSet(
+              notification_data.GetChangedTypes()), std::string());
   FOR_EACH_OBSERVER(SyncNotifierObserver, observer_list_,
                     OnIncomingNotification(type_payloads));
 }
 
+void P2PNotifier::OnOutgoingNotification() {}
+
+void P2PNotifier::SendNotificationDataForTest(
+    const P2PNotificationData& notification_data) {
+  SendNotificationData(notification_data);
+}
+
+void P2PNotifier::SendNotificationData(
+    const P2PNotificationData& notification_data) {
+  DCHECK(parent_message_loop_proxy_->BelongsToCurrentThread());
+  notifier::Notification notification;
+  notification.channel = kSyncNotificationChannel;
+  notification.data = notification_data.ToString();
+  VLOG(1) << "Sending XMPP notification: " << notification.ToString();
+  talk_mediator_->SendNotification(notification);
+}
+
 }  // namespace sync_notifier