for ossrs#133, decode the h264 NALU from rtp.

trunk/src/app/srs_app_rtsp.cpp

@@ -33,19 +33,24 @@ using namespace std;
 #include <srs_kernel_log.hpp>
 #include <srs_app_utility.hpp>
 #include <srs_core_autofree.hpp>
+#include <srs_kernel_stream.hpp>
+#include <srs_kernel_buffer.hpp>
 
 #ifdef SRS_AUTO_STREAM_CASTER
 
-SrsRtpConn::SrsRtpConn(SrsRtspConn* r, int p)
+SrsRtpConn::SrsRtpConn(SrsRtspConn* r, int p, int sid)
 {
     rtsp = r;
     _port = p;
+    stream_id = sid;
     listener = new SrsUdpListener(this, p);
+    cache = new SrsRtpPacket();
 }
 
 SrsRtpConn::~SrsRtpConn()
 {
     srs_freep(listener);
+    srs_freep(cache);
 }
 
 int SrsRtpConn::port()
@@ -61,6 +66,53 @@ int SrsRtpConn::listen()
 int SrsRtpConn::on_udp_packet(sockaddr_in* from, char* buf, int nb_buf)
 {
     int ret = ERROR_SUCCESS;
+
+    if (true) {
+        SrsStream stream;
+
+        if ((ret = stream.initialize(buf, nb_buf)) != ERROR_SUCCESS) {
+            return ret;
+        }
+
+        SrsRtpPacket pkt;
+        if ((ret = pkt.decode(&stream)) != ERROR_SUCCESS) {
+            srs_error("rtsp: decode rtp packet failed. ret=%d", ret);
+            return ret;
+        }
+
+        if (pkt.chunked) {
+            if (!cache) {
+                cache = new SrsRtpPacket();
+            }
+            cache->copy(&pkt);
+            cache->payload->append(pkt.payload->bytes(), pkt.payload->length());
+            if (!cache->completed) {
+                srs_trace("rtsp: rtp chunked %dB, vt=%d/%u, sts=%u/%#x/%#x, paylod=%dB", 
+                    nb_buf, cache->version, cache->payload_type, cache->sequence_number, cache->timestamp, cache->ssrc, 
+                    cache->payload->length()
+                );
+                return ret;
+            }
+        } else {
+            srs_freep(cache);
+            cache = new SrsRtpPacket();
+            cache->reap(&pkt);
+        }
+    }
+
+    srs_trace("rtsp: rtp %dB, vt=%d/%u, sts=%u/%#x/%#x, paylod=%dB, chunked=%d", 
+        nb_buf, cache->version, cache->payload_type, cache->sequence_number, cache->timestamp, cache->ssrc, 
+        cache->payload->length(), cache->chunked
+    );
+
+    // always free it.
+    SrsAutoFree(SrsRtpPacket, cache);
+
+    if ((ret = rtsp->on_rtp_packet(cache)) != ERROR_SUCCESS) {
+        srs_error("rtsp: process rtp packet failed. ret=%d", ret);
+        return ret;
+    }
+
     return ret;
 }
 
@@ -162,10 +214,10 @@ int SrsRtspConn::do_cycle()
             SrsRtpConn* rtp = NULL;
             if (req->stream_id == video_id) {
                 srs_freep(video_rtp);
-                rtp = video_rtp = new SrsRtpConn(this, lpm);
+                rtp = video_rtp = new SrsRtpConn(this, lpm, video_id);
             } else {
                 srs_freep(audio_rtp);
-                rtp = audio_rtp = new SrsRtpConn(this, lpm);
+                rtp = audio_rtp = new SrsRtpConn(this, lpm, audio_id);
             }
             if ((ret = rtp->listen()) != ERROR_SUCCESS) {
                 srs_error("rtsp: rtp listen at port=%d failed. ret=%d", lpm, ret);
@@ -210,6 +262,12 @@ int SrsRtspConn::do_cycle()
     return ret;
 }
 
