Updated version to 1.2.23. Fixes for scope lookup and rtmpe playback

Author: mondain

client/pom.xml

@@ -3,7 +3,7 @@
     <parent>
         <groupId>org.red5</groupId>
         <artifactId>red5-parent</artifactId>
-        <version>1.2.22</version>
+        <version>1.2.23</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 1.2.22";
+    public static final String VERSION = "Red5 Client 1.2.23";
 
     /**
      * 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>1.2.22</version>
+        <version>1.2.23</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>red5-server-common</artifactId>

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

@@ -55,12 +55,12 @@ public final class Red5 {
     /**
      * Server version with revision
      */
-    public static final String VERSION = "Red5 Server 1.2.22";
+    public static final String VERSION = "Red5 Server 1.2.23";
 
     /**
      * Server version for fmsVer requests
      */
-    public static final String FMS_VERSION = "RED5/1,2,22,0";
+    public static final String FMS_VERSION = "RED5/1,2,23,0";
 
     /**
      * Server capabilities

common/src/main/java/org/red5/server/net/rtmp/RTMPConnection.java

@@ -49,11 +49,13 @@
 import org.red5.server.exception.ClientRejectedException;
 import org.red5.server.net.protocol.RTMPDecodeState;
 import org.red5.server.net.rtmp.codec.RTMP;
+import org.red5.server.net.rtmp.event.AudioData;
 import org.red5.server.net.rtmp.event.BytesRead;
 import org.red5.server.net.rtmp.event.ChunkSize;
 import org.red5.server.net.rtmp.event.ClientBW;
 import org.red5.server.net.rtmp.event.ClientInvokeEvent;
 import org.red5.server.net.rtmp.event.ClientNotifyEvent;
+import org.red5.server.net.rtmp.event.IRTMPEvent;
 import org.red5.server.net.rtmp.event.Invoke;
 import org.red5.server.net.rtmp.event.Notify;
 import org.red5.server.net.rtmp.event.Ping;
@@ -1366,6 +1368,10 @@ protected void writingMessage(Packet message) {
             }
             old.incrementAndGet();
         }
+        // XXX(paul) work-around for RTMPE issue with Mina messageSent callback
+        if (isEncrypted()) {
+            writtenMessages.incrementAndGet();
+        }
     }
 
     /**
@@ -1602,7 +1608,10 @@ public void onSuccess(Packet packet) {
      *            Message to mark
      */
     public void messageSent(Packet message) {
-        if (message.getMessage() instanceof VideoData) {
+        //log.info("messageSent: {}", message);
+        IRTMPEvent event = message.getMessage();
+        if (event instanceof VideoData) {
+            log.info("Video message sent");
             Number streamId = message.getHeader().getStreamId();
             AtomicInteger pending = pendingVideos.get(streamId.doubleValue());
             if (isTrace) {
@@ -1611,6 +1620,12 @@ public void messageSent(Packet message) {
             if (pending != null) {
                 pending.decrementAndGet();
             }
+        } else if (event instanceof AudioData) {
+            log.info("Audio message sent");
+        } else if (event instanceof Notify) {
+            log.info("Notify message sent");
+        } else {
+            log.warn("Message sent: {} data type: {}", event.getType(), event.getDataType());
         }
         writtenMessages.incrementAndGet();
     }

common/src/main/java/org/red5/server/net/rtmp/RTMPMinaConnection.java

@@ -327,15 +327,14 @@ public void setIoSession(IoSession protocolSession) {
     public void write(Packet out) {
         if (ioSession != null) {
             final Semaphore lock = getLock();
-            if (log.isTraceEnabled()) {
-                log.trace("Write lock wait count: {} closed: {}", lock.getQueueLength(), isClosed());
-            }
+            //if (log.isTraceEnabled()) {
+            //    log.trace("Write lock wait count: {} closed: {}", lock.getQueueLength(), isClosed());
+            //}
             while (!isClosed()) {
                 boolean acquired = false;
                 try {
                     acquired = lock.tryAcquire(10, TimeUnit.MILLISECONDS);
-                    if (acquired) {
-                        // attempt write if not closing
+                    if (acquired) { // attempt write if not closing 
                         if (!ioSession.isClosing()) {
                             if (log.isTraceEnabled()) {
                                 log.trace("Writing message");
@@ -348,12 +347,11 @@ public void write(Packet out) {
                 } catch (InterruptedException e) {
                     log.warn("Interrupted while waiting for write lock. State: {}", RTMP.states[state.getState()], e);
                     if (log.isInfoEnabled()) {
-                        // further debugging to assist with possible connection problems
+                        // further debugging to assist with possible connection problems 
                         log.info("Session id: {} in queue size: {} pending msgs: {} last ping/pong: {}", getSessionId(), currentQueueSize(), getPendingMessages(), getLastPingSentAndLastPongReceivedInterval());
                         log.info("Available permits - decoder: {} encoder: {}", decoderLock.availablePermits(), encoderLock.availablePermits());
                     }
-                    String exMsg = e.getMessage();
-                    // if the exception cause is null break out of here to prevent looping until closed
+                    String exMsg = e.getMessage(); // if the exception cause is null break out of here to prevent looping until closed 
                     if (exMsg == null || exMsg.indexOf("null") >= 0) {
                         log.debug("Exception writing to connection: {}", this);
                         break;

common/src/main/java/org/red5/server/scope/Scope.java

@@ -1446,7 +1446,7 @@ public IBasicScope getBasicScope(ScopeType type, String name) {
                 // if its broadcast type then allow an alias match in addition to the name match
                 if (type == ScopeType.BROADCAST) {
                     // checks publish and subscribe aliases
-                    scope = stream().filter(child -> child.getType().equals(type) && (name.equals(child.getName()) || name.equals(((IBroadcastScope) child).getClientBroadcastStream().getAlias()) || ((IBroadcastScope) child).getClientBroadcastStream().containsAlias(name))).findFirst();
+                    scope = stream().filter(child -> child.getType().equals(type) && (name.equals(child.getName()) || ((IBroadcastScope) child).getClientBroadcastStream().containsAlias(name))).findFirst();
                 } else {
                     scope = stream().filter(child -> child.getType().equals(type) && name.equals(child.getName())).findFirst();
                 }

common/src/main/java/org/red5/server/stream/PlayEngine.java

@@ -83,6 +83,10 @@ public final class PlayEngine implements IFilter, IPushableConsumer, IPipeConnec
 
     private static final Logger log = Red5LoggerFactory.getLogger(PlayEngine.class);
 
+    private static boolean isDebug = log.isDebugEnabled();
+
+    private static boolean isTrace = log.isTraceEnabled();
+
     private final AtomicReference<IMessageInput> msgInReference = new AtomicReference<>();
 
     private final AtomicReference<IMessageOutput> msgOutReference = new AtomicReference<>();
@@ -280,7 +284,7 @@ void setMessageOut(IMessageOutput msgOut) {
      * Start stream
      */
     public void start() {
-        if (log.isDebugEnabled()) {
+        if (isDebug) {
             log.debug("start - subscriber stream state: {}", (subscriberStream != null ? subscriberStream.getState() : null));
         }
         switch (subscriberStream.getState()) {
@@ -290,7 +294,7 @@ public void start() {
                 IMessageOutput out = consumerService.getConsumerOutput(subscriberStream);
                 if (msgOutReference.compareAndSet(null, out)) {
                     out.subscribe(this, null);
-                } else if (log.isDebugEnabled()) {
+                } else if (isDebug) {
                     log.debug("Message output was already set for stream: {}", subscriberStream);
                 }
                 break;
@@ -389,7 +393,7 @@ public void play(IPlayItem item, boolean withReset) throws StreamNotFoundExcepti
         IMessage msg = null;
         currentItem.set(item);
         long itemLength = item.getLength();
-        if (log.isDebugEnabled()) {
+        if (isDebug) {
             log.debug("Play decision is {} (0=Live, 1=File, 2=Wait, 3=N/A) item length: {}", playDecision, itemLength);
         }
         switch (playDecision) {
@@ -431,7 +435,7 @@ public void play(IPlayItem item, boolean withReset) throws StreamNotFoundExcepti
                 in = providerService.getLiveProviderInput(thisScope, itemName, true);
                 if (msgInReference.compareAndSet(null, in)) {
                     if (type == -1 && itemLength >= 0) {
-                        if (log.isDebugEnabled()) {
+                        if (isDebug) {
                             log.debug("Creating wait job for {}", itemLength);
                         }
                         // Wait given timeout for stream to be published
@@ -443,7 +447,7 @@ public void execute(ISchedulingService service) {
                             }
                         });
                     } else if (type == -2) {
-                        if (log.isDebugEnabled()) {
+                        if (isDebug) {
                             log.debug("Creating wait job");
                         }
                         // Wait x seconds for the stream to be published
@@ -456,7 +460,7 @@ public void execute(ISchedulingService service) {
                     } else {
                         connectToProvider(itemName);
                     }
-                } else if (log.isDebugEnabled()) {
+                } else if (isDebug) {
                     log.debug("Message input already set for {}", itemName);
                 }
                 break;
@@ -756,7 +760,7 @@ public void seek(int position) throws IllegalStateException, OperationNotSupport
      *             If stream is in stopped state
      */
     public void stop() throws IllegalStateException {
-        if (log.isDebugEnabled()) {
+        if (isDebug) {
             log.debug("stop - subscriber stream state: {}", (subscriberStream != null ? subscriberStream.getState() : null));
         }
         // allow stop if playing or paused
@@ -805,7 +809,7 @@ public void stop() throws IllegalStateException {
      * Close stream
      */
     public void close() {
-        if (log.isDebugEnabled()) {
+        if (isDebug) {
             log.debug("close");
         }
         if (!subscriberStream.getState().equals(StreamState.CLOSED)) {
@@ -824,7 +828,7 @@ public void close() {
             if (out != null) {
                 List<IConsumer> consumers = out.getConsumers();
                 // assume a list of 1 in most cases
-                if (log.isDebugEnabled()) {
+                if (isDebug) {
                     log.debug("Message out consumers: {}", consumers.size());
                 }
                 if (!consumers.isEmpty()) {
@@ -965,7 +969,7 @@ private void doPushMessage(Status status) {
      *            The message to send.
      */
     private void doPushMessage(AbstractMessage message) {
-        if (log.isTraceEnabled()) {
+        if (isTrace) {
             String msgType = message.getMessageType();
             log.trace("doPushMessage: {}", msgType);
         }
@@ -1019,7 +1023,7 @@ private void sendMessage(RTMPMessage messageIn) {
         event.setSourceType(eventIn.getSourceType());
         // instance the outgoing message
         RTMPMessage messageOut = RTMPMessage.build(event, eventTime);
-        if (log.isTraceEnabled()) {
+        if (isTrace) {
             log.trace("Source type - in: {} out: {}", eventIn.getSourceType(), messageOut.getBody().getSourceType());
             long delta = System.currentTimeMillis() - playbackStart;
             log.trace("sendMessage: streamStartTS {}, length {}, streamOffset {}, timestamp {} last timestamp {} delta {} buffered {}", new Object[] { streamStartTS.get(), currentItem.get().getLength(), streamOffset, eventTime, lastMessageTs, delta, lastMessageTs - delta });
@@ -1032,7 +1036,7 @@ private void sendMessage(RTMPMessage messageIn) {
             long length = currentItem.get().getLength();
             if (length >= 0) {
                 int duration = eventTime - streamStartTS.get();
-                if (log.isTraceEnabled()) {
+                if (isTrace) {
                     log.trace("sendMessage duration={} length={}", duration, length);
                 }
                 if (duration - streamOffset >= length) {
@@ -1052,7 +1056,7 @@ private void sendMessage(RTMPMessage messageIn) {
                 // subtract the offset time of when the stream started playing for the client
                 eventTime -= startTs;
                 messageOut.getBody().setTimestamp(eventTime);
-                if (log.isTraceEnabled()) {
+                if (isTrace) {
                     log.trace("sendMessage (updated): streamStartTS={}, length={}, streamOffset={}, timestamp={}", new Object[] { startTs, currentItem.get().getLength(), streamOffset, eventTime });
                 }
             }
@@ -1150,7 +1154,7 @@ private void sendStopStatus(IPlayItem item) {
      * @param bytes
      */
     private void sendOnPlayStatus(String code, int duration, long bytes) {
-        if (log.isDebugEnabled()) {
+        if (isDebug) {
             log.debug("Sending onPlayStatus - code: {} duration: {} bytes: {}", code, duration, bytes);
         }
         // create the buffer
@@ -1204,7 +1208,7 @@ private void sendTransitionStatus() {
     private void sendCompleteStatus() {
         // may be the correct duration
         int duration = (lastMessageTs > 0) ? Math.max(0, lastMessageTs - streamStartTS.get()) : 0;
-        if (log.isDebugEnabled()) {
+        if (isDebug) {
             log.debug("sendCompleteStatus - duration: {} bytes sent: {}", duration, bytesSent.get());
         }
         sendOnPlayStatus(StatusCodes.NS_PLAY_COMPLETE, duration, bytesSent.get());
@@ -1417,7 +1421,7 @@ public void onPipeConnectionEvent(PipeConnectionEvent event) {
                 }
                 break;
             default:
-                if (log.isDebugEnabled()) {
+                if (isDebug) {
                     log.debug("Unhandled pipe event: {}", event);
                 }
         }
@@ -1447,10 +1451,11 @@ public void pushMessage(IPipe pipe, IMessage message) throws IOException {
             RTMPMessage rtmpMessage = (RTMPMessage) message;
             IRTMPEvent body = rtmpMessage.getBody();
             if (body instanceof IStreamData) {
+                final String subscribedStreamName = subscriberStream.getBroadcastStreamPublishName();
                 // the subscriber paused 
                 if (subscriberStream.getState() == StreamState.PAUSED) {
                     if (log.isInfoEnabled() && shouldLogPacketDrop()) {
-                        log.info("Dropping packet because we are paused. sessionId={} stream={} count={}", sessionId, subscriberStream.getBroadcastStreamPublishName(), droppedPacketsCount);
+                        log.info("Dropping packet because we are paused. sessionId={} stream={} count={}", sessionId, subscribedStreamName, droppedPacketsCount);
                     }
                     videoFrameDropper.dropPacket(rtmpMessage);
                     return;
@@ -1469,7 +1474,7 @@ public void pushMessage(IPipe pipe, IMessage message) throws IOException {
                                     droppedPacketsCount++;
                                     if (log.isInfoEnabled() && shouldLogPacketDrop()) {
                                         // client disabled video or the app doesn't have enough bandwidth allowed for this stream
-                                        log.info("Drop packet. Failed to acquire token or no video. sessionId={} stream={} count={}", sessionId, subscriberStream.getBroadcastStreamPublishName(), droppedPacketsCount);
+                                        log.info("Drop packet. Failed to acquire token or no video. sessionId={} stream={} count={}", sessionId, subscribedStreamName, droppedPacketsCount);
                                     }
                                     return;
                                 }
@@ -1480,14 +1485,14 @@ public void pushMessage(IPipe pipe, IMessage message) throws IOException {
                                 // pending video messages and drop video packets until the queue is below the threshold.
                                 // only check for frame dropping if the codec supports it
                                 long pendingVideos = pendingVideoMessages();
-                                if (log.isTraceEnabled()) {
-                                    log.trace("Pending messages sessionId={} pending={} threshold={} sequential={} stream={}, count={}", new Object[] { sessionId, pendingVideos, maxPendingVideoFrames, numSequentialPendingVideoFrames, subscriberStream.getBroadcastStreamPublishName(), droppedPacketsCount });
+                                if (isTrace) {
+                                    log.trace("Pending messages sessionId={} stream={} pending={} threshold={} sequential={} dropped={}", new Object[] { sessionId, subscribedStreamName, pendingVideos, maxPendingVideoFrames, numSequentialPendingVideoFrames, droppedPacketsCount });
                                 }
                                 if (!videoFrameDropper.canSendPacket(rtmpMessage, pendingVideos)) {
                                     // drop frame as it depends on other frames that were dropped before
                                     droppedPacketsCount++;
                                     if (log.isInfoEnabled() && shouldLogPacketDrop()) {
-                                        log.info("Frame dropper says to drop packet. sessionId={} stream={} count={}", sessionId, subscriberStream.getBroadcastStreamPublishName(), droppedPacketsCount);
+                                        log.info("Frame dropper says to drop packet. sessionId={} stream={} dropped={}", sessionId, subscribedStreamName, droppedPacketsCount);
                                     }
                                     return;
                                 }
@@ -1501,7 +1506,7 @@ public void pushMessage(IPipe pipe, IMessage message) throws IOException {
                                 if (pendingVideos > maxPendingVideoFrames || numSequentialPendingVideoFrames > maxSequentialPendingVideoFrames) {
                                     droppedPacketsCount++;
                                     if (log.isInfoEnabled() && shouldLogPacketDrop()) {
-                                        log.info("Drop packet. Pending above threshold. sessionId={} pending={} threshold={} sequential={} stream={} count={}", new Object[] { sessionId, pendingVideos, maxPendingVideoFrames, numSequentialPendingVideoFrames, subscriberStream.getBroadcastStreamPublishName(), droppedPacketsCount });
+                                        log.info("Drop packet. Pending above threshold. sessionId={} stream={} pending={} threshold={} sequential={} dropped={}", new Object[] { sessionId, subscribedStreamName, pendingVideos, maxPendingVideoFrames, numSequentialPendingVideoFrames, droppedPacketsCount });
                                     }
                                     // drop because the client has insufficient bandwidth
                                     long now = System.currentTimeMillis();
@@ -1570,7 +1575,7 @@ private long pendingVideoMessages() {
                 return (Long) pendingRequest.getResult();
             }
         }
-        return 0;
+        return 0L;
     }
 
     /**