Adding tests for meter service.

Change-Id: Ic220fdaa89b421661019f1a6ef23c7b02eef127c

@@ -22,10 +22,9 @@ import org.onosproject.core.CoreService;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.meter.Band;
 import org.onosproject.net.meter.DefaultBand;
-import org.onosproject.net.meter.DefaultMeter;
+import org.onosproject.net.meter.DefaultMeterRequest;
 import org.onosproject.net.meter.Meter;
-import org.onosproject.net.meter.MeterId;
-import org.onosproject.net.meter.MeterOperation;
+import org.onosproject.net.meter.MeterRequest;
 import org.onosproject.net.meter.MeterService;
 
 import java.util.Collections;
@@ -41,7 +40,7 @@ public class AddMeter extends AbstractShellCommand {
             required = true, multiValued = false)
     String uri = null;
 
-    private final String appId = "org.onosproject.cli.addMeter";
+    private final String appId = "org.onosproject.cli.meterCmd";
 
     @Override
     protected void execute() {
@@ -50,25 +49,20 @@ public class AddMeter extends AbstractShellCommand {
 
         DeviceId deviceId = DeviceId.deviceId(uri);
 
-        MeterId meterId = service.allocateMeterId();
-
         Band band = DefaultBand.builder()
                         .ofType(Band.Type.DROP)
                         .withRate(500)
                         .build();
 
 
-        Meter meter = DefaultMeter.builder()
+        MeterRequest request = DefaultMeterRequest.builder()
                 .forDevice(deviceId)
                 .fromApp(coreService.registerApplication(appId))
-                .withId(meterId)
                 .withUnit(Meter.Unit.KB_PER_SEC)
                 .withBands(Collections.singleton(band))
-                .build();
-
-        MeterOperation op = new MeterOperation(meter, MeterOperation.Type.ADD, null);
+                .add();
 
-        service.addMeter(op);
+        service.submit(request);
 
     }
 }

@@ -15,6 +15,7 @@
  */
 package org.onosproject.net.meter;
 
+import com.google.common.base.Objects;
 import com.google.common.collect.ImmutableSet;
 import org.onosproject.core.ApplicationId;
 import org.onosproject.net.DeviceId;
