Moving meter store implementation to use map events

Change-Id: I338473b7286d7b9e5cdfb938f16c7b6155d4cbb5

@@ -20,7 +20,6 @@ import org.onosproject.net.DeviceId;
 
 import java.util.Collection;
 import java.util.Collections;
-import java.util.Optional;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
@@ -37,7 +36,6 @@ public final class DefaultMeter implements Meter, MeterEntry  {
     private final boolean burst;
     private final Collection<Band> bands;
     private final DeviceId deviceId;
-    private final Optional<MeterContext> context;
 
     private MeterState state;
     private long life;
@@ -47,14 +45,13 @@ public final class DefaultMeter implements Meter, MeterEntry  {
 
     private DefaultMeter(DeviceId deviceId, MeterId id, ApplicationId appId,
                         Unit unit, boolean burst,
-                        Collection<Band> bands, Optional<MeterContext> context) {
+                        Collection<Band> bands) {
         this.deviceId = deviceId;
         this.id = id;
         this.appId = appId;
         this.unit = unit;
         this.burst = burst;
         this.bands = bands;
-        this.context = context;
     }
 
     @Override
@@ -88,11 +85,6 @@ public final class DefaultMeter implements Meter, MeterEntry  {
     }
 
     @Override
-    public Optional<MeterContext> context() {
-        return null;
-    }
-
-    @Override
     public MeterState state() {
         return state;
     }
@@ -154,7 +146,6 @@ public final class DefaultMeter implements Meter, MeterEntry  {
         private boolean burst = false;
         private Collection<Band> bands;
         private DeviceId deviceId;
-        private Optional<MeterContext> context;
 
 
         @Override
@@ -194,19 +185,13 @@ public final class DefaultMeter implements Meter, MeterEntry  {
         }
 
         @Override
-        public Meter.Builder withContext(MeterContext context) {
-            this.context = Optional.<MeterContext>ofNullable(context);
-            return this;
-        }
-
-        @Override
         public DefaultMeter build() {
             checkNotNull(deviceId, "Must specify a device");
             checkNotNull(bands, "Must have bands.");
             checkArgument(bands.size() > 0, "Must have at least one band.");
             checkNotNull(appId, "Must have an application id");
             checkNotNull(id, "Must specify a meter id");
-            return new DefaultMeter(deviceId, id, appId, unit, burst, bands, context);
+            return new DefaultMeter(deviceId, id, appId, unit, burst, bands);
         }
 
 

@@ -19,7 +19,6 @@ import org.onosproject.core.ApplicationId;
 import org.onosproject.net.DeviceId;
 
 import java.util.Collection;
-import java.util.Optional;
 
 /**
  * Represents a generalized meter to be deployed on a device.
@@ -81,14 +80,6 @@ public interface Meter {
     Collection<Band> bands();
 
     /**
-     * Obtains an optional context.
-     *
-     * @return optional; which will be empty if there is no context.
-     * Otherwise it will return the context.
-     */
-    Optional<MeterContext> context();
-
-    /**
      * Fetches the state of this meter.
      *
      * @return a meter state
@@ -177,8 +168,6 @@ public interface Meter {
          */
         Builder withBands(Collection<Band> bands);
 
-        Builder withContext(MeterContext context);
-
         /**
          * Builds the meter based on the specified parameters.
          *

@@ -16,23 +16,23 @@
 package org.onosproject.incubator.net.meter;
 
 /**
- * Created by ash on 01/08/15.
+ * A context permitting the application to be notified when the
+ * meter installation has been successful.
  */
 public interface MeterContext {
 
     /**
      * Invoked on successful installation of the meter.
      *
-     * @param op a meter operation
+     * @param op a meter
      */
-    default void onSuccess(MeterOperation op) {
-    }
+    default void onSuccess(Meter op) {}
 
     /**
      * Invoked when error is encountered while installing a meter.
      *
-     * @param op a meter operation
+     * @param op a meter
      * @param reason the reason why it failed
      */
-    default void onError(MeterOperation op, MeterFailReason reason) {}
+    default void onError(Meter op, MeterFailReason reason) {}
 }

@@ -20,28 +20,19 @@ import org.onosproject.event.AbstractEvent;
 /**
  * Entity that represents Meter events.
  */
-public class MeterEvent extends AbstractEvent<MeterEvent.Type, MeterOperation> {
+public class MeterEvent extends AbstractEvent<MeterEvent.Type, Meter> {
 
 
-    private final MeterFailReason reason;
-
     public enum Type {
-
-        /**
-         * Signals that a meter has been added.
-         */
-        METER_UPDATED,
-
         /**
-         * Signals that a meter update failed.
+         * A meter addition was requested.
          */
-        METER_OP_FAILED,
+        METER_ADD_REQ,
 
         /**
-         * A meter operation was requested.
+         * A meter removal was requested.
          */
-        METER_OP_REQ,
-
+        METER_REM_REQ
     }
 
 
@@ -50,32 +41,22 @@ public class MeterEvent extends AbstractEvent<MeterEvent.Type, MeterOperation> {
      * current time.
      *
      * @param type  meter event type
-     * @param op event subject
+     * @param meter event subject
      */
-    public MeterEvent(Type type, MeterOperation op) {
-        super(type, op);
-        this.reason = null;
+    public MeterEvent(Type type, Meter meter) {
+        super(type, meter);
     }
 
     /**
      * Creates an event of a given type and for the specified meter and time.
      *
      * @param type  meter event type
-     * @param op event subject
+     * @param meter event subject
      * @param time  occurrence time
      */
-    public MeterEvent(Type type, MeterOperation op, long time) {
-        super(type, op, time);
-        this.reason = null;
-    }
-
-    public MeterEvent(Type type, MeterOperation op, MeterFailReason reason) {
-        super(type, op);
-        this.reason = reason;
+    public MeterEvent(Type type, Meter meter, long time) {
+        super(type, meter, time);
     }
 
-    public MeterFailReason reason() {
-        return reason;
-    }
 
 }

@@ -17,11 +17,15 @@ package org.onosproject.incubator.net.meter;
 
 import com.google.common.base.MoreObjects;
 
+import java.util.Optional;
+
 /**
  * Representation of an operation on the meter table.
  */
 public class MeterOperation {
 
+    private final Optional<MeterContext> context;
+
     /**
      * Tyoe of meter operation.
      */
@@ -35,9 +39,10 @@ public class MeterOperation {
     private final Type type;
 
 
-    public MeterOperation(Meter meter, Type type) {
+    public MeterOperation(Meter meter, Type type, MeterContext context) {
         this.meter = meter;
         this.type = type;
+        this.context = Optional.ofNullable(context);
     }
 
     /**
@@ -58,6 +63,16 @@ public class MeterOperation {
         return meter;
     }
 
+    /**
+     * Returns a context which allows application to
+     * be notified on the success value of this operation.
+     *
+     * @return a meter context
+     */
+    public Optional<MeterContext> context() {
+        return this.context;
+    }
+
     @Override
     public String toString() {
         return MoreObjects.toStringHelper(this)

@@ -30,28 +30,21 @@ public interface MeterService
      *
      * @param meter a meter.
      */
-    void addMeter(Meter meter);
+    void addMeter(MeterOperation meter);
 
     /**
      * Updates a meter by adding statistic information to the meter.
      *
      * @param meter an updated meter
      */
-    void updateMeter(Meter meter);
+    void updateMeter(MeterOperation meter);
 
     /**
      * Remove a meter from the system and the dataplane.
      *
      * @param meter a meter to remove
      */
-    void removeMeter(Meter meter);
-
-    /**
-     * Remove a meter from the system and the dataplane by meter id.
-     *
-     * @param id a meter id
-     */
-    void removeMeter(MeterId id);
+    void removeMeter(MeterOperation meter);
 
     /**
      * Fetch the meter by the meter id.

@@ -18,6 +18,7 @@ package org.onosproject.incubator.net.meter;
 import org.onosproject.store.Store;
 
 import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
 
 /**
  * Entity that stores and distributed meter objects.
@@ -28,25 +29,29 @@ public interface MeterStore extends Store<MeterEvent, MeterStoreDelegate> {
      * Adds a meter to the store.
      *
      * @param meter a meter
+     * @return a future indicating the result of the store operation
      */
-    void storeMeter(Meter meter);
+    CompletableFuture<MeterStoreResult> storeMeter(Meter meter);
 
     /**
      * Deletes a meter from the store.
      *
      * @param meter a meter
+     * @return a future indicating the result of the store operation
      */
-    void deleteMeter(Meter meter);
+    CompletableFuture<MeterStoreResult> deleteMeter(Meter meter);
 
     /**
      * Updates a meter whose meter id is the same as the passed meter.
      *
      * @param meter a new meter
+     * @return a future indicating the result of the store operation
      */
-    void updateMeter(Meter meter);
+    CompletableFuture<MeterStoreResult> updateMeter(Meter meter);
 
     /**
      * Updates a given meter's state with the provided state.
+     *
      * @param meter a meter
      */
     void updateMeterState(Meter meter);

+/*
+ * Copyright 2015 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.incubator.net.meter;
+
+import java.util.Optional;
+
+/**
+ * An entity used to indicate whether the store operation passed.
+ */
+public final class MeterStoreResult {
+
+
+    private final Type type;
+    private final Optional<MeterFailReason> reason;
+
+    public enum Type {
+        SUCCESS,
+        FAIL
+    }
+
+    private MeterStoreResult(Type type, MeterFailReason reason) {
+        this.type = type;
+        this.reason = Optional.ofNullable(reason);
+    }
+
+    public Type type() {
+        return type;
+    }
+
+    public Optional<MeterFailReason> reason() {
+        return reason;
+    }
+
+    /**
+     * A successful store opertion.
+     *
+     * @return a meter store result
+     */
+    public static MeterStoreResult success() {
+        return new MeterStoreResult(Type.SUCCESS, null);
+    }
+
+    /**
+     * A failed store operation.
+     *
+     * @param reason a failure reason
+     * @return a meter store result
+     */
+    public static MeterStoreResult fail(MeterFailReason reason) {
+        return new MeterStoreResult(Type.FAIL, reason);
+    }
+
+}

@@ -21,6 +21,7 @@ import org.apache.felix.scr.annotations.Deactivate;
 import org.apache.felix.scr.annotations.Reference;
 import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.apache.felix.scr.annotations.Service;
+import org.onlab.util.TriConsumer;
 import org.onosproject.incubator.net.meter.DefaultMeter;
 import org.onosproject.incubator.net.meter.Meter;
 import org.onosproject.incubator.net.meter.MeterEvent;
@@ -35,6 +36,7 @@ import org.onosproject.incubator.net.meter.MeterService;
 import org.onosproject.incubator.net.meter.MeterState;
 import org.onosproject.incubator.net.meter.MeterStore;
 import org.onosproject.incubator.net.meter.MeterStoreDelegate;
+import org.onosproject.incubator.net.meter.MeterStoreResult;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.provider.AbstractListenerProviderRegistry;
 import org.onosproject.net.provider.AbstractProviderService;
@@ -44,7 +46,6 @@ import org.slf4j.Logger;
 
 import java.util.Collection;
 
-import static com.google.common.base.Preconditions.checkNotNull;
 import static org.slf4j.LoggerFactory.getLogger;
 
 
@@ -69,11 +70,29 @@ public class MeterManager extends AbstractListenerProviderRegistry<MeterEvent, M
 
     private AtomicCounter meterIdCounter;
 
+    private TriConsumer<MeterOperation, MeterStoreResult, Throwable> onComplete;
+
     @Activate
     public void activate() {
         meterIdCounter = storageService.atomicCounterBuilder()
                 .withName(meterIdentifier)
                 .build();
+
+        onComplete = (op, result, error) ->
+            {
+                op.context().ifPresent(c -> {
+                    if (error != null) {
+                        c.onError(op.meter(), MeterFailReason.UNKNOWN);
+                    } else {
+                        if (result.reason().isPresent()) {
+                            c.onError(op.meter(), result.reason().get());
+                        } else {
+                            c.onSuccess(op.meter());
+                        }
+                    }
+                });
+
+            };
         log.info("Started");
     }
 
@@ -88,31 +107,27 @@ public class MeterManager extends AbstractListenerProviderRegistry<MeterEvent, M
     }
 
     @Override
-    public void addMeter(Meter meter) {
-        DefaultMeter m = (DefaultMeter) meter;
+    public void addMeter(MeterOperation op) {
+        DefaultMeter m = (DefaultMeter) op.meter();
         m.setState(MeterState.PENDING_ADD);
-        store.storeMeter(m);
+        store.storeMeter(m).whenComplete((result, error) ->
+                                                 onComplete.accept(op, result, error));
     }
 
     @Override
-    public void updateMeter(Meter meter) {
-        DefaultMeter m = (DefaultMeter) meter;
+    public void updateMeter(MeterOperation op) {
+        DefaultMeter m = (DefaultMeter) op.meter();
         m.setState(MeterState.PENDING_ADD);
-        store.updateMeter(m);
+        store.updateMeter(m).whenComplete((result, error) ->
+                                                  onComplete.accept(op, result, error));
     }
 
     @Override
-    public void removeMeter(Meter meter) {
-        DefaultMeter m = (DefaultMeter) meter;
+    public void removeMeter(MeterOperation op) {
+        DefaultMeter m = (DefaultMeter) op.meter();
         m.setState(MeterState.PENDING_REMOVE);
-        store.deleteMeter(m);
-    }
-
-    @Override
-    public void removeMeter(MeterId id) {
-        DefaultMeter meter = (DefaultMeter) store.getMeter(id);
-        checkNotNull(meter, "No such meter {}", id);
-        removeMeter(meter);
+        store.deleteMeter(m).whenComplete((result, error) ->
+                                                  onComplete.accept(op, result, error));
     }
 
     @Override
@@ -155,17 +170,18 @@ public class MeterManager extends AbstractListenerProviderRegistry<MeterEvent, M
 
         @Override
         public void notify(MeterEvent event) {
-            DeviceId deviceId = event.subject().meter().deviceId();
-            MeterProvider p = getProvider(event.subject().meter().deviceId());
+            DeviceId deviceId = event.subject().deviceId();
+            MeterProvider p = getProvider(event.subject().deviceId());
             switch (event.type()) {
-                case METER_UPDATED:
-                    break;
-                case METER_OP_FAILED:
-                    event.subject().meter().context().ifPresent(c ->
-                        c.onError(event.subject(), event.reason()));
+                case METER_ADD_REQ:
+                    p.performMeterOperation(deviceId, new MeterOperation(event.subject(),
+                                                                         MeterOperation.Type.ADD,
+                                                                         null));
                     break;
-                case METER_OP_REQ:
-                    p.performMeterOperation(deviceId, event.subject());
+                case METER_REM_REQ:
+                    p.performMeterOperation(deviceId, new MeterOperation(event.subject(),
+                                                                         MeterOperation.Type.REMOVE,
+                                                                         null));
                     break;
                 default:
                     log.warn("Unknown meter event {}", event.type());

@@ -15,38 +15,40 @@
  */
 package org.onosproject.incubator.store.meter.impl;
 
+import com.google.common.collect.Collections2;
+import com.google.common.collect.Maps;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Deactivate;
-import org.apache.felix.scr.annotations.Property;
 import org.apache.felix.scr.annotations.Reference;
 import org.apache.felix.scr.annotations.ReferenceCardinality;
-import org.onlab.util.KryoNamespace;
-import org.onlab.util.Tools;
 import org.onosproject.cluster.ClusterService;
 import org.onosproject.cluster.NodeId;
-import org.onosproject.incubator.net.meter.DefaultBand;
 import org.onosproject.incubator.net.meter.DefaultMeter;
 import org.onosproject.incubator.net.meter.Meter;
 import org.onosproject.incubator.net.meter.MeterEvent;
 import org.onosproject.incubator.net.meter.MeterFailReason;
 import org.onosproject.incubator.net.meter.MeterId;
 import org.onosproject.incubator.net.meter.MeterOperation;
+import org.onosproject.incubator.net.meter.MeterState;
 import org.onosproject.incubator.net.meter.MeterStore;
 import org.onosproject.incubator.net.meter.MeterStoreDelegate;
+import org.onosproject.incubator.net.meter.MeterStoreResult;
 import org.onosproject.mastership.MastershipService;
 import org.onosproject.store.AbstractStore;
-import org.onosproject.store.cluster.messaging.ClusterCommunicationService;
-import org.onosproject.store.cluster.messaging.MessageSubject;
+import org.onosproject.store.serializers.KryoNamespaces;
 import org.onosproject.store.service.ConsistentMap;
+import org.onosproject.store.service.MapEvent;
+import org.onosproject.store.service.MapEventListener;
 import org.onosproject.store.service.Serializer;
+import org.onosproject.store.service.StorageException;
 import org.onosproject.store.service.StorageService;
+import org.onosproject.store.service.Versioned;
 import org.slf4j.Logger;
 
+import java.util.Arrays;
 import java.util.Collection;
-import java.util.Objects;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.stream.Collectors;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 
 import static org.slf4j.LoggerFactory.getLogger;
 
@@ -60,54 +62,37 @@ public class DistributedMeterStore extends AbstractStore<MeterEvent, MeterStoreD
     private Logger log = getLogger(getClass());
 
     private static final String METERSTORE = "onos-meter-store";
-    private static final int MESSAGE_HANDLER_THREAD_POOL_SIZE = 8;
-
-    private static final MessageSubject UPDATE_METER = new MessageSubject("peer-mod-meter");
-
-
-    @Property(name = "msgHandlerPoolSize", intValue = MESSAGE_HANDLER_THREAD_POOL_SIZE,
-            label = "Number of threads in the message handler pool")
-    private int msgPoolSize;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     private StorageService storageService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
-    private ClusterCommunicationService clusterCommunicationService;
-
-    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     private MastershipService mastershipService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     private ClusterService clusterService;
 
-    private ConsistentMap<MeterId, Meter> meters;
+    private ConsistentMap<MeterId, MeterData> meters;
     private NodeId local;
-    private KryoNamespace kryoNameSpace;
 
-    private Serializer serializer;
+    private MapEventListener mapListener = new InternalMapEventListener();
+
+    private Map<MeterId, CompletableFuture<MeterStoreResult>> futures =
+            Maps.newConcurrentMap();
 
     @Activate
     public void activate() {
 
         local = clusterService.getLocalNode().id();
 
-        kryoNameSpace =
-                KryoNamespace.newBuilder()
-                                .register(DefaultMeter.class)
-                                .register(DefaultBand.class)
-                                .build();
-
-        serializer = Serializer.using(kryoNameSpace);
 
-        meters = storageService.<MeterId, Meter>consistentMapBuilder()
+        meters = storageService.<MeterId, MeterData>consistentMapBuilder()
                     .withName(METERSTORE)
-                    .withSerializer(serializer)
+                    .withSerializer(Serializer.using(Arrays.asList(KryoNamespaces.API),
+                                                     MeterData.class))
                     .build();
 
-        ExecutorService executors = Executors.newFixedThreadPool(
-                msgPoolSize, Tools.groupedThreads("onos/store/meter", "message-handlers"));
-        registerMessageHandlers(executors);
+        meters.addListener(mapListener);
 
         log.info("Started");
     }
@@ -115,159 +100,133 @@ public class DistributedMeterStore extends AbstractStore<MeterEvent, MeterStoreD
     @Deactivate
     public void deactivate() {
 
-
+        meters.removeListener(mapListener);
         log.info("Stopped");
     }
 
-    private void registerMessageHandlers(ExecutorService executor) {
-        clusterCommunicationService.<MeterEvent>addSubscriber(UPDATE_METER, kryoNameSpace::deserialize,
-                                                              this::notifyDelegate, executor);
-
-    }
-
 
     @Override
-    public void storeMeter(Meter meter) {
-        NodeId master = mastershipService.getMasterFor(meter.deviceId());
-
-        meters.put(meter.id(), meter);
-
-        MeterEvent event = new MeterEvent(MeterEvent.Type.METER_OP_REQ,
-                                          new MeterOperation(meter, MeterOperation.Type.ADD));
-        if (Objects.equals(local, master)) {
-            notifyDelegate(event);
-        } else {
-            clusterCommunicationService.unicast(
-                    event,
-                    UPDATE_METER,
-                    serializer::encode,
-                    master
-            ).whenComplete((result, error) -> {
-                if (error != null) {
-                    log.warn("Failed to install meter {} because {} on {}",
-                             meter, error, master);
-
-                    // notify app of failure
-                    meter.context().ifPresent(c -> c.onError(
-                            event.subject(), MeterFailReason.UNKNOWN));
-                }
-            });
+    public CompletableFuture<MeterStoreResult> storeMeter(Meter meter) {
+        CompletableFuture<MeterStoreResult> future = new CompletableFuture<>();
+        futures.put(meter.id(), future);
+        MeterData data = new MeterData(meter, null, local);
+
+        try {
+            meters.put(meter.id(), data);
+        } catch (StorageException e) {
+            future.completeExceptionally(e);
         }
 
+        return future;
+
     }
 
     @Override
-    public void deleteMeter(Meter meter) {
+    public CompletableFuture<MeterStoreResult> deleteMeter(Meter meter) {
+        CompletableFuture<MeterStoreResult> future = new CompletableFuture<>();
+        futures.put(meter.id(), future);
 
-        NodeId master = mastershipService.getMasterFor(meter.deviceId());
+        MeterData data = new MeterData(meter, null, local);
 
         // update the state of the meter. It will be pruned by observing
         // that it has been removed from the dataplane.
-        meters.put(meter.id(), meter);
-
-        MeterEvent event = new MeterEvent(MeterEvent.Type.METER_OP_REQ,
-                                          new MeterOperation(meter, MeterOperation.Type.REMOVE));
-        if (Objects.equals(local, master)) {
-            notifyDelegate(event);
-        } else {
-            clusterCommunicationService.unicast(
-                    event,
-                    UPDATE_METER,
-                    serializer::encode,
-                    master
-            ).whenComplete((result, error) -> {
-                if (error != null) {
-                    log.warn("Failed to delete meter {} because {} on {}",
-                             meter, error, master);
-
-                    // notify app of failure
-                    meter.context().ifPresent(c -> c.onError(
-                            event.subject(), MeterFailReason.UNKNOWN));
-                }
-            });
+        try {
+            meters.put(meter.id(), data);
+        } catch (StorageException e) {
+            future.completeExceptionally(e);
         }
 
+
+        return future;
     }
 
     @Override
-    public void updateMeter(Meter meter) {
-
-        NodeId master = mastershipService.getMasterFor(meter.deviceId());
-
-        meters.put(meter.id(), meter);
-
-        MeterEvent event = new MeterEvent(MeterEvent.Type.METER_OP_REQ,
-                                          new MeterOperation(meter, MeterOperation.Type.MODIFY));
-        if (Objects.equals(local, master)) {
-            notifyDelegate(event);
-        } else {
-            clusterCommunicationService.unicast(
-                    event,
-                    UPDATE_METER,
-                    serializer::encode,
-                    master
-            ).whenComplete((result, error) -> {
-                if (error != null) {
-                    log.warn("Failed to update meter {} because {} on {}",
-                             meter, error, master);
-
-                    // notify app of failure
-                    meter.context().ifPresent(c -> c.onError(
-                            event.subject(), MeterFailReason.UNKNOWN));
-                }
-            });
+    public CompletableFuture<MeterStoreResult> updateMeter(Meter meter) {
+        CompletableFuture<MeterStoreResult> future = new CompletableFuture<>();
+        futures.put(meter.id(), future);
+
+        MeterData data = new MeterData(meter, null, local);
+        try {
+            meters.put(meter.id(), data);
+        } catch (StorageException e) {
+            future.completeExceptionally(e);
         }
-
+        return future;
     }
 
     @Override
     public void updateMeterState(Meter meter) {
-        meters.compute(meter.id(), (id, v) -> {
-            DefaultMeter m = (DefaultMeter) v;
+        meters.computeIfPresent(meter.id(), (id, v) -> {
+            DefaultMeter m = (DefaultMeter) v.meter();
             m.setState(meter.state());
             m.setProcessedPackets(meter.packetsSeen());
             m.setProcessedBytes(meter.bytesSeen());
             m.setLife(meter.life());
+            // TODO: Prune if drops to zero.
             m.setReferenceCount(meter.referenceCount());
-            return m;
+            return new MeterData(m, null, v.origin());
         });
     }
 
     @Override
     public Meter getMeter(MeterId meterId) {
-        return meters.get(meterId).value();
+        MeterData data = Versioned.valueOrElse(meters.get(meterId), null);
+        return data == null ? null : data.meter();
     }
 
     @Override
     public Collection<Meter> getAllMeters() {
-        return meters.values().stream()
-                .map(v -> v.value()).collect(Collectors.toSet());
+        return Collections2.transform(meters.asJavaMap().values(),
+                                      MeterData::meter);
     }
 
     @Override
     public void failedMeter(MeterOperation op, MeterFailReason reason) {
-        NodeId master = mastershipService.getMasterFor(op.meter().deviceId());
-        meters.remove(op.meter().id());
-
-        MeterEvent event = new MeterEvent(MeterEvent.Type.METER_OP_FAILED, op, reason);
-        if (Objects.equals(local, master)) {
-            notifyDelegate(event);
-        } else {
-            clusterCommunicationService.unicast(
-                    event,
-                    UPDATE_METER,
-                    serializer::encode,
-                    master
-            ).whenComplete((result, error) -> {
-                if (error != null) {
-                    log.warn("Failed to delete failed meter {} because {} on {}",
-                             op.meter(), error, master);
-
-                    // Can't do any more...
-                }
-            });
-        }
+        meters.computeIfPresent(op.meter().id(), (k, v) ->
+                new MeterData(v.meter(), reason, v.origin()));
+    }
+
+    private class InternalMapEventListener implements MapEventListener<MeterId, MeterData> {
+        @Override
+        public void event(MapEvent<MeterId, MeterData> event) {
+            MeterData data = event.value().value();
+            NodeId master = mastershipService.getMasterFor(data.meter().deviceId());
+            switch (event.type()) {
+                case INSERT:
+                case UPDATE:
+                        switch (data.meter().state()) {
+                            case PENDING_ADD:
+                            case PENDING_REMOVE:
+                                if (!data.reason().isPresent() && local.equals(master)) {
+                                    notifyDelegate(
+                                            new MeterEvent(data.meter().state() == MeterState.PENDING_ADD ?
+                                                    MeterEvent.Type.METER_ADD_REQ : MeterEvent.Type.METER_REM_REQ,
+                                                                  data.meter()));
+                                } else if (data.reason().isPresent() && local.equals(data.origin())) {
+                                    MeterStoreResult msr = MeterStoreResult.fail(data.reason().get());
+                                    //TODO: No future -> no friend
+                                    futures.get(data.meter().id()).complete(msr);
+                                }
+                                break;
+                            case ADDED:
+                            case REMOVED:
+                                if (local.equals(data.origin())) {
+                                    futures.get(data.meter().id()).complete(MeterStoreResult.success());
+                                }
+                                break;
+                            default:
+                                log.warn("Unknown meter state type {}", data.meter().state());
+                        }
+                    break;
+                case REMOVE:
+                    //Only happens at origin so we do not need to care.
+                    break;
+                default:
+                    log.warn("Unknown Map event type {}", event.type());
+            }
 
+        }
     }
 
+
 }

+/*
+ * Copyright 2015 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.incubator.store.meter.impl;
+
+import org.onosproject.cluster.NodeId;
+import org.onosproject.incubator.net.meter.Meter;
+import org.onosproject.incubator.net.meter.MeterFailReason;
+
+import java.util.Optional;
+
+/**
+ * A class representing the meter information stored in the meter store.
+ */
+public class MeterData {
+
+    private final Meter meter;
+    private final Optional<MeterFailReason> reason;
+    private final NodeId origin;
+
+    public MeterData(Meter meter, MeterFailReason reason, NodeId origin) {
+        this.meter = meter;
+        this.reason = Optional.ofNullable(reason);
+        this.origin = origin;
+    }
+
+    public Meter meter() {
+        return meter;
+    }
+
+    public Optional<MeterFailReason> reason() {
+        return this.reason;
+    }
+
+    public NodeId origin() {
+        return this.origin;
+    }
+
+
+}

+/*
+ * Copyright 2015 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.onlab.util;
+
+/**
+ * A consumer that accepts three arguments.
+ */
+public interface TriConsumer<U, V, W> {
+
+    /**
+     * Applies the given arguments to the function.
+     */
+    void accept(U arg1, V arg2, W arg3);
+
+}
\ No newline at end of file