Fix for WsSession leak; Add extra Tomcat timeout properties

Author: mondain

client/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <groupId>org.red5</groupId>
         <artifactId>red5-parent</artifactId>
-        <version>2.0.15</version>
+        <version>2.0.16</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>red5-client</artifactId>

client/src/main/java/org/red5/client/Red5Client.java

@@ -18,7 +18,7 @@ public final class Red5Client {
     /**
      * Current server version with revision
      */
-    public static final String VERSION = "Red5 Client 2.0.15";
+    public static final String VERSION = "Red5 Client 2.0.16";
 
     /**
      * Create a new Red5Client object using the connection local to the current thread A bit of magic that lets you access the red5 scope

common/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <groupId>org.red5</groupId>
         <artifactId>red5-parent</artifactId>
-        <version>2.0.15</version>
+        <version>2.0.16</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>red5-server-common</artifactId>
@@ -105,7 +105,7 @@
         <dependency>
             <groupId>net.engio</groupId>
             <artifactId>mbassador</artifactId>
-            <version>2.0.15</version>
+            <version>2.0.16</version>
         </dependency> -->
         <dependency>
             <groupId>junit</groupId>

common/src/main/java/org/red5/server/api/Red5.java

@@ -57,12 +57,12 @@ public final class Red5 {
     /**
      * Server version with revision
      */
-    public static final String VERSION = "Red5 Server 2.0.15";
+    public static final String VERSION = "Red5 Server 2.0.16";
 
     /**
      * Server version for fmsVer requests
      */
-    public static final String FMS_VERSION = "RED5/2,0,15,0";
+    public static final String FMS_VERSION = "RED5/2,0,16,0";
 
     /**
      * Server capabilities

io/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <groupId>org.red5</groupId>
         <artifactId>red5-parent</artifactId>
-        <version>2.0.15</version>
+        <version>2.0.16</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>red5-io</artifactId>

pom.xml

@@ -24,7 +24,7 @@
     <name>Red5</name>
     <description>The Red5 server</description>
     <groupId>org.red5</groupId>
-    <version>2.0.15</version>
+    <version>2.0.16</version>
     <url>https://github.com/Red5/red5-server</url>
     <inceptionYear>2005</inceptionYear>
     <organization>
@@ -99,7 +99,7 @@
         <red5-io.version>${project.version}</red5-io.version>
         <red5-server-common.version>${project.version}</red5-server-common.version>
         <red5-service.version>${project.version}</red5-service.version>
-        <slf4j.version>2.0.15</slf4j.version>
+        <slf4j.version>2.0.16</slf4j.version>
         <logback.version>1.5.6</logback.version>
         <bc.version>1.79</bc.version>
         <mina.version>2.0.23</mina.version>

server/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <groupId>org.red5</groupId>
         <artifactId>red5-parent</artifactId>
-        <version>2.0.15</version>
+        <version>2.0.16</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>red5-server</artifactId>

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

@@ -9,7 +9,6 @@
 
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
-import java.lang.ref.WeakReference;
 import java.nio.ByteBuffer;
 import java.util.Arrays;
 import java.util.Collections;
@@ -28,6 +27,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.tomcat.websocket.Constants;
 import org.apache.tomcat.websocket.WsSession;
+import org.red5.net.websocket.model.WSMessage;
 import org.red5.server.AttributeStore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -50,9 +50,7 @@ public class WebSocketConnection extends AttributeStore implements Comparable<We
 
     private static final Logger log = LoggerFactory.getLogger(WebSocketConnection.class);
 
-    private static final boolean isTrace = log.isTraceEnabled();
-
-    private static final boolean isDebug = log.isDebugEnabled();
+    private static final boolean isTrace = log.isTraceEnabled(), isDebug = log.isDebugEnabled();
 
     // Sending async on windows times out
     private static boolean useAsync;
@@ -69,7 +67,7 @@ public class WebSocketConnection extends AttributeStore implements Comparable<We
     private final WsSession wsSession;
 
     // reference to the scope for manager access
-    private WeakReference<WebSocketScope> scope;
+    private final WebSocketScope scope;
 
     // unique identifier for the session
     private final String wsSessionId;
@@ -111,7 +109,9 @@ public class WebSocketConnection extends AttributeStore implements Comparable<We
     public WebSocketConnection(WebSocketScope scope, Session session) {
         log.debug("New WebSocket - scope: {} session: {}", scope, session);
         // set the scope for ease of use later
-        this.scope = new WeakReference<>(scope);
+        this.scope = scope;
+        // add the session to the user props
+        session.getUserProperties().put(WSConstants.WS_SCOPE, scope);
         // set our path
         path = scope.getPath();
         if (isDebug) {
@@ -167,15 +167,20 @@ public WebSocketConnection(WebSocketScope scope, Session session) {
         // add the timeouts to the user props
         userProps.put(Constants.READ_IDLE_TIMEOUT_MS, readTimeout);
         userProps.put(Constants.WRITE_IDLE_TIMEOUT_MS, sendTimeout);
-        // set the close timeout to 5 seconds
-        userProps.put(Constants.SESSION_CLOSE_TIMEOUT_PROPERTY, TimeUnit.SECONDS.toMillis(5));
+        // write timeout used when sending WebSocket messages in blocking mode
+        userProps.put(Constants.BLOCKING_SEND_TIMEOUT_PROPERTY, Long.getLong(Constants.BLOCKING_SEND_TIMEOUT_PROPERTY, 8000L).longValue());
+        // write timeout Tomcat uses when writing a session close message when the close is abnormal
+        userProps.put(Constants.ABNORMAL_SESSION_CLOSE_SEND_TIMEOUT_PROPERTY, Long.getLong(Constants.ABNORMAL_SESSION_CLOSE_SEND_TIMEOUT_PROPERTY, 10000L).longValue());
+        // time Tomcat waits for a peer to send a WebSocket session close message after Tomcat has sent a close message
+        // to the peer
+        userProps.put(Constants.SESSION_CLOSE_TIMEOUT_PROPERTY, Long.getLong(Constants.SESSION_CLOSE_TIMEOUT_PROPERTY, 5000L).longValue());
         if (isDebug) {
             log.debug("userProps: {}", userProps);
         }
         // set maximum messages size to 10,000 bytes
         session.setMaxTextMessageBufferSize(10000);
-        // set maximum idle timeout to 30 seconds (read timeout)
-        session.setMaxIdleTimeout(readTimeout);
+        // set maximum idle timeout to the largest of the read and send timeouts
+        session.setMaxIdleTimeout(Math.max(readTimeout, sendTimeout));
     }
 
     /**
@@ -320,6 +325,16 @@ public void sendPong(byte[] buf) throws IllegalArgumentException, IOException {
         }
     }
 
+    /**
+     * Send a received message to the scope.
+     *
+     * @param wsMessage
+     *            WSMessage
+     */
+    public void onReceive(WSMessage wsMessage) {
+        scope.onMessage(wsMessage);
+    }
+
     /**
      * Close the connection.
      */
@@ -386,7 +401,7 @@ public static void setUseAsync(boolean useAsync) {
      * @return WebSocketScope
      */
     public WebSocketScope getScope() {
-        return scope != null ? scope.get() : null;
+        return scope;
     }
 
     /**

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

@@ -11,6 +11,7 @@
 import java.io.IOException;
 import java.io.UnsupportedEncodingException;
 import java.nio.ByteBuffer;
+import java.util.Map;
 
 import org.apache.mina.core.buffer.IoBuffer;
 import org.red5.net.websocket.WSConstants;
@@ -36,12 +37,8 @@ public class DefaultWebSocketEndpoint extends Endpoint {
 
     private final Logger log = LoggerFactory.getLogger(DefaultWebSocketEndpoint.class);
 
-    @SuppressWarnings("unused")
     private final boolean isDebug = log.isDebugEnabled(), isTrace = log.isTraceEnabled();
 
-    // websocket scope where connections connect
-    private WebSocketScope scope;
-
     /**
      * TODO: Currently, Tomcat uses an Endpoint instance once - however the java doc of endpoint says: "Each instance
      * of a websocket endpoint is guaranteed not to be called by more than one thread at a time per active connection."
@@ -56,10 +53,15 @@ public void onOpen(Session session, EndpointConfig config) {
         if (isDebug) {
             log.debug("Session opened: {}\n{}", session.getId(), session.getRequestParameterMap());
         }
+        // user props from session
+        Map<String, Object> userProps = session.getUserProperties();
         // get ws scope from user props
-        scope = (WebSocketScope) config.getUserProperties().get(WSConstants.WS_SCOPE);
+        WebSocketScope scope = (WebSocketScope) userProps.get(WSConstants.WS_SCOPE);
+        if (isDebug) {
+            log.debug("onOpen - session: {} props: {} scope: {}", session.getId(), userProps, scope);
+        }
         // get ws connection from session user props
-        WebSocketConnection conn = (WebSocketConnection) session.getUserProperties().get(WSConstants.WS_CONNECTION);
+        WebSocketConnection conn = (WebSocketConnection) userProps.get(WSConstants.WS_CONNECTION);
         if (conn == null) {
             log.warn("WebSocketConnection null at onOpen for {}", session.getId());
         }
@@ -72,11 +74,18 @@ public void onOpen(Session session, EndpointConfig config) {
     public void onClose(Session session, CloseReason closeReason) {
         final String sessionId = session.getId();
         log.debug("Session closed: {}", sessionId);
+        // user props from session
+        Map<String, Object> userProps = session.getUserProperties();
+        // get ws scope from user props
+        WebSocketScope scope = (WebSocketScope) userProps.get(WSConstants.WS_SCOPE);
+        if (isDebug) {
+            log.debug("onClose - session: {} props: {} scope: {}", session.getId(), userProps, scope);
+        }
         WebSocketConnection conn = null;
         // getting the sessions user properties on a closed connection will throw an exception when it checks state
         try {
             // ensure we grab the scope from the session if its null
-            conn = (WebSocketConnection) session.getUserProperties().get(WSConstants.WS_CONNECTION);
+            conn = (WebSocketConnection) userProps.get(WSConstants.WS_CONNECTION);
             // if we don't get it from the session, try the scope lookup
             if (conn == null) {
                 log.warn("Connection for id: {} was not found in the session onClose", sessionId);
@@ -137,8 +146,7 @@ public void onMessage(String message) {
                 try {
                     // create a websocket message and add the current connection for listener access
                     WSMessage wsMessage = new WSMessage(message, conn);
-                    // fire the message off to the scope for handling
-                    scope.onMessage(wsMessage);
+                    conn.onReceive(wsMessage);
                 } catch (UnsupportedEncodingException e) {
                     log.warn("Exception on message", e);
                 }
@@ -167,8 +175,7 @@ public void onMessage(ByteBuffer message) {
                 conn.updateReadBytes(message.limit());
                 // create a websocket message and add the current connection for listener access
                 WSMessage wsMessage = new WSMessage(IoBuffer.wrap(message), conn);
-                // fire the message off to the scope for handling
-                scope.onMessage(wsMessage);
+                conn.onReceive(wsMessage);
             } else {
                 log.debug("Connection null or not connected", conn);
             }

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

@@ -39,10 +39,7 @@ public class WsHttpUpgradeHandler implements InternalHttpUpgradeHandler {
 
     private Logger log = LoggerFactory.getLogger(WsHttpUpgradeHandler.class); // must not be static
 
-    private final boolean isTrace = log.isTraceEnabled();
-
-    @SuppressWarnings("unused")
-    private final boolean isDebug = log.isDebugEnabled();
+    private final boolean isTrace = log.isTraceEnabled(), isDebug = log.isDebugEnabled();
 
     private static final StringManager sm = StringManager.getManager(WsHttpUpgradeHandler.class);
 
@@ -83,6 +80,10 @@ public class WsHttpUpgradeHandler implements InternalHttpUpgradeHandler {
 
     private long lastReadBytes, lastWrittenBytes;
 
+    private transient WebSocketScopeManager manager;
+
+    private transient WebSocketScope scope;
+
     public WsHttpUpgradeHandler() {
         applicationClassLoader = Thread.currentThread().getContextClassLoader();
     }
@@ -110,6 +111,12 @@ public void preInit(Endpoint ep, EndpointConfig endpointConfig, DefaultWsServerC
         } else {
             log.debug("pre-init without http session");
         }
+        // user props
+        Map<String, Object> userProps = endpointConfig.getUserProperties();
+        // get the ws scope manager from user props
+        manager = (WebSocketScopeManager) userProps.get(WSConstants.WS_MANAGER);
+        // get ws scope from user props
+        scope = (WebSocketScope) userProps.get(WSConstants.WS_SCOPE);
     }
 
     @Override
@@ -141,10 +148,6 @@ public void init(WebConnection connection) {
                 wsFrame = new WsFrameServer(socketWrapper, upgradeInfo, wsSession, transformation, applicationClassLoader);
                 // WsFrame adds the necessary final transformations. Copy the completed transformation chain to the remote end point.
                 wsRemoteEndpointServer.setTransformation(wsFrame.getTransformation());
-                // get the ws scope manager from user props
-                WebSocketScopeManager manager = (WebSocketScopeManager) endpointConfig.getUserProperties().get(WSConstants.WS_MANAGER);
-                // get ws scope from user props
-                WebSocketScope scope = (WebSocketScope) endpointConfig.getUserProperties().get(WSConstants.WS_SCOPE);
                 // create a ws connection instance
                 WebSocketConnection conn = new WebSocketConnection(scope, wsSession);
                 // set ip and port
@@ -243,7 +246,7 @@ public void destroy() {
     }
 
     private void onError(Throwable throwable) {
-        if (log.isDebugEnabled()) {
+        if (isDebug) {
             log.debug("onError for ws id: {}", wsSession.getId(), throwable);
         }
         // Need to call onError using the web application's class loader
@@ -258,7 +261,7 @@ private void onError(Throwable throwable) {
     }
 
     private void close(CloseReason cr) {
-        if (log.isDebugEnabled()) {
+        if (isDebug) {
             log.debug("close for ws id: {} reason: {}", wsSession.getId(), cr);
         }
         /*
@@ -268,6 +271,15 @@ private void close(CloseReason cr) {
          * recover from whatever messed up state the client put the connection into.
          */
         wsSession.onClose(cr);
+        // null these so that we don't try to use them again
+        wsSession = null;
+        connection = null;
+        upgradeInfo = null;
+        // null the socket wrapper so that we don't try to use it again
+        if (socketWrapper != null) {
+            socketWrapper.close();
+            socketWrapper = null;
+        }
     }
 
     @Override
@@ -290,8 +302,6 @@ public void timeoutAsync(long now) {
         if (wsSession != null) {
             try {
                 final String wsSessionId = wsSession.getId();
-                // get scope from endpoint config
-                WebSocketScope scope = (WebSocketScope) endpointConfig.getUserProperties().get(WSConstants.WS_SCOPE);
                 // do lookup by session id, skips need for session user props
                 WebSocketConnection conn = scope.getConnectionBySessionId(wsSessionId);
                 // if we don't get it from the scope, try the session lookup