Improve Websocket notification interface

Improve websocket notification interface

Issue-ID: CCSDK-3315
Signed-off-by: Ravi Pendurty <ravi.pendurty@highstreet-technologies.com>
Change-Id: I0ded865adddb546ade98df4760e0a32ec964295a
Signed-off-by: Ravi Pendurty <ravi.pendurty@highstreet-technologies.com>

9 files changed, 639 insertions, 192 deletions

diff --git a/sdnr/wt/websocketmanager/provider/pom.xml b/sdnr/wt/websocketmanager/provider/pom.xml
index 0366ed2a4..fcdaa5e2f 100644
--- a/sdnr/wt/websocketmanager/provider/pom.xml
+++ b/sdnr/wt/websocketmanager/provider/pom.xml
@@ -57,6 +57,11 @@
             <version>${project.version}</version>
         </dependency>
         <dependency>
+            <groupId>org.opendaylight.mdsal.model</groupId>
+            <artifactId>ietf-topology</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
             <groupId>${project.groupId}</groupId>
             <artifactId>sdnr-wt-yang-utils</artifactId>
             <version>${project.version}</version>
@@ -136,7 +141,7 @@
         </dependency>
         <dependency>
             <groupId>${project.groupId}</groupId>
-            <artifactId>sdnr-wt-devicemanager-provider</artifactId>
+            <artifactId>sdnr-wt-devicemanager-core-provider</artifactId>
             <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
diff --git a/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/WebSocketManagerProvider.java b/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/WebSocketManagerProvider.java
index 0b6e9b453..610001775 100644
--- a/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/WebSocketManagerProvider.java
+++ b/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/WebSocketManagerProvider.java
@@ -24,6 +24,7 @@ import org.onap.ccsdk.features.sdnr.wt.websocketmanager.model.data.NotificationO
 import org.onap.ccsdk.features.sdnr.wt.yang.mapper.YangToolsMapperHelper;
 import org.opendaylight.mdsal.dom.api.DOMNotification;
 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev130715.DateAndTime;
+import org.opendaylight.yang.gen.v1.urn.tbd.params.xml.ns.yang.network.topology.rev131021.NodeId;
 import org.opendaylight.yangtools.yang.binding.Notification;
 import org.opendaylight.yangtools.yang.common.QName;
 import org.osgi.service.http.HttpService;
@@ -84,24 +85,44 @@ public class WebSocketManagerProvider implements WebsocketManagerService, AutoCl
 
 
     @Override
