Consistently ordered notification support for single partition scenario.

Change-Id: I6d959fafb879aa89885c2fb758aa73efd4b47cb0

@@ -17,6 +17,8 @@
 package org.onosproject.store.consistent.impl;
 package org.onosproject.store.consistent.impl;
 
 
 
 
+import java.util.function.Consumer;
+
 import net.kuujo.copycat.cluster.ClusterConfig;
 import net.kuujo.copycat.cluster.ClusterConfig;
 import net.kuujo.copycat.cluster.internal.coordinator.ClusterCoordinator;
 import net.kuujo.copycat.cluster.internal.coordinator.ClusterCoordinator;
 import net.kuujo.copycat.cluster.internal.coordinator.CoordinatorConfig;
 import net.kuujo.copycat.cluster.internal.coordinator.CoordinatorConfig;
@@ -81,4 +83,22 @@ public interface Database extends DatabaseProxy<String, byte[]>, Resource<Databa
       .addStartupTask(() -> coordinator.open().thenApply(v -> null))
       .addStartupTask(() -> coordinator.open().thenApply(v -> null))
       .addShutdownTask(coordinator::close);
       .addShutdownTask(coordinator::close);
   }
   }
+
+  /**
+   * Tells whether the database supports change notifications.
+   * @return true if notifications are supported; false otherwise
+   */
+  boolean hasChangeNotificationSupport();
+
+  /**
+   * Registers a new consumer of StateMachineUpdates.
+   * @param consumer consumer to register
+   */
+  void registerConsumer(Consumer<StateMachineUpdate> consumer);
+
+  /**
+   * Unregisters a consumer of StateMachineUpdates.
+   * @param consumer consumer to unregister
+   */
+  void unregisterConsumer(Consumer<StateMachineUpdate> consumer);
 }
 }
\ No newline at end of file

@@ -34,8 +34,12 @@ import java.util.stream.Collectors;
 import java.util.Set;
 import java.util.Set;
 
 
 import org.onlab.util.HexString;
 import org.onlab.util.HexString;
+import org.onlab.util.SharedExecutors;
 import org.onlab.util.Tools;
 import org.onlab.util.Tools;
 import org.onosproject.core.ApplicationId;
 import org.onosproject.core.ApplicationId;
+
+import static org.onosproject.store.consistent.impl.StateMachineUpdate.Target.MAP;
+
 import org.onosproject.store.service.AsyncConsistentMap;
 import org.onosproject.store.service.AsyncConsistentMap;
 import org.onosproject.store.service.ConsistentMapException;
 import org.onosproject.store.service.ConsistentMapException;
 import org.onosproject.store.service.MapEvent;
 import org.onosproject.store.service.MapEvent;
@@ -101,6 +105,17 @@ public class DefaultAsyncConsistentMap<K, V> implements AsyncConsistentMap<K, V>
         this.readOnly = readOnly;
         this.readOnly = readOnly;
         this.purgeOnUninstall = purgeOnUninstall;
         this.purgeOnUninstall = purgeOnUninstall;
         this.eventPublisher = eventPublisher;
         this.eventPublisher = eventPublisher;
