Moving meter store implementation to use map events
Change-Id: I338473b7286d7b9e5cdfb938f16c7b6155d4cbb5
diff --git a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/DefaultMeter.java b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/DefaultMeter.java
index 138bb18..177688c 100644
--- a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/DefaultMeter.java
+++ b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/DefaultMeter.java
@@ -20,7 +20,6 @@
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 @@
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 @@
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 @@
}
@Override
- public Optional<MeterContext> context() {
- return null;
- }
-
- @Override
public MeterState state() {
return state;
}
@@ -154,7 +146,6 @@
private boolean burst = false;
private Collection<Band> bands;
private DeviceId deviceId;
- private Optional<MeterContext> context;
@Override
@@ -194,19 +185,13 @@
}
@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);
}
diff --git a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/Meter.java b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/Meter.java
index 14593df..37c1105 100644
--- a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/Meter.java
+++ b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/Meter.java
@@ -19,7 +19,6 @@
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 @@
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 @@
*/
Builder withBands(Collection<Band> bands);
- Builder withContext(MeterContext context);
-
/**
* Builds the meter based on the specified parameters.
*
diff --git a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterContext.java b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterContext.java
index 919cf63..b112b9a 100644
--- a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterContext.java
+++ b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterContext.java
@@ -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) {}
}
diff --git a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterEvent.java b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterEvent.java
index 9439145..808a3e8 100644
--- a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterEvent.java
+++ b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterEvent.java
@@ -20,28 +20,19 @@
/**
* 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 {
+ /**
+ * A meter addition was requested.
+ */
+ METER_ADD_REQ,
/**
- * Signals that a meter has been added.
+ * A meter removal was requested.
*/
- METER_UPDATED,
-
- /**
- * Signals that a meter update failed.
- */
- METER_OP_FAILED,
-
- /**
- * A meter operation was requested.
- */
- METER_OP_REQ,
-
+ METER_REM_REQ
}
@@ -50,32 +41,22 @@
* 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, Meter meter, long time) {
+ super(type, meter, time);
}
- public MeterEvent(Type type, MeterOperation op, MeterFailReason reason) {
- super(type, op);
- this.reason = reason;
- }
-
- public MeterFailReason reason() {
- return reason;
- }
}
diff --git a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterOperation.java b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterOperation.java
index ff7e988..8354888 100644
--- a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterOperation.java
+++ b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterOperation.java
@@ -17,11 +17,15 @@
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 @@
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 @@
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)
diff --git a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterService.java b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterService.java
index b8702dd..c7eb93a 100644
--- a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterService.java
+++ b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterService.java
@@ -30,28 +30,21 @@
*
* @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.
diff --git a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterStore.java b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterStore.java
index 232eedd..54e5658 100644
--- a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterStore.java
+++ b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterStore.java
@@ -18,6 +18,7 @@
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 @@
* 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);
diff --git a/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterStoreResult.java b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterStoreResult.java
new file mode 100644
index 0000000..8ad5aa6
--- /dev/null
+++ b/incubator/api/src/main/java/org/onosproject/incubator/net/meter/MeterStoreResult.java
@@ -0,0 +1,66 @@
+/*
+ * 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);
+ }
+
+}
diff --git a/incubator/net/src/main/java/org/onosproject/incubator/net/meter/impl/MeterManager.java b/incubator/net/src/main/java/org/onosproject/incubator/net/meter/impl/MeterManager.java
index b0726d6..47450c7 100644
--- a/incubator/net/src/main/java/org/onosproject/incubator/net/meter/impl/MeterManager.java
+++ b/incubator/net/src/main/java/org/onosproject/incubator/net/meter/impl/MeterManager.java
@@ -21,6 +21,7 @@
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.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 java.util.Collection;
-import static com.google.common.base.Preconditions.checkNotNull;
import static org.slf4j.LoggerFactory.getLogger;
@@ -69,11 +70,29 @@
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 @@
}
@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 @@
@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:
+ case METER_ADD_REQ:
+ p.performMeterOperation(deviceId, new MeterOperation(event.subject(),
+ MeterOperation.Type.ADD,
+ null));
break;
- case METER_OP_FAILED:
- event.subject().meter().context().ifPresent(c ->
- c.onError(event.subject(), event.reason()));
- 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());
diff --git a/incubator/store/src/main/java/org/onosproject/incubator/store/meter/impl/DistributedMeterStore.java b/incubator/store/src/main/java/org/onosproject/incubator/store/meter/impl/DistributedMeterStore.java
index b8f2080..6477e68 100644
--- a/incubator/store/src/main/java/org/onosproject/incubator/store/meter/impl/DistributedMeterStore.java
+++ b/incubator/store/src/main/java/org/onosproject/incubator/store/meter/impl/DistributedMeterStore.java
@@ -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 @@
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 @@
@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());
+ public CompletableFuture<MeterStoreResult> storeMeter(Meter meter) {
+ CompletableFuture<MeterStoreResult> future = new CompletableFuture<>();
+ futures.put(meter.id(), future);
+ MeterData data = new MeterData(meter, null, local);
- 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));
- }
- });
+ 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) {
+ public CompletableFuture<MeterStoreResult> updateMeter(Meter meter) {
+ CompletableFuture<MeterStoreResult> future = new CompletableFuture<>();
+ futures.put(meter.id(), future);
- 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));
- }
- });
+ 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());
+ }
+
+ }
+ }
+
+
}
diff --git a/incubator/store/src/main/java/org/onosproject/incubator/store/meter/impl/MeterData.java b/incubator/store/src/main/java/org/onosproject/incubator/store/meter/impl/MeterData.java
new file mode 100644
index 0000000..cfc1402
--- /dev/null
+++ b/incubator/store/src/main/java/org/onosproject/incubator/store/meter/impl/MeterData.java
@@ -0,0 +1,52 @@
+/*
+ * 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;
+ }
+
+
+}
diff --git a/utils/misc/src/main/java/org/onlab/util/TriConsumer.java b/utils/misc/src/main/java/org/onlab/util/TriConsumer.java
new file mode 100644
index 0000000..754c7da
--- /dev/null
+++ b/utils/misc/src/main/java/org/onlab/util/TriConsumer.java
@@ -0,0 +1,28 @@
+/*
+ * 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