-    public void sendNotification(Notification notification, String nodeId, QName eventType) {
+    public void sendNotification(Notification notification, NodeId nodeId, QName eventType) {
+        if(!assertNotificationType(notification, eventType)){
+            return;
+        }
         this.sendNotification(notification, nodeId, eventType, YangToolsMapperHelper.getTime(notification,Instant.now()));
     }
 
+    public static boolean assertNotificationType(Notification notification, QName eventType) {
+        final String yangTypeName = eventType.getLocalName();
+        final Class<?> cls = notification.getClass();
+        final String clsNameToTest = YangToolsMapperHelper.toCamelCaseClassName(yangTypeName);
+        if(cls.getSimpleName().equals(clsNameToTest)) {
+            return true;
+        }
+        Class<?>[] ifs = cls.getInterfaces();
+        for(Class<?> clsif:ifs) {
+            if(clsif.getSimpleName().equals(clsNameToTest)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+
     @Override
-    public void sendNotification(Notification notification, String nodeId, QName eventType, DateAndTime eventTime) {
-        WebSocketManagerSocket.broadCast(new NotificationOutput(notification, nodeId, eventType, eventTime));
+    public void sendNotification(Notification notification, NodeId nodeId, QName eventType, DateAndTime eventTime) {
+        WebSocketManagerSocket.broadCast(new NotificationOutput(notification, nodeId.getValue(), eventType, eventTime));
 
     }
 
     @Override
-    public void sendNotification(DOMNotification notification, String nodeId, QName eventType) {
+    public void sendNotification(DOMNotification notification, NodeId nodeId, QName eventType) {
         LOG.warn("not yet implemented");
 
     }
 
     @Override
-    public void sendNotification(DOMNotification notification, String nodeId, QName eventType, DateAndTime eventTime) {
+    public void sendNotification(DOMNotification notification, NodeId nodeId, QName eventType, DateAndTime eventTime) {
         LOG.warn("not yet implemented");
 
     }
diff --git a/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/WebSocketManagerSocket.java b/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/WebSocketManagerSocket.java
index 945de3c1f..a642bda69 100644
--- a/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/WebSocketManagerSocket.java
+++ b/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/WebSocketManagerSocket.java
@@ -25,6 +25,10 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Random;
 import java.util.Set;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 import org.eclipse.jetty.websocket.api.Session;
@@ -41,7 +45,7 @@ import org.slf4j.LoggerFactory;
 
 public class WebSocketManagerSocket extends WebSocketAdapter {
 
-    private static final Logger LOG = LoggerFactory.getLogger(WebSocketManagerSocket.class.getName());
+    private static final Logger LOG = LoggerFactory.getLogger(WebSocketManagerSocket.class);
     public static final String MSG_KEY_DATA = "data";
     public static final DataType MSG_KEY_SCOPES = DataType.scopes;
     public static final String MSG_KEY_PARAM = "param";
@@ -54,7 +58,47 @@ public class WebSocketManagerSocket extends WebSocketAdapter {
     private static final Pattern PATTERN_SCOPEREGISTRATION =
             Pattern.compile(REGEX_SCOPEREGISTRATION, Pattern.MULTILINE);
     private static final Random RND = new Random();
+    private static final long SEND_MESSAGE_TIMEOUT_MILLIS = 1500;
+    private static final int QUEUE_SIZE = 100;
 
+    private final Thread sendingSyncThread;
+    private final ArrayBlockingQueue<String> messageQueue;
+    private boolean closed;
+
+    private final Runnable sendingRunner = new Runnable() {
+        @Override
+        public void run() {
+            LOG.debug("isrunning");
+            while (!closed) {
+                try {
+
+                    String message = messageQueue.poll();
+                    if (message != null) {
+                        WebSocketManagerSocket.this.session.getRemote().sendStringByFuture(message)
+                                .get(SEND_MESSAGE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+                        LOG.info("message sent");
+                    }
+                } catch (InterruptedException | ExecutionException | TimeoutException e) {
+                    LOG.warn("problem pushing message: ", e);
+                }
+
+                if (messageQueue.isEmpty()) {
+                    trySleep(1000);
+                }
+
+            }
+            LOG.debug("isstopped");
+
+        };
+    };
+
+    private static void trySleep(int sleepMs) {
+        try {
+            Thread.sleep(sleepMs);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
 
     /**
      * list of all sessionids
@@ -80,6 +124,8 @@ public class WebSocketManagerSocket extends WebSocketAdapter {
 
     public WebSocketManagerSocket() {
         this.myUniqueSessionId = _genSessionId();
+        this.sendingSyncThread = new Thread(this.sendingRunner);
+        this.messageQueue = new ArrayBlockingQueue<>(QUEUE_SIZE);
     }
 
     @Override
@@ -112,6 +158,8 @@ public class WebSocketManagerSocket extends WebSocketAdapter {
     @Override
     public void onWebSocketConnect(Session sess) {
         this.session = sess;
+        closed = false;
+        this.sendingSyncThread.start();
         clientList.put(String.valueOf(this.hashCode()), this);
         LOG.debug("client connected from " + this.getRemoteAdr());
     }
@@ -119,13 +167,14 @@ public class WebSocketManagerSocket extends WebSocketAdapter {
     @Override
     public void onWebSocketClose(int statusCode, String reason) {
         clientList.remove(String.valueOf(this.hashCode()));
+        this.sendingSyncThread.interrupt();
+        closed = true;
         LOG.debug("client disconnected from " + this.getRemoteAdr());
     }
 
     @Override
     public void onWebSocketError(Throwable cause) {
         LOG.debug("error caused on " + this.getRemoteAdr() + " :" + cause.getMessage());
-        // super.onWebSocketError(cause);
     }
 
     private String getRemoteAdr() {
@@ -146,12 +195,12 @@ public class WebSocketManagerSocket extends WebSocketAdapter {
     private boolean manageClientRequest(String request) {
         boolean ret = false;
         final Matcher matcher = PATTERN_SCOPEREGISTRATION.matcher(request);
-        if(!matcher.find()) {
+        if (!matcher.find()) {
             return false;
         }
         try {
             ScopeRegistration registration = mapper.readValue(request, ScopeRegistration.class);
-            if (registration!=null && registration.validate() && registration.isType(MSG_KEY_SCOPES)) {
+            if (registration != null && registration.validate() && registration.isType(MSG_KEY_SCOPES)) {
                 ret = true;
                 String sessionId = this.getSessionId();
                 UserScopes clientDto = new UserScopes();
@@ -188,9 +237,9 @@ public class WebSocketManagerSocket extends WebSocketAdapter {
     public void send(String msg) {
         try {
             LOG.trace("sending {}", msg);
-            this.session.getRemote().sendString(msg);
-        } catch (Exception e) {
-            LOG.warn("problem sending message: " + e.getMessage());
+            this.messageQueue.put(msg);
+        } catch (InterruptedException e) {
+            LOG.warn("problem putting message into sending queue: " + e.getMessage());
         }
     }
 
@@ -200,7 +249,7 @@ public class WebSocketManagerSocket extends WebSocketAdapter {
 
     private void sendToAll(NotificationOutput output) {
         try {
-            this.sendToAll(output.getNodeId(), output.getType(), mapper.writeValueAsString(output));
+            sendToAll(output.getNodeId(), output.getType(), mapper.writeValueAsString(output));
         } catch (JsonProcessingException e) {
             LOG.warn("problem serializing noitifcation: ", e);
         }
diff --git a/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/utils/RateFilter.java b/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/utils/RateFilter.java
deleted file mode 100644
index 5f3a5af2c..000000000
--- a/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/utils/RateFilter.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP : ccsdk features
- * ================================================================================
- * Copyright (C) 2020 highstreet technologies GmbH Intellectual Property.
- * All rights reserved.
- * ================================================================================
- * 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.
- * ============LICENSE_END=========================================================
- *
- */
-package org.onap.ccsdk.features.sdnr.wt.websocketmanager.utils;
-
-import java.time.Duration;
-import java.time.Instant;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Problems of to many notifications during mount of thousand of devices:
- * <ul>
- * <li>Overload ODLUX with notification flood -> ODLUX App can not control notifications rate
- * <li>Notification processing blocks user -> App design with notifications popups
- * </ul>
- * Rate filter
- * <ul>
- * <li>Do not use a thread -> Do nothing if there are no notifications
- * <li>Parameter1 integrationTime : Measurement or integration time for period
- * <li>Parameter2 readMaxCount : Specifies event number per interval indicating overload
- * <li>Start measurement on event received that comes later then
- * </ul>
- *
- * <pre>
- *  Example (e: Event received, rateMaxCount=3)
- *         eee                           e  e e e e e  e  e e e e    e         e                e
- *  ---//--|--------------|-----//-------|--------------|--------------|--------------|---//----|--------------|
- *         P1             P2             P1             P2             P3             P7        P1
- *Overload              no             no            yes            yes             no              no
- *
- *
- *Intention to use:
- *   1. Construct with parameters for WS stream to handle
- *   2.
- * </pre>
- */
-
-public class RateFilter {
-
-    private static final Logger LOG = LoggerFactory.getLogger(RateFilter.class.getName());
-
-    private final Duration integrationTime; // Integration time to measure event rate
-    private final long rateMaxCount; //Rate for dropping packets
-    private Instant timeStampPeriodStart; //Time stamp period beginn
-    private Instant timeStampLastEvent; //Measurement interval
-    private long rateCount; // >0: integration running 0: no integration running
-    private boolean overload; //true means in overload status. Change at end of period only.
-    private GetNow get;
-
-    /**
-     * Allow testing with own timestamp provider
-     */
-    public interface GetNow {
-        Instant now();
-    }
-
-    public RateFilter(Duration integrationTime, long rateMaxCount, GetNow getNowMethod) {
-        this.integrationTime = integrationTime;
-        this.rateMaxCount = rateMaxCount;
-        this.get = getNowMethod;
-        this.timeStampLastEvent = Instant.MIN;
-    }
-
-    public RateFilter(Duration integrationTime, long rateMaxCount) {
-        this(integrationTime, rateMaxCount, () -> Instant.now());
-    }
-
-    public synchronized boolean getOverloadStatus() {
-        return overload;
-    }
-
-    /**
-     * Handle filter on event received
-     */
-    public synchronized void filterEvent() {
-        final Instant now = get.now();
-        final Duration durationSinceLastEvent = Duration.between(timeStampLastEvent, now);
-        this.timeStampLastEvent = now;
-
-        if (durationSinceLastEvent.compareTo(integrationTime) >= 0) {
-            //No measurement. Sync and start with period
-            LOG.debug("Sync");
-            timeStampPeriodStart = now;
-            rateCount = 1; //Reset event count .. is part of the
-        } else {
-            //Within period
-            Duration durationPeriod = Duration.between(timeStampPeriodStart, now);
-            rateCount++;
-            boolean endOfPeriod = durationPeriod.compareTo(integrationTime) >= 0;
-            LOG.debug("Period start{}: now:{} end:{} dur:{} int:{}", timeStampPeriodStart, now, endOfPeriod, durationPeriod, integrationTime);
-            if (endOfPeriod) {
-                //Only if end of Period
-                overload = rateCount > rateMaxCount;
-                LOG.debug("Reset overload {}", overload);
-                timeStampPeriodStart = timeStampPeriodStart.plus(integrationTime);
-                rateCount = 0;
-            }
-        }
-    }
-
-    @Override
-    public String toString() {
-        StringBuilder builder = new StringBuilder();
-        builder.append("RateFilter [integrationTime=");
-        builder.append(integrationTime);
-        builder.append(", rateMaxCount=");
-        builder.append(rateMaxCount);
-        builder.append(", timeStampPeriodStart=");
-        builder.append(timeStampPeriodStart);
-        builder.append(", timeStampLastEvent=");
-        builder.append(timeStampLastEvent);
-        builder.append(", rateCount=");
-        builder.append(rateCount);
-        builder.append(", overload=");
-        builder.append(overload);
-        builder.append("]");
-        return builder.toString();
-    }
-}
diff --git a/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/utils/RateFilterManager.java b/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/utils/RateFilterManager.java
new file mode 100644
index 000000000..7ffa29e89
--- /dev/null
+++ b/sdnr/wt/websocketmanager/provider/src/main/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager/utils/RateFilterManager.java
@@ -0,0 +1,323 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP : ccsdk features
+ * ================================================================================
+ * Copyright (C) 2020 highstreet technologies GmbH Intellectual Property.
+ * All rights reserved.
+ * ================================================================================
+ * 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.
+ * ============LICENSE_END=========================================================
+ *
+ */
+package org.onap.ccsdk.features.sdnr.wt.websocketmanager.utils;
+
+import java.io.Closeable;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Problems of to many notifications during mount of thousand of devices:
+ * <ul>
+ * <li>Overload ODLUX with notification flood -> ODLUX App can not control notifications rate
+ * <li>Notification processing blocks user -> App design with notifications popups
+ * </ul>
+ * Rate filter requirements
+ * <ul>
+ * <li>Use a single thread
+ * <li>Parameter1 integrationTime : Measurement or integration time for period
+ * <li>Parameter2 readMaxCount : Specifies event number per interval indicating overload
+ * <li>Start measurement on event received that comes later then
+ * </ul>
+ *
+ * <pre>
+ *  Example for behavior (e: Event received, rateMaxCount=3)
+ *         eee                           e  e e e e e  e  e e e e    e         e                e
+ *  ---//--|--------------|-----//-------|--------------|--------------|--------------|---//----|--------------|
+ *         P1             P2             P1             P2             P3             P7        P1
+ *Overload              no             no            yes            yes             no              no
+ * </pre>
+ *
+ * Interface to use:
+ * <ul>
+ * <li>construct RateFilterManager. Parameters are integration time and function to get the actual time
+ * <li>RateFilterManager.getRateFilter() provides rateFilter object for a stream to count events and provide overload
+ * status.
+ * <li>rateFilter.event() count the events during measurement period
+ * <li>rateFilter.getOverloadStatus() indicates status
+ * <li>rateFilter.close() to release this object
+ * </ul>
+ */
+
+public class RateFilterManager implements Closeable {
+
+    private static final Logger LOG = LoggerFactory.getLogger(RateFilterManager.class.getName());
+    private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("dd.mm.yy hh:mm:ss_SSS")
+            .withLocale(Locale.GERMAN).withZone(ZoneId.systemDefault());
+    private static final long CLIENTS_NUMBER_WARNLEVEL = 1000;
+
+    //Configuration
+    private final Duration integrationTime; // Integration time to measure event rate
+    private GetNow get; //Provides actual system time
+    private final Map<Long, RateFilter> rateFilterList;
+    @SuppressWarnings("unused")
+    private final Timer timerTask;
+
+    /**
+     * Allow testing with own timestamp provider Provide actual system time.
+     */
+    public interface GetNow {
+        Instant now();
+    }
+
+    /**
+     * Constructor with all parameters, intended to be used for unit test
+     *
+     * @param integrationTime is the interval length for counting events.
+     * @param rateMaxCountDefault if event count exceed this limit, status changes to overload.
+     * @param startTimer true start time with intervall time
+     * @param get function to provide actual system time.
+     */
+    public RateFilterManager(Duration integrationTime, boolean startTimer, GetNow get) {
+        this.integrationTime = integrationTime;
+        this.get = get;
+
+        this.rateFilterList = Collections.synchronizedMap(new HashMap<Long, RateFilter>());
+        this.timerTask = startTimer ? startTimerTask(integrationTime) : null;
+    }
+
+    /**
+     * Get RateFilter manager
+     *
+     * @param integrationTime is the time to measure events
+     * @param rateMaxCountDefault if exceeded state overload is true
+     */
+    public RateFilterManager(Duration integrationTime) {
+        this(integrationTime, true, () -> Instant.now());
+    }
+
+    /**
+     */
+    /**
+     * Get a specific rate filter for one stream. Use close() to release.
+     *
+     * @param ratePerMinute Rate per Minute for this filter. If 0 never overloaded.
+     * @return RateFilter object for each event stream.
+     * @throws IllegalArgumentException on negative rate
+     */
+    public synchronized RateFilter getRateFilter(long maxRatePerMinute) throws IllegalArgumentException {
+        long maxEventsPerIntegration = convertRPMToMaxCount(maxRatePerMinute);
+        if (maxEventsPerIntegration < 0)
+            throw new IllegalArgumentException(
+                    "Resulting in illegal maxEventsPerIntegration=" + maxEventsPerIntegration);
+        return getRateFilterInstance(maxEventsPerIntegration);
+    }
+
+    @Override
+    public void close() {
+        if (timerTask != null) {
+            timerTask.cancel();
+            timerTask.purge();
+        }
+        rateFilterList.clear();
+    }
+
+    /**
+     * Function to get a new Ratefilter for a connection
+     *
+     * @param maxEventsPerIntegration
+     * @return reference to object with filter status
+     */
+    private RateFilter getRateFilterInstance(long maxEventsPerIntegration) {
+        RateFilter rateFilter;
+        synchronized (rateFilterList) {
+            rateFilter = rateFilterList.get(maxEventsPerIntegration);
+            if (rateFilter == null) {
+                rateFilter = new RateFilter(maxEventsPerIntegration);
+                synchronized (rateFilterList) {
+                    rateFilterList.put(maxEventsPerIntegration, rateFilter);
+                }
+            } else {
+                if (rateFilter.addClient() > CLIENTS_NUMBER_WARNLEVEL)
+                    LOG.warn("Warnlevel {} exceeded for client connections", CLIENTS_NUMBER_WARNLEVEL);
+            }
+        }
+        return rateFilter;
+    }
+
+    private Timer startTimerTask(Duration integrationTime) {
+        long milliseconds = integrationTime.toMillis();
+        LOG.debug("Start startTimerTask with {} ms", milliseconds);
+        Timer time = new Timer();
+        time.scheduleAtFixedRate(new TimeoutHandler(), 0L, milliseconds);
+        return time;
+    }
+
+    private class TimeoutHandler extends TimerTask {
+        @Override
+        public void run() {
+            LOG.debug("Run timeout task at {}", f(get.now()));
+            synchronized (rateFilterList) {
+                rateFilterList.forEach((k, f) -> f.timer());
+            }
+        }
+    }
+
+    /**
+     * Provide nice debug output for Instant and Duration
+     *
+     * @param i with instant
+     * @return output string
+     */
+    private static String f(Instant i) {
+        return i != null ? FORMATTER.format(i) : "null";
+    }
+
+    /**
+     * Convert a rate per minute into events per integration time.
+     *
+     * @param ratePerMinute
+     * @return events per integration time.
+     */
+    private long convertRPMToMaxCount(long ratePerMinute) {
+        return ratePerMinute * integrationTime.toSeconds() / TimeUnit.MINUTES.toSeconds(1);
+    }
+
+    /**
+     * Ratefilter class contains status informaton for one event stream.
+     */
+    public class RateFilter implements Closeable {
+        private final long maxEventsPerIntegration; //uuid and maximum of events without overload
+        private Long clients; // Number of clients for this filter.
+        private long rateCount; // number of events during integration period
+        private boolean overload; //true means in overload status. Change at end of period only.
+
+        /**
+         * Create a new Filter
+         *
+         * @param maxEventsPerIntegration >= 1 characteristics and uuid of this filter. < 1 switched off
+         * @see {@link #close}
+         */
+        private RateFilter(long maxEventsPerIntegration) {
+            synchronized (this) {
+                this.clients = 1L;
+                this.maxEventsPerIntegration = maxEventsPerIntegration;
+                this.rateCount = 0;
+            }
+        }
+
+        /**
+         * Add a client to this filter
+         *
+         * @return number of clients, handled by this filter
+         * @see {@link #close}
+         */
+        private synchronized long addClient() {
+            if (clients >= 1) {
+                ++clients;
+            } else {
+                LOG.warn("Misalligned open/close for {} with number {}", maxEventsPerIntegration, clients);
+            }
+            return clients;
+        }
+
+        /**
+         * Provide actual overload status
+         *
+         * @return status true means overloaded false not overloaded
+         */
+        public synchronized boolean getOverloadStatus() {
+            return overload;
+        }
+
+        /**
+         * Handle filter on event received
+         */
+        public synchronized void event() {
+            rateCount++;
+            LOG.debug("event rc:{}", rateCount);
+        }
+
+        /**
+         * Called if measurement period ends. Device if overload and reset counter.
+         */
+        public synchronized void timer() {
+            //Change overload only at end of period
+            //Always inactive if maxEventsPerIntegration== 0
+            if (maxEventsPerIntegration > 0) {
+                overload = rateCount > maxEventsPerIntegration;
+            }
+            rateCount = 0;
+            LOG.debug("Timer ol:{} rc:{}", overload, rateCount);
+        }
+
+        /**
+         * Get maximum events allowed per integration period
+         *
+         * @return 1 ...
+         */
+        public synchronized long getMaxEventsPerIntegration() {
+            return maxEventsPerIntegration;
+        }
+
+        /**
+         * Get number of client streams.
+         *
+         * @return 1 ...
+         */
+        public synchronized long getClients() {
+            return clients;
+        }
+
+        @Override
+        public void close() {
+            synchronized (rateFilterList) {
+                if (clients == 1) {
+                    LOG.debug("Close and remove last client {}", maxEventsPerIntegration);
+                    rateFilterList.remove(this.maxEventsPerIntegration);
+                    clients--;
+                } else if (clients > 1) {
+                    LOG.debug("Close one client of {} for events {}", clients, maxEventsPerIntegration);
+                    clients--;
+                } else {
+                    LOG.warn("Misaligned new/close for events {}", maxEventsPerIntegration);
+                }
+            }
+        }
+
+        @Override
+        public String toString() {
+            StringBuilder builder = new StringBuilder();
+            builder.append("RateFilter [maxEventsPerIntegration=");
+            builder.append(maxEventsPerIntegration);
+            builder.append(", clients=");
+            builder.append(clients);
+            builder.append(", rateCount=");
+            builder.append(rateCount);
+            builder.append(", overload=");
+            builder.append(overload);
+            builder.append("]");
+            return builder.toString();
+        }
+    }
+}
diff --git a/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/AkkaConfigTest.java b/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/AkkaConfigTest.java
index f3cf09545..df04c388f 100644
--- a/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/AkkaConfigTest.java
+++ b/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/AkkaConfigTest.java
@@ -20,13 +20,11 @@ package org.onap.ccsdk.features.sdnr.wt.websocketmanager2.test;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
-import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileNotFoundException;
-import java.io.FileReader;
 import java.io.IOException;
 import java.net.URISyntaxException;
-import java.nio.file.Paths;
+import java.nio.file.Files;
 import org.junit.Test;
 import org.onap.ccsdk.features.sdnr.wt.websocketmanager.utils.AkkaConfig;
 
@@ -58,19 +56,6 @@ public class AkkaConfigTest {
     public static String loadResourceContentAsString(String resourceName)
             throws URISyntaxException, FileNotFoundException, IOException {
 
-        StringBuilder sb = new StringBuilder();
-
-        ClassLoader classLoader = AkkaConfigTest.class.getClassLoader();
-        File file = Paths.get(classLoader.getResource(resourceName).toURI()).toFile();
-        try (BufferedReader br = new BufferedReader(new FileReader(file))) {
-            String line = br.readLine();
-
-            while (line != null) {
-                sb.append(line);
-                sb.append(System.lineSeparator());
-                line = br.readLine();
-            }
-        }
-        return sb.toString();
+        return Files.readString(new File("src/test/resources/"+resourceName).toPath());
     }
 }
diff --git a/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/RateFilterTest.java b/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/RateFilterTest.java
index f4fab6810..d5a940f73 100644
--- a/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/RateFilterTest.java
+++ b/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/RateFilterTest.java
@@ -21,10 +21,15 @@
  */
 package org.onap.ccsdk.features.sdnr.wt.websocketmanager2.test;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
 import java.time.Duration;
 import java.time.Instant;
+import java.util.Timer;
+import java.util.TimerTask;
 import org.junit.Test;
-import org.onap.ccsdk.features.sdnr.wt.websocketmanager.utils.RateFilter;
+import org.onap.ccsdk.features.sdnr.wt.websocketmanager.utils.RateFilterManager;
+import org.onap.ccsdk.features.sdnr.wt.websocketmanager.utils.RateFilterManager.RateFilter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -33,11 +38,13 @@ import org.slf4j.LoggerFactory;
  *
  * <pre>
  *  Testcase (e: 17 Event received, rateMaxCount=3)
- *         eee                           e  e e e e e  e  e e e e    e         e                e
- *  ---//--|--------------|-----//-------|--------------|--------------|--------------|---//----|--------------|
- *         P1:1           P2:1           P1:2           P2:2           P3:2          P4:2       P1:3
- *         1000-1002      2000           3500 10 millis
- *Overload              no             no            yes            yes             no              no
+ *         1 3                           4  5 6 7 8 9 10 11    14   15        16               17     18
+ *     t            t           t          t              t              t              t         t              t
+ *           eee                           e  e e e e e  e  e e e e    e         e                e      e
+ *    ---//--|--------------|-----//-------|--------------|--------------|--------------|---//----|--------------|
+ *           P1:1           P2:1           P1:2           P2:2           P3:2          P4:2       P1:3
+ * ms  500   1000-1002      2000           3500           4500           5500           6500      7500           8500
+ *Overload              no             no            yes                      yes     no              no
  * </pre>
  *
  */
@@ -45,37 +52,140 @@ public class RateFilterTest {
 
     private static final Logger LOG = LoggerFactory.getLogger(RateFilterTest.class.getName());
 
-    private static int MILLIS = 1000;
-    private static long[] now = { 1000, 1001, 1002, //P1:1 0-2
-            3500, 3550, 3560, 3570, 3580, 3590, 3800, //P1:2 3500 3-9
-            4510, 4520, 4530, 4540, 4900, //P2:2 4500 10-14
-            5700, //P3:2 5500 15
-            7000, 8000};//P1:3 16-17
+    private static int INTEGRATIONTIMEMILLIS = 1000;
+    private static long EVENTS_PER_INTERVALL = 4;
+    private static long RATE_PER_MINUTE = EVENTS_PER_INTERVALL * 60;
+    /* Negative event time indicates timer event */
+    private static long[] now = {-500, 1000, 1010, 1020, //P1:1 1-3
+            -1500, -2500, -3500, 3500, 3550, 3560, 3570, 3580, 3590, 3800, //P1:2 3500 4-10
+            -4500, 4510, 4520, 4530, 4540, 4900, //P2:2 4500 11-15
+            -5500, 5700, //P3:2 5500 16
+            -6500, -7500, 7500, 8000};//P1:3 17-18
+    private static boolean[] overload = {false, false, false, false, //P1:1 1-3
+            false, false, false, false, false, false, false, false, false, false, //P1:2 3500 4-10
+            true, true, true, true, true, true, //P2:2 4500 11-15
+            true, true, //P3:2 5500 16
+            false, false, false, false};//P1:3 17-18
+
     private static int idx;
+    private static long millis;
 
     @Test
-    public void test() {
-        RateFilter rateFilter = new RateFilter(Duration.ofMillis(MILLIS), 4, () -> getNow());
+    public void testStates() {
+        reset();
+        RateFilterManager rateFilterManager =
+                new RateFilterManager(Duration.ofMillis(INTEGRATIONTIMEMILLIS), false, () -> getNow());
+        RateFilter rateFilter = rateFilterManager.getRateFilter(RATE_PER_MINUTE);
         LOG.info("Init done");
+        assertEquals("Events per integration period", EVENTS_PER_INTERVALL, rateFilter.getMaxEventsPerIntegration());
 
-        for (int t=0; t < 20; t++) {
-            LOG.info("{}", t);
-            rateFilter.filterEvent();
-            LOG.info("{}", rateFilter.getOverloadStatus());
+        for (int t = 1; t < 30; t++) {
+            boolean expected = tick();
+            if (millis < 0) {
+                LOG.info("{} - timer {}", t, millis);
+                rateFilter.timer();
+            } else {
+                LOG.info("{} - event {}", t, millis);
+                rateFilter.event();
+            }
+            LOG.info("Overload={} {}", rateFilter.getOverloadStatus(), expected);
+            assertEquals("Filter activity", expected, rateFilter.getOverloadStatus());
         }
+        rateFilter.close();
+    }
+
+    @Test
+    public void testThread() throws InterruptedException {
+        LOG.info("testThread");
+        reset();
+        RateFilterManager rateFilterManager = new RateFilterManager(Duration.ofMillis(INTEGRATIONTIMEMILLIS));
+        RateFilter rateFilter = rateFilterManager.getRateFilter(RATE_PER_MINUTE);
+
+        tick();
+        Thread.sleep(2000);
+
+        Object objectYouNeedToLockOn = new Object();
+        Timer timer = new Timer();
+        timer.scheduleAtFixedRate(new TimerTask() {
+            long localMillis;
+
+            @Override
+            public void run() {
+                long xLocalMillis = localMillis += 10;
+                long xMillis = Math.abs(millis);
+                if (xLocalMillis >= xMillis) {
+                    LOG.info("aTime:{} Millis:{} Idx={}", xLocalMillis, xMillis, idx);
+                    boolean expected = tick();
+                    if (millis > 0) {
+                        //Skip negatives .. handled by timer
+                        rateFilter.event();
+                        boolean actual = rateFilter.getOverloadStatus();
+                        LOG.info("bTime:{} Millis:{} Idx={} Overload={} Expected={} {}", xLocalMillis, xMillis, idx,
+                                actual, expected, actual == expected ? "" : "XXXX");
+                        if (idx >= 30) {
+                            LOG.info("Test is ending");
+                            synchronized (objectYouNeedToLockOn) {
+                                objectYouNeedToLockOn.notify();
+                            }
+                            timer.cancel();
+                        }
+                        assertEquals("Filter activity", expected, rateFilter.getOverloadStatus());
+                    }
+                }
+            }
+        }, 0, 10);
+        synchronized (objectYouNeedToLockOn) {
+            objectYouNeedToLockOn.wait();
+        }
+        //rateFilter.close();
+        LOG.info("Test end");
+    }
+
+    @Test
+    public void testMultipleClients() {
+        RateFilterManager rateFilterManager = new RateFilterManager(Duration.ofMillis(INTEGRATIONTIMEMILLIS));
+        RateFilter rateFilter1 = rateFilterManager.getRateFilter(RATE_PER_MINUTE);
+        assertEquals("Multiple clients", 1, rateFilter1.getClients());
+        RateFilter rateFilter2 = rateFilterManager.getRateFilter(RATE_PER_MINUTE);
+        assertEquals("Multiple clients", 2, rateFilter1.getClients());
+        RateFilter rateFilter3 = rateFilterManager.getRateFilter(RATE_PER_MINUTE);
+        assertEquals("Multiple clients", 3, rateFilter1.getClients());
+
+        assertEquals("Similar instances", rateFilter1, rateFilter3);
+
+        RateFilter rateFilterOther = rateFilterManager.getRateFilter(2*RATE_PER_MINUTE);
+        assertNotEquals("Different instances", rateFilter1, rateFilterOther);
+        rateFilterOther.close();
+
+        rateFilter3.close();
+        assertEquals("Multiple clients", 2, rateFilter1.getClients());
+        rateFilter2.close();
+        assertEquals("Multiple clients", 1, rateFilter1.getClients());
+        rateFilter1.close();
+        assertEquals("Multiple clients", 0, rateFilter1.getClients());
+
+        rateFilterManager.close();
+    }
+
+    private Instant getNow() {
+        LOG.debug("Now:{}", millis);
+        return Instant.ofEpochMilli(Math.abs(millis));
+    }
 
+    private void reset() {
+        idx = 0;
     }
 
-    Instant getNow() {
-        long res;
+    private boolean tick() {
         if (idx < now.length) {
-            res = now[idx];
+            millis = now[idx];
         } else {
             int lastIdx = now.length - 1;
-            res = now[lastIdx] + (idx - lastIdx) * MILLIS;
+            millis = now[lastIdx] + (idx - lastIdx) * INTEGRATIONTIMEMILLIS;
         }
+        boolean expected = idx < overload.length ? overload[idx] : false;
         idx++;
-        return Instant.ofEpochMilli(res);
+        return expected;
     }
 
 }
diff --git a/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/UserScopeTest.java b/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/UserScopeTest.java
new file mode 100644
index 000000000..885ded348
--- /dev/null
+++ b/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/UserScopeTest.java
@@ -0,0 +1,76 @@
+/*
+ * ============LICENSE_START========================================================================
+ * ONAP : ccsdk feature sdnr wt
+ * =================================================================================================
+ * Copyright (C) 2019 highstreet technologies GmbH Intellectual Property. All rights reserved.
+ * =================================================================================================
+ * 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.
+ * ============LICENSE_END==========================================================================
+ */
+package org.onap.ccsdk.features.sdnr.wt.websocketmanager2.test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import java.util.Arrays;
+import java.util.List;
+import org.junit.Test;
+import org.onap.ccsdk.features.sdnr.wt.websocketmanager.model.data.ReducedSchemaInfo;
+import org.onap.ccsdk.features.sdnr.wt.websocketmanager.model.data.SchemaInfo;
+import org.onap.ccsdk.features.sdnr.wt.websocketmanager.model.data.Scope;
+import org.onap.ccsdk.features.sdnr.wt.websocketmanager.utils.UserScopes;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.devicemanager.rev190109.ObjectCreationNotification;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.devicemanager.rev190109.ProblemNotification;
+import org.opendaylight.yangtools.yang.common.QName;
+
+public class UserScopeTest {
+
+
+    @Test
+    public void testAllNodes() {
+        UserScopes scopes1 = new UserScopes();
+        scopes1.setScopes(Arrays.asList(buildScope(null, ProblemNotification.QNAME)));
+
+        assertTrue(scopes1.hasScope(new ReducedSchemaInfo(ProblemNotification.QNAME)));
+        assertFalse(scopes1.hasScope("RoadmA", new ReducedSchemaInfo(ObjectCreationNotification.QNAME)));
+
+        assertTrue(scopes1.hasScope("RoadmA", new ReducedSchemaInfo(ProblemNotification.QNAME)));
+
+    }
+
+    @Test
+    public void testRevisionStar() {
+        UserScopes scopes1 = new UserScopes();
+        scopes1.setScopes(
+                Arrays.asList(buildScope(null, ProblemNotification.QNAME.getNamespace().toString(), "*", null)));
+
+        assertTrue(scopes1.hasScope(new ReducedSchemaInfo(ProblemNotification.QNAME)));
+        assertTrue(scopes1.hasScope("RoadmA", new ReducedSchemaInfo(ObjectCreationNotification.QNAME)));
+
+        assertTrue(scopes1.hasScope("RoadmA", new ReducedSchemaInfo(ProblemNotification.QNAME)));
+
+    }
+
+    private static final Scope buildScope(String nodeId, String namespace, String revision,
+            List<String> notifications) {
+        Scope scope = new Scope();
+        scope.setNodeId(nodeId);
+        scope.setSchema(new SchemaInfo(namespace, revision, notifications));
+        return scope;
+    }
+
+    private static final Scope buildScope(String nodeId, QName qname) {
+        Scope scope = new Scope();
+        scope.setNodeId(nodeId);
+        scope.setSchema(new SchemaInfo(qname));
+        return scope;
+    }
+
+}
diff --git a/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/WebsockerProviderTest.java b/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/WebsockerProviderTest.java
index bc3cd10f8..2e6462462 100644
--- a/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/WebsockerProviderTest.java
+++ b/sdnr/wt/websocketmanager/provider/src/test/java/org/onap/ccsdk/features/sdnr/wt/websocketmanager2/test/WebsockerProviderTest.java
@@ -17,11 +17,17 @@
  */
 package org.onap.ccsdk.features.sdnr.wt.websocketmanager2.test;
 
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import org.junit.Test;
 import org.mockito.Mockito;
 import org.onap.ccsdk.features.sdnr.wt.websocketmanager.WebSocketManagerProvider;
 import org.opendaylight.mdsal.binding.api.RpcProviderService;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.devicemanager.rev190109.ObjectCreationNotification;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.devicemanager.rev190109.ProblemNotification;
+import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.devicemanager.rev190109.ProblemNotificationBuilder;
+import org.opendaylight.yangtools.yang.binding.Notification;
 import org.osgi.service.http.HttpService;
 
 public class WebsockerProviderTest extends Mockito {
@@ -42,4 +48,14 @@ public class WebsockerProviderTest extends Mockito {
 
     }
 
+    @Test
+    public void testTypeAssertion() {
+
+        Notification problemNotification = new ProblemNotificationBuilder().build();
+        assertTrue(WebSocketManagerProvider.assertNotificationType(problemNotification, ProblemNotification.QNAME));
+        assertFalse(
+                WebSocketManagerProvider.assertNotificationType(problemNotification, ObjectCreationNotification.QNAME));
+
+    }
+
 }