Code clean up in ConsistentMap and LeaderElector resources

Change-Id: I1834188393f19e37394c32047538e6027522a13d

@@ -15,7 +15,6 @@
  */
 package org.onosproject.store.primitives.resources.impl;
 
-import io.atomix.catalyst.util.Listener;
 import io.atomix.copycat.client.CopycatClient;
 import io.atomix.resource.Resource;
 import io.atomix.resource.ResourceTypeInfo;
@@ -32,6 +31,21 @@ import java.util.function.Predicate;
 
 import org.onlab.util.Match;
 import org.onosproject.store.primitives.TransactionId;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Clear;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.ContainsKey;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.ContainsValue;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.EntrySet;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Get;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.IsEmpty;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.KeySet;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Listen;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Size;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionCommit;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionPrepare;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.TransactionRollback;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Unlisten;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.UpdateAndGet;
+import org.onosproject.store.primitives.resources.impl.AtomixConsistentMapCommands.Values;
 import org.onosproject.store.service.AsyncConsistentMap;
 import org.onosproject.store.service.MapEvent;
 import org.onosproject.store.service.MapEventListener;
@@ -76,48 +90,48 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
 
     @Override
     public CompletableFuture<Boolean> isEmpty() {
-        return submit(new AtomixConsistentMapCommands.IsEmpty());
+        return submit(new IsEmpty());
     }
 
     @Override
     public CompletableFuture<Integer> size() {
-        return submit(new AtomixConsistentMapCommands.Size());
+        return submit(new Size());
     }
 
     @Override
     public CompletableFuture<Boolean> containsKey(String key) {
-        return submit(new AtomixConsistentMapCommands.ContainsKey(key));
+        return submit(new ContainsKey(key));
     }
 
     @Override
     public CompletableFuture<Boolean> containsValue(byte[] value) {
-        return submit(new AtomixConsistentMapCommands.ContainsValue(value));
+        return submit(new ContainsValue(value));
     }
 
     @Override
     public CompletableFuture<Versioned<byte[]>> get(String key) {
-        return submit(new AtomixConsistentMapCommands.Get(key));
+        return submit(new Get(key));
     }
 
     @Override
     public CompletableFuture<Set<String>> keySet() {
-        return submit(new AtomixConsistentMapCommands.KeySet());
+        return submit(new KeySet());
     }
 
     @Override
     public CompletableFuture<Collection<Versioned<byte[]>>> values() {
-        return submit(new AtomixConsistentMapCommands.Values());
+        return submit(new Values());
     }
 
     @Override
     public CompletableFuture<Set<Entry<String, Versioned<byte[]>>>> entrySet() {
-        return submit(new AtomixConsistentMapCommands.EntrySet());
+        return submit(new EntrySet());
     }
 
     @Override
     @SuppressWarnings("unchecked")
     public CompletableFuture<Versioned<byte[]>> put(String key, byte[] value) {
-        return submit(new AtomixConsistentMapCommands.UpdateAndGet(key, value, Match.ANY, Match.ANY))
+        return submit(new UpdateAndGet(key, value, Match.ANY, Match.ANY))
                 .whenComplete((r, e) -> throwIfLocked(r.status()))
                 .thenApply(v -> v.oldValue());
     }
@@ -125,7 +139,7 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
     @Override
     @SuppressWarnings("unchecked")
     public CompletableFuture<Versioned<byte[]>> putAndGet(String key, byte[] value) {
-        return submit(new AtomixConsistentMapCommands.UpdateAndGet(key, value, Match.ANY, Match.ANY))
+        return submit(new UpdateAndGet(key, value, Match.ANY, Match.ANY))
                 .whenComplete((r, e) -> throwIfLocked(r.status()))
                 .thenApply(v -> v.newValue());
     }
