implement save meta,need more test

Conflicts:

	trunk/src/app/srs_app_source.cpp
	trunk/src/app/srs_app_source.hpp
	trunk/src/protocol/srs_rtmp_stack.cpp

@@ -259,13 +259,35 @@ int SrsFlvSegment::write_metadata(SrsSharedPtrMessage* metadata)
     }
     }
     
     
     // to flv file.
     // to flv file.
-    if ((ret = enc->write_metadata(18, payload, size)) != ERROR_SUCCESS) {
+    if ((ret = enc->write_metadata(0, payload, size)) != ERROR_SUCCESS) {
         return ret;
         return ret;
     }
     }
     
     
     return ret;
     return ret;
 }
 }
 
 
+int SrsFlvSegment::write_data(SrsSharedPtrMessage* shared_data)
+{
+    int ret = ERROR_SUCCESS;
+
+    SrsSharedPtrMessage* data = shared_data->copy();
+    SrsAutoFree(SrsSharedPtrMessage, data);
+
+    if ((jitter->correct(data, jitter_algorithm)) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+    char* payload = data->payload;
+    int size = data->size;
+    int64_t timestamp = plan->filter_timestamp(data->timestamp);
+    if ((ret = enc->write_metadata( timestamp , payload, size)) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+
+    return ret;
+}
+
 int SrsFlvSegment::write_audio(SrsSharedPtrMessage* shared_audio)
 int SrsFlvSegment::write_audio(SrsSharedPtrMessage* shared_audio)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
@@ -291,6 +313,8 @@ int SrsFlvSegment::write_audio(SrsSharedPtrMessage* shared_audio)
     return ret;
     return ret;
 }
 }
 
 
+
+
 int SrsFlvSegment::write_video(SrsSharedPtrMessage* shared_video)
 int SrsFlvSegment::write_video(SrsSharedPtrMessage* shared_video)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
@@ -605,6 +629,17 @@ int SrsDvrPlan::on_meta_data(SrsSharedPtrMessage* shared_metadata)
     return segment->write_metadata(shared_metadata);
     return segment->write_metadata(shared_metadata);
 }
 }
 
 
+int SrsDvrPlan::on_data(SrsSharedPtrMessage* shared_metadata)
+{
+    int ret = ERROR_SUCCESS;
+
+    if (!dvr_enabled) {
+        return ret;
+    }
+
+    return segment->write_data(shared_metadata);
+}
+
 int SrsDvrPlan::on_audio(SrsSharedPtrMessage* shared_audio)
 int SrsDvrPlan::on_audio(SrsSharedPtrMessage* shared_audio)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
@@ -1038,6 +1073,11 @@ int SrsDvr::on_meta_data(SrsOnMetaDataPacket* m)
     return ret;
     return ret;
 }
 }
 
 
