Updated master to commit id 6ee8aa3e67ce89908a8c93aa9445c6f71a18f986

Change-Id: I94b055ee2f298daf71e2ec794fd0f2495bd8081f

1 files changed, 634 insertions, 0 deletions

diff --git a/framework/src/onos/core/net/src/main/java/org/onosproject/net/statistic/impl/FlowStatisticManager.java b/framework/src/onos/core/net/src/main/java/org/onosproject/net/statistic/impl/FlowStatisticManager.java
new file mode 100644
index 00000000..6515ef31
--- /dev/null
+++ b/framework/src/onos/core/net/src/main/java/org/onosproject/net/statistic/impl/FlowStatisticManager.java
@@ -0,0 +1,634 @@
+/*

+ * 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.statistic.impl;

+

+import com.google.common.base.MoreObjects;

+import com.google.common.base.Predicate;

+import com.google.common.collect.ImmutableSet;

+import org.apache.felix.scr.annotations.Activate;

+import org.apache.felix.scr.annotations.Component;

+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.onosproject.cli.Comparators;

+import org.onosproject.net.ConnectPoint;

+import org.onosproject.net.Device;

+import org.onosproject.net.Port;

+import org.onosproject.net.PortNumber;

+import org.onosproject.net.device.DeviceService;

+import org.onosproject.net.flow.DefaultTypedFlowEntry;

+import org.onosproject.net.flow.FlowEntry;

+import org.onosproject.net.flow.FlowRule;

+import org.onosproject.net.flow.FlowRuleEvent;

+import org.onosproject.net.flow.FlowRuleListener;

+import org.onosproject.net.flow.FlowRuleService;

+import org.onosproject.net.flow.TypedStoredFlowEntry;

+import org.onosproject.net.flow.instructions.Instruction;

+import org.onosproject.net.statistic.DefaultLoad;

+import org.onosproject.net.statistic.FlowStatisticService;

+import org.onosproject.net.statistic.Load;

+import org.onosproject.net.statistic.FlowStatisticStore;

+import org.onosproject.net.statistic.SummaryFlowEntryWithLoad;

+import org.onosproject.net.statistic.TypedFlowEntryWithLoad;

+

+import org.slf4j.Logger;

+

+import java.util.ArrayList;

+import java.util.HashMap;

+import java.util.List;

+import java.util.Map;

+import java.util.Objects;

+import java.util.Set;

+import java.util.TreeMap;

+import java.util.stream.Collectors;

+

+import static com.google.common.base.Preconditions.checkNotNull;

+import static org.onosproject.security.AppGuard.checkPermission;

+import static org.slf4j.LoggerFactory.getLogger;

+import static org.onosproject.security.AppPermission.Type.*;

+

+/**

+ * Provides an implementation of the Flow Statistic Service.

+ */

+@Component(immediate = true, enabled = true)

+@Service

+public class FlowStatisticManager implements FlowStatisticService {

+    private final Logger log = getLogger(getClass());

+

+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)

+    protected FlowRuleService flowRuleService;

+

+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)

+    protected FlowStatisticStore flowStatisticStore;

+

+    @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)

+    protected DeviceService deviceService;

+

+    private final InternalFlowRuleStatsListener frListener = new InternalFlowRuleStatsListener();

+

+    @Activate

+    public void activate() {

+        flowRuleService.addListener(frListener);

+        log.info("Started");

+    }

+

+    @Deactivate

+    public void deactivate() {

+        flowRuleService.removeListener(frListener);

+        log.info("Stopped");

+    }

+

+    @Override

+    public Map<ConnectPoint, SummaryFlowEntryWithLoad> loadSummary(Device device) {

+        checkPermission(STATISTIC_READ);

+

+        Map<ConnectPoint, SummaryFlowEntryWithLoad> summaryLoad = new TreeMap<>(Comparators.CONNECT_POINT_COMPARATOR);

+

+        if (device == null) {

+            return summaryLoad;

+        }

+

+        List<Port> ports = new ArrayList<>(deviceService.getPorts(device.id()));

+

+        for (Port port : ports) {

+            ConnectPoint cp = new ConnectPoint(device.id(), port.number());

+            SummaryFlowEntryWithLoad sfe = loadSummaryPortInternal(cp);

+            summaryLoad.put(cp, sfe);

+        }

+

+        return summaryLoad;

+    }

+

+    @Override

