support cache last gop for video

@@ -45,7 +45,8 @@ url: rtmp://127.0.0.1:1935/live/livestream
 * nginx v1.5.0: 139524 lines <br/>
 * nginx v1.5.0: 139524 lines <br/>
 
 
 ### History
 ### History
-* v0.2, 2013-10-23, v0.2 released. 10125 lines.
+* v0.3, 2013-10-27, support cache last gop for client fast startup.
+* v0.2, 2013-10-25, v0.2 released. 10125 lines.
 * v0.2, 2013-10-25, support flash publish.
 * 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-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 time jitter detect and correct algorithm(SrsConsumer::jitter_correct).

@@ -145,7 +145,9 @@ int SrsClient::do_cycle()
 				return ret;
 				return ret;
 			}
 			}
 			srs_info("start to publish stream %s success", req->stream.c_str());
 			srs_info("start to publish stream %s success", req->stream.c_str());
-			return streaming_publish(source, true);
+			ret = streaming_publish(source, true);
+			source->on_unpublish();
+			return ret;
 		}
 		}
 		case SrsClientFlashPublish: {
 		case SrsClientFlashPublish: {
 			srs_verbose("flash start to publish stream %s.", req->stream.c_str());
 			srs_verbose("flash start to publish stream %s.", req->stream.c_str());
@@ -155,7 +157,9 @@ int SrsClient::do_cycle()
 				return ret;
 				return ret;
 			}
 			}
 			srs_info("flash start to publish stream %s success", req->stream.c_str());
 			srs_info("flash start to publish stream %s success", req->stream.c_str());
-			return streaming_publish(source, false);
+			ret = streaming_publish(source, false);
+			source->on_unpublish();
+			return ret;
 		}
 		}
 		default: {
 		default: {
 			ret = ERROR_SYSTEM_CLIENT_INVALID;
 			ret = ERROR_SYSTEM_CLIENT_INVALID;

@@ -31,6 +31,35 @@ SrsCodec::~SrsCodec()
 {
 {
 }
 }
 
 
+bool SrsCodec::video_is_keyframe(int8_t* data, int size)
+{
+	// E.4.3.1 VIDEODATA
+	// Frame Type UB [4]
+	// Type of video frame. The following values are defined:
+	// 	1 = key frame (for AVC, a seekable frame)
+	// 	2 = inter frame (for AVC, a non-seekable frame)
+	// 	3 = disposable inter frame (H.263 only)
+	// 	4 = generated key frame (reserved for server use only)
+	// 	5 = video info/command frame
+	//
+	// AVCPacketType IF CodecID == 7 UI8
+	// The following values are defined:
+	// 	0 = AVC sequence header
+	// 	1 = AVC NALU
+	// 	2 = AVC end of sequence (lower level NALU sequence ender is
+	// 		not required or supported)
+	
+	// 2bytes required.
+	if (size < 1) {
+		return false;
+	}
+
+	char frame_type = *(char*)data;
+	frame_type = (frame_type >> 4) & 0x0F;
+	
+	return frame_type == 1;
+}
+
 bool SrsCodec::video_is_sequence_header(int8_t* data, int size)
 bool SrsCodec::video_is_sequence_header(int8_t* data, int size)
 {
 {
 	// E.4.3.1 VIDEODATA
 	// E.4.3.1 VIDEODATA

@@ -40,10 +40,26 @@ public:
 	SrsCodec();
 	SrsCodec();
 	virtual ~SrsCodec();
 	virtual ~SrsCodec();
 public:
 public:
+	/**
+	* only check the frame_type, not check the codec type.
+	*/
+	virtual bool video_is_keyframe(int8_t* data, int size);
+	/**
+	* check codec h264, keyframe, sequence header
+	*/
 	virtual bool video_is_sequence_header(int8_t* data, int size);
 	virtual bool video_is_sequence_header(int8_t* data, int size);
+	/**
+	* check codec aac, sequence header
+	*/
 	virtual bool audio_is_sequence_header(int8_t* data, int size);
 	virtual bool audio_is_sequence_header(int8_t* data, int size);
-private:
+	/**
+	* check codec h264.
+	*/
 	virtual bool video_is_h264(int8_t* data, int size);
 	virtual bool video_is_h264(int8_t* data, int size);
+private:
+	/**
+	* check codec aac.
+	*/
 	virtual bool audio_is_aac(int8_t* data, int size);
 	virtual bool audio_is_aac(int8_t* data, int size);
 };
 };
 
 

@@ -64,6 +64,11 @@ SrsConsumer::~SrsConsumer()
 	source->on_consumer_destroy(this);
 	source->on_consumer_destroy(this);
 }
 }
 
 