+int SrsDvr::on_data(SrsSharedPtrMessage* shared_data)
+{
+    return plan->on_data(shared_data);
+}
+
 int SrsDvr::on_audio(SrsSharedPtrMessage* shared_audio)
 int SrsDvr::on_audio(SrsSharedPtrMessage* shared_audio)
 {
 {
     return plan->on_audio(shared_audio);
     return plan->on_audio(shared_audio);

@@ -141,6 +141,10 @@ public:
     */
     */
     virtual int write_metadata(SrsSharedPtrMessage* metadata);
     virtual int write_metadata(SrsSharedPtrMessage* metadata);
     /**
     /**
+    * @param shared_data, directly ptr, copy it if need to save it.
+    */
+    virtual int write_data(SrsSharedPtrMessage* shared_data);
+    /**
     * @param shared_audio, directly ptr, copy it if need to save it.
     * @param shared_audio, directly ptr, copy it if need to save it.
     */
     */
     virtual int write_audio(SrsSharedPtrMessage* shared_audio);
     virtual int write_audio(SrsSharedPtrMessage* shared_audio);
@@ -221,6 +225,10 @@ public:
     */
     */
     virtual int on_meta_data(SrsSharedPtrMessage* shared_metadata);
     virtual int on_meta_data(SrsSharedPtrMessage* shared_metadata);
     /**
     /**
+    * when got metadata.
+    */
+    virtual int on_data(SrsSharedPtrMessage* shared_metadata);
+    /**
     * @param shared_audio, directly ptr, copy it if need to save it.
     * @param shared_audio, directly ptr, copy it if need to save it.
     */
     */
     virtual int on_audio(SrsSharedPtrMessage* shared_audio);
     virtual int on_audio(SrsSharedPtrMessage* shared_audio);
@@ -328,6 +336,11 @@ public:
     */
     */
     virtual int on_meta_data(SrsOnMetaDataPacket* m);
     virtual int on_meta_data(SrsOnMetaDataPacket* m);
     /**
     /**
+    * the data packets to dvr.
+    * @param shared_data, directly ptr, copy it if need to save it.
+    */
+    virtual int on_data(SrsSharedPtrMessage* shared_data);
+    /**
     * mux the audio packets to dvr.
     * mux the audio packets to dvr.
     * @param shared_audio, directly ptr, copy it if need to save it.
     * @param shared_audio, directly ptr, copy it if need to save it.
     */
     */

@@ -271,7 +271,7 @@ int SrsFlvStreamEncoder::write_video(int64_t timestamp, char* data, int size)
 
 
 int SrsFlvStreamEncoder::write_metadata(int64_t timestamp, char* data, int size)
 int SrsFlvStreamEncoder::write_metadata(int64_t timestamp, char* data, int size)
 {
 {
-    return enc->write_metadata(SrsCodecFlvTagScript, data, size);
+    return enc->write_metadata(timestamp, data, size);
 }
 }
 
 
 bool SrsFlvStreamEncoder::has_cache()
 bool SrsFlvStreamEncoder::has_cache()

@@ -1063,6 +1063,7 @@ int SrsRtmpConn::process_publish_message(SrsSource* source, SrsCommonMessage* ms
     
     
     // process onMetaData
     // process onMetaData
     if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
     if (msg->header.is_amf0_data() || msg->header.is_amf3_data()) {
+#if 0
         SrsPacket* pkt = NULL;
         SrsPacket* pkt = NULL;
         if ((ret = rtmp->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
         if ((ret = rtmp->decode_message(msg, &pkt)) != ERROR_SUCCESS) {
             srs_error("decode onMetaData message failed. ret=%d", ret);
             srs_error("decode onMetaData message failed. ret=%d", ret);
@@ -1082,6 +1083,14 @@ int SrsRtmpConn::process_publish_message(SrsSource* source, SrsCommonMessage* ms
         
         
         srs_info("ignore AMF0/AMF3 data message.");
         srs_info("ignore AMF0/AMF3 data message.");
         return ret;
         return ret;
+#else
+        if ((ret = source->on_data(msg)) != ERROR_SUCCESS) {
+                       srs_error("source process onMetaData message failed. ret=%d", ret);
+                       return ret;
+                   }
+                   srs_info("process onMetaData message success.");
+                   return ret;
+#endif
     }
     }
     
     
     return ret;
     return ret;

@@ -1579,6 +1579,33 @@ int SrsSource::on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata
     return ret;
     return ret;
 }
 }
 
 
+int SrsSource::on_data(SrsCommonMessage* shared_data)
+{
+          int ret = ERROR_SUCCESS;
+           last_packet_time = shared_data->header.timestamp;
+
+           // convert shared_audio to msg, user should not use shared_audio again.
+           // the payload is transfer to msg, and set to NULL in shared_audio.
+           SrsSharedPtrMessage msg;
+           if ((ret = msg.create(shared_data)) != ERROR_SUCCESS) {
+               srs_error("initialize the audio failed. ret=%d", ret);
+               return ret;
+           }
+#ifdef SRS_AUTO_DVR
+    if ((ret = dvr->on_data(&msg)) != ERROR_SUCCESS) {
+        srs_warn("dvr process data message failed, ignore and disable dvr. ret=%d", ret);
+
+        // unpublish, ignore ret.
+        dvr->on_unpublish();
+
+        // ignore.
+        ret = ERROR_SUCCESS;
+    }
+#endif
+
+           return ret;
+}
+
 int SrsSource::on_audio(SrsCommonMessage* shared_audio)
 int SrsSource::on_audio(SrsCommonMessage* shared_audio)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;

@@ -555,6 +555,8 @@ public:
     virtual bool can_publish(bool is_edge);
     virtual bool can_publish(bool is_edge);
     virtual int on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);
     virtual int on_meta_data(SrsCommonMessage* msg, SrsOnMetaDataPacket* metadata);
 public:
 public:
+    virtual int on_data(SrsCommonMessage* data);
+public:
     virtual int on_audio(SrsCommonMessage* audio);
     virtual int on_audio(SrsCommonMessage* audio);
 private:
 private:
     virtual int on_audio_imp(SrsSharedPtrMessage* audio);
     virtual int on_audio_imp(SrsSharedPtrMessage* audio);

@@ -416,13 +416,13 @@ int SrsFlvEncoder::write_header(char flv_header[9])
     return ret;
     return ret;
 }
 }
 
 
-int SrsFlvEncoder::write_metadata(char type, char* data, int size)
+int SrsFlvEncoder::write_metadata(int64_t timestamp, char* data, int size)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
     srs_assert(data);
     srs_assert(data);
     
     
-    if ((ret = write_metadata_to_cache(type, data, size, tag_header)) != ERROR_SUCCESS) {
+    if ((ret = write_metadata_to_cache(timestamp, data, size, tag_header)) != ERROR_SUCCESS) {
         return ret;
         return ret;
     }
     }
     
     
@@ -528,7 +528,7 @@ int SrsFlvEncoder::write_tags(SrsSharedPtrMessage** msgs, int count)
                 return ret;
                 return ret;
             }
             }
         } else {
         } else {
-            if ((ret = write_metadata_to_cache(SrsCodecFlvTagScript, msg->payload, msg->size, cache)) != ERROR_SUCCESS) {
+            if ((ret = write_metadata_to_cache(msg->timestamp, msg->payload, msg->size, cache)) != ERROR_SUCCESS) {
                 return ret;
                 return ret;
             }
             }
         }
         }