@@ -151,6 +152,26 @@ public final class DefaultMeter implements Meter, MeterEntry  {
                 .add("bands", bands).toString();
     }
 
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        DefaultMeter that = (DefaultMeter) o;
+        return Objects.equal(id, that.id) &&
+                Objects.equal(appId, that.appId) &&
+                Objects.equal(unit, that.unit) &&
+                Objects.equal(deviceId, that.deviceId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(id, appId, unit, deviceId);
+    }
+
     public static final class Builder implements Meter.Builder {
 
         private MeterId id;

+/*
+ * 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.net.meter;
+
+import com.google.common.collect.ImmutableSet;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.net.DeviceId;
+
+import java.util.Collection;
+import java.util.Optional;
+
+import static com.google.common.base.MoreObjects.toStringHelper;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * A default implementation of a meter.
+ */
+public final class DefaultMeterRequest implements MeterRequest {
+
+
+
+    private final ApplicationId appId;
+    private final Meter.Unit unit;
+    private final boolean burst;
+    private final Collection<Band> bands;
+    private final DeviceId deviceId;
+    private final Optional<MeterContext> context;
+    private final Type op;
+
+    private DefaultMeterRequest(DeviceId deviceId, ApplicationId appId,
+                                Meter.Unit unit, boolean burst,
+                                Collection<Band> bands, MeterContext context, Type op) {
+        this.deviceId = deviceId;
+        this.appId = appId;
+        this.unit = unit;
+        this.burst = burst;
+        this.bands = bands;
+        this.context = Optional.ofNullable(context);
+        this.op = op;
+    }
+
+    @Override
+    public DeviceId deviceId() {
+        return deviceId;
+    }
+
+    @Override
+    public ApplicationId appId() {
+        return appId;
+    }
+
+    @Override
+    public Meter.Unit unit() {
+        return unit;
+    }
+
+    @Override
+    public boolean isBurst() {
+        return burst;
+    }
+
+    @Override
+    public Collection<Band> bands() {
+        return bands;
+    }
+
+    @Override
+    public Optional<MeterContext> context() {
+        return context;
+    }
+
+
+
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    @Override
+    public String toString() {
+        return toStringHelper(this)
+                .add("device", deviceId)
+                .add("appId", appId.name())
+                .add("unit", unit)
+                .add("isBurst", burst)
+                .add("bands", bands).toString();
+    }
+
+    public static final class Builder implements MeterRequest.Builder {
+
+        private ApplicationId appId;
+        private Meter.Unit unit = Meter.Unit.KB_PER_SEC;
+        private boolean burst = false;
+        private Collection<Band> bands;
+        private DeviceId deviceId;
+        private MeterContext context;
+
+
+        @Override
+        public MeterRequest.Builder forDevice(DeviceId deviceId) {
+            this.deviceId = deviceId;
+            return this;
+        }
+
+        @Override
+        public MeterRequest.Builder fromApp(ApplicationId appId) {
+            this.appId = appId;
+            return this;
+        }
+
+        @Override
+        public MeterRequest.Builder  withUnit(Meter.Unit unit) {
+            this.unit = unit;
+            return this;
+        }
+
+        @Override
+        public MeterRequest.Builder  burst() {
+            this.burst = true;
+            return this;
+        }
+
+        @Override
+        public MeterRequest.Builder  withBands(Collection<Band> bands) {
+            this.bands = ImmutableSet.copyOf(bands);
+            return this;
+        }
+
+        @Override
+        public MeterRequest.Builder withContext(MeterContext context) {
+            this.context = context;
+            return this;
+        }
+
+        @Override
+        public MeterRequest add() {
+            validate();
+            return new DefaultMeterRequest(deviceId, appId, unit, burst, bands,
+                                           context, Type.ADD);
+        }
+
+        @Override
+        public MeterRequest remove() {
+            validate();
+            return new DefaultMeterRequest(deviceId, appId, unit, burst, bands,
+                                           context, Type.REMOVE);
+        }
+
+        private void validate() {
+            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");
+        }
+
+
+    }
+}

@@ -130,7 +130,7 @@ public interface Meter {
         /**
          * Assigns the id to this meter.
          *
-         * @param id a meter id
+         * @param id a e
          * @return this
          */
         Builder withId(MeterId id);

@@ -26,7 +26,7 @@ public interface MeterContext {
      *
      * @param op a meter
      */
-    default void onSuccess(Meter op) {}
+    default void onSuccess(MeterRequest op) {}
 
     /**
      * Invoked when error is encountered while installing a meter.
@@ -34,5 +34,5 @@ public interface MeterContext {
      * @param op a meter
      * @param reason the reason why it failed
      */
-    default void onError(Meter op, MeterFailReason reason) {}
+    default void onError(MeterRequest op, MeterFailReason reason) {}
 }

@@ -16,15 +16,13 @@
 package org.onosproject.net.meter;
 
 import com.google.common.base.MoreObjects;
-
-import java.util.Optional;
+import com.google.common.base.Objects;
 
 /**
  * Representation of an operation on the meter table.
  */
 public class MeterOperation {
 
-    private final Optional<MeterContext> context;
 
     /**
      * Tyoe of meter operation.
@@ -39,10 +37,9 @@ public class MeterOperation {
     private final Type type;
 
 
-    public MeterOperation(Meter meter, Type type, MeterContext context) {
+    public MeterOperation(Meter meter, Type type) {
         this.meter = meter;
         this.type = type;
-        this.context = Optional.ofNullable(context);
     }
 
     /**
@@ -63,16 +60,6 @@ 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)
@@ -80,4 +67,22 @@ public class MeterOperation {
                 .add("type", type)
                 .toString();
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (o == null || getClass() != o.getClass()) {
+            return false;
+        }
+        MeterOperation that = (MeterOperation) o;
+        return Objects.equal(meter, that.meter) &&
+                Objects.equal(type, that.type);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(meter, 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.net.meter;
+
+import org.onosproject.core.ApplicationId;
+import org.onosproject.net.DeviceId;
+
+import java.util.Collection;
+import java.util.Optional;
+
+/**
+ * Represents a generalized meter request to be deployed on a device.
+ */
+public interface MeterRequest {
+
+    enum Type {
+        ADD,
+        MODIFY,
+        REMOVE
+    }
+
+    /**
+     * The target device for this meter.
+     *
+     * @return a device id
+     */
+    DeviceId deviceId();
+
+    /**
+     * The id of the application which created this meter.
+     *
+     * @return an application id
+     */
+    ApplicationId appId();
+
+    /**
+     * The unit used within this meter.
+     *
+     * @return the unit
+     */
+    Meter.Unit unit();
+
+    /**
+     * Signals whether this meter applies to bursts only.
+     *
+     * @return a boolean
+     */
+    boolean isBurst();
+
+    /**
+     * The collection of bands to apply on the dataplane.
+     *
+     * @return a collection of bands.
+     */
+    Collection<Band> bands();
+
+    /**
+     * Returns the callback context for this meter.
+     *
+     * @return an optional meter context
+     */
+    Optional<MeterContext> context();
+
+    /**
+     * A meter builder.
+     */
+    interface Builder {
+
+        /**
+         * Assigns the target device for this meter.
+         *
+         * @param deviceId a device id
+         * @return this
+         */
+        Builder forDevice(DeviceId deviceId);
+
+        /**
+         * Assigns the application that built this meter.
+         *
+         * @param appId an application id
+         * @return this
+         */
+        Builder fromApp(ApplicationId appId);
+
+        /**
+         * Assigns the @See Unit to use for this meter.
+         * Defaults to kb/s
+         *
+         * @param unit a unit
+         * @return this
+         */
+        Builder withUnit(Meter.Unit unit);
+
+        /**
+         * Sets this meter as applicable to burst traffic only.
+         * Defaults to false.
+         *
+         * @return this
+         */
+        Builder burst();
+
+        /**
+         * Assigns bands to this meter. There must be at least one band.
+         *
+         * @param bands a collection of bands
+         * @return this
+         */
+        Builder withBands(Collection<Band> bands);
+
+        /**
+         * Assigns an execution context for this meter request.
+         *
+         * @param context a meter context
+         * @return this
+         */
+        Builder withContext(MeterContext context);
+
+        /**
+         * Requests the addition of a meter.
+         *
+         * @return a meter request
+         */
+        MeterRequest add();
+
+        /**
+         * Requests the removal of a meter.
+         *
+         * @return a meter request
+         */
+        MeterRequest remove();
+
+    }
+
+}

@@ -30,23 +30,18 @@ public interface MeterService
     /**
      * Adds a meter to the system and performs it installation.
      *
-     * @param meter a meter.
+     * @param meter a meter
+     * @return a meter (with a meter id)
      */
-    void addMeter(MeterOperation meter);
-
-    /**
-     * Updates a meter by adding statistic information to the meter.
-     *
-     * @param meter an updated meter
-     */
-    void updateMeter(MeterOperation meter);
+    Meter submit(MeterRequest meter);
 
     /**
      * Remove a meter from the system and the dataplane.
      *
      * @param meter a meter to remove
+     * @param meterId the meter id of the meter to remove.
      */
-    void removeMeter(MeterOperation meter);
+    void withdraw(MeterRequest meter, MeterId meterId);
 
     /**
      * Fetch the meter by the meter id.
@@ -63,10 +58,4 @@ public interface MeterService
      */
     Collection<Meter> getAllMeters();
 
-    /**
-     * Allocate a meter id which must be used to create the meter.
-     *
-     * @return a meter id
-     */
-    MeterId allocateMeterId();
 }

+/*
+ * 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.net.meter;
+
+import com.google.common.testing.EqualsTester;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Collections;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.onosproject.net.NetTestTools.APP_ID;
+import static org.onosproject.net.NetTestTools.did;
+
+
+/**
+ * DefaultMeter Tests.
+ */
+public class DefaultMeterTest {
+
+    private Meter m1;
+    private Meter sameAsm1;
+    private Meter m2;
+
+    @Before
+    public void setup() {
+
+        Band band = DefaultBand.builder()
+                .ofType(Band.Type.DROP)
+                .withRate(500)
+                .build();
+
+        m1 = DefaultMeter.builder()
+                .forDevice(did("1"))
+                .fromApp(APP_ID)
+                .withId(MeterId.meterId(1))
+                .withUnit(Meter.Unit.KB_PER_SEC)
+                .withBands(Collections.singletonList(band))
+                .build();
+
+        sameAsm1 = DefaultMeter.builder()
+                .forDevice(did("1"))
+                .fromApp(APP_ID)
+                .withId(MeterId.meterId(1))
+                .withUnit(Meter.Unit.KB_PER_SEC)
+                .withBands(Collections.singletonList(band))
+                .build();
+
+        m2 = DefaultMeter.builder()
+                .forDevice(did("2"))
+                .fromApp(APP_ID)
+                .withId(MeterId.meterId(2))
+                .withUnit(Meter.Unit.KB_PER_SEC)
+                .withBands(Collections.singletonList(band))
+                .build();
+
+    }
+
+    @Test
+    public void testEquality() {
+        new EqualsTester()
+                .addEqualityGroup(m1, sameAsm1)
+                .addEqualityGroup(m2).testEquals();
+    }
+
+    @Test
+    public void testConstruction() {
+        DefaultMeter m = (DefaultMeter) m1;
+
+        assertThat(m.deviceId(), is(did("1")));
+        assertThat(m.appId(), is(APP_ID));
+        assertThat(m.id(), is(MeterId.meterId(1)));
+        assertThat(m.isBurst(), is(false));
+
+        assertThat(m.life(), is(0L));
+        assertThat(m.bytesSeen(), is(0L));
+        assertThat(m.packetsSeen(), is(0L));
+        assertThat(m.referenceCount(), is(0L));
+
+    }
+
+
+
+
+}

+/*
+ * 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.net.meter;
+
+import com.google.common.testing.EqualsTester;
+import org.junit.Test;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.net.DeviceId;
+
+import java.util.Collection;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.onlab.junit.ImmutableClassChecker.assertThatClassIsImmutableBaseClass;
+
+/**
+ * Unit tests for the MeterOperationTest object.
+ */
+public class MeterOperationTest {
+
+
+
+    /**
+     * Checks that the MeterOperation class is immutable.
+     */
+    @Test
+    public void testImmutability() {
+        assertThatClassIsImmutableBaseClass(MeterOperation.class);
+    }
+
+    @Test
+    public void testEquality() {
+        final Meter m1 = new TestMeter();
+        final Meter m2 = new TestMeter();
+        final MeterOperation op1 = new MeterOperation(m1,
+                                                      MeterOperation.Type.ADD);
+        final MeterOperation sameAsOp1 = new MeterOperation(m1,
+                                                            MeterOperation.Type.ADD);
+        final MeterOperation op2 = new MeterOperation(m2,
+                                            MeterOperation.Type.ADD);
+
+        new EqualsTester()
+                .addEqualityGroup(op1, sameAsOp1)
+                .addEqualityGroup(op2)
+                .testEquals();
+    }
+
+    @Test
+    public void testConstruction() {
+        final Meter m1 = new TestMeter();
+
+        final MeterOperation op = new MeterOperation(m1, MeterOperation.Type.ADD);
+
+        assertThat(op.meter(), is(m1));
+    }
+
+    private static final class TestMeter implements Meter {
+
+        @Override
+        public DeviceId deviceId() {
+            return null;
+        }
+
+        @Override
+        public MeterId id() {
+            return null;
+        }
+
+        @Override
+        public ApplicationId appId() {
+            return null;
+        }
+
+        @Override
+        public Unit unit() {
+            return null;
+        }
+
+        @Override
+        public boolean isBurst() {
+            return false;
+        }
+
+        @Override
+        public Collection<Band> bands() {
+            return null;
+        }
+
+        @Override
+        public MeterState state() {
+            return null;
+        }
+
+        @Override
+        public long life() {
+            return 0;
+        }
+
+        @Override
+        public long referenceCount() {
+            return 0;
+        }
+
+        @Override
+        public long packetsSeen() {
+            return 0;
+        }
+
+        @Override
+        public long bytesSeen() {
+            return 0;
+        }
+    }
+
+}

@@ -21,6 +21,7 @@ import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -38,6 +39,7 @@ public final class TestConsistentMap<K, V> extends ConsistentMapAdapter<K, V> {
     private final List<MapEventListener<K, V>> listeners;
     private final HashMap<K, V> map;
     private final String mapName;
+    private final AtomicLong counter = new AtomicLong(0);
 
     private TestConsistentMap(String mapName) {
         map = new HashMap<>();
@@ -46,7 +48,7 @@ public final class TestConsistentMap<K, V> extends ConsistentMapAdapter<K, V> {
     }
 
     private Versioned<V> version(V v) {
-        return new Versioned<>(v, 1, System.currentTimeMillis());
+        return new Versioned<>(v, counter.incrementAndGet(), System.currentTimeMillis());
     }
 
     /**
@@ -115,8 +117,12 @@ public final class TestConsistentMap<K, V> extends ConsistentMapAdapter<K, V> {
 
     @Override
     public Versioned<V> put(K key, V value) {
-        Versioned<V> result = version(map.put(key, value));
-        notifyListeners(mapName, INSERT, key, result);
+        Versioned<V> result = version(value);
+        if (map.put(key, value) == null) {
+            notifyListeners(mapName, INSERT, key, result);
+        } else {
+            notifyListeners(mapName, UPDATE, key, result);
+        }
         return result;
     }
 
@@ -213,6 +219,11 @@ public final class TestConsistentMap<K, V> extends ConsistentMapAdapter<K, V> {
         listeners.remove(listener);
     }
 
+    @Override
+    public Map<K, V> asJavaMap() {
+        return map;
+    }
+
     public static Builder builder() {
         return new Builder();
     }

@@ -32,6 +32,7 @@ import org.onosproject.net.meter.MeterOperation;
 import org.onosproject.net.meter.MeterProvider;
 import org.onosproject.net.meter.MeterProviderRegistry;
 import org.onosproject.net.meter.MeterProviderService;
+import org.onosproject.net.meter.MeterRequest;
 import org.onosproject.net.meter.MeterService;
 import org.onosproject.net.meter.MeterState;
 import org.onosproject.net.meter.MeterStore;
@@ -72,7 +73,7 @@ public class MeterManager extends AbstractListenerProviderRegistry<MeterEvent, M
 
     private AtomicCounter meterIdCounter;
 
-    private TriConsumer<MeterOperation, MeterStoreResult, Throwable> onComplete;
+    private TriConsumer<MeterRequest, MeterStoreResult, Throwable> onComplete;
 
     @Activate
     public void activate() {
@@ -82,16 +83,16 @@ public class MeterManager extends AbstractListenerProviderRegistry<MeterEvent, M
 
         store.setDelegate(delegate);
 
-        onComplete = (op, result, error) ->
+        onComplete = (request, result, error) ->
             {
-                op.context().ifPresent(c -> {
+                request.context().ifPresent(c -> {
                     if (error != null) {
-                        c.onError(op.meter(), MeterFailReason.UNKNOWN);
+                        c.onError(request, MeterFailReason.UNKNOWN);
                     } else {
                         if (result.reason().isPresent()) {
-                            c.onError(op.meter(), result.reason().get());
+                            c.onError(request, result.reason().get());
                         } else {
-                            c.onSuccess(op.meter());
+                            c.onSuccess(request);
                         }
                     }
                 });
@@ -112,27 +113,42 @@ public class MeterManager extends AbstractListenerProviderRegistry<MeterEvent, M
     }
 
     @Override
-    public void addMeter(MeterOperation op) {
-        DefaultMeter m = (DefaultMeter) op.meter();
+    public Meter submit(MeterRequest request) {
+
+        Meter.Builder mBuilder = DefaultMeter.builder()
+                .forDevice(request.deviceId())
+                .fromApp(request.appId())
+                .withBands(request.bands())
+                .withId(allocateMeterId())
+                .withUnit(request.unit());
+
+        if (request.isBurst()) {
+            mBuilder.burst();
+        }
+        DefaultMeter m = (DefaultMeter) mBuilder.build();
         m.setState(MeterState.PENDING_ADD);
         store.storeMeter(m).whenComplete((result, error) ->
-                                                 onComplete.accept(op, result, error));
+                                                 onComplete.accept(request, result, error));
+        return m;
     }
 
     @Override
-    public void updateMeter(MeterOperation op) {
-        DefaultMeter m = (DefaultMeter) op.meter();
-        m.setState(MeterState.PENDING_ADD);
-        store.updateMeter(m).whenComplete((result, error) ->
-                                                  onComplete.accept(op, result, error));
-    }
+    public void withdraw(MeterRequest request, MeterId meterId) {
+        Meter.Builder mBuilder = DefaultMeter.builder()
+                .forDevice(request.deviceId())
+                .fromApp(request.appId())
+                .withBands(request.bands())
+                .withId(meterId)
+                .withUnit(request.unit());
+
+        if (request.isBurst()) {
+            mBuilder.burst();
+        }
 
-    @Override
-    public void removeMeter(MeterOperation op) {
-        DefaultMeter m = (DefaultMeter) op.meter();
+        DefaultMeter m = (DefaultMeter) mBuilder.build();
         m.setState(MeterState.PENDING_REMOVE);
         store.deleteMeter(m).whenComplete((result, error) ->
-                                                  onComplete.accept(op, result, error));
+                                                  onComplete.accept(request, result, error));
     }
 
     @Override
@@ -145,10 +161,9 @@ public class MeterManager extends AbstractListenerProviderRegistry<MeterEvent, M
         return store.getAllMeters();
     }
 
-    @Override
-    public MeterId allocateMeterId() {
+    private MeterId allocateMeterId() {
         // FIXME: This will break one day.
-        return MeterId.meterId((int) meterIdCounter.getAndIncrement());
+        return MeterId.meterId((int) meterIdCounter.incrementAndGet());
     }
 
     private class InternalMeterProviderService
@@ -185,8 +200,7 @@ public class MeterManager extends AbstractListenerProviderRegistry<MeterEvent, M
                 if (m.state() == MeterState.PENDING_ADD) {
                     provider().performMeterOperation(m.deviceId(),
                                                      new MeterOperation(m,
-                                                                        MeterOperation.Type.ADD,
-                                                                        null));
+                                                                        MeterOperation.Type.ADD));
                 } else {
                     store.deleteMeterNow(m);
                 }
@@ -203,13 +217,11 @@ public class MeterManager extends AbstractListenerProviderRegistry<MeterEvent, M
             switch (event.type()) {
                 case METER_ADD_REQ:
                     p.performMeterOperation(deviceId, new MeterOperation(event.subject(),
-                                                                         MeterOperation.Type.ADD,
-                                                                         null));
+                                                                         MeterOperation.Type.ADD));
                     break;
                 case METER_REM_REQ:
                     p.performMeterOperation(deviceId, new MeterOperation(event.subject(),
-                                                                         MeterOperation.Type.REMOVE,
-                                                                         null));
+                                                                         MeterOperation.Type.REMOVE));
                     break;
                 default:
                     log.warn("Unknown meter event {}", 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.net.meter.impl;
+
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.onlab.junit.TestUtils;
+import org.onlab.packet.IpAddress;
+import org.onosproject.cluster.ClusterServiceAdapter;
+import org.onosproject.cluster.ControllerNode;
+import org.onosproject.cluster.DefaultControllerNode;
+import org.onosproject.cluster.NodeId;
+import org.onosproject.common.event.impl.TestEventDispatcher;
+import org.onosproject.core.ApplicationId;
+import org.onosproject.core.DefaultApplicationId;
+import org.onosproject.incubator.store.meter.impl.DistributedMeterStore;
+import org.onosproject.mastership.MastershipServiceAdapter;
+import org.onosproject.net.DeviceId;
+import org.onosproject.net.meter.Band;
+import org.onosproject.net.meter.DefaultBand;
+import org.onosproject.net.meter.DefaultMeter;
+import org.onosproject.net.meter.DefaultMeterRequest;
+import org.onosproject.net.meter.Meter;
+import org.onosproject.net.meter.MeterId;
+import org.onosproject.net.meter.MeterOperation;
+import org.onosproject.net.meter.MeterOperations;
+import org.onosproject.net.meter.MeterProvider;
+import org.onosproject.net.meter.MeterProviderRegistry;
+import org.onosproject.net.meter.MeterProviderService;
+import org.onosproject.net.meter.MeterRequest;
+import org.onosproject.net.meter.MeterService;
+import org.onosproject.net.meter.MeterState;
+import org.onosproject.net.provider.AbstractProvider;
+import org.onosproject.net.provider.ProviderId;
+import org.onosproject.store.service.TestStorageService;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.onosproject.net.NetTestTools.APP_ID;
+import static org.onosproject.net.NetTestTools.did;
+import static org.onosproject.net.NetTestTools.injectEventDispatcher;
+
+/**
+ * Meter manager tests.
+ */
+public class MeterManagerTest {
+
+    private static final ProviderId PID = new ProviderId("of", "foo");
+    private static final NodeId NID_LOCAL = new NodeId("local");
+    private static final IpAddress LOCALHOST = IpAddress.valueOf("127.0.0.1");
+
+    private MeterService service;
+    private MeterManager manager;
+    private DistributedMeterStore meterStore;
+    private MeterProviderRegistry registry;
+    private MeterProviderService providerService;
+
+    private TestProvider provider;
+
+    private ApplicationId appId;
+
+
+    private Meter m1;
+    private Meter m2;
+    private MeterRequest.Builder m1Request;
+    private MeterRequest.Builder m2Request;
+
+    private Map<MeterId, Meter> meters = Maps.newHashMap();
+
+    @Before
+    public void setup() throws Exception {
+        meterStore = new DistributedMeterStore();
+        TestUtils.setField(meterStore, "storageService", new TestStorageService());
+        TestUtils.setField(meterStore, "clusterService", new TestClusterService());
+        TestUtils.setField(meterStore, "mastershipService", new TestMastershipService());
+        meterStore.activate();
+
+        manager = new MeterManager();
+        manager.store = meterStore;
+        TestUtils.setField(manager, "storageService", new TestStorageService());
+        injectEventDispatcher(manager, new TestEventDispatcher());
+        service = manager;
+        registry = manager;
+
+        manager.activate();
+
+        provider = new TestProvider(PID);
+        providerService = registry.register(provider);
+
+        appId = new TestApplicationId(0, "MeterManagerTest");
+
+        assertTrue("provider should be registered",
+                   registry.getProviders().contains(provider.id()));
+
+        Band band = DefaultBand.builder()
+                .ofType(Band.Type.DROP)
+                .withRate(500)
+                .build();
+
+        m1 = DefaultMeter.builder()
+                .forDevice(did("1"))
+                .fromApp(APP_ID)
+                .withId(MeterId.meterId(1))
+                .withUnit(Meter.Unit.KB_PER_SEC)
+                .withBands(Collections.singletonList(band))
+                .build();
+
+        m2 = DefaultMeter.builder()
+                .forDevice(did("2"))
+                .fromApp(APP_ID)
+                .withId(MeterId.meterId(2))
+                .withUnit(Meter.Unit.KB_PER_SEC)
+                .withBands(Collections.singletonList(band))
+                .build();
+
+        m1Request = DefaultMeterRequest.builder()
+                .forDevice(did("1"))
+                .fromApp(APP_ID)
+                .withUnit(Meter.Unit.KB_PER_SEC)
+                .withBands(Collections.singletonList(band));
+
+        m2Request = DefaultMeterRequest.builder()
+                .forDevice(did("2"))
+                .fromApp(APP_ID)
+                .withUnit(Meter.Unit.KB_PER_SEC)
+                .withBands(Collections.singletonList(band));
+
+
+    }
+
+    @After
+    public void tearDown() {
+        registry.unregister(provider);
+        assertFalse("provider should not be registered",
+                    registry.getProviders().contains(provider.id()));
+
+        manager.deactivate();
+        injectEventDispatcher(manager, null);
+
+    }
+
+    @Test
+    public void testAddition() {
+        manager.submit(m1Request.add());
+
+        assertTrue("The meter was not added", manager.getAllMeters().size() == 1);
+
+        assertThat(manager.getMeter(MeterId.meterId(1)), is(m1));
+    }
+
+    @Test
+    public void testRemove() {
+        manager.submit(m1Request.add());
+        manager.withdraw(m1Request.remove(), m1.id());
+
+        assertThat(manager.getMeter(MeterId.meterId(1)).state(),
+                   is(MeterState.PENDING_REMOVE));
+
+        providerService.pushMeterMetrics(m1.deviceId(), Collections.emptyList());
+
+        assertTrue("The meter was not removed", manager.getAllMeters().size() == 0);
+
+    }
+
+
+
+    public class TestApplicationId extends DefaultApplicationId {
+        public TestApplicationId(int id, String name) {
+            super(id, name);
+        }
+    }
+
+    private class TestProvider extends AbstractProvider implements MeterProvider {
+
+        protected TestProvider(ProviderId id) {
+            super(PID);
+        }
+
+        @Override
+        public void performMeterOperation(DeviceId deviceId, MeterOperations meterOps) {
+            //Currently unused.
+        }
+
+        @Override
+        public void performMeterOperation(DeviceId deviceId, MeterOperation meterOp) {
+            meters.put(meterOp.meter().id(), meterOp.meter());
+        }
+    }
+
+    private final class TestClusterService extends ClusterServiceAdapter {
+
+        ControllerNode local = new DefaultControllerNode(NID_LOCAL, LOCALHOST);
+
+        @Override
+        public ControllerNode getLocalNode() {
+            return local;
+        }
+
+        @Override
+        public Set<ControllerNode> getNodes() {
+            return Sets.newHashSet();
+        }
+
+    }
+
+    private class TestMastershipService extends MastershipServiceAdapter {
+        @Override
+        public NodeId getMasterFor(DeviceId deviceId) {
+            return NID_LOCAL;
+        }
+    }
+}

@@ -143,7 +143,9 @@ public class DistributedMeterStore extends AbstractStore<MeterEvent, MeterStoreD
         // update the state of the meter. It will be pruned by observing
         // that it has been removed from the dataplane.
         try {
-            meters.put(meter.id(), data);
+            if (meters.computeIfPresent(meter.id(), (k, v) -> data) == null) {
+                future.complete(MeterStoreResult.success());
+            }
         } catch (StorageException e) {
             future.completeExceptionally(e);
         }
@@ -159,7 +161,9 @@ public class DistributedMeterStore extends AbstractStore<MeterEvent, MeterStoreD
 
         MeterData data = new MeterData(meter, null, local);
         try {
-            meters.put(meter.id(), data);
+            if (meters.computeIfPresent(meter.id(), (k, v) -> data) == null) {
+                future.complete(MeterStoreResult.fail(MeterFailReason.INVALID_METER));
+            }
         } catch (StorageException e) {
             future.completeExceptionally(e);
         }
@@ -227,6 +231,10 @@ public class DistributedMeterStore extends AbstractStore<MeterEvent, MeterStoreD
                                 }
                                 break;
                             case ADDED:
+                                if (local.equals(data.origin()) && data.meter().state() == MeterState.PENDING_ADD) {
+                                    futures.remove(data.meter().id()).complete(MeterStoreResult.success());
+                                }
+                                break;
                             case REMOVED:
                                 if (local.equals(data.origin()) && data.meter().state() == MeterState.PENDING_REMOVE) {
                                     futures.remove(data.meter().id()).complete(MeterStoreResult.success());