@@ -133,14 +147,14 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
     @Override
     @SuppressWarnings("unchecked")
     public CompletableFuture<Versioned<byte[]>> putIfAbsent(String key, byte[] value) {
-        return submit(new AtomixConsistentMapCommands.UpdateAndGet(key, value, Match.NULL, Match.ANY))
+        return submit(new UpdateAndGet(key, value, Match.NULL, Match.ANY))
                 .whenComplete((r, e) -> throwIfLocked(r.status()))
                 .thenApply(v -> v.oldValue());
     }
     @Override
     @SuppressWarnings("unchecked")
     public CompletableFuture<Versioned<byte[]>> remove(String key) {
-        return submit(new AtomixConsistentMapCommands.UpdateAndGet(key, null, Match.ANY, Match.ANY))
+        return submit(new UpdateAndGet(key, null, Match.ANY, Match.ANY))
                 .whenComplete((r, e) -> throwIfLocked(r.status()))
                 .thenApply(v -> v.oldValue());
     }
@@ -148,7 +162,7 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
     @Override
     @SuppressWarnings("unchecked")
     public CompletableFuture<Boolean> remove(String key, byte[] value) {
-        return submit(new AtomixConsistentMapCommands.UpdateAndGet(key, null, Match.ifValue(value), Match.ANY))
+        return submit(new UpdateAndGet(key, null, Match.ifValue(value), Match.ANY))
                 .whenComplete((r, e) -> throwIfLocked(r.status()))
                 .thenApply(v -> v.updated());
     }
@@ -156,7 +170,7 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
     @Override
     @SuppressWarnings("unchecked")
     public CompletableFuture<Boolean> remove(String key, long version) {
-        return submit(new AtomixConsistentMapCommands.UpdateAndGet(key, null, Match.ANY, Match.ifValue(version)))
+        return submit(new UpdateAndGet(key, null, Match.ANY, Match.ifValue(version)))
                 .whenComplete((r, e) -> throwIfLocked(r.status()))
                 .thenApply(v -> v.updated());
     }
@@ -164,7 +178,7 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
     @Override
     @SuppressWarnings("unchecked")
     public CompletableFuture<Versioned<byte[]>> replace(String key, byte[] value) {
-        return submit(new AtomixConsistentMapCommands.UpdateAndGet(key, value, Match.NOT_NULL, Match.ANY))
+        return submit(new UpdateAndGet(key, value, Match.NOT_NULL, Match.ANY))
                 .whenComplete((r, e) -> throwIfLocked(r.status()))
                 .thenApply(v -> v.oldValue());
     }
