for ossrs#250, merge avc to codec. use queue to dequeue.

Author: winlinvip

trunk/configure

@@ -366,7 +366,7 @@ ModuleLibIncs=(${SRS_OBJS_DIR})
 MODULE_FILES=("srs_kernel_error" "srs_kernel_log" "srs_kernel_stream"
         "srs_kernel_utility" "srs_kernel_flv" "srs_kernel_codec" "srs_kernel_file" 
         "srs_kernel_consts" "srs_kernel_aac" "srs_kernel_mp3" "srs_kernel_ts"
-        "srs_kernel_avc" "srs_kernel_buffer")
+        "srs_kernel_buffer")
 KERNEL_INCS="src/kernel"; MODULE_DIR=${KERNEL_INCS} . auto/modules.sh
 KERNEL_OBJS="${MODULE_OBJS[@]}"
 #

trunk/ide/srs_upp/srs_upp.upp

@@ -20,8 +20,6 @@ file
 	kernel readonly separator,
 	../../src/kernel/srs_kernel_aac.hpp,
 	../../src/kernel/srs_kernel_aac.cpp,
-	../../src/kernel/srs_kernel_avc.hpp,
-	../../src/kernel/srs_kernel_avc.cpp,
 	../../src/kernel/srs_kernel_buffer.hpp,
 	../../src/kernel/srs_kernel_buffer.cpp,
 	../../src/kernel/srs_kernel_codec.hpp,

trunk/src/app/srs_app_hls.cpp

@@ -47,7 +47,7 @@ using namespace std;
 #include <srs_rtmp_sdk.hpp>
 #include <srs_app_pithy_print.hpp>
 #include <srs_kernel_utility.hpp>
-#include <srs_kernel_avc.hpp>
+#include <srs_kernel_codec.hpp>
 #include <srs_kernel_file.hpp>
 #include <srs_rtmp_buffer.hpp>
 #include <srs_kernel_ts.hpp>

trunk/src/app/srs_app_mpegts_udp.cpp

@@ -57,6 +57,68 @@ ISrsUdpHandler::~ISrsUdpHandler()
 {
 }
 
+SrsMpegtsQueue::SrsMpegtsQueue()
+{
+    nb_audios = nb_videos = 0;
+}
+
+SrsMpegtsQueue::~SrsMpegtsQueue()
+{
+    std::map<int64_t, SrsSharedPtrMessage*>::iterator it;
+    for (it = msgs.begin(); it != msgs.end(); ++it) {
+        SrsSharedPtrMessage* msg = it->second;
+        srs_freep(msg);
+    }
+    msgs.clear();
+}
+
+int SrsMpegtsQueue::push(SrsSharedPtrMessage* msg)
+{
+    int ret = ERROR_SUCCESS;
+
+    if (msgs.find(msg->timestamp) != msgs.end()) {
+        srs_warn("mpegts: free the msg for dts exists, dts=%"PRId64, msg->timestamp);
+        srs_freep(msg);
+        return ret;
+    }
+
+    if (msg->is_audio()) {
+        nb_audios++;
+    }
+
+    if (msg->is_video()) {
+        nb_videos++;
+    }
+
+    msgs[msg->timestamp] = msg;
+
+    return ret;
+}
+
+SrsSharedPtrMessage* SrsMpegtsQueue::dequeue()
+{
+    // got 2+ videos and audios, ok to dequeue.
+    bool av_ok = nb_videos >= 2 && nb_audios >= 2;
+    // 100 videos about 30s, while 300 audios about 30s
+    bool av_overflow = nb_videos > 100 || nb_audios > 300;
+
+    if (av_ok || av_overflow) {
+        std::map<int64_t, SrsSharedPtrMessage*>::iterator it = msgs.begin();
+        SrsSharedPtrMessage* msg = it->second;
+        msgs.erase(it);
+
+        if (msg->is_audio()) {
+            nb_audios--;
+        }
+
+        if (msg->is_video()) {
+            nb_videos--;
+        }
+    }
+
+    return NULL;
+}
+
 SrsMpegtsOverUdp::SrsMpegtsOverUdp(SrsConfDirective* c)
 {
     stream = new SrsStream();
@@ -72,6 +134,7 @@ SrsMpegtsOverUdp::SrsMpegtsOverUdp(SrsConfDirective* c)
     h264_sps_changed = false;
     h264_pps_changed = false;
     h264_sps_pps_sent = false;
+    queue = new SrsMpegtsQueue();
 }
 
 SrsMpegtsOverUdp::~SrsMpegtsOverUdp()