+        this.database.registerConsumer(update -> {
+            SharedExecutors.getSingleThreadExecutor().execute(() -> {
+                if (update.target() == MAP) {
+                    Result<UpdateResult<String, byte[]>> result = update.output();
+                    if (result.success() && result.value().mapName().equals(name)) {
+                        MapEvent<K, V> mapEvent = result.value().<K, V>map(this::dK, serializer::decode).toMapEvent();
+                        notifyLocalListeners(mapEvent);
+                    }
+                }
+            });
+        });
     }
     }
 
 
     /**
     /**
@@ -322,7 +337,11 @@ public class DefaultAsyncConsistentMap<K, V> implements AsyncConsistentMap<K, V>
                     value == null ? null : serializer.encode(value))
                     value == null ? null : serializer.encode(value))
                 .thenApply(this::unwrapResult)
                 .thenApply(this::unwrapResult)
                 .thenApply(r -> r.<K, V>map(this::dK, serializer::decode))
                 .thenApply(r -> r.<K, V>map(this::dK, serializer::decode))
-                .whenComplete((r, e) -> notifyListeners(r != null ? r.toMapEvent() : null));
+                .whenComplete((r, e) -> {
+                    if (r != null && e == null && !database.hasChangeNotificationSupport()) {
+                        notifyListeners(r.toMapEvent());
+                    }
+                });
     }
     }
 
 
     private <T> T unwrapResult(Result<T> result) {
     private <T> T unwrapResult(Result<T> result) {
@@ -363,8 +382,10 @@ public class DefaultAsyncConsistentMap<K, V> implements AsyncConsistentMap<K, V>
     }
     }
 
 
     protected void notifyLocalListeners(MapEvent<K, V> event) {
     protected void notifyLocalListeners(MapEvent<K, V> event) {
+        if (event != null) {
             listeners.forEach(listener -> listener.event(event));
             listeners.forEach(listener -> listener.event(event));
         }
         }
+    }
 
 
     protected void notifyRemoteListeners(MapEvent<K, V> event) {
     protected void notifyRemoteListeners(MapEvent<K, V> event) {
         if (eventPublisher != null) {
         if (eventPublisher != null) {

@@ -21,23 +21,29 @@ import net.kuujo.copycat.resource.internal.AbstractResource;
 import net.kuujo.copycat.resource.internal.ResourceManager;
 import net.kuujo.copycat.resource.internal.ResourceManager;
 import net.kuujo.copycat.state.internal.DefaultStateMachine;
 import net.kuujo.copycat.state.internal.DefaultStateMachine;
 import net.kuujo.copycat.util.concurrent.Futures;
 import net.kuujo.copycat.util.concurrent.Futures;
+import net.kuujo.copycat.util.function.TriConsumer;
 
 
 import java.util.Collection;
 import java.util.Collection;
 import java.util.Map;
 import java.util.Map;
 import java.util.Set;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.function.Supplier;
 
 
 import org.onosproject.cluster.NodeId;
 import org.onosproject.cluster.NodeId;
 import org.onosproject.store.service.Transaction;
 import org.onosproject.store.service.Transaction;
 import org.onosproject.store.service.Versioned;
 import org.onosproject.store.service.Versioned;
 
 
+import com.google.common.collect.Sets;
+
 /**
 /**
  * Default database.
  * Default database.
  */
  */
 public class DefaultDatabase extends AbstractResource<Database> implements Database {
 public class DefaultDatabase extends AbstractResource<Database> implements Database {
     private final StateMachine<DatabaseState<String, byte[]>> stateMachine;
     private final StateMachine<DatabaseState<String, byte[]>> stateMachine;
     private DatabaseProxy<String, byte[]> proxy;
     private DatabaseProxy<String, byte[]> proxy;
+    private final Set<Consumer<StateMachineUpdate>> consumers = Sets.newCopyOnWriteArraySet();
+    private final TriConsumer<String, Object, Object> watcher = new InternalStateMachineWatcher();
 
 
     @SuppressWarnings({ "unchecked", "rawtypes" })
     @SuppressWarnings({ "unchecked", "rawtypes" })
     public DefaultDatabase(ResourceManager context) {
     public DefaultDatabase(ResourceManager context) {
@@ -46,6 +52,14 @@ public class DefaultDatabase extends AbstractResource<Database> implements Datab
                 DatabaseState.class,
                 DatabaseState.class,
                 DefaultDatabaseState.class,
                 DefaultDatabaseState.class,
                 DefaultDatabase.class.getClassLoader());
                 DefaultDatabase.class.getClassLoader());
+        this.stateMachine.addStartupTask(() -> {
+            stateMachine.registerWatcher(watcher);
+            return CompletableFuture.completedFuture(null);
+        });
+        this.stateMachine.addShutdownTask(() -> {
+            stateMachine.unregisterWatcher(watcher);
+            return CompletableFuture.completedFuture(null);
+        });
     }
     }
 
 
     /**
     /**
@@ -209,4 +223,27 @@ public class DefaultDatabase extends AbstractResource<Database> implements Datab
         }
         }
         return false;
         return false;
     }
     }
+
+    @Override
+    public void registerConsumer(Consumer<StateMachineUpdate> consumer) {
+        consumers.add(consumer);
+    }
+
+    @Override
+    public void unregisterConsumer(Consumer<StateMachineUpdate> consumer) {
+        consumers.remove(consumer);
+    }
+
+    @Override
+    public boolean hasChangeNotificationSupport() {
+        return true;
+    }
+
+    private class InternalStateMachineWatcher implements TriConsumer<String, Object, Object> {
+        @Override
+        public void accept(String name, Object input, Object output) {
+            StateMachineUpdate update = new StateMachineUpdate(name, input, output);
+            consumers.forEach(consumer -> consumer.accept(update));
+        }
+    }
 }
 }

@@ -25,6 +25,7 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.Collectors;
 
 
 import org.onosproject.cluster.NodeId;
 import org.onosproject.cluster.NodeId;
@@ -363,4 +364,17 @@ public class PartitionedDatabase implements Database {
     protected void setTransactionManager(TransactionManager transactionManager) {
     protected void setTransactionManager(TransactionManager transactionManager) {
         this.transactionManager = transactionManager;
         this.transactionManager = transactionManager;
     }
     }
+
+    @Override
+    public boolean hasChangeNotificationSupport() {
+        return false;
+    }
+
+    @Override
+    public void registerConsumer(Consumer<StateMachineUpdate> consumer) {
+    }
+
+    @Override
+    public void unregisterConsumer(Consumer<StateMachineUpdate> consumer) {
+    }
 }
 }

+package org.onosproject.store.consistent.impl;
+
+/**
+ * Representation of a state machine update.
+ */
+public class StateMachineUpdate {
+
+    /**
+     * Target data structure type this update is for.
+     */
+    enum Target {
+        /**
+         * Update is for a map.
+         */
+        MAP,
+
+        /**
+         * Update is for a non-map data structure.
+         */
+        OTHER
+    }
+
+    private final String operationName;
+    private final Object input;
+    private final Object output;
+
+    public StateMachineUpdate(String operationName, Object input, Object output) {
+        this.operationName = operationName;
+        this.input = input;
+        this.output = output;
+    }
+
+    public Target target() {
+        // FIXME: This check is brittle
+        if (operationName.contains("mapUpdate")) {
+            return Target.MAP;
+        } else {
+            return Target.OTHER;
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T> T input() {
+        return (T) input;
+    }
+
+    @SuppressWarnings("unchecked")
+    public <T> T output() {
+        return (T) output;
+    }
+}
\ No newline at end of file