+int SrsConsumer::get_time()
+{
+	return (int)last_pkt_correct_time;
+}
+
 int SrsConsumer::enqueue(SrsSharedPtrMessage* msg)
 int SrsConsumer::enqueue(SrsSharedPtrMessage* msg)
 {
 {
 	int ret = ERROR_SUCCESS;
 	int ret = ERROR_SUCCESS;
@@ -144,10 +149,12 @@ int SrsConsumer::jitter_correct(SrsSharedPtrMessage* msg)
 SrsSource::SrsSource(std::string _stream_url)
 SrsSource::SrsSource(std::string _stream_url)
 {
 {
 	stream_url = _stream_url;
 	stream_url = _stream_url;
-	cache_metadata = NULL;
-	cache_sh_video = NULL;
-	cache_sh_audio = NULL;
 	codec = new SrsCodec();
 	codec = new SrsCodec();
+	
+	cache_metadata = cache_sh_video = cache_sh_audio = NULL;
+	
+	cached_video_count = 0;
+	enable_gop_cache = true;
 }
 }
 
 
 SrsSource::~SrsSource()
 SrsSource::~SrsSource()
@@ -159,6 +166,8 @@ SrsSource::~SrsSource()
 	}
 	}
 	consumers.clear();
 	consumers.clear();
 	
 	
+	clear_gop_cache();
+	
 	srs_freep(cache_metadata);
 	srs_freep(cache_metadata);
 	srs_freep(cache_sh_video);
 	srs_freep(cache_sh_video);
 	srs_freep(cache_sh_audio);
 	srs_freep(cache_sh_audio);
@@ -246,8 +255,16 @@ int SrsSource::on_audio(SrsCommonMessage* audio)
 	if (codec->audio_is_sequence_header(msg->payload, msg->size)) {
 	if (codec->audio_is_sequence_header(msg->payload, msg->size)) {
 		srs_freep(cache_sh_audio);
 		srs_freep(cache_sh_audio);
 		cache_sh_audio = msg->copy();
 		cache_sh_audio = msg->copy();
+		return ret;
 	}
 	}
 	
 	
+	// cache the last gop packets
+	if ((ret = cache_last_gop(msg)) != ERROR_SUCCESS) {
+		srs_error("shrink gop cache failed. ret=%d", ret);
+		return ret;
+	}
+	srs_verbose("cache gop success.");
+	
 	return ret;
 	return ret;
 }
 }
 
 
@@ -282,7 +299,15 @@ int SrsSource::on_video(SrsCommonMessage* video)
 	if (codec->video_is_sequence_header(msg->payload, msg->size)) {
 	if (codec->video_is_sequence_header(msg->payload, msg->size)) {
 		srs_freep(cache_sh_video);
 		srs_freep(cache_sh_video);
 		cache_sh_video = msg->copy();
 		cache_sh_video = msg->copy();
+		return ret;
 	}
 	}
+
+	// cache the last gop packets
+	if ((ret = cache_last_gop(msg)) != ERROR_SUCCESS) {
+		srs_error("shrink gop cache failed. ret=%d", ret);
+		return ret;
+	}
+	srs_verbose("cache gop success.");
 	
 	
 	return ret;
 	return ret;
 }
 }
@@ -312,6 +337,16 @@ int SrsSource::on_video(SrsCommonMessage* video)
 	}
 	}
 	srs_info("dispatch audio sequence header success");
 	srs_info("dispatch audio sequence header success");
 	
 	
+	std::vector<SrsSharedPtrMessage*>::iterator it;
+	for (it = gop_cache.begin(); it != gop_cache.end(); ++it) {
+		SrsSharedPtrMessage* msg = *it;
+		if ((ret = consumer->enqueue(msg->copy())) != ERROR_SUCCESS) {
+			srs_error("dispatch cached gop failed. ret=%d", ret);
+			return ret;
+		}
+	}
+	srs_trace("dispatch cached gop success. count=%d, duration=%d", (int)gop_cache.size(), consumer->get_time());
+	
 	return ret;
 	return ret;
 }
 }
 
 