+int SrsRtspConn::on_rtp_packet(SrsRtpPacket* pkt)
+{
+    int ret = ERROR_SUCCESS;
+    return ret;
+}
+
 int SrsRtspConn::cycle()
 {
     // serve the rtsp client.

trunk/src/app/srs_app_rtsp.hpp

@@ -45,6 +45,7 @@ class SrsRtspConn;
 class SrsRtspStack;
 class SrsRtspCaster;
 class SrsConfDirective;
+class SrsRtpPacket;
 
 /**
 * a rtp connection which transport a stream.
@@ -54,9 +55,11 @@ class SrsRtpConn: public ISrsUdpHandler
 private:
     SrsUdpListener* listener;
     SrsRtspConn* rtsp;
+    SrsRtpPacket* cache;
+    int stream_id;
     int _port;
 public:
-    SrsRtpConn(SrsRtspConn* r, int p);
+    SrsRtpConn(SrsRtspConn* r, int p, int sid);
     virtual ~SrsRtpConn();
 public:
     virtual int port();
@@ -103,6 +106,9 @@ class SrsRtspConn : public ISrsThreadHandler
     virtual int serve();
 private:
     virtual int do_cycle();
+// internal methods
+public:
+    virtual int on_rtp_packet(SrsRtpPacket* pkt);
 // interface ISrsThreadHandler
 public:
     virtual int cycle();

trunk/src/kernel/srs_kernel_error.hpp

@@ -144,6 +144,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #define ERROR_RTMP_MIC_CACHE_OVERFLOW       2041
 #define ERROR_RTSP_TOKEN_NOT_NORMAL         2042
 #define ERROR_RTSP_REQUEST_HEADER_EOF       2043
+#define ERROR_RTP_HEADER_CORRUPT            2044
 //                                           
 // system control message, 
 // not an error, but special control logic.

trunk/src/protocol/srs_rtsp_stack.cpp

@@ -34,6 +34,7 @@ using namespace std;
 #include <srs_kernel_consts.hpp>
 #include <srs_core_autofree.hpp>
 #include <srs_kernel_utility.hpp>
+#include <srs_kernel_stream.hpp>
 
 #ifdef SRS_AUTO_STREAM_CASTER
 
@@ -118,6 +119,121 @@ std::string srs_generate_rtsp_method_str(SrsRtspMethod method)
     }
 }
 
+SrsRtpPacket::SrsRtpPacket()
+{
+    version = 2;
+    padding = 0;
+    extension = 0;
+    csrc_count = 0;
+    marker = 1;
+
+    payload_type = 0;
+    sequence_number = 0;
+    timestamp = 0;
+    ssrc = 0;
+
+    payload = new SrsSimpleBuffer();
+    chunked = false;
+    completed = false;
+}
+
+SrsRtpPacket::~SrsRtpPacket()
+{
+    srs_freep(payload);
+}
+
+void SrsRtpPacket::copy(SrsRtpPacket* src)
+{
+    version = src->version;
+    padding = src->padding;
+    extension = src->extension;
+    csrc_count = src->csrc_count;
+    marker = src->marker;
+    payload_type = src->payload_type;
+    sequence_number = src->sequence_number;
+    timestamp = src->timestamp;
+    ssrc = src->ssrc;
+
+    chunked = src->chunked;
+    completed = src->completed;
+}
+
+void SrsRtpPacket::reap(SrsRtpPacket* src)
+{
+    copy(src);
+
+    payload = src->payload;
+    src->payload = NULL;
+}
+
+int SrsRtpPacket::decode(SrsStream* stream)
+{
+    int ret = ERROR_SUCCESS;
+
+    // 12bytes header, atleast 2bytes content.
+    if (!stream->require(14)) {
+        ret = ERROR_RTP_HEADER_CORRUPT;
+        srs_error("rtsp: rtp header corrupt. ret=%d", ret);
+        return ret;
+    }
+
+    int8_t vv = stream->read_1bytes();
+    version = (vv >> 6) & 0x03;
+    padding = (vv >> 5) & 0x01;
+    extension = (vv >> 4) & 0x01;
+    csrc_count = vv & 0x0f;
+
+    int8_t mv = stream->read_1bytes();
+    marker = (mv >> 7) & 0x01;
+    payload_type = mv & 0x7f;
+
+    sequence_number = stream->read_2bytes();
+    timestamp = stream->read_4bytes();
+    ssrc = stream->read_4bytes();
+
+    // frame type
+    // 0... .... reserverd
+    // .11. .... NALU[0]&0x60
+    // ...1 11.. FU indicator
+    // .... ..00 reserverd
+    int8_t ftv = stream->read_1bytes();
+    int8_t nalu_0x60 = ftv & 0x60;
+    int8_t fu_indicator = ftv & 0x1c;
+
+    // nri, whatever
+    // 10.. .... first chunk.
+    // 00.. .... continous chunk.
+    // 01.. .... last chunk.
+    // ...1 1111 NALU[0]&0x1f
+    int8_t nriv = stream->read_1bytes();
+    bool first_chunk = (nriv & 0xC0) == 0x80;
+    bool last_chunk = (nriv & 0xC0) == 0x40;
+    bool contious_chunk = (nriv & 0xC0) == 0x00;
+    int8_t nalu_0x1f = nriv & 0x1f;
+
+    // chunked, generate the first byte NALU.
+    if (fu_indicator == 0x1c && (first_chunk || last_chunk || contious_chunk)) {
+        chunked = true;
+        completed = last_chunk;
+
+        // generate and append the first byte NALU.
+        if (first_chunk) {
+            int8_t nalu_byte0 = nalu_0x60 | nalu_0x1f;
+            payload->append((char*)&nalu_byte0, 1);
+        }
+
+        payload->append(stream->data() + stream->pos(), stream->size() - stream->pos());
+        return ret;
+    }
+
+    // no chunked, append to payload.
+    stream->skip(-2);
+    payload->append(stream->data() + stream->pos(), stream->size() - stream->pos());
+    completed = true;
+
+    return ret;
+}
+
 SrsRtspSdp::SrsRtspSdp()
 {
     state = SrsRtspSdpStateOthers;