Message Stream IDs copied from the request packets.

vbence · May 15, 2015 · 08798ff · 08798ff
1 parent 4740dc3
commit 08798ff
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 14 deletions.
diff --git a/src/org/czentral/incubator/streamm/PublisherAppInstance.java b/src/org/czentral/incubator/streamm/PublisherAppInstance.java
@@ -132,7 +132,7 @@ protected void sendError(MessageInfo mi, RTMPCommand command, Object object, Obj
         response.writeNumber(command.getTxid());
         response.writeMixed(object);
         response.writeMixed(information);
-        context.writeCommand(mi.chunkStreamID, response);
+        context.writeCommand(mi.chunkStreamID, mi.messageStreamID, response);
     }
 
     protected void sendSuccess(MessageInfo mi, RTMPCommand command) {
@@ -141,7 +141,7 @@ protected void sendSuccess(MessageInfo mi, RTMPCommand command) {
         response.writeNumber(command.getTxid());
         response.writeMixed(null);
         response.writeMixed(null);
-        context.writeCommand(mi.chunkStreamID, response);
+        context.writeCommand(mi.chunkStreamID, mi.messageStreamID, response);
     }
 
     protected void cmdDeleteStream(MessageInfo mi, RTMPCommand command) {
@@ -150,7 +150,7 @@ protected void cmdDeleteStream(MessageInfo mi, RTMPCommand command) {
         response.writeNumber(command.getTxid());
         response.writeMixed(null);
         response.writeMixed(null);
-        context.writeCommand(mi.chunkStreamID, response);
+        context.writeCommand(mi.chunkStreamID, mi.messageStreamID, response);
         context.terminate();
     }
 
@@ -160,7 +160,7 @@ protected void cmdCreateStream(MessageInfo mi, RTMPCommand command) {
         response.writeNumber(command.getTxid());
         response.writeMixed(null);
         response.writeMixed(1d);
-        context.writeCommand(mi.chunkStreamID, response);
+        context.writeCommand(mi.chunkStreamID, mi.messageStreamID, response);
     }
 
     protected void cmdConnect(MessageInfo mi, RTMPCommand command) {
@@ -184,10 +184,21 @@ protected void cmdConnect(MessageInfo mi, RTMPCommand command) {
         arg2.put("objectEncoding", 0d);
         response.writeObject(arg2);
 
-        context.writeCommand(mi.chunkStreamID, response);
+        context.writeCommand(mi.chunkStreamID, mi.messageStreamID, response);
     }
 
     protected void cmdFCPublish(MessageInfo mi, RTMPCommand command) {
+        fcpublishTxID = command.getTxid();
+        sendSuccess(mi, command);
+    }
+
+    protected void cmdPublish(MessageInfo mi, RTMPCommand command) {
+
+        Object arg1 = command.getArguments().get(0);
+        if (!(arg1 instanceof String)) {
+            sendSuccess(mi, command);
+            return;
+        }
 
         String streamName = (String)command.getArguments().get(0);
         int separatorPos = streamName.indexOf(STREAM_NAME_SEPARATOR);
@@ -234,11 +245,7 @@ protected void cmdFCPublish(MessageInfo mi, RTMPCommand command) {
         context.getLimit().assemblyBufferSize = 256 * 1024;
 
         streamID = clientStreamID;
-        fcpublishTxID = command.getTxid();
-        sendSuccess(mi, command);
-    }
-
-    protected void cmdPublish(MessageInfo mi, RTMPCommand command) {
+
         AMFPacket response = new AMFPacket();
         response.writeString("onStatus");
         response.writeNumber(fcpublishTxID);
@@ -254,7 +261,7 @@ protected void cmdPublish(MessageInfo mi, RTMPCommand command) {
         arg2.put("objectEncoding", 0d);
         response.writeObject(arg2);
 
-        context.writeCommand(mi.chunkStreamID, response);
+        context.writeCommand(mi.chunkStreamID, mi.messageStreamID, response);
     }
 
     @Override

diff --git a/src/org/czentral/minirtmp/ApplicationContext.java b/src/org/czentral/minirtmp/ApplicationContext.java
@@ -153,7 +153,7 @@ protected void processMessage(MessageInfo mi, byte[] buffer, int payloadOffset,
                         response.writeString("_error");
                         response.writeMixed(null);
                         response.writeMixed(null);
-                        writeCommand(mi.chunkStreamID, response);
+                        writeCommand(mi.chunkStreamID, mi.messageStreamID, response);
                         return;
                     }
                     applicationInstance.onConnect(this);
@@ -166,7 +166,7 @@ protected void processMessage(MessageInfo mi, byte[] buffer, int payloadOffset,
                     response.writeString("_error");
                     response.writeMixed(null);
                     response.writeMixed(null);
-                    writeCommand(mi.chunkStreamID, response);
+                    writeCommand(mi.chunkStreamID, mi.messageStreamID, response);
                 }
             } else {
                 applicationInstance.invokeCommand(mi, command);
@@ -205,7 +205,7 @@ protected RTMPCommand readData(AMFDecoder decoder) {
         return new RTMPCommand("", 0d, null, arguments);
     }
 
-    public void writeCommand(int streamID, AMFPacket message) {
+    public void writeCommand(int streamID, int messageStreamID, AMFPacket message) {
         byte[] header = new byte[12];
         header[0] = (byte)((streamID) & 0x3f);
 
@@ -218,6 +218,11 @@ public void writeCommand(int streamID, AMFPacket message) {
         // message type
         header[7] = 0x14;
 
+        header[8] = (byte) ((messageStreamID >> 24) & 0xff);
+        header[9] = (byte) ((messageStreamID >> 16) & 0xff);
+        header[10] = (byte) ((messageStreamID >> 8) & 0xff);
+        header[11] = (byte) ((messageStreamID) & 0xff);
+
         // writing
         try {
             outputStream.write(header);