Added createStream with parameter for AMS support.

Red5 · Oct 12, 2015 · b1429a1 · b1429a1
1 parent 87d2954
commit b1429a1
Show file tree

Hide file tree

Showing 5 changed files with 49 additions and 8 deletions.
diff --git a/src/main/java/org/red5/server/api/stream/IStreamService.java b/src/main/java/org/red5/server/api/stream/IStreamService.java
@@ -35,6 +35,14 @@ public interface IStreamService extends IScopeService {
 	 */
 	public Number createStream();
 
+	/**
+	 * Create a stream and return a corresponding id.
+	 * 
+	 * @param streamId  Stream id
+	 * @return     ID of created stream
+	 */
+	public Number createStream(Number streamId);	
+
 	/**
 	 * Close the stream but not deallocate the resources.
 	 * 

diff --git a/src/main/java/org/red5/server/net/rtmp/event/Ping.java b/src/main/java/org/red5/server/net/rtmp/event/Ping.java
@@ -277,7 +277,14 @@ protected void releaseInternal() {
 	public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
 		super.readExternal(in);
 		eventType = in.readShort();
-		value2 = (Number) in.readObject();
+		switch (eventType) {
+			case PING_CLIENT:
+			case PONG_SERVER:
+				value2 = (Number) in.readInt();
+				break;
+			default:
+				value2 = (Number) in.readDouble();				
+		}
 		value3 = in.readInt();
 		value4 = in.readInt();
 	}
@@ -286,7 +293,14 @@ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundExcept
 	public void writeExternal(ObjectOutput out) throws IOException {
 		super.writeExternal(out);
 		out.writeShort(eventType);
-		out.writeObject(value2);
+		switch (eventType) {
+			case PING_CLIENT:
+			case PONG_SERVER:
+				out.writeDouble(value2.intValue());
+				break;
+			default:
+				out.writeDouble(value2.doubleValue());				
+		}
 		out.writeInt(value3);
 		out.writeInt(value4);
 	}

diff --git a/src/main/java/org/red5/server/net/rtmp/message/Header.java b/src/main/java/org/red5/server/net/rtmp/message/Header.java
@@ -218,7 +218,7 @@ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundExcept
 		dataType = in.readByte();
 		channelId = in.readInt();
 		size = in.readInt();
-		streamId = (Number) in.readObject();
+		streamId = (Number) in.readDouble();
 		timerBase = in.readInt();
 		timerDelta = in.readInt();
 		extendedTimestamp = in.readInt();
@@ -228,7 +228,7 @@ public void writeExternal(ObjectOutput out) throws IOException {
 		out.writeByte(dataType);
 		out.writeInt(channelId);
 		out.writeInt(size);
-		out.writeObject(streamId);
+		out.writeDouble(streamId.doubleValue());
 		out.writeInt(timerBase);
 		out.writeInt(timerDelta);
 		out.writeInt(extendedTimestamp);
@@ -240,7 +240,7 @@ public void writeExternal(ObjectOutput out) throws IOException {
 	@Override
 	public String toString() {
 		// if its new and props are un-set, just return that message
-		if ((channelId + dataType + size + streamId.intValue()) > 0) {
+		if ((channelId + dataType + size + streamId.doubleValue()) > 0d) {
 			return "Header [streamId=" + streamId + ", channelId=" + channelId + ", dataType=" + dataType + ", timerBase=" + timerBase + ", timerDelta=" + timerDelta + ", size=" + size + ", extendedTimestamp=" + extendedTimestamp + "]";
 		} else {
 			return "empty";

diff --git a/src/main/java/org/red5/server/net/rtmp/status/Status.java b/src/main/java/org/red5/server/net/rtmp/status/Status.java
@@ -223,15 +223,15 @@ public void serialize(Output output) {
 	}
 
 	public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-		clientid = (Number) in.readObject();
+		clientid = (Number) in.readDouble();
 		code = (String) in.readObject();
 		description = (String) in.readObject();
 		details = (String) in.readObject();
 		level = (String) in.readObject();
 	}
 
 	public void writeExternal(ObjectOutput out) throws IOException {
-		out.writeObject(clientid);
+		out.writeDouble(clientid.doubleValue());
 		out.writeObject(code);
 		out.writeObject(description);
 		out.writeObject(details);

diff --git a/src/main/java/org/red5/server/stream/StreamService.java b/src/main/java/org/red5/server/stream/StreamService.java
@@ -74,6 +74,7 @@ protected Boolean initialValue() {
 	/** {@inheritDoc} */
 	public Number createStream() {
 		IConnection conn = Red5.getConnectionLocal();
+		log.trace("createStream connection: {}", conn.getSessionId());
 		if (conn instanceof IStreamCapableConnection) {
 			Number streamId = ((IStreamCapableConnection) conn).reserveStreamId();
 			if (log.isTraceEnabled()) {
@@ -84,6 +85,24 @@ public Number createStream() {
 		return -1;
 	}
 
+	/** {@inheritDoc} */
+	public Number createStream(Number streamId) {
+		IConnection conn = Red5.getConnectionLocal();
+		log.trace("createStream stream id: {} connection: {}", streamId, conn.getSessionId());
+		if (conn instanceof IStreamCapableConnection) {
+			if (streamId.doubleValue() > 0d) {
+				streamId = ((IStreamCapableConnection) conn).reserveStreamId(streamId);				
+			} else {
+				streamId = ((IStreamCapableConnection) conn).reserveStreamId();
+			}
+			if (log.isTraceEnabled()) {
+				log.trace("Stream id: {} created for {}", streamId, conn.getSessionId());
+			}
+			return streamId;
+		}
+		return -1;
+	}	
+
 	/** {@inheritDoc} */
 	public void initStream(Number streamId) {
 		IConnection conn = Red5.getConnectionLocal();
@@ -149,7 +168,7 @@ public void closeStream() {
 	 * @param streamId stream ID (number: 1,2,...)
 	 */
 	public void closeStream(IConnection conn, Number streamId) {
-		log.info("closeStream: streamId={}, connection={}", streamId, conn);
+		log.info("closeStream  stream id: {} connection: {}", streamId, conn.getSessionId());
 		if (conn instanceof IStreamCapableConnection) {
 			IStreamCapableConnection scConn = (IStreamCapableConnection) conn;
 			IClientStream stream = scConn.getStreamById(streamId);