Fix for recording avc/aac with queue

Red5 · Jul 27, 2017 · 2f581bc · 2f581bc
1 parent bde6e7d
commit 2f581bc
Show file tree

Hide file tree

Showing 2 changed files with 42 additions and 20 deletions.
diff --git a/pom.xml b/pom.xml
@@ -3,7 +3,7 @@
     <parent>
         <groupId>org.red5</groupId>
         <artifactId>red5-parent</artifactId>
-        <version>1.0.9-RELEASE</version>
+        <version>1.0.10-M1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>red5-server</artifactId>

diff --git a/src/main/java/org/red5/server/net/rtmp/RTMPMinaIoHandler.java b/src/main/java/org/red5/server/net/rtmp/RTMPMinaIoHandler.java
@@ -54,22 +54,41 @@ public class RTMPMinaIoHandler extends IoHandlerAdapter {
     @Override
     public void sessionCreated(IoSession session) throws Exception {
         log.debug("Session created RTMP");
-        // add rtmpe filter, rtmp protocol filter is added upon successful handshake
-        session.getFilterChain().addFirst("rtmpeFilter", new RTMPEIoFilter());
-        // create a connection
-        RTMPMinaConnection conn = createRTMPMinaConnection();
-        // add session to the connection
-        conn.setIoSession(session);
-        // add the handler
-        conn.setHandler(handler);
-        // add the connections session id for look up using the connection manager
-        session.setAttribute(RTMPConnection.RTMP_SESSION_ID, conn.getSessionId());
-        // create an inbound handshake
-        InboundHandshake handshake = new InboundHandshake();
-        // set whether or not unverified will be allowed
-        handshake.setUnvalidatedConnectionAllowed(((RTMPHandler) handler).isUnvalidatedConnectionAllowed()); 
-        // add the in-bound handshake, defaults to non-encrypted mode
-        session.setAttribute(RTMPConnection.RTMP_HANDSHAKE, handshake);
+        // check for the rtmpe filter and add if not there
+        if (!session.getFilterChain().contains("rtmpeFilter")) {
+            // add rtmpe filter, rtmp protocol filter is added upon successful handshake
+            session.getFilterChain().addFirst("rtmpeFilter", new RTMPEIoFilter());
+        }
+        // connection instance
+        RTMPMinaConnection conn = null;
+        String sessionId = null;
+        // check to ensure the connection instance doesnt already exist
+        if (!session.containsAttribute(RTMPConnection.RTMP_SESSION_ID)) {
+            // create a connection
+            conn = createRTMPMinaConnection();
+            // add session to the connection
+            conn.setIoSession(session);
+            // add the handler
+            conn.setHandler(handler);
+            // get the session id
+            sessionId = conn.getSessionId();
+            // add the connections session id for look up using the connection manager
+            session.setAttribute(RTMPConnection.RTMP_SESSION_ID, sessionId);
+            // create an inbound handshake
+            InboundHandshake handshake = new InboundHandshake();
+            // set whether or not unverified will be allowed
+            handshake.setUnvalidatedConnectionAllowed(((RTMPHandler) handler).isUnvalidatedConnectionAllowed());
+            // add the in-bound handshake, defaults to non-encrypted mode
+            session.setAttribute(RTMPConnection.RTMP_HANDSHAKE, handshake);
+            log.debug("Created: {}", sessionId);
+        } else {
+            sessionId = (String) session.getAttribute(RTMPConnection.RTMP_SESSION_ID);
+            log.warn("Session previously created: {} id: {}", session.getId(), sessionId);
+            RTMPConnManager connManager = (RTMPConnManager) RTMPConnManager.getInstance();
+            if ((RTMPMinaConnection) connManager.getConnectionBySessionId(sessionId) == null) {
+                log.warn("Connection lookup failed for {}", sessionId);
+            }
+        }
     }
 
     /** {@inheritDoc} */
@@ -220,7 +239,8 @@ private void forceClose(final IoSession session) {
      * Close and clean-up the IoSession.
      * 
      * @param session
-     * @param immediately close without waiting for the write queue to flush
+     * @param immediately
+     *            close without waiting for the write queue to flush
      */
     @SuppressWarnings("deprecation")
     private void cleanSession(final IoSession session, boolean immediately) {
@@ -276,14 +296,16 @@ public void operationComplete(CloseFuture future) {
     /**
      * Setter for handler.
      *
-     * @param handler RTMP events handler
+     * @param handler
+     *            RTMP events handler
      */
     public void setHandler(IRTMPHandler handler) {
         this.handler = handler;
     }
 
     /**
-     * @param codecFactory the codecFactory to set
+     * @param codecFactory
+     *            the codecFactory to set
      */
     @Deprecated
     public void setCodecFactory(ProtocolCodecFactory codecFactory) {