commit 0463cf98ee58802f991b756e904d3b8a6faa3992
author Madan Jampani <madan@onlab.us> Wed May 04 14:46:08 2016 -0700
committer Gerrit Code Review <gerrit@onlab.us> Thu May 05 21:36:37 2016 +0000
tree a8d3984165ecf80031daf53b01f63ded162e3b17
parent 4abe844367210fab8cc9e816fda0cd9a2fa05b85

New API for specifying an executor when registering a map listener

Change-Id: I1fc92e0a3da576d88d5ece4a666af8ad1c1fb9d8

core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMap.java

diff --git a/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMap.java b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMap.java
index 2cb4a50..9bad652 100644
--- a/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMap.java
+++ b/core/store/primitives/src/main/java/org/onosproject/store/primitives/resources/impl/AtomixConsistentMap.java
@@ -22,10 +22,12 @@
 import java.util.Collection;
 import java.util.ConcurrentModificationException;
 import java.util.List;
+import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Properties;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -54,7 +56,9 @@
 import org.onosproject.store.service.MapEventListener;
 import org.onosproject.store.service.MapTransaction;
 import org.onosproject.store.service.Versioned;
+
 import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 
 /**
@@ -65,7 +69,7 @@
     implements AsyncConsistentMap<String, byte[]> {
 
     private final Set<Consumer<Status>> statusChangeListeners = Sets.newCopyOnWriteArraySet();
-    private final Set<MapEventListener<String, byte[]>> mapEventListeners = Sets.newCopyOnWriteArraySet();
+    private final Map<MapEventListener<String, byte[]>, Executor> mapEventListeners = Maps.newIdentityHashMap();
 
     public static final String CHANGE_SUBJECT = "changeEvents";
 
@@ -87,7 +91,8 @@
     }
 
     private void handleEvent(List<MapEvent<String, byte[]>> events) {
-        events.forEach(event -> mapEventListeners.forEach(listener -> listener.event(event)));
+        events.forEach(event ->
+            mapEventListeners.forEach((listener, executor) -> executor.execute(() -> listener.event(event))));
     }
 
     @Override
@@ -250,18 +255,19 @@
     }
 
     @Override
-    public synchronized CompletableFuture<Void> addListener(MapEventListener<String, byte[]> listener) {
+    public synchronized CompletableFuture<Void> addListener(MapEventListener<String, byte[]> listener,
+                                                            Executor executor) {
         if (mapEventListeners.isEmpty()) {
-            return submit(new Listen()).thenRun(() -> mapEventListeners.add(listener));
+            return submit(new Listen()).thenRun(() -> mapEventListeners.putIfAbsent(listener, executor));
         } else {
-            mapEventListeners.add(listener);
+            mapEventListeners.put(listener, executor);
             return CompletableFuture.completedFuture(null);
         }
     }
 
     @Override
     public synchronized CompletableFuture<Void> removeListener(MapEventListener<String, byte[]> listener) {
-        if (mapEventListeners.remove(listener) && mapEventListeners.isEmpty()) {
+        if (mapEventListeners.remove(listener) != null && mapEventListeners.isEmpty()) {
             return submit(new Unlisten()).thenApply(v -> null);
         }
         return CompletableFuture.completedFuture(null);