Fixes and test file added.

Author: vbence

.idea/modules/stream-m_test.iml

@@ -28,8 +28,6 @@ class RTMPStreamProcessor implements Processor {
 
     protected boolean finished = false;
 
-    protected Map<Integer, MessageInfo> lastMessages = new HashMap<Integer, MessageInfo>();
-    
     private final int DEFAULT_CHUNK_SIZE = 128;
 
     protected ResourceLimit limit;
@@ -39,6 +37,7 @@ class RTMPStreamProcessor implements Processor {
     protected RtmpReader reader = new RtmpReader(DEFAULT_CHUNK_SIZE);
 
     protected RtmpPacket lastPacket = null;
+    protected int chunkOffset = 0;
 
     public RTMPStreamProcessor(ResourceLimit limit, ChunkProcessor chunkProcessor) {
         this.limit = limit;
@@ -90,29 +89,21 @@ public int processPacket(byte[] buffer, int offset, int length) {
         int processed = 0;
 
         ByteBuffer bb = ByteBuffer.wrap(buffer, offset, length);
-        if (lastPacket == null || lastPacket.chunkOffset >= reader.getChunkSize() || lastPacket.messageOffset >= lastPacket.messageSize) {
+        if (lastPacket == null || chunkOffset >= reader.getChunkSize() || lastPacket.messageOffset + chunkOffset >= lastPacket.messageSize) {
             try {
                 Optional<RtmpPacket> op = reader.read(bb);
                 if (!op.isPresent()) {
                     return 0;
                 }
                 lastPacket = op.get();
+                chunkOffset = 0;
                 processed += lastPacket.headerLength;
-                //System.out.println(op.get());
             } catch (RtmpException e) {
                 e.printStackTrace();
                 throw new RuntimeException(e);
-                //return 0;
             }
         }
-        //if (bb.remaining() < lastPacket.payloadLegth) {
-        //    return 0;
-        //}
-
-        //System.err.println("code: " + code + ", sid: " + sid + ", type: " + type + ", length: " + payloadLength);
-        //System.err.println(HexDump.prettyPrintHex(buffer, bufferOffset, sidLength + headLength + payloadLength));
-        //System.err.println(HexDump.prettyPrintHex(buffer, bufferOffset, Math.min(16, sidLength + headLength + payloadLength)));
-        
+
         // change chunk size command processed
         if (lastPacket.messageType == 0x01) {
             if (bb.remaining() < 4) {
@@ -125,30 +116,24 @@ public int processPacket(byte[] buffer, int offset, int length) {
             reader.setChunkSize(newChunkSize);
         }
 
-        //chunkProcessor.processChunk(lastMessage, buffer, readOffset, payloadLength);
-        //MessageInfo messageInfo = buildInfo(p);
-        int chunkRemaining = (int)Math.min(lastPacket.messageSize-lastPacket.messageOffset, reader.getChunkSize() - lastPacket.chunkOffset);
+        int chunkRemaining = (int)Math.min(lastPacket.messageSize-(lastPacket.messageOffset+chunkOffset), reader.getChunkSize() - chunkOffset);
         int byteCount = Math.min(chunkRemaining, bb.remaining());
         if (byteCount > 0) {
-            chunkProcessor.processChunk(buildInfo(lastPacket), bb.array(), bb.position(), byteCount);
+            chunkProcessor.processChunk(buildInfo(lastPacket, chunkOffset), bb.array(), bb.position(), byteCount);
 
             finished = !chunkProcessor.alive();
 
-            lastPacket = lastPacket.transposed(byteCount);
+            chunkOffset += byteCount;
 
             processed += byteCount;
         }
 
-        //System.out.printf("len: %d, newLen: %d%n", sidLength + headLength + payloadLength, p.messageSize + p.headerLength);
-        //return sidLength + headLength + payloadLength;
-        //System.out.printf("len: %d, newLen: %d (%d + %d)%n", sidLength + headLength + payloadLength, p.headerLength + p.payloadLegth, p.headerLength, p.payloadLegth);
         return processed;
     }
 
-    private MessageInfo buildInfo(RtmpPacket p) {
+    private static MessageInfo buildInfo(RtmpPacket p, int chunkOffset) {
         MessageInfo mi = new MessageInfo(p.sid, p.messageType, (int)p.messageSize);
-        mi.offset = (int)p.messageOffset;
-        mi.length = (int)p.messageSize;
+        mi.offset = (int)p.messageOffset + chunkOffset;
         mi.calculatedTimestamp = p.absoluteTimestamp;
         return mi;
     }

src/main/java/org/czentral/minirtmp/RTMPStreamProcessor.java

@@ -32,7 +32,6 @@ public class RtmpPacket {
 
     long absoluteTimestamp;
     long messageOffset;
-    long chunkOffset;
     long headerLength;
 
     protected RtmpPacket() {
@@ -76,31 +75,6 @@ public boolean hasExtendedTimestamp() {
         return packetTimestamp == 0xffffff;
     }
 
-    private RtmpPacket(RtmpPacket other) {
-        this.chunkType = other.chunkType;
-        this.sid = other.sid;
-        this.packetTimestamp = other.packetTimestamp;
-        this.messageSize = other.messageSize;
-        this.messageType = other.messageType;
-        this.messageStreamId = other.messageStreamId;
-        this.absoluteTimestamp = other.absoluteTimestamp;
-        this.messageOffset = other.messageOffset;
-        this.chunkOffset = other.chunkOffset;
-        this.headerLength = other.headerLength;
-    }
-
-    /**
-     * Accounts part of the packet being processed by creating a new packet with modified messageOffset and payloadLength.
-     * @param byteCount
-     * @return
-     */
-    public RtmpPacket transposed(int byteCount) {
-        RtmpPacket result = new RtmpPacket(this);
-        result.messageOffset += byteCount;
-        result.chunkOffset += byteCount;
-        return result;
-    }
-
     @Override
     public String toString() {
         return "RtmpPacket{" +
@@ -112,7 +86,6 @@ public String toString() {
                 ", messageStreamId=" + messageStreamId +
                 ", absoluteTimestamp=" + absoluteTimestamp +
                 ", messageOffset=" + messageOffset +
-                ", chunkOffset=" + chunkOffset +
                 ", headerLength=" + headerLength +
                 '}';
     }

src/main/java/org/czentral/minirtmp/RtmpPacket.java

@@ -17,8 +17,6 @@
 
 package org.czentral.minirtmp;
 
-import org.czentral.incubator.streamm.HexDump;
-
 import java.nio.ByteBuffer;
 import java.util.HashMap;
 import java.util.Map;
@@ -29,10 +27,6 @@ public class RtmpReader {
 
     private long chunkSize;
 
-    private byte[] lastBytes = new byte[16 * 1024];
-    private ByteBuffer lastBuffer = ByteBuffer.wrap(lastBytes);
-    private RtmpPacket lastParsedPacket = null;
-
     public RtmpReader(long chunkSize) {
         this.chunkSize = chunkSize;
     }
@@ -48,7 +42,11 @@ public Optional<RtmpPacket> read(ByteBuffer buffer) throws RtmpException {
         RtmpPacket lastPacket = lastPackets.getOrDefault(p.sid, null);
 
         boolean newMessage = lastPacket == null ||
-                lastPacket.messageOffset == lastPacket.messageSize;
+                lastPacket.messageOffset + chunkSize >= lastPacket.messageSize;
+
+        if (!newMessage) {
+            p.messageOffset = lastPacket.messageOffset + chunkSize;
+        }
 
         if (lastPacket != null) {
             p.absoluteTimestamp = lastPacket.absoluteTimestamp;
@@ -68,15 +66,6 @@ public Optional<RtmpPacket> read(ByteBuffer buffer) throws RtmpException {
 
         } else {
             if (p.chunkType > 0) {
-                //System.err.println("Last good packet --------------------------");
-                //System.err.println(lastParsedPacket);
-                //System.err.print(HexDump.prettyPrintHex(lastBytes, 0, lastBuffer.limit()));
-                byte[] b = new byte[1024];
-                ByteBuffer temp = buffer.duplicate();
-                temp.position(originalOffset);
-                int length = Math.min(b.length, temp.remaining());
-                temp.get(b, 0, length);
-                System.err.print(HexDump.prettyPrintHex(b, 0, length));
                 throw new RtmpException(String.format("Missing previous chunk (chunkId: %d, type: %d)", p.sid, p.chunkType));
             }
         }
@@ -97,19 +86,10 @@ public Optional<RtmpPacket> read(ByteBuffer buffer) throws RtmpException {
             }
         }
 
-        if (!newMessage) {
-            p.messageOffset = lastPacket.messageOffset;
-        }
         p.headerLength = buffer.position() - originalOffset;
 
         lastPackets.put(p.sid, p);
 
-        ByteBuffer temp = buffer.duplicate();
-        temp.position(originalOffset);
-        lastBuffer.limit(Math.min(temp.limit()-temp.position(), lastBytes.length));
-        temp.get(lastBytes, 0, lastBuffer.limit());
-        lastParsedPacket = p;
-
         return Optional.of(p);
     }
 

src/main/java/org/czentral/minirtmp/RtmpReader.java

@@ -17,15 +17,11 @@
 
 package org.czentral.minirtmp;
 
-import org.czentral.minihttp.ChunkedInputStream;
 import org.junit.Test;
 
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
-import java.nio.file.Files;
-
-import static org.junit.Assert.*;
 
 public class RTMPStreamProcessorTest {
 
@@ -36,14 +32,12 @@ public void process() throws Exception {
         limit.assemblyBufferSize = 200 * 1024;
         limit.chunkStreamCount = 100;
 
-        String pathname = "httpresp.pcap";
+        String pathname = "rtmp-in.pcap";
         byte[] buffer = loadFile(pathname);
-        //int start = 0xc938;
         int start = 0xc5f;
-        //int end = 0xe938;
         int end = 0x16c7 + 0x28;
 
-        for (int i=0xc6b; i<end; i++) {
+        for (int i=start; i<end; i++) {
             TestChunkProcessor cp = new TestChunkProcessor();
             RTMPStreamProcessor o = new RTMPStreamProcessor(limit, cp);
             try {
@@ -60,7 +54,9 @@ public void process() throws Exception {
     }
 
     private byte[] loadFile(String pathname) throws IOException {
-        File file = new File(pathname);
+        //File file = new File(pathname);
+        ClassLoader classLoader = getClass().getClassLoader();
+        File file = new File(classLoader.getResource(pathname).getFile());
         FileInputStream fis = new FileInputStream(file);
         byte[] buffer = new byte[(int)file.length()];
         int offset = 0;