Implemented internal ws conn pinger on a single thread

Author: mondain

server/src/main/java/org/red5/net/websocket/WebSocketScope.java

@@ -12,6 +12,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CopyOnWriteArraySet;
 
 import org.red5.net.websocket.listener.IWebSocketDataListener;
 import org.red5.net.websocket.model.WSMessage;
@@ -36,7 +37,8 @@ public class WebSocketScope implements InitializingBean, DisposableBean {
 
     protected ConcurrentSkipListSet<WebSocketConnection> conns = new ConcurrentSkipListSet<>();
 
-    protected ConcurrentSkipListSet<IWebSocketDataListener> listeners = new ConcurrentSkipListSet<>();
+    // this has very few entries, possibly only one, COWAS is fine here and won't incur Comparable requirements
+    protected CopyOnWriteArraySet<IWebSocketDataListener> listeners = new CopyOnWriteArraySet<>();
 
     protected IScope scope;
 
@@ -180,9 +182,7 @@ public void addConnection(WebSocketConnection conn) {
         // prevent false failed logging when a connection is already registered
         if (conns.add(conn)) {
             log.debug("Added connection: {}", conn);
-            for (IWebSocketDataListener listener : listeners) {
-                listener.onWSConnect(conn);
-            }
+            listeners.forEach(listener -> listener.onWSConnect(conn));
         } else {
             log.debug("Add connection skipped, already registered: {}", conn);
         }
@@ -197,9 +197,7 @@ public void removeConnection(WebSocketConnection conn) {
         // prevent false failed logging when a connection isnt registered
         if (conns.remove(conn)) {
             log.debug("Removed connection: {}", conn);
-            for (IWebSocketDataListener listener : listeners) {
-                listener.onWSDisconnect(conn);
-            }
+            listeners.forEach(listener -> listener.onWSDisconnect(conn));
         } else {
             log.debug("Remove connection skipped, not registered: {}", conn);
         }

server/src/main/java/org/red5/net/websocket/WebSocketScopeManager.java

@@ -11,6 +11,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
 import org.red5.net.websocket.listener.DefaultWebSocketDataListener;
@@ -32,6 +34,15 @@ public class WebSocketScopeManager {
 
     private static final Logger log = LoggerFactory.getLogger(WebSocketScopeManager.class);
 
+    // used to ping WS connections
+    private static final byte[] PING_BYTES = "PING!".getBytes();
+
+    // one executor per scope manager
+    private ExecutorService executor = Executors.newFixedThreadPool(1);
+
+    // future for the ws pinger
+    private Future<?> pingFuture;
+
     // reference to the owning application scope
     private IScope appScope;
 
@@ -47,6 +58,9 @@ public class WebSocketScopeManager {
     // whether or not to copy listeners from parent to child on create
     protected boolean copyListeners = true;
 
+    // value for the websocket ping period/interval
+    public static long websocketPingInterval = 5000L;
+
     public void addListener(IWebSocketScopeListener listener) {
         scopeListeners.add(listener);
     }
@@ -136,6 +150,37 @@ public boolean addWebSocketScope(WebSocketScope webSocketScope) {
         if (scopes.putIfAbsent(path, webSocketScope) == null) {
             log.info("addWebSocketScope: {}", webSocketScope);
             notifyListeners(WebSocketEvent.SCOPE_ADDED, webSocketScope);
+            // ensure the ping future exists, if not spawn it
+            if (pingFuture == null || pingFuture.isDone()) {
+                pingFuture = executor.submit(() -> {
+                    final String oldName = Thread.currentThread().getName();
+                    Thread.currentThread().setName("WebSocketPinger");
+                    do {
+                        scopes.forEach((sName, wsScope) -> {
+                            log.trace("start pinging scope: {}", sName);
+                            wsScope.getConns().forEach(wsConn -> {
+                                // ping connected websocket
+                                if (wsConn.isConnected()) {
+                                    log.debug("pinging ws: {} on scope: {}", wsConn.getWsSessionId(), sName);
+                                    try {
+                                        wsConn.sendPing(PING_BYTES);
+                                    } catch (Exception e) {
+                                    }
+                                }
+                            });
+                            log.trace("finished pinging scope: {}", sName);
+                        });
+                        // sleep for interval
+                        try {
+                            Thread.sleep(websocketPingInterval);
+                        } catch (InterruptedException e1) {
+                        }
+                    } while (!scopes.isEmpty());
+                    // reset ping future
+                    pingFuture = null;
+                    Thread.currentThread().setName(oldName);
+                });
+            }
             return true;
         }
         return false;
@@ -165,7 +210,9 @@ public boolean removeWebSocketScope(WebSocketScope webSocketScope) {
      */
     public void addConnection(WebSocketConnection conn) {
         WebSocketScope scope = getScope(conn);
-        scope.addConnection(conn);
+        if (scope != null) {
+            scope.addConnection(conn);
+        }
     }
 
     /**
@@ -180,7 +227,7 @@ public void removeConnection(WebSocketConnection conn) {
             if (scope != null) {
                 scope.removeConnection(conn);
                 if (!scope.isValid()) {
-                    // scope is not valid. delete this.
+                    // scope is not valid, delete it
                     removeWebSocketScope(scope);
                 }
             }
@@ -346,6 +393,9 @@ private WebSocketScope getScope(WebSocketConnection conn) {
      * Stops this manager and the scopes contained within.
      */
     public void stop() {
+        if (pingFuture != null && !pingFuture.isCancelled()) {
+            pingFuture.cancel(true);
+        }
         for (WebSocketScope scope : scopes.values()) {
             scope.unregister();
         }

server/src/main/java/org/red5/net/websocket/server/DefaultWebSocketEndpoint.java

@@ -52,8 +52,6 @@ public class DefaultWebSocketEndpoint extends Endpoint {
      * onMessage, onClose methods to ensure the room thread always gets the correct instance of the variable holder.
      */
 
-    //private ThreadLocal<WebSocketConnection> connectionLocal = new ThreadLocal<>();
-
     @Override
     public void onOpen(Session session, EndpointConfig config) {
         log.debug("Session opened: {}\n{}", session.getId(), session.getRequestParameterMap());

server/src/main/java/org/red5/net/websocket/server/UpgradeUtil.java

@@ -178,8 +178,6 @@ public static void doUpgrade(DefaultWsServerContainer sc, HttpServletRequest req
                 ep = new PojoEndpointServer(pathParams, clazz);
                 // Need to make path params available to POJO
                 perSessionServerEndpointConfig.getUserProperties().put("org.apache.tomcat.websocket.pojo.PojoEndpoint.pathParams", pathParams);
-                // removed in 8.5.x post .61
-                //perSessionServerEndpointConfig.getUserProperties().put(org.apache.tomcat.websocket.pojo.Constants.POJO_PATH_PARAM_KEY, pathParams);
             }
         } catch (InstantiationException e) {
             throw new ServletException(e);

server/src/main/java/org/red5/net/websocket/server/WsFrameServer.java

@@ -50,17 +50,6 @@ private void onDataAvailable() throws IOException {
         if (log.isDebugEnabled()) {
             log.debug("wsFrameServer.onDataAvailable - session {}", wsSession.getId());
         }
-        /*
-        // set connection local to the message handler so WSMessage will contain the connection
-        DefaultWebSocketEndpoint ep = (DefaultWebSocketEndpoint) wsSession.getLocal();
-        if (ep.getConnectionLocal() == null) {
-            if (log.isDebugEnabled()) {
-                log.debug("Endpoint had no connection local for session: {}", wsSession.getId());
-            }
-            WebSocketConnection conn = (WebSocketConnection) wsSession.getUserProperties().get(WSConstants.WS_CONNECTION);
-            ep.setConnectionLocal(conn);
-        }
-        */
         // handle input
         if (isOpen() && inputBuffer.hasRemaining() && !isSuspended()) {
             // There might be a data that was left in the buffer when the read has been suspended.
@@ -84,10 +73,6 @@ private void onDataAvailable() throws IOException {
             }
             processInputBuffer();
         }
-        /*
-        // clear thread local
-        ((DefaultWebSocketEndpoint) wsSession.getLocal()).setConnectionLocal(null);
-        */
     }
 
     @Override