Added Meter object accounting

Change-Id: I2a3b88ffd1756b738e197943b3e02771f5729e45

@@ -80,4 +80,11 @@ public interface MeterStore extends Store<MeterEvent, MeterStoreDelegate> {
      * @param reason a failure reason
      * @param reason a failure reason
      */
      */
     void failedMeter(MeterOperation op, MeterFailReason reason);
     void failedMeter(MeterOperation op, MeterFailReason reason);
+
+    /**
+     * Delete this meter immediately.
+     * @param m a meter
+     */
+    void deleteMeterNow(Meter m);
+
 }
 }

@@ -45,6 +45,8 @@ import org.onosproject.store.service.StorageService;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 
 
 import java.util.Collection;
 import java.util.Collection;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 
 import static org.slf4j.LoggerFactory.getLogger;
 import static org.slf4j.LoggerFactory.getLogger;
 
 
@@ -170,7 +172,25 @@ public class MeterManager extends AbstractListenerProviderRegistry<MeterEvent, M
 
 
         @Override
         @Override
         public void pushMeterMetrics(DeviceId deviceId, Collection<Meter> meterEntries) {
         public void pushMeterMetrics(DeviceId deviceId, Collection<Meter> meterEntries) {
-            meterEntries.forEach(m -> store.updateMeterState(m));
+            //FIXME: FOLLOWING CODE CANNOT BE TESTED UNTIL SOMETHING THAT
+            //FIXME: IMPLEMENTS METERS EXISTS
+            Map<MeterId, Meter> storedMeterMap = store.getAllMeters().stream()
+                    .collect(Collectors.toMap(Meter::id, m -> m));
+
+            meterEntries.stream()
+                    .filter(m -> storedMeterMap.remove(m.id()) != null)
+                    .forEach(m -> store.updateMeterState(m));
+
+            storedMeterMap.values().stream().forEach(m -> {
+                if (m.state() == MeterState.PENDING_ADD) {
+                    provider().performMeterOperation(m.deviceId(),
+                                                     new MeterOperation(m,
+                                                                        MeterOperation.Type.ADD,
+                                                                        null));
+                } else {
+                    store.deleteMeterNow(m);
+                }
+            });
         }
         }
     }
     }
 
 

@@ -198,6 +198,12 @@ public class DistributedMeterStore extends AbstractStore<MeterEvent, MeterStoreD
                 new MeterData(v.meter(), reason, v.origin()));
                 new MeterData(v.meter(), reason, v.origin()));
     }
     }
 
 
+    @Override
+    public void deleteMeterNow(Meter m) {
+        futures.remove(m.id());
+        meters.remove(m.id());
+    }
+
     private class InternalMapEventListener implements MapEventListener<MeterId, MeterData> {
     private class InternalMapEventListener implements MapEventListener<MeterId, MeterData> {
         @Override
         @Override
         public void event(MapEvent<MeterId, MeterData> event) {
         public void event(MapEvent<MeterId, MeterData> event) {
@@ -217,12 +223,12 @@ public class DistributedMeterStore extends AbstractStore<MeterEvent, MeterStoreD
                                 } else if (data.reason().isPresent() && local.equals(data.origin())) {
                                 } else if (data.reason().isPresent() && local.equals(data.origin())) {
                                     MeterStoreResult msr = MeterStoreResult.fail(data.reason().get());
                                     MeterStoreResult msr = MeterStoreResult.fail(data.reason().get());
                                     //TODO: No future -> no friend
                                     //TODO: No future -> no friend
-                                    futures.remove(data.meter().id()).complete(msr);
+                                    futures.get(data.meter().id()).complete(msr);
                                 }
                                 }
                                 break;
                                 break;
                             case ADDED:
                             case ADDED:
                             case REMOVED:
                             case REMOVED:
-                                if (local.equals(data.origin())) {
+                                if (local.equals(data.origin()) && data.meter().state() == MeterState.PENDING_REMOVE) {
                                     futures.remove(data.meter().id()).complete(MeterStoreResult.success());
                                     futures.remove(data.meter().id()).complete(MeterStoreResult.success());
                                 }
                                 }
                                 break;
                                 break;

@@ -306,6 +306,9 @@ public class OpenFlowControllerImpl implements OpenFlowController {
                 case PORT:
                 case PORT:
                     executorMsgs.submit(new OFMessageHandler(dpid, reply));
                     executorMsgs.submit(new OFMessageHandler(dpid, reply));
                     break;
                     break;
+                case METER:
+                    executorMsgs.submit(new OFMessageHandler(dpid, reply));
+                    break;
                 case EXPERIMENTER:
                 case EXPERIMENTER:
                     if (reply instanceof OFCalientFlowStatsReply) {
                     if (reply instanceof OFCalientFlowStatsReply) {
                         // Convert Calient flow statistics to regular flow stats
                         // Convert Calient flow statistics to regular flow stats
@@ -353,6 +356,7 @@ public class OpenFlowControllerImpl implements OpenFlowController {
                     }
                     }
                     break;
                     break;
                 default:
                 default:
+                    log.warn("Discarding unknown stats reply type {}", reply.getStatsType());
                     break;
                     break;
             }
             }
             break;
             break;