@@ -325,3 +360,58 @@ void SrsSource::on_consumer_destroy(SrsConsumer* consumer)
 	srs_info("handle consumer destroy success.");
 	srs_info("handle consumer destroy success.");
 }
 }
 
 
+void SrsSource::on_unpublish()
+{
+	clear_gop_cache();
+	srs_trace("clear cache when unpublish.");
+}
+
+int SrsSource::cache_last_gop(SrsSharedPtrMessage* msg)
+{
+	int ret = ERROR_SUCCESS;
+	
+	if (!enable_gop_cache) {
+		srs_verbose("gop cache is disabled.");
+		return ret;
+	}
+	
+	// got video, update the video count if acceptable
+	if (msg->header.is_video()) {
+		cached_video_count++;
+	}
+	
+	// no acceptable video or pure audio, disable the cache.
+	if (cached_video_count == 0) {
+		srs_verbose("ignore any frame util got a h264 video frame.");
+		return ret;
+	}
+	
+	// clear gop cache when got key frame
+	if (msg->header.is_video() && codec->video_is_keyframe(msg->payload, msg->size)) {
+		srs_info("clear gop cache when got keyframe. vcount=%d, count=%d",
+			cached_video_count, (int)gop_cache.size());
+			
+		clear_gop_cache();
+		
+		// curent msg is video frame, so we set to 1.
+		cached_video_count = 1;
+	}
+	
+	// cache the frame.
+	gop_cache.push_back(msg->copy());
+	
+	return ret;
+}
+
+void SrsSource::clear_gop_cache()
+{
+	std::vector<SrsSharedPtrMessage*>::iterator it;
+	for (it = gop_cache.begin(); it != gop_cache.end(); ++it) {
+		SrsSharedPtrMessage* msg = *it;
+		srs_freep(msg);
+	}
+	gop_cache.clear();
+
+	cached_video_count = 0;
+}
+

@@ -55,6 +55,10 @@ public:
 	virtual ~SrsConsumer();
 	virtual ~SrsConsumer();
 public:
 public:
 	/**
 	/**
+	* get current client time, the last packet time.
+	*/
+	virtual int get_time();
+	/**
 	* enqueue an shared ptr message.
 	* enqueue an shared ptr message.
 	*/
 	*/
 	virtual int enqueue(SrsSharedPtrMessage* msg);
 	virtual int enqueue(SrsSharedPtrMessage* msg);
@@ -92,6 +96,22 @@ private:
 	SrsCodec* codec;
 	SrsCodec* codec;
 	std::string stream_url;
 	std::string stream_url;
 	std::vector<SrsConsumer*> consumers;
 	std::vector<SrsConsumer*> consumers;
+// gop cache for client fast startup.
+private:
+	/**
+	* if disabled the gop cache,
+	* the client will wait for the next keyframe for h264,
+	* and will be black-screen.
+	*/
+	bool enable_gop_cache;
+	/**
+	* the video frame count, avoid cache for pure audio stream.
+	*/
+	int cached_video_count;
+	/**
+	* cached gop.
+	*/
+	std::vector<SrsSharedPtrMessage*> gop_cache;
 private:
 private:
 	SrsSharedPtrMessage* cache_metadata;
 	SrsSharedPtrMessage* cache_metadata;
 	// the cached video sequence header.
 	// the cached video sequence header.
@@ -108,6 +128,15 @@ public:
 public:
 public:
 	virtual int create_consumer(SrsConsumer*& consumer);
 	virtual int create_consumer(SrsConsumer*& consumer);
 	virtual void on_consumer_destroy(SrsConsumer* consumer);
 	virtual void on_consumer_destroy(SrsConsumer* consumer);
+	virtual void on_unpublish();
+private:
+	/**
+	* only for h264 codec
+	* 1. cache the gop when got h264 video packet.
+	* 2. clear gop when got keyframe.
+	*/
+	virtual int cache_last_gop(SrsSharedPtrMessage* msg);
+	virtual void clear_gop_cache();
 };
 };
 
 
 #endif
 #endif