@@ -82,6 +145,7 @@ SrsMpegtsOverUdp::~SrsMpegtsOverUdp()
     srs_freep(stream);
     srs_freep(context);
     srs_freep(avc);
+    srs_freep(queue);
 }
 
 int SrsMpegtsOverUdp::on_udp_packet(sockaddr_in* from, char* buf, int nb_buf)
@@ -280,11 +344,14 @@ int SrsMpegtsOverUdp::on_ts_video(SrsTsMessage* msg, SrsStream* avs)
 
         // it may be return error, but we must process all packets.
         if ((ret = write_h264_raw_frame(frame, frame_size, dts, pts)) != ERROR_SUCCESS) {
-            if (ret = ERROR_H264_DROP_BEFORE_SPS_PPS) {
+            if (ret == ERROR_H264_DROP_BEFORE_SPS_PPS) {
                 continue;
             }
             return ret;
         }
+
+        // for video, drop others with same pts/dts.
+        break;
     }
 
     return ret;
@@ -399,15 +466,28 @@ int SrsMpegtsOverUdp::rtmp_write_packet(char type, u_int32_t timestamp, char* da
     SrsSharedPtrMessage* msg = NULL;
 
     if ((ret = srs_rtmp_create_msg(type, timestamp, data, size, stream_id, &msg)) != ERROR_SUCCESS) {
+        srs_error("mpegts: create shared ptr msg failed. ret=%d", ret);
         return ret;
     }
-
     srs_assert(msg);
-    
-    // send out encoded msg.
-    if ((ret = client->send_and_free_message(msg, stream_id)) != ERROR_SUCCESS) {
+
+    // push msg to queue.
+    if ((ret = queue->push(msg)) != ERROR_SUCCESS) {
+        srs_error("mpegts: push msg to queue failed. ret=%d", ret);
         return ret;
     }
+
+    // for all ready msg, dequeue and send out.
+    for (;;) {
+        if ((msg = queue->dequeue()) == NULL) {
+            break;
+        }
+    
+        // send out encoded msg.
+        if ((ret = client->send_and_free_message(msg, stream_id)) != ERROR_SUCCESS) {
+            return ret;
+        }
+    }
 
     return ret;
 }

trunk/src/app/srs_app_mpegts_udp.hpp

@@ -34,6 +34,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 
 class sockaddr_in;
 #include <string>
+#include <map>
 
 class SrsStream;
 class SrsTsContext;
@@ -43,6 +44,7 @@ class SrsRtmpClient;
 class SrsStSocket;
 class SrsRequest;
 class SrsRawH264Stream;
+class SrsSharedPtrMessage;
 
 #include <srs_app_st.hpp>
 #include <srs_kernel_ts.hpp>
@@ -68,6 +70,26 @@ class ISrsUdpHandler
     virtual int on_udp_packet(sockaddr_in* from, char* buf, int nb_buf) = 0;
 };
 
+/**
+* the queue for mpegts over udp to send packets.
+* for the aac in mpegts contains many flv packets in a pes packet,
+* we must recalc the timestamp.
+*/
+class SrsMpegtsQueue
+{
+private:
+    // key: dts, value: msg.
+    std::map<int64_t, SrsSharedPtrMessage*> msgs;
+    int nb_audios;
+    int nb_videos;
+public:
+    SrsMpegtsQueue();
+    virtual ~SrsMpegtsQueue();
+public:
+    virtual int push(SrsSharedPtrMessage* msg);
+    virtual SrsSharedPtrMessage* dequeue();
+};
+
 /**
 * the mpegts over udp stream caster.
 */
@@ -92,6 +114,7 @@ class SrsMpegtsOverUdp : virtual public ISrsTsHandler
     std::string h264_pps;
     bool h264_pps_changed;
     bool h264_sps_pps_sent;
+    SrsMpegtsQueue* queue;
 public:
     SrsMpegtsOverUdp(SrsConfDirective* c);
     virtual ~SrsMpegtsOverUdp();

trunk/src/app/srs_app_source.cpp

@@ -40,7 +40,7 @@ using namespace std;
 #include <srs_kernel_stream.hpp>
 #include <srs_app_edge.hpp>
 #include <srs_kernel_utility.hpp>
-#include <srs_kernel_avc.hpp>
+#include <srs_kernel_codec.hpp>
 #include <srs_rtmp_msg_array.hpp>
 
 #define CONST_MAX_JITTER_MS         500