Distribute failover event with topic

Change-Id: I8629e7e19ebd4a18f95b32ad3ce1eba7ddf4ecc6

@@ -47,4 +47,9 @@ public class TestStorageService extends StorageServiceAdapter {
     public TransactionContextBuilder transactionContextBuilder() {
     public TransactionContextBuilder transactionContextBuilder() {
         throw new UnsupportedOperationException("transactionContextBuilder");
         throw new UnsupportedOperationException("transactionContextBuilder");
     }
     }
+
+    @Override
+    public <T> Topic<T> getTopic(String name, Serializer serializer) {
+        return new TestTopic(name);
+    }
 }
 }

+/*
+ * Copyright 2016-present Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.store.service;
+
+import com.google.common.collect.Sets;
+
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+/**
+ * Test implementation of topic.
+ */
+public class TestTopic<T> implements Topic<T> {
+    private final String name;
+    private final Set<Consumer<T>> callbacks = Sets.newConcurrentHashSet();
+
+    public TestTopic(String name) {
+        this.name = name;
+    }
+
+    @Override
+    public CompletableFuture<Void> publish(T message) {
+        callbacks.forEach(c -> c.accept(message));
+        return CompletableFuture.completedFuture(null);
+    }
+
+    @Override
+    public CompletableFuture<Void> subscribe(Consumer<T> callback) {
+        callbacks.add(callback);
+        return CompletableFuture.completedFuture(null);
+    }
+
+    @Override
+    public CompletableFuture<Void> unsubscribe(Consumer<T> callback) {
+        callbacks.remove(callback);
+        return CompletableFuture.completedFuture(null);
+    }
+
+    @Override
+    public String name() {
+        return name;
+    }
+
+    @Override
+    public Type primitiveType() {
+        return Type.TOPIC;
+    }
+}

@@ -62,6 +62,7 @@ import org.onosproject.store.service.MapEventListener;
 import org.onosproject.store.service.MultiValuedTimestamp;
 import org.onosproject.store.service.MultiValuedTimestamp;
 import org.onosproject.store.service.Serializer;
 import org.onosproject.store.service.Serializer;
 import org.onosproject.store.service.StorageService;
 import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Topic;
 import org.onosproject.store.service.Versioned;
 import org.onosproject.store.service.Versioned;
 import org.osgi.service.component.ComponentContext;
 import org.osgi.service.component.ComponentContext;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
@@ -145,6 +146,8 @@ public class DistributedGroupStore
 
 
     private KryoNamespace clusterMsgSerializer;
     private KryoNamespace clusterMsgSerializer;
 
 
+    private static Topic<GroupStoreMessage> groupTopic;
+
     @Property(name = "garbageCollect", boolValue = GARBAGE_COLLECT,
     @Property(name = "garbageCollect", boolValue = GARBAGE_COLLECT,
             label = "Enable group garbage collection")
             label = "Enable group garbage collection")
     private boolean garbageCollect = GARBAGE_COLLECT;
     private boolean garbageCollect = GARBAGE_COLLECT;
@@ -210,6 +213,9 @@ public class DistributedGroupStore
         log.debug("Current size of pendinggroupkeymap:{}",
         log.debug("Current size of pendinggroupkeymap:{}",
                   auditPendingReqQueue.size());
                   auditPendingReqQueue.size());
 
 
+        groupTopic = getOrCreateGroupTopic(serializer);
+        groupTopic.subscribe(this::processGroupMessage);
+
         log.info("Started");
         log.info("Started");
     }
     }
 
 
@@ -237,6 +243,14 @@ public class DistributedGroupStore
         }
         }
     }
     }
 
 
+    private Topic<GroupStoreMessage> getOrCreateGroupTopic(Serializer serializer) {
+        if (groupTopic == null) {
+            return storageService.getTopic("group-failover-notif", serializer);
+        } else {
+            return groupTopic;
+        }
+    };
+
     /**
     /**
      * Returns the group store eventual consistent key map.
      * Returns the group store eventual consistent key map.
      *
      *
@@ -1109,6 +1123,16 @@ public class DistributedGroupStore
         }
         }
     }
     }
 
 
+    private void processGroupMessage(GroupStoreMessage message) {
+        if (message.type() == GroupStoreMessage.Type.FAILOVER) {
+            // FIXME: groupStoreEntriesByKey inaccessible here
+            getGroupIdTable(message.deviceId()).values()
+                    .stream()
+                    .filter((storedGroup) -> (storedGroup.appCookie().equals(message.appCookie())))
+                    .findFirst().ifPresent(group -> notifyDelegate(new GroupEvent(Type.GROUP_BUCKET_FAILOVER, group)));
+        }
+    }
+
     private void process(GroupStoreMessage groupOp) {
     private void process(GroupStoreMessage groupOp) {
         log.debug("Received remote group operation {} request for device {}",
         log.debug("Received remote group operation {} request for device {}",
                   groupOp.type(),
                   groupOp.type(),
@@ -1314,13 +1338,12 @@ public class DistributedGroupStore
 
 
     @Override
     @Override
     public void notifyOfFailovers(Collection<Group> failoverGroups) {
     public void notifyOfFailovers(Collection<Group> failoverGroups) {
-        List<GroupEvent> failoverEvents = new ArrayList<>();
         failoverGroups.forEach(group -> {
         failoverGroups.forEach(group -> {
             if (group.type() == Group.Type.FAILOVER) {
             if (group.type() == Group.Type.FAILOVER) {
-                failoverEvents.add(new GroupEvent(GroupEvent.Type.GROUP_BUCKET_FAILOVER, group));
+                groupTopic.publish(GroupStoreMessage.createGroupFailoverMsg(
+                        group.deviceId(), group));
             }
             }
         });
         });
-        notifyDelegate(failoverEvents);
     }
     }
 
 
     private void garbageCollect(DeviceId deviceId,
     private void garbageCollect(DeviceId deviceId,

@@ -40,7 +40,8 @@ public final class GroupStoreMessage {
     public enum Type {
     public enum Type {
         ADD,
         ADD,
         UPDATE,
         UPDATE,
-        DELETE
+        DELETE,
+        FAILOVER
     }
     }
 
 
     private GroupStoreMessage(Type type,
     private GroupStoreMessage(Type type,
@@ -119,6 +120,18 @@ public final class GroupStoreMessage {
                                      null);
                                      null);
     }
     }
 
 
+    public static GroupStoreMessage createGroupFailoverMsg(DeviceId deviceId,
+                                                           GroupDescription desc) {
+        return new GroupStoreMessage(Type.FAILOVER,
+                                     deviceId,
+                                     desc.appCookie(),
+                                     desc,
+                                     null,
+                                     null,
+                                     desc.appCookie());
+    }
+
+
     /**
     /**
      * Returns the device identifier of this group request.
      * Returns the device identifier of this group request.
      *
      *