@@ -563,7 +563,7 @@ int SrsFlvEncoder::write_tags(SrsSharedPtrMessage** msgs, int count)
 }
 }
 #endif
 #endif
 
 
-int SrsFlvEncoder::write_metadata_to_cache(char type, char* data, int size, char* cache)
+int SrsFlvEncoder::write_metadata_to_cache(int64_t timestamp, char* data, int size, char* cache)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
@@ -582,10 +582,11 @@ int SrsFlvEncoder::write_metadata_to_cache(char type, char* data, int size, char
     if ((ret = tag_stream->initialize(cache, 11)) != ERROR_SUCCESS) {
     if ((ret = tag_stream->initialize(cache, 11)) != ERROR_SUCCESS) {
         return ret;
         return ret;
     }
     }
-    tag_stream->write_1bytes(type);
+    tag_stream->write_1bytes(SrsCodecFlvTagScript);
     tag_stream->write_3bytes(size);
     tag_stream->write_3bytes(size);
-    tag_stream->write_3bytes(0x00);
-    tag_stream->write_1bytes(0x00);
+    tag_stream->write_3bytes((int32_t)timestamp);
+    // default to little-endian
+    tag_stream->write_1bytes((timestamp >> 24) & 0xFF);
     tag_stream->write_3bytes(0x00);
     tag_stream->write_3bytes(0x00);
     
     
     return ret;
     return ret;

@@ -460,14 +460,13 @@ public:
     virtual int write_header(char flv_header[9]);
     virtual int write_header(char flv_header[9]);
     /**
     /**
     * write flv metadata. 
     * write flv metadata. 
-    * @param type, the type of data, or other message type.
-    *       @see SrsCodecFlvTag
+    * @param timestamp
     * @param data, the amf0 metadata which serialize from:
     * @param data, the amf0 metadata which serialize from:
     *   AMF0 string: onMetaData,
     *   AMF0 string: onMetaData,
     *   AMF0 object: the metadata object.
     *   AMF0 object: the metadata object.
     * @remark assert data is not NULL.
     * @remark assert data is not NULL.
     */
     */
-    virtual int write_metadata(char type, char* data, int size);
+    virtual int write_metadata(int64_t timestamp, char* data, int size);
     /**
     /**
     * write audio/video packet.
     * write audio/video packet.
     * @remark assert data is not NULL.
     * @remark assert data is not NULL.
@@ -499,7 +498,7 @@ public:
     virtual int write_tags(SrsSharedPtrMessage** msgs, int count);
     virtual int write_tags(SrsSharedPtrMessage** msgs, int count);
 #endif
 #endif
 private:
 private:
-    virtual int write_metadata_to_cache(char type, char* data, int size, char* cache);
+    virtual int write_metadata_to_cache(int64_t timestamp, char* data, int size, char* cache);
     virtual int write_audio_to_cache(int64_t timestamp, char* data, int size, char* cache);
     virtual int write_audio_to_cache(int64_t timestamp, char* data, int size, char* cache);
     virtual int write_video_to_cache(int64_t timestamp, char* data, int size, char* cache);
     virtual int write_video_to_cache(int64_t timestamp, char* data, int size, char* cache);
     virtual int write_pts_to_cache(int size, char* cache);
     virtual int write_pts_to_cache(int size, char* cache);

@@ -1659,7 +1659,7 @@ int srs_flv_write_tag(srs_flv_t flv, char type, int32_t time, char* data, int si
     } else if (type == SRS_RTMP_TYPE_VIDEO) {
     } else if (type == SRS_RTMP_TYPE_VIDEO) {
         return context->enc.write_video(time, data, size);
         return context->enc.write_video(time, data, size);
     } else {
     } else {
-        return context->enc.write_metadata(type, data, size);
+        return context->enc.write_metadata(time, data, size);
     }
     }
 
 
     return ret;
     return ret;

@@ -691,9 +691,25 @@ int SrsProtocol::do_decode_message(SrsMessageHeader& header, SrsStream* stream,
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
     
     
     SrsPacket* packet = NULL;
     SrsPacket* packet = NULL;
-    
+	if (header.is_amf0_data() || header.is_amf3_data()) {
+		if (header.is_amf3_data() && stream->require(1)) {
+			srs_verbose("skip 1bytes to decode DATA3 ");
+			stream->skip(1);
+		}
+		std::string name;
+		if ((ret = srs_amf0_read_string(stream, name)) != ERROR_SUCCESS) {
+			srs_error(
+					"decode AMF0/AMF3 script name failed. message type:%d ret=%d",
+					header.message_type, ret);
+			return ret;
+		}
+		srs_verbose("do docode script data:%s", name);
+		*ppacket = packet = new SrsOnMetaDataPacket();
+		return ret;
+	}
+
     // decode specified packet type
     // decode specified packet type
-    if (header.is_amf0_command() || header.is_amf3_command() || header.is_amf0_data() || header.is_amf3_data()) {
+    if (header.is_amf0_command() || header.is_amf3_command()) {
         srs_verbose("start to decode AMF0/AMF3 command message.");
         srs_verbose("start to decode AMF0/AMF3 command message.");
         
         
         // skip 1bytes to decode the amf3 command.
         // skip 1bytes to decode the amf3 command.

@@ -460,7 +460,7 @@ VOID TEST(KernelFlvTest, FlvEncoderWriteMetadata)
     };
     };
     char pts[] = { (char)0x00, (char)0x00, (char)0x00, (char)19 };
     char pts[] = { (char)0x00, (char)0x00, (char)0x00, (char)19 };
     
     
-    ASSERT_TRUE(ERROR_SUCCESS == enc.write_metadata(18, md, 8));
+    ASSERT_TRUE(ERROR_SUCCESS == enc.write_metadata(0, md, 8));
     ASSERT_TRUE(11 + 8 + 4 == fs.offset);
     ASSERT_TRUE(11 + 8 + 4 == fs.offset);
     
     
     EXPECT_TRUE(srs_bytes_equals(tag_header, fs.data, 11));
     EXPECT_TRUE(srs_bytes_equals(tag_header, fs.data, 11));