@@ -172,7 +186,7 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
     @Override
     @SuppressWarnings("unchecked")
     public CompletableFuture<Boolean> replace(String key, byte[] oldValue, byte[] newValue) {
-        return submit(new AtomixConsistentMapCommands.UpdateAndGet(key,
+        return submit(new UpdateAndGet(key,
                 newValue,
                 Match.ifValue(oldValue),
                 Match.ANY))
@@ -183,7 +197,7 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
     @Override
     @SuppressWarnings("unchecked")
     public CompletableFuture<Boolean> replace(String key, long oldVersion, byte[] newValue) {
-        return submit(new AtomixConsistentMapCommands.UpdateAndGet(key,
+        return submit(new UpdateAndGet(key,
                 newValue,
                 Match.ANY,
                 Match.ifValue(oldVersion)))
@@ -193,7 +207,7 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
 
     @Override
     public CompletableFuture<Void> clear() {
-        return submit(new AtomixConsistentMapCommands.Clear())
+        return submit(new Clear())
                 .whenComplete((r, e) -> throwIfLocked(r))
                 .thenApply(v -> null);
     }
@@ -224,7 +238,7 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
             }
             Match<byte[]> valueMatch = r1 == null ? Match.NULL : Match.ANY;
             Match<Long> versionMatch = r1 == null ? Match.ANY : Match.ifValue(r1.version());
-            return submit(new AtomixConsistentMapCommands.UpdateAndGet(key,
+            return submit(new UpdateAndGet(key,
                     computedValue.get(),
                     valueMatch,
                     versionMatch))
@@ -235,18 +249,18 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
 
     @Override
     public synchronized CompletableFuture<Void> addListener(MapEventListener<String, byte[]> listener) {
-        if (!mapEventListeners.isEmpty()) {
+        if (mapEventListeners.isEmpty()) {
+            return submit(new Listen()).thenRun(() -> mapEventListeners.add(listener));
+        } else {
             mapEventListeners.add(listener);
             return CompletableFuture.completedFuture(null);
         }
-        mapEventListeners.add(listener);
-        return submit(new AtomixConsistentMapCommands.Listen()).thenApply(v -> null);
     }
 
     @Override
     public synchronized CompletableFuture<Void> removeListener(MapEventListener<String, byte[]> listener) {
         if (mapEventListeners.remove(listener) && mapEventListeners.isEmpty()) {
-            return submit(new AtomixConsistentMapCommands.Unlisten()).thenApply(v -> null);
+            return submit(new Unlisten()).thenApply(v -> null);
         }
         return CompletableFuture.completedFuture(null);
     }
@@ -259,45 +273,17 @@ public class AtomixConsistentMap extends Resource<AtomixConsistentMap>
 
     @Override
     public CompletableFuture<Boolean> prepare(MapTransaction<String, byte[]> transaction) {
-        return submit(new AtomixConsistentMapCommands.TransactionPrepare(transaction))
-                .thenApply(v -> v == PrepareResult.OK);
+        return submit(new TransactionPrepare(transaction)).thenApply(v -> v == PrepareResult.OK);
     }
 
     @Override
     public CompletableFuture<Void> commit(TransactionId transactionId) {
-        return submit(new AtomixConsistentMapCommands.TransactionCommit(transactionId))
-                .thenApply(v -> null);
+        return submit(new TransactionCommit(transactionId)).thenApply(v -> null);
     }
 
     @Override
     public CompletableFuture<Void> rollback(TransactionId transactionId) {
-        return submit(new AtomixConsistentMapCommands.TransactionRollback(transactionId))
+        return submit(new TransactionRollback(transactionId))
                 .thenApply(v -> null);
     }
-
-    /**
-     * Change listener context.
-     */
-    private final class ChangeListener implements Listener<MapEvent<String, byte[]>> {
-        private final MapEventListener<String, byte[]> listener;
-
-        private ChangeListener(MapEventListener<String, byte[]> listener) {
-            this.listener = listener;
-        }
-
-        @Override
-        public void accept(MapEvent<String, byte[]> event) {
-            listener.event(event);
-        }
-
-        @Override
-        public void close() {
-            synchronized (AtomixConsistentMap.this) {
-                mapEventListeners.remove(listener);
-                if (mapEventListeners.isEmpty()) {
-                    submit(new AtomixConsistentMapCommands.Unlisten());
-                }
-            }
-        }
-    }
 }
\ No newline at end of file

@@ -29,6 +29,15 @@ import java.util.function.Consumer;
 import org.onosproject.cluster.Leadership;
 import org.onosproject.cluster.NodeId;
 import org.onosproject.event.Change;
+import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands.Anoint;
+import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands.GetAllLeaderships;
+import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands.GetElectedTopics;
+import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands.GetLeadership;
+import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands.Listen;
+import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands.Promote;
+import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands.Run;
+import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands.Unlisten;
+import org.onosproject.store.primitives.resources.impl.AtomixLeaderElectorCommands.Withdraw;
 import org.onosproject.store.service.AsyncLeaderElector;
 
 import com.google.common.collect.Sets;
@@ -70,22 +79,22 @@ public class AtomixLeaderElector extends Resource<AtomixLeaderElector>
 
     @Override
     public CompletableFuture<Leadership> run(String topic, NodeId nodeId) {
-        return submit(new AtomixLeaderElectorCommands.Run(topic, nodeId));
+        return submit(new Run(topic, nodeId));
     }
 
     @Override
     public CompletableFuture<Void> withdraw(String topic) {
-        return submit(new AtomixLeaderElectorCommands.Withdraw(topic));
+        return submit(new Withdraw(topic));
     }
 
     @Override
     public CompletableFuture<Boolean> anoint(String topic, NodeId nodeId) {
-        return submit(new AtomixLeaderElectorCommands.Anoint(topic, nodeId));
+        return submit(new Anoint(topic, nodeId));
     }
 
     @Override
     public CompletableFuture<Boolean> promote(String topic, NodeId nodeId) {
-        return submit(new AtomixLeaderElectorCommands.Promote(topic, nodeId));
+        return submit(new Promote(topic, nodeId));
     }
 
     @Override
@@ -95,69 +104,32 @@ public class AtomixLeaderElector extends Resource<AtomixLeaderElector>
 
     @Override
     public CompletableFuture<Leadership> getLeadership(String topic) {
-        return submit(new AtomixLeaderElectorCommands.GetLeadership(topic));
+        return submit(new GetLeadership(topic));
     }
 
     @Override
     public CompletableFuture<Map<String, Leadership>> getLeaderships() {
-        return submit(new AtomixLeaderElectorCommands.GetAllLeaderships());
+        return submit(new GetAllLeaderships());
     }
 
     public CompletableFuture<Set<String>> getElectedTopics(NodeId nodeId) {
-        return submit(new AtomixLeaderElectorCommands.GetElectedTopics(nodeId));
-    }
-
-    /**
-     * Leadership change listener context.
-     */
-    private final class LeadershipChangeListener implements Listener<Change<Leadership>> {
-        private final Consumer<Change<Leadership>> listener;
-
-        private LeadershipChangeListener(Consumer<Change<Leadership>> listener) {
-            this.listener = listener;
-        }
-
-        @Override
-        public void accept(Change<Leadership> change) {
-            listener.accept(change);
-        }
-
-        @Override
-        public void close() {
-            synchronized (AtomixLeaderElector.this) {
-                submit(new AtomixLeaderElectorCommands.Unlisten());
-            }
-        }
+        return submit(new GetElectedTopics(nodeId));
     }
 
     @Override
-    public CompletableFuture<Void> addChangeListener(Consumer<Change<Leadership>> consumer) {
+    public synchronized CompletableFuture<Void> addChangeListener(Consumer<Change<Leadership>> consumer) {
+        if (leadershipChangeListeners.isEmpty()) {
+            return submit(new Listen()).thenRun(() -> leadershipChangeListeners.add(consumer));
+        } else {
             leadershipChangeListeners.add(consumer);
-        return setupListener();
-    }
-
-    @Override
-    public CompletableFuture<Void> removeChangeListener(Consumer<Change<Leadership>> consumer) {
-        leadershipChangeListeners.remove(consumer);
-        return teardownListener();
-    }
-
-    private CompletableFuture<Void> setupListener() {
-        if (listener == null && !leadershipChangeListeners.isEmpty()) {
-            Consumer<Change<Leadership>> changeConsumer = change -> {
-                leadershipChangeListeners.forEach(consumer -> consumer.accept(change));
-            };
-            return submit(new AtomixLeaderElectorCommands.Listen())
-                    .thenAccept(v -> listener = new LeadershipChangeListener(changeConsumer));
-        }
             return CompletableFuture.completedFuture(null);
         }
+    }
 
-    private CompletableFuture<Void> teardownListener() {
-        if (listener != null && leadershipChangeListeners.isEmpty()) {
-            listener.close();
-            listener = null;
-            return submit(new AtomixLeaderElectorCommands.Unlisten());
+    @Override
+    public synchronized CompletableFuture<Void> removeChangeListener(Consumer<Change<Leadership>> consumer) {
+        if (leadershipChangeListeners.remove(listener) && leadershipChangeListeners.isEmpty()) {
+            return submit(new Unlisten()).thenApply(v -> null);
         }
         return CompletableFuture.completedFuture(null);
     }