support flash publish

README.md

@@ -26,9 +26,12 @@ Compare:<br/>
 * nginx v1.5.0: 139524 lines <br/>
 
 Features:<br/>
+* v0.2, 2013-10-23, v0.2 released.
+* v0.2, 2013-10-25, support flash publish.
 * v0.2, 2013-10-25, support h264/avc codec by rtmp complex handshake(SrsComplexHandshake).
 * v0.2, 2013-10-24, support time jitter detect and correct algorithm(SrsConsumer::jitter_correct).
-* v0.2, 2013-10-24, support decode the video/audio codec type(SrsCodec), cache the h264/avc sequence header.
+* v0.2, 2013-10-24, support decode codec type(SrsCodec) to cache the h264/avc sequence header.
+* v0.1, 2013-10-23, v0.1 released.
 * v0.1, 2013-10-23, support basic amf0 codec, simplify the api using c-style api.
 * v0.1, 2013-10-23, support shared ptr msg(SrsSharedPtrMessage) for zero memory copy.
 * v0.1, 2013-10-22, support vp6 codec with rtmp protocol specified simple handshake.

trunk/src/core/srs_core_client.cpp

@@ -137,15 +137,25 @@ int SrsClient::do_cycle()
 			srs_info("start to play stream %s success", req->stream.c_str());
 			return streaming_play(source);
 		}
-		case SrsClientPublish: {
+		case SrsClientFMLEPublish: {
-			srs_verbose("start to publish stream %s.", req->stream.c_str());
+			srs_verbose("FMLE start to publish stream %s.", req->stream.c_str());
 			
-			if ((ret = rtmp->start_publish(res->stream_id)) != ERROR_SUCCESS) {
+			if ((ret = rtmp->start_fmle_publish(res->stream_id)) != ERROR_SUCCESS) {
 				srs_error("start to publish stream failed. ret=%d", ret);
 				return ret;
 			}
 			srs_info("start to publish stream %s success", req->stream.c_str());
-			return streaming_publish(source);
+			return streaming_publish(source, true);
+		}
+		case SrsClientFlashPublish: {
+			srs_verbose("flash start to publish stream %s.", req->stream.c_str());
+			
+			if ((ret = rtmp->start_flash_publish(res->stream_id)) != ERROR_SUCCESS) {
+				srs_error("flash start to publish stream failed. ret=%d", ret);
+				return ret;
+			}
+			srs_info("flash start to publish stream %s success", req->stream.c_str());
+			return streaming_publish(source, false);
 		}
 		default: {
 			ret = ERROR_SYSTEM_CLIENT_INVALID;
@@ -237,7 +247,7 @@ int SrsClient::streaming_play(SrsSource* source)
 	return ret;
 }
 
-int SrsClient::streaming_publish(SrsSource* source)
+int SrsClient::streaming_publish(SrsSource* source, bool is_fmle)
 {
 	int ret = ERROR_SUCCESS;
 	
@@ -293,6 +303,12 @@ int SrsClient::streaming_publish(SrsSource* source)
 				return ret;
 			}
 			
+			// flash unpublish.
+			if (!is_fmle) {
+				srs_trace("flash publish finished.");
+				return ret;
+			}
+		
 			SrsPacket* pkt = msg->get_packet();
 			if (dynamic_cast<SrsFMLEStartPacket*>(pkt)) {
 				SrsFMLEStartPacket* unpublish = dynamic_cast<SrsFMLEStartPacket*>(pkt);

trunk/src/core/srs_core_client.hpp

@@ -54,7 +54,7 @@ protected:
 	virtual int do_cycle();
 private:
 	virtual int streaming_play(SrsSource* source);
-	virtual int streaming_publish(SrsSource* source);
+	virtual int streaming_publish(SrsSource* source, bool is_fmle);
 	virtual int get_peer_ip();
 };
 

trunk/src/core/srs_core_protocol.cpp

@@ -1792,7 +1792,7 @@ int SrsPublishPacket::decode(SrsStream* stream)
 		return ret;
 	}
 	
-	if ((ret = srs_amf0_read_string(stream, type)) != ERROR_SUCCESS) {
+	if (!stream->empty() && (ret = srs_amf0_read_string(stream, type)) != ERROR_SUCCESS) {
 		srs_error("amf0 decode publish type failed. ret=%d", ret);
 		return ret;
 	}

trunk/src/core/srs_core_protocol.hpp

@@ -623,6 +623,7 @@ public:
 	double transaction_id;
 	SrsAmf0Null* command_object;
 	std::string stream_name;
+	// optional, default to live.
 	std::string type;
 public:
 	SrsPublishPacket();

trunk/src/core/srs_core_rtmp.cpp

@@ -395,7 +395,6 @@ int SrsRtmp::identify_client(int stream_id, SrsClientType& type, std::string& st
 			return identify_fmle_publish_client(
 				dynamic_cast<SrsFMLEStartPacket*>(pkt), type, stream_name);
 		}
-		// TODO: identify the flash publish client.
 		
 		srs_trace("ignore AMF0/AMF3 command message.");
 	}
@@ -517,7 +516,7 @@ int SrsRtmp::start_play(int stream_id)
 	return ret;
 }
 