+    public SummaryFlowEntryWithLoad loadSummary(Device device, PortNumber pNumber) {

+        checkPermission(STATISTIC_READ);

+

+        ConnectPoint cp = new ConnectPoint(device.id(), pNumber);

+        return loadSummaryPortInternal(cp);

+    }

+

+    @Override

+    public Map<ConnectPoint, List<TypedFlowEntryWithLoad>> loadAllByType(Device device,

+                                                                  TypedStoredFlowEntry.FlowLiveType liveType,

+                                                                  Instruction.Type instType) {

+        checkPermission(STATISTIC_READ);

+

+        Map<ConnectPoint, List<TypedFlowEntryWithLoad>> allLoad = new TreeMap<>(Comparators.CONNECT_POINT_COMPARATOR);

+

+        if (device == null) {

+            return allLoad;

+        }

+

+        List<Port> ports = new ArrayList<>(deviceService.getPorts(device.id()));

+

+        for (Port port : ports) {

+            ConnectPoint cp = new ConnectPoint(device.id(), port.number());

+            List<TypedFlowEntryWithLoad> tfel = loadAllPortInternal(cp, liveType, instType);

+            allLoad.put(cp, tfel);

+        }

+

+        return allLoad;

+    }

+

+    @Override

+    public List<TypedFlowEntryWithLoad> loadAllByType(Device device, PortNumber pNumber,

+                                               TypedStoredFlowEntry.FlowLiveType liveType,

+                                               Instruction.Type instType) {

+        checkPermission(STATISTIC_READ);

+

+        ConnectPoint cp = new ConnectPoint(device.id(), pNumber);

+        return loadAllPortInternal(cp, liveType, instType);

+    }

+

+    @Override

+    public Map<ConnectPoint, List<TypedFlowEntryWithLoad>> loadTopnByType(Device device,

+                                                                   TypedStoredFlowEntry.FlowLiveType liveType,

+                                                                   Instruction.Type instType,

+                                                                   int topn) {

+        checkPermission(STATISTIC_READ);

+

+        Map<ConnectPoint, List<TypedFlowEntryWithLoad>> allLoad = new TreeMap<>(Comparators.CONNECT_POINT_COMPARATOR);

+

+        if (device == null) {

+            return allLoad;

+        }

+

+        List<Port> ports = new ArrayList<>(deviceService.getPorts(device.id()));

+

+        for (Port port : ports) {

+            ConnectPoint cp = new ConnectPoint(device.id(), port.number());

+            List<TypedFlowEntryWithLoad> tfel = loadTopnPortInternal(cp, liveType, instType, topn);

+            allLoad.put(cp, tfel);

+        }

+

+        return allLoad;

+    }

+

+    @Override

+    public List<TypedFlowEntryWithLoad> loadTopnByType(Device device, PortNumber pNumber,

+                                                TypedStoredFlowEntry.FlowLiveType liveType,

+                                                Instruction.Type instType,

+                                                int topn) {

+        checkPermission(STATISTIC_READ);

+

+        ConnectPoint cp = new ConnectPoint(device.id(), pNumber);

+        return loadTopnPortInternal(cp, liveType, instType, topn);

+    }

+

