Merge pull request #338 from Red5/bug/R5SI-786

Bug/r5 si 786

Author: mondain

client/src/test/java/org/red5/client/net/rtmp/codec/RTMPClientProtocolDecoderTest.java

@@ -2,6 +2,8 @@
 
 import static org.junit.Assert.*;
 
+import java.util.List;
+
 import org.apache.mina.core.buffer.IoBuffer;
 import org.junit.After;
 import org.junit.Before;
@@ -89,4 +91,16 @@ public void testDecodeBufferExTS() {
         assertTrue(packet.getMessage() instanceof AudioData);
     }
 
+    @Test
+    public void testExtendedTImestampPartialPacket() {
+        //Buffer contains 2 complete objects and 1 incomplete object.
+        byte[] buf = IOUtils.hexStringToByteArray("03ffffff00004b090100000005584fce270100002800000042419e1e45152c236f0000030000030000030000030000030000030000030000030000030000030000030000030000030000030000030000030000030000030000049c03ffffff000008080100000005584fd1af01211004608c1c03ffffff000049090100000005");
+        RTMPConnection conn = RTMPConnManager.getInstance().createConnection(RTMPMinaConnection.class);
+        conn.setStateCode(RTMP.STATE_CONNECTED);
+        RTMPClientProtocolDecoder decoder = new RTMPClientProtocolDecoder();
+        List<Object> objects = decoder.decodeBuffer(conn, IoBuffer.wrap(buf));
+        //RTMPDecodeState state = conn.getDecoderState();
+        assertTrue(objects.size() == 2);
+
+    }
 }

common/src/main/java/org/red5/server/net/rtmp/codec/RTMPProtocolDecoder.java

@@ -364,24 +364,14 @@ public Header decodeHeader(ChunkHeader chh, RTMPDecodeState state, IoBuffer in,
         // represents "packet" header length via "format" only 1 byte in the chunk header is needed here
         int headerLength = RTMPUtils.getHeaderLength(headerSize);
         headerLength += chh.getSize() - 1;
-        if (in.remaining() < headerLength || in.remaining() < 3) {
+        //If remaining bytes is less than known headerLength return null and set decoder state.
+        //This length does not include 4-byte extended timestamp field if present.
+        if (in.remaining() < headerLength) {
             state.bufferDecoding(headerLength - in.remaining());
             in.position(startPostion);
             return null;
-        } else {
-            int currentPostition = in.position();
-            // medium int is 3 bytes
-            int timeBase = RTMPUtils.readUnsignedMediumInt(in);
-            in.position(currentPostition);
-            if (timeBase >= MEDIUM_INT_MAX) {
-                headerLength += 4;
-                if (in.remaining() < headerLength) {
-                    state.bufferDecoding(headerLength - in.remaining());
-                    in.position(startPostion);
-                    return null;
-                }
-            }
         }
+
         Header lastHeader = rtmp.getLastReadHeader(channelId);
         if (log.isTraceEnabled()) {
             log.trace("{} lastHeader: {}", Header.HeaderType.values()[headerSize], lastHeader);
@@ -403,11 +393,7 @@ public Header decodeHeader(ChunkHeader chh, RTMPDecodeState state, IoBuffer in,
         //        if (log.isTraceEnabled()) {
         //            log.trace("headerLength: {}", headerLength);
         //        }
-        if (in.remaining() < headerLength) {
-            //log.trace("Header too small (hlen: {}), buffering. remaining: {}", headerLength, remaining);
-            state.bufferDecoding(headerLength);
-            return null;
-        }
+
         int timeBase = 0, timeDelta = 0;
         Header header = new Header();
         header.setChannelId(channelId);
@@ -420,6 +406,12 @@ public Header decodeHeader(ChunkHeader chh, RTMPDecodeState state, IoBuffer in,
                 header.setStreamId(RTMPUtils.readReverseInt(in));
                 // read the extended timestamp if we have the indication that it exists
                 if (timeBase >= MEDIUM_INT_MAX) {
+                    headerLength += 4;
+                    if (in.remaining() < 4) {
+                        state.bufferDecoding(headerLength - in.remaining());
+                        in.position(startPostion);
+                        return null;
+                    }
                     long ext = in.getUnsignedInt();
                     timeBase = (int) (ext ^ (ext >>> 32));
                     if (log.isTraceEnabled()) {
@@ -440,6 +432,12 @@ public Header decodeHeader(ChunkHeader chh, RTMPDecodeState state, IoBuffer in,
                 header.setStreamId(lastHeader.getStreamId());
                 // read the extended timestamp if we have the indication that it exists
                 if (timeDelta >= MEDIUM_INT_MAX) {
+                    headerLength += 4;
+                    if (in.remaining() < 4) {
+                        state.bufferDecoding(headerLength - in.remaining());
+                        in.position(startPostion);
+                        return null;
+                    }
                     long ext = in.getUnsignedInt();
                     timeDelta = (int) (ext ^ (ext >>> 32));
                     header.setExtended(true);
@@ -457,6 +455,12 @@ public Header decodeHeader(ChunkHeader chh, RTMPDecodeState state, IoBuffer in,
                 header.setStreamId(lastHeader.getStreamId());
                 // read the extended timestamp if we have the indication that it exists
                 if (timeDelta >= MEDIUM_INT_MAX) {
+                    headerLength += 4;
+                    if (in.remaining() < 4) {
+                        state.bufferDecoding(headerLength - in.remaining());
+                        in.position(startPostion);
+                        return null;
+                    }
                     long ext = in.getUnsignedInt();
                     timeDelta = (int) (ext ^ (ext >>> 32));
                     header.setExtended(true);
@@ -475,6 +479,12 @@ public Header decodeHeader(ChunkHeader chh, RTMPDecodeState state, IoBuffer in,
                 // This field is present in Type 3 chunks when the most recent Type 0, 1, or 2 chunk for the same chunk stream ID
                 // indicated the presence of an extended timestamp field
                 if (lastHeader.isExtended()) {
+                    headerLength += 4;
+                    if (in.remaining() < 4) {
+                        state.bufferDecoding(headerLength - in.remaining());
+                        in.position(startPostion);
+                        return null;
+                    }
                     long ext = in.getUnsignedInt();
                     int timeExt = (int) (ext ^ (ext >>> 32));
                     if (log.isTraceEnabled()) {