-int SrsRtmp::start_publish(int stream_id)
+int SrsRtmp::start_fmle_publish(int stream_id)
 {
 	int ret = ERROR_SUCCESS;
 	
@@ -625,6 +624,8 @@ int SrsRtmp::start_publish(int stream_id)
 		srs_info("send onStatus(NetStream.Publish.Start) message success.");
 	}
 	
+	srs_info("FMLE publish success.");
+	
 	return ret;
 }
 
@@ -686,6 +687,34 @@ int SrsRtmp::fmle_unpublish(int stream_id, double unpublish_tid)
 	return ret;
 }
 
+int SrsRtmp::start_flash_publish(int stream_id)
+{
+	int ret = ERROR_SUCCESS;
+	
+	// publish response onStatus(NetStream.Publish.Start)
+	if (true) {
+		SrsCommonMessage* msg = new SrsCommonMessage();
+		SrsOnStatusCallPacket* pkt = new SrsOnStatusCallPacket();
+		
+		pkt->data->set(StatusLevel, new SrsAmf0String(StatusLevelStatus));
+		pkt->data->set(StatusCode, new SrsAmf0String(StatusCodePublishStart));
+		pkt->data->set(StatusDescription, new SrsAmf0String("Started publishing stream."));
+		pkt->data->set(StatusClientId, new SrsAmf0String(RTMP_SIG_CLIENT_ID));
+		
+		msg->set_packet(pkt, stream_id);
+		
+		if ((ret = protocol->send_message(msg)) != ERROR_SUCCESS) {
+			srs_error("send onStatus(NetStream.Publish.Start) message failed. ret=%d", ret);
+			return ret;
+		}
+		srs_info("send onStatus(NetStream.Publish.Start) message success.");
+	}
+	
+	srs_info("flash publish success.");
+	
+	return ret;
+}
+
 int SrsRtmp::identify_create_stream_client(SrsCreateStreamPacket* req, int stream_id, SrsClientType& type, std::string& stream_name)
 {
 	int ret = ERROR_SUCCESS;
@@ -731,6 +760,11 @@ int SrsRtmp::identify_create_stream_client(SrsCreateStreamPacket* req, int strea
 			srs_trace("identity client type=play, stream_name=%s", stream_name.c_str());
 			return ret;
 		}
+		if (dynamic_cast<SrsPublishPacket*>(pkt)) {
+			srs_info("identify client by publish, falsh publish.");
+			return identify_flash_publish_client(
+				dynamic_cast<SrsPublishPacket*>(pkt), type, stream_name);
+		}
 		
 		srs_trace("ignore AMF0/AMF3 command message.");
 	}
@@ -742,7 +776,7 @@ int SrsRtmp::identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsClientType
 {
 	int ret = ERROR_SUCCESS;
 	
-	type = SrsClientPublish;
+	type = SrsClientFMLEPublish;
 	stream_name = req->stream_name;
 	
 	// releaseStream response
@@ -762,3 +796,13 @@ int SrsRtmp::identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsClientType
 	return ret;
 }
 
+int SrsRtmp::identify_flash_publish_client(SrsPublishPacket* req, SrsClientType& type, std::string& stream_name)
+{
+	int ret = ERROR_SUCCESS;
+	
+	type = SrsClientFlashPublish;
+	stream_name = req->stream_name;
+	
+	return ret;
+}
+

trunk/src/core/srs_core_rtmp.hpp

@@ -40,6 +40,7 @@ class SrsCommonMessage;
 class SrsCreateStreamPacket;
 class SrsFMLEStartPacket;
 class SrsComplexHandshake;
+class SrsPublishPacket;
 
 /**
 * the original request from client.
@@ -85,7 +86,8 @@ enum SrsClientType
 {
 	SrsClientUnknown,
 	SrsClientPlay,
-	SrsClientPublish,
+	SrsClientFMLEPublish,
+	SrsClientFlashPublish,
 };
 
 /**
@@ -146,15 +148,21 @@ public:
 	* onFCPublish(NetStream.Publish.Start)
 	* onStatus(NetStream.Publish.Start)
 	*/
-	virtual int start_publish(int stream_id);
+	virtual int start_fmle_publish(int stream_id);
 	/**
 	* process the FMLE unpublish event.
 	* @unpublish_tid the unpublish request transaction id.
 	*/
 	virtual int fmle_unpublish(int stream_id, double unpublish_tid);
+	/**
+	* when client type is publish, response with packets:
+	* onStatus(NetStream.Publish.Start)
+	*/
+	virtual int start_flash_publish(int stream_id);
 private:
 	virtual int identify_create_stream_client(SrsCreateStreamPacket* req, int stream_id, SrsClientType& type, std::string& stream_name);
 	virtual int identify_fmle_publish_client(SrsFMLEStartPacket* req, SrsClientType& type, std::string& stream_name);
+	virtual int identify_flash_publish_client(SrsPublishPacket* req, SrsClientType& type, std::string& stream_name);
 };
 
 #endif