+    private SummaryFlowEntryWithLoad loadSummaryPortInternal(ConnectPoint cp) {

+        checkPermission(STATISTIC_READ);

+

+        Set<FlowEntry> currentStats;

+        Set<FlowEntry> previousStats;

+

+        TypedStatistics typedStatistics;

+        synchronized (flowStatisticStore) {

+             currentStats = flowStatisticStore.getCurrentFlowStatistic(cp);

+            if (currentStats == null) {

+                return new SummaryFlowEntryWithLoad(cp, new DefaultLoad());

+            }

+            previousStats = flowStatisticStore.getPreviousFlowStatistic(cp);

+            if (previousStats == null) {

+                return new SummaryFlowEntryWithLoad(cp, new DefaultLoad());

+            }

+            // copy to local flow entry

+            typedStatistics = new TypedStatistics(currentStats, previousStats);

+

+            // Check for validity of this stats data

+            checkLoadValidity(currentStats, previousStats);

+        }

+

+        // current and previous set is not empty!

+        Set<FlowEntry> currentSet = typedStatistics.current();

+        Set<FlowEntry> previousSet = typedStatistics.previous();

+        Load totalLoad = new DefaultLoad(aggregateBytesSet(currentSet), aggregateBytesSet(previousSet),

+                TypedFlowEntryWithLoad.avgPollInterval());

+

+        Map<FlowRule, TypedStoredFlowEntry> currentMap;

+        Map<FlowRule, TypedStoredFlowEntry> previousMap;

+

+        currentMap = typedStatistics.currentImmediate();

+        previousMap = typedStatistics.previousImmediate();

+        Load immediateLoad = new DefaultLoad(aggregateBytesMap(currentMap), aggregateBytesMap(previousMap),

+                TypedFlowEntryWithLoad.shortPollInterval());

+

+        currentMap = typedStatistics.currentShort();

+        previousMap = typedStatistics.previousShort();

+        Load shortLoad = new DefaultLoad(aggregateBytesMap(currentMap), aggregateBytesMap(previousMap),

+                TypedFlowEntryWithLoad.shortPollInterval());

+

+        currentMap = typedStatistics.currentMid();

+        previousMap = typedStatistics.previousMid();

+        Load midLoad = new DefaultLoad(aggregateBytesMap(currentMap), aggregateBytesMap(previousMap),

+                TypedFlowEntryWithLoad.midPollInterval());

+

+        currentMap = typedStatistics.currentLong();

+        previousMap = typedStatistics.previousLong();

+        Load longLoad = new DefaultLoad(aggregateBytesMap(currentMap), aggregateBytesMap(previousMap),

+                TypedFlowEntryWithLoad.longPollInterval());

+

+        currentMap = typedStatistics.currentUnknown();

+        previousMap = typedStatistics.previousUnknown();

+        Load unknownLoad = new DefaultLoad(aggregateBytesMap(currentMap), aggregateBytesMap(previousMap),

+                TypedFlowEntryWithLoad.avgPollInterval());

+

+        return new SummaryFlowEntryWithLoad(cp, totalLoad, immediateLoad, shortLoad, midLoad, longLoad, unknownLoad);

+    }

+

+    private List<TypedFlowEntryWithLoad> loadAllPortInternal(ConnectPoint cp,

+                                                             TypedStoredFlowEntry.FlowLiveType liveType,

+                                                             Instruction.Type instType) {

+        checkPermission(STATISTIC_READ);

+

+        List<TypedFlowEntryWithLoad> retTFEL = new ArrayList<>();

+

+        Set<FlowEntry> currentStats;

+        Set<FlowEntry> previousStats;

+

+        TypedStatistics typedStatistics;

+        synchronized (flowStatisticStore) {

+            currentStats = flowStatisticStore.getCurrentFlowStatistic(cp);

+            if (currentStats == null) {

+                return retTFEL;

+            }

+            previousStats = flowStatisticStore.getPreviousFlowStatistic(cp);

+            if (previousStats == null) {

+                return retTFEL;

+            }

+            // copy to local flow entry set

+            typedStatistics = new TypedStatistics(currentStats, previousStats);

+

+            // Check for validity of this stats data

+            checkLoadValidity(currentStats, previousStats);

+        }

+

+        // current and previous set is not empty!

+        boolean isAllLiveType = (liveType == null ? true : false); // null is all live type

+        boolean isAllInstType = (instType == null ? true : false); // null is all inst type

+

+        Map<FlowRule, TypedStoredFlowEntry> currentMap;

+        Map<FlowRule, TypedStoredFlowEntry> previousMap;

+

+        if (isAllLiveType || liveType == TypedStoredFlowEntry.FlowLiveType.IMMEDIATE_FLOW) {

+            currentMap = typedStatistics.currentImmediate();

+            previousMap = typedStatistics.previousImmediate();

+

+            List<TypedFlowEntryWithLoad> fel = typedFlowEntryLoadByInstInternal(cp, currentMap, previousMap,

+                    isAllInstType, instType, TypedFlowEntryWithLoad.shortPollInterval());

+            if (fel.size() > 0) {

+                retTFEL.addAll(fel);

+            }

+        }

+

+        if (isAllLiveType || liveType == TypedStoredFlowEntry.FlowLiveType.SHORT_FLOW) {

+            currentMap = typedStatistics.currentShort();

+            previousMap = typedStatistics.previousShort();

+

+            List<TypedFlowEntryWithLoad> fel = typedFlowEntryLoadByInstInternal(cp, currentMap, previousMap,

+                    isAllInstType, instType, TypedFlowEntryWithLoad.shortPollInterval());

+            if (fel.size() > 0) {

+                retTFEL.addAll(fel);

+            }

+        }

+

+        if (isAllLiveType || liveType == TypedStoredFlowEntry.FlowLiveType.MID_FLOW) {

+            currentMap = typedStatistics.currentMid();

+            previousMap = typedStatistics.previousMid();

+

+            List<TypedFlowEntryWithLoad> fel = typedFlowEntryLoadByInstInternal(cp, currentMap, previousMap,

+                    isAllInstType, instType, TypedFlowEntryWithLoad.midPollInterval());

+            if (fel.size() > 0) {

+                retTFEL.addAll(fel);

+            }

+        }

+

+        if (isAllLiveType || liveType == TypedStoredFlowEntry.FlowLiveType.LONG_FLOW) {

+            currentMap = typedStatistics.currentLong();

+            previousMap = typedStatistics.previousLong();

+

+            List<TypedFlowEntryWithLoad> fel = typedFlowEntryLoadByInstInternal(cp, currentMap, previousMap,

+                    isAllInstType, instType, TypedFlowEntryWithLoad.longPollInterval());

+            if (fel.size() > 0) {

+                retTFEL.addAll(fel);

+            }

+        }

+

+        if (isAllLiveType || liveType == TypedStoredFlowEntry.FlowLiveType.UNKNOWN_FLOW) {

+            currentMap = typedStatistics.currentUnknown();

+            previousMap = typedStatistics.previousUnknown();

+

+            List<TypedFlowEntryWithLoad> fel = typedFlowEntryLoadByInstInternal(cp, currentMap, previousMap,

+                    isAllInstType, instType, TypedFlowEntryWithLoad.avgPollInterval());

+            if (fel.size() > 0) {

+                retTFEL.addAll(fel);

+            }

+        }

+

+        return retTFEL;

+    }

+

+    private List<TypedFlowEntryWithLoad> typedFlowEntryLoadByInstInternal(ConnectPoint cp,

+                                                                      Map<FlowRule, TypedStoredFlowEntry> currentMap,

+                                                                      Map<FlowRule, TypedStoredFlowEntry> previousMap,

+                                                                      boolean isAllInstType,

+                                                                      Instruction.Type instType,

+                                                                      int liveTypePollInterval) {

+        List<TypedFlowEntryWithLoad> fel = new ArrayList<>();

+

+        for (TypedStoredFlowEntry tfe : currentMap.values()) {

+            if (isAllInstType ||

+                    tfe.treatment().allInstructions().stream().

+                            filter(i -> i.type() == instType).

+                            findAny().isPresent()) {

+                long currentBytes = tfe.bytes();

+                long previousBytes = previousMap.getOrDefault(tfe, new DefaultTypedFlowEntry((FlowRule) tfe)).bytes();

+                Load fLoad = new DefaultLoad(currentBytes, previousBytes, liveTypePollInterval);

+                fel.add(new TypedFlowEntryWithLoad(cp, tfe, fLoad));

+            }

+        }

+

+        return fel;

+    }

+

+    private List<TypedFlowEntryWithLoad> loadTopnPortInternal(ConnectPoint cp,

+                                                             TypedStoredFlowEntry.FlowLiveType liveType,

+                                                             Instruction.Type instType,

+                                                             int topn) {

+        List<TypedFlowEntryWithLoad> fel = loadAllPortInternal(cp, liveType, instType);

+

+        // Sort with descending order of load

+        List<TypedFlowEntryWithLoad> tfel =

+                fel.stream().sorted(Comparators.TYPEFLOWENTRY_WITHLOAD_COMPARATOR).

+                        limit(topn).collect(Collectors.toList());

+

+        return tfel;

+    }

+

+    private long aggregateBytesSet(Set<FlowEntry> setFE) {

+        return setFE.stream().mapToLong(FlowEntry::bytes).sum();

+    }

+

+    private long aggregateBytesMap(Map<FlowRule, TypedStoredFlowEntry> mapFE) {

+        return mapFE.values().stream().mapToLong(FlowEntry::bytes).sum();

+    }

+

+    /**

+     * Internal data class holding two set of typed flow entries.

+     */

+    private static class TypedStatistics {

+        private final ImmutableSet<FlowEntry> currentAll;

+        private final ImmutableSet<FlowEntry> previousAll;

+

+        private final Map<FlowRule, TypedStoredFlowEntry> currentImmediate = new HashMap<>();

+        private final Map<FlowRule, TypedStoredFlowEntry> previousImmediate = new HashMap<>();

+

+        private final Map<FlowRule, TypedStoredFlowEntry> currentShort = new HashMap<>();

+        private final Map<FlowRule, TypedStoredFlowEntry> previousShort = new HashMap<>();

+

+        private final Map<FlowRule, TypedStoredFlowEntry> currentMid = new HashMap<>();

+        private final Map<FlowRule, TypedStoredFlowEntry> previousMid = new HashMap<>();

+

+        private final Map<FlowRule, TypedStoredFlowEntry> currentLong = new HashMap<>();

+        private final Map<FlowRule, TypedStoredFlowEntry> previousLong = new HashMap<>();

+

+        private final Map<FlowRule, TypedStoredFlowEntry> currentUnknown = new HashMap<>();

+        private final Map<FlowRule, TypedStoredFlowEntry> previousUnknown = new HashMap<>();

+

+        public TypedStatistics(Set<FlowEntry> current, Set<FlowEntry> previous) {

+            this.currentAll = ImmutableSet.copyOf(checkNotNull(current));

+            this.previousAll = ImmutableSet.copyOf(checkNotNull(previous));

+

+            currentAll.forEach(fe -> {

+                TypedStoredFlowEntry tfe = TypedFlowEntryWithLoad.newTypedStoredFlowEntry(fe);

+

+                switch (tfe.flowLiveType()) {

+                    case IMMEDIATE_FLOW:

+                        currentImmediate.put(fe, tfe);

+                        break;

+                    case SHORT_FLOW:

+                        currentShort.put(fe, tfe);

+                        break;

+                    case MID_FLOW:

+                        currentMid.put(fe, tfe);

+                        break;

+                    case LONG_FLOW:

+                        currentLong.put(fe, tfe);

+                        break;

+                    default:

+                        currentUnknown.put(fe, tfe);

+                        break;

+                }

+            });

+

+            previousAll.forEach(fe -> {

+                TypedStoredFlowEntry tfe = TypedFlowEntryWithLoad.newTypedStoredFlowEntry(fe);

+

+                switch (tfe.flowLiveType()) {

+                    case IMMEDIATE_FLOW:

+                        if (currentImmediate.containsKey(fe)) {

+                            previousImmediate.put(fe, tfe);

+                        } else if (currentShort.containsKey(fe)) {

+                            previousShort.put(fe, tfe);

+                        } else if (currentMid.containsKey(fe)) {

+                            previousMid.put(fe, tfe);

+                        } else if (currentLong.containsKey(fe)) {

+                            previousLong.put(fe, tfe);

+                        } else {

+                            previousUnknown.put(fe, tfe);

+                        }

+                        break;

+                    case SHORT_FLOW:

+                        if (currentShort.containsKey(fe)) {

+                            previousShort.put(fe, tfe);

+                        } else if (currentMid.containsKey(fe)) {

+                            previousMid.put(fe, tfe);

+                        } else if (currentLong.containsKey(fe)) {

+                            previousLong.put(fe, tfe);

+                        } else {

+                            previousUnknown.put(fe, tfe);

+                        }

+                        break;

+                    case MID_FLOW:

+                        if (currentMid.containsKey(fe)) {

+                            previousMid.put(fe, tfe);

+                        } else if (currentLong.containsKey(fe)) {

+                            previousLong.put(fe, tfe);

+                        } else {

+                            previousUnknown.put(fe, tfe);

+                        }

+                        break;

+                    case LONG_FLOW:

+                        if (currentLong.containsKey(fe)) {

+                            previousLong.put(fe, tfe);

+                        } else {

+                            previousUnknown.put(fe, tfe);

+                        }

+                        break;

+                    default:

+                        previousUnknown.put(fe, tfe);

+                        break;

+                }

+            });

+        }

+

+        /**

+         * Returns flow entries as the current value.

+         *

+         * @return flow entries as the current value

+         */

+        public ImmutableSet<FlowEntry> current() {

+            return currentAll;

+        }

+

+        /**

+         * Returns flow entries as the previous value.

+         *

+         * @return flow entries as the previous value

+         */

+        public ImmutableSet<FlowEntry> previous() {

+            return previousAll;

+        }

+

+        public Map<FlowRule, TypedStoredFlowEntry> currentImmediate() {

+            return currentImmediate;

+        }

+        public Map<FlowRule, TypedStoredFlowEntry> previousImmediate() {

+            return previousImmediate;

+        }

+        public Map<FlowRule, TypedStoredFlowEntry> currentShort() {

+            return currentShort;

+        }

+        public Map<FlowRule, TypedStoredFlowEntry> previousShort() {

+            return previousShort;

+        }

+        public Map<FlowRule, TypedStoredFlowEntry> currentMid() {

+            return currentMid;

+        }

+        public Map<FlowRule, TypedStoredFlowEntry> previousMid() {

+            return previousMid;

+        }

+        public Map<FlowRule, TypedStoredFlowEntry> currentLong() {

+            return currentLong;

+        }

+        public Map<FlowRule, TypedStoredFlowEntry> previousLong() {

+            return previousLong;

+        }

+        public Map<FlowRule, TypedStoredFlowEntry> currentUnknown() {

+            return currentUnknown;

+        }

+        public Map<FlowRule, TypedStoredFlowEntry> previousUnknown() {

+            return previousUnknown;

+        }

+

+        /**

+         * Validates values are not empty.

+         *

+         * @return false if either of the sets is empty. Otherwise, true.

+         */

+        public boolean isValid() {

+            return !(currentAll.isEmpty() || previousAll.isEmpty());

+        }

+

+        @Override

+        public int hashCode() {

+            return Objects.hash(currentAll, previousAll);

+        }

+

+        @Override

+        public boolean equals(Object obj) {

+            if (this == obj) {

+                return true;

+            }

+            if (!(obj instanceof TypedStatistics)) {

+                return false;

+            }

+            final TypedStatistics other = (TypedStatistics) obj;

+            return Objects.equals(this.currentAll, other.currentAll) &&

+                    Objects.equals(this.previousAll, other.previousAll);

+        }

+

+        @Override

+        public String toString() {

+            return MoreObjects.toStringHelper(this)

+                    .add("current", currentAll)

+                    .add("previous", previousAll)

+                    .toString();

+        }

+    }

+

+    private void checkLoadValidity(Set<FlowEntry> current, Set<FlowEntry> previous) {

+        current.stream().forEach(c -> {

+            FlowEntry f = previous.stream().filter(p -> c.equals(p)).

+                    findAny().orElse(null);

+            if (f != null && c.bytes() < f.bytes()) {

+                log.debug("FlowStatisticManager:checkLoadValidity():" +

+                        "Error: " + c + " :Previous bytes=" + f.bytes() +

+                        " is larger than current bytes=" + c.bytes() + " !!!");

+            }

+        });

+

+    }

+

+    /**

+     * Creates a predicate that checks the instruction type of a flow entry is the same as

+     * the specified instruction type.

+     *

+     * @param instType instruction type to be checked

+     * @return predicate

+     */

+    private static Predicate<FlowEntry> hasInstructionType(Instruction.Type instType) {

+        return new Predicate<FlowEntry>() {

+            @Override

+            public boolean apply(FlowEntry flowEntry) {

+                List<Instruction> allInstructions = flowEntry.treatment().allInstructions();

+

+                return allInstructions.stream().filter(i -> i.type() == instType).findAny().isPresent();

+            }

+        };

+    }

+

+    /**

+     * Internal flow rule event listener for FlowStatisticManager.

+     */

+    private class InternalFlowRuleStatsListener implements FlowRuleListener {

+

+        @Override

+        public void event(FlowRuleEvent event) {

+            FlowRule rule = event.subject();

+            switch (event.type()) {

+                case RULE_ADDED:

+                    if (rule instanceof FlowEntry) {

+                        flowStatisticStore.addFlowStatistic((FlowEntry) rule);

+                    }

+                    break;

+                case RULE_UPDATED:

+                    flowStatisticStore.updateFlowStatistic((FlowEntry) rule);

+                    break;

+                case RULE_ADD_REQUESTED:

+                    break;

+                case RULE_REMOVE_REQUESTED:

+                    break;

+                case RULE_REMOVED:

+                    flowStatisticStore.removeFlowStatistic(rule);

+                    break;

+                default:

+                    log.warn("Unknown flow rule event {}", event);

+            }

+        }

+    }

+}