fix #133, support push rtsp to srs. 2.0.120.

README.md

@@ -1,10 +1,6 @@
 #Simple-RTMP-Server
 
-<<<<<<< HEAD
 SRS/2.0，开发代号：[ZhouGuowen](https://github.com/winlinvip/simple-rtmp-server/wiki/v1_CN_Product#release20)
-=======
-SRS/1.0，开发代号：[HuKaiqun](https://github.com/winlinvip/simple-rtmp-server/wiki/v1_CN_Product#release10)
->>>>>>> 1.0release
 
 SRS定位是运营级的互联网直播服务器集群，追求更好的概念完整性和最简单实现的代码。
 
@@ -493,6 +489,8 @@ Supported operating systems and hardware:
 [#250](https://github.com/winlinvip/simple-rtmp-server/issues/250).
 1. Rewrite HLS(h.264+aac/mp3) streaming, read
 [#304](https://github.com/winlinvip/simple-rtmp-server/issues/304).
+1. Support push RTSP to SRS, read
+[#133](https://github.com/winlinvip/simple-rtmp-server/issues/133).
 1. [no-plan] Support <500ms latency, FRSC(Fast RTMP-compatible Stream Channel tech).
 1. [no-plan] Support RTMP 302 redirect [#92](https://github.com/winlinvip/simple-rtmp-server/issues/92).
 1. [no-plan] Support multiple processes, for both origin and edge
@@ -531,6 +529,7 @@ Supported operating systems and hardware:
 
 ### SRS 2.0 history
 
+* v2.0, 2015-02-18, fix [#133](https://github.com/winlinvip/simple-rtmp-server/issues/133), support push rtsp to srs. 2.0.120.
 * v2.0, 2015-02-17, the join maybe failed, should use a variable to ensure thread terminated. 2.0.119.
 * v2.0, 2015-02-15, for [#304](https://github.com/winlinvip/simple-rtmp-server/issues/304), support config default acodec/vcodec. 2.0.118.
 * v2.0, 2015-02-15, for [#304](https://github.com/winlinvip/simple-rtmp-server/issues/304), rewrite hls/ts code, support h.264+mp3 for hls. 2.0.117.

trunk/src/app/srs_app_mpegts_udp.cpp

@@ -629,7 +629,6 @@ int SrsMpegtsOverUdp::connect()
         return ret;
     }
 
-
     return ret;
 }
 

trunk/src/app/srs_app_rtsp.cpp

@@ -35,6 +35,12 @@ using namespace std;
 #include <srs_core_autofree.hpp>
 #include <srs_kernel_stream.hpp>
 #include <srs_kernel_buffer.hpp>
+#include <srs_rtmp_sdk.hpp>
+#include <srs_rtmp_amf0.hpp>
+#include <srs_rtmp_utility.hpp>
+#include <srs_kernel_utility.hpp>
+#include <srs_raw_avc.hpp>
+#include <srs_kernel_codec.hpp>
 
 #ifdef SRS_AUTO_STREAM_CASTER
 
@@ -100,15 +106,15 @@ int SrsRtpConn::on_udp_packet(sockaddr_in* from, char* buf, int nb_buf)
         }
     }
 
-    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, 
+    srs_trace("rtsp: rtp #%d %dB, vt=%d/%u, sts=%u/%u/%#x, paylod=%dB, chunked=%d", 
+        stream_id, 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) {
+    if ((ret = rtsp->on_rtp_packet(cache, stream_id)) != ERROR_SUCCESS) {
         srs_error("rtsp: process rtp packet failed. ret=%d", ret);
         return ret;
     }
@@ -116,9 +122,59 @@ int SrsRtpConn::on_udp_packet(sockaddr_in* from, char* buf, int nb_buf)
     return ret;
 }
 
+SrsRtspAudioCache::SrsRtspAudioCache()
+{
+    dts = NULL;
+    audio_samples = NULL;
+    payload = NULL;
+}
+
+SrsRtspAudioCache::~SrsRtspAudioCache()
+{
+    srs_freep(audio_samples);
+    srs_freep(payload);
+}
+
+SrsRtspJitter::SrsRtspJitter()
+{
+    delta = 0;
+    previous_timestamp = 0;
+    pts = 0;
+}
+
+SrsRtspJitter::~SrsRtspJitter()
+{
+}
+
+int64_t SrsRtspJitter::timestamp()
+{
+    return pts;
+}
+
+int SrsRtspJitter::correct(int64_t& ts)
+{
+    int ret = ERROR_SUCCESS;
+
+    if (previous_timestamp == 0) {
+        previous_timestamp = ts;
+    }
+
+    delta = srs_max(0, ts - previous_timestamp);
+    if (delta > 90000) {
+        delta = 0;
+    }
+
+    previous_timestamp = ts;
+
+    ts = pts + delta;
+    pts = ts;    
+
+    return ret;
+}
+
 SrsRtspConn::SrsRtspConn(SrsRtspCaster* c, st_netfd_t fd, std::string o)
 {
-    output = o;
+    output_template = o;
 
     session = "";
     video_rtp = NULL;
@@ -129,6 +185,18 @@ SrsRtspConn::SrsRtspConn(SrsRtspCaster* c, st_netfd_t fd, std::string o)
     skt = new SrsStSocket(fd);
     rtsp = new SrsRtspStack(skt);
     trd = new SrsThread("rtsp", this, 0, false);
+
+    req = NULL;
+    io = NULL;
+    client = NULL;
+    stream_id = 0;
+    vjitter = new SrsRtspJitter();
+    ajitter = new SrsRtspJitter();
+
+    avc = new SrsRawH264Stream();
+    aac = new SrsRawAacStream();
+    acodec = new SrsRawAacStreamCodec();
+    acache = new SrsRtspAudioCache();
 }
 
 SrsRtspConn::~SrsRtspConn()
@@ -142,6 +210,13 @@ SrsRtspConn::~SrsRtspConn()
     srs_freep(trd);
     srs_freep(skt);
     srs_freep(rtsp);
+    
+    close();
+
+    srs_freep(vjitter);
+    srs_freep(ajitter);
+    srs_freep(acodec);
+    srs_freep(acache);
 }
 
 int SrsRtspConn::serve()
@@ -179,6 +254,18 @@ int SrsRtspConn::do_cycle()
                 return ret;
             }
         } else if (req->is_announce()) {
+            if (rtsp_tcUrl.empty()) {
+                rtsp_tcUrl = req->uri;
+            }
+            size_t pos = string::npos;
+            if ((pos = rtsp_tcUrl.rfind(".sdp")) != string::npos) {
+                rtsp_tcUrl = rtsp_tcUrl.substr(0, pos);
+            }
+            if ((pos = rtsp_tcUrl.rfind("/")) != string::npos) {
+                rtsp_stream = rtsp_tcUrl.substr(pos + 1);
+                rtsp_tcUrl = rtsp_tcUrl.substr(0, pos);
+            }
+
             srs_assert(req->sdp);
             video_id = ::atoi(req->sdp->video_stream_id.c_str());
             audio_id = ::atoi(req->sdp->audio_stream_id.c_str());
@@ -186,13 +273,13 @@ int SrsRtspConn::do_cycle()
             audio_codec = req->sdp->audio_codec;
             audio_sample_rate = ::atoi(req->sdp->audio_sample_rate.c_str());
             audio_channel = ::atoi(req->sdp->audio_channel.c_str());
-            sps = req->sdp->video_sps;
-            pps = req->sdp->video_pps;
-            asc = req->sdp->audio_sh;
-            srs_trace("rtsp: video(#%d, %s, %s/%s), audio(#%d, %s, %s/%s, %dHZ %dchannels)", 
+            h264_sps = req->sdp->video_sps;
+            h264_pps = req->sdp->video_pps;
+            aac_specific_config = req->sdp->audio_sh;
+            srs_trace("rtsp: video(#%d, %s, %s/%s), audio(#%d, %s, %s/%s, %dHZ %dchannels), %s/%s", 
                 video_id, video_codec.c_str(), req->sdp->video_protocol.c_str(), req->sdp->video_transport_format.c_str(), 
                 audio_id, audio_codec.c_str(), req->sdp->audio_protocol.c_str(), req->sdp->audio_transport_format.c_str(),
-                audio_sample_rate, audio_channel
+                audio_sample_rate, audio_channel, rtsp_tcUrl.c_str(), rtsp_stream.c_str()
             );
 
             SrsRtspResponse* res = new SrsRtspResponse(req->seq);
@@ -262,9 +349,38 @@ int SrsRtspConn::do_cycle()
     return ret;
 }
 
-int SrsRtspConn::on_rtp_packet(SrsRtpPacket* pkt)
+int SrsRtspConn::on_rtp_packet(SrsRtpPacket* pkt, int stream_id)
 {
     int ret = ERROR_SUCCESS;
+
+    // ensure rtmp connected.
+    if ((ret = connect()) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+    if (stream_id == video_id) {
+        // rtsp tbn is ts tbn.
+        int64_t pts = pkt->timestamp;
+        if ((ret = vjitter->correct(pts)) != ERROR_SUCCESS) {
+            srs_error("rtsp: correct by jitter failed. ret=%d", ret);
+            return ret;
+        }
+
+        // TODO: FIXME: set dts to pts, please finger out the right dts.
+        int64_t dts = pts;
+
+        return on_rtp_video(pkt, dts, pts);
+    } else {
+        // rtsp tbn is ts tbn.
+        int64_t pts = pkt->timestamp;
+        if ((ret = ajitter->correct(pts)) != ERROR_SUCCESS) {
+            srs_error("rtsp: correct by jitter failed. ret=%d", ret);
+            return ret;
+        }
+
+        return on_rtp_audio(pkt, pts);
+    }
+
     return ret;
 }
 
@@ -307,6 +423,336 @@ void SrsRtspConn::on_thread_stop()
     caster->remove(this);
 }
 
+int SrsRtspConn::on_rtp_video(SrsRtpPacket* pkt, int64_t dts, int64_t pts)
+{
+    int ret = ERROR_SUCCESS;
+
+    if ((ret = kickoff_audio_cache(pkt, dts)) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+    if ((ret = write_h264_ipb_frame(pkt->payload->bytes(), pkt->payload->length(), dts / 90, pts / 90)) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+    return ret;
+}
+
+int SrsRtspConn::on_rtp_audio(SrsRtpPacket* pkt, int64_t dts)
+{
+    int ret = ERROR_SUCCESS;
+
+    if ((ret = kickoff_audio_cache(pkt, dts)) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+    // cache current audio to kickoff.
+    acache->dts = dts;
+    acache->audio_samples = pkt->audio_samples;
+    acache->payload = pkt->payload;
+
+    pkt->audio_samples = NULL;
+    pkt->payload = NULL;
+
+    return ret;
+}
+
+int SrsRtspConn::kickoff_audio_cache(SrsRtpPacket* pkt, int64_t dts)
+{
+    int ret = ERROR_SUCCESS;
+
+    // nothing to kick off.
+    if (!acache->payload) {
+        return ret;
+    }
+
+    if (dts - acache->dts > 0 && acache->audio_samples->nb_sample_units > 0) {
+        int64_t delta = (dts - acache->dts) / acache->audio_samples->nb_sample_units;
+        for (int i = 0; i < acache->audio_samples->nb_sample_units; i++) {
+            char* frame = acache->audio_samples->sample_units[i].bytes;
+            int nb_frame = acache->audio_samples->sample_units[i].size;
+            int64_t timestamp = (acache->dts + delta * i) / 90;
+            acodec->aac_packet_type = 1;
+            if ((ret = write_audio_raw_frame(frame, nb_frame, acodec, timestamp)) != ERROR_SUCCESS) {
+                return ret;
+            }
+        }
+    }
+
+    acache->dts = 0;
+    srs_freep(acache->audio_samples);
+    srs_freep(acache->payload);
+
+    return ret;
+}
+
+int SrsRtspConn::write_sequence_header()
+{
+    int ret = ERROR_SUCCESS;
+
+    // use the current dts.
+    int64_t dts = vjitter->timestamp() / 90;
+
+    // send video sps/pps
+    if ((ret = write_h264_sps_pps(dts, dts)) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+    // generate audio sh by audio specific config.
+    if (true) {
+        std::string sh = aac_specific_config;
+
+        SrsAvcAacCodec dec;
+        if ((ret = dec.audio_aac_sequence_header_demux((char*)sh.c_str(), (int)sh.length())) != ERROR_SUCCESS) {
+            return ret;
+        }
+
+        acodec->sound_format = SrsCodecAudioAAC;
+        acodec->sound_type = (dec.aac_channels == 2)? SrsCodecAudioSoundTypeStereo : SrsCodecAudioSoundTypeMono;
+        acodec->sound_size = SrsCodecAudioSampleSize16bit;
+        acodec->aac_packet_type = 0;
+
+        static int aac_sample_rates[] = {
+            96000, 88200, 64000, 48000,
+            44100, 32000, 24000, 22050,
+            16000, 12000, 11025,  8000,
+            7350,     0,     0,    0
+        };
+        switch (aac_sample_rates[dec.aac_sample_rate]) {
+            case 11025:
+                acodec->sound_rate = SrsCodecAudioSampleRate11025;
+                break;
+            case 22050:
+                acodec->sound_rate = SrsCodecAudioSampleRate22050;
+                break;
+            case 44100:
+                acodec->sound_rate = SrsCodecAudioSampleRate44100;
+                break;
+            default:
+                break;
+        };
+
+        if ((ret = write_audio_raw_frame((char*)sh.data(), (int)sh.length(), acodec, dts)) != ERROR_SUCCESS) {
+            return ret;
+        }
+    }
+
+    return ret;
+}
+
+int SrsRtspConn::write_h264_sps_pps(u_int32_t dts, u_int32_t pts)
+{
+    int ret = ERROR_SUCCESS;
+    
+    // h264 raw to h264 packet.
+    std::string sh;
+    if ((ret = avc->mux_sequence_header(h264_sps, h264_pps, dts, pts, sh)) != ERROR_SUCCESS) {
+        return ret;
+    }
+    
+    // h264 packet to flv packet.
+    int8_t frame_type = SrsCodecVideoAVCFrameKeyFrame;
+    int8_t avc_packet_type = SrsCodecVideoAVCTypeSequenceHeader;
+    char* flv = NULL;
+    int nb_flv = 0;
+    if ((ret = avc->mux_avc2flv(sh, frame_type, avc_packet_type, dts, pts, &flv, &nb_flv)) != ERROR_SUCCESS) {
+        return ret;
+    }
+    
+    // the timestamp in rtmp message header is dts.
+    u_int32_t timestamp = dts;
+    if ((ret = rtmp_write_packet(SrsCodecFlvTagVideo, timestamp, flv, nb_flv)) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+    return ret;
+}
+
+int SrsRtspConn::write_h264_ipb_frame(char* frame, int frame_size, u_int32_t dts, u_int32_t pts) 
+{
+    int ret = ERROR_SUCCESS;
+
+    std::string ibp;
+    int8_t frame_type;
+    if ((ret = avc->mux_ipb_frame(frame, frame_size, dts, pts, ibp, frame_type)) != ERROR_SUCCESS) {
+        return ret;
+    }
+    
+    int8_t avc_packet_type = SrsCodecVideoAVCTypeNALU;
+    char* flv = NULL;
+    int nb_flv = 0;
+    if ((ret = avc->mux_avc2flv(ibp, frame_type, avc_packet_type, dts, pts, &flv, &nb_flv)) != ERROR_SUCCESS) {
+        return ret;
+    }
+    
+    // the timestamp in rtmp message header is dts.
+    u_int32_t timestamp = dts;
+    return rtmp_write_packet(SrsCodecFlvTagVideo, timestamp, flv, nb_flv);
+}
+
+int SrsRtspConn::write_audio_raw_frame(char* frame, int frame_size, SrsRawAacStreamCodec* codec, u_int32_t dts)
+{
+    int ret = ERROR_SUCCESS;
+
+    char* data = NULL;
+    int size = 0;
+    if ((ret = aac->mux_aac2flv(frame, frame_size, codec, dts, &data, &size)) != ERROR_SUCCESS) {
+        return ret;
+    }
+    
+    return rtmp_write_packet(SrsCodecFlvTagAudio, dts, data, size);
+}
+
+int SrsRtspConn::rtmp_write_packet(char type, u_int32_t timestamp, char* data, int size)
+{
+    int ret = ERROR_SUCCESS;
+    
+    SrsSharedPtrMessage* msg = NULL;
+
+    if ((ret = srs_rtmp_create_msg(type, timestamp, data, size, stream_id, &msg)) != ERROR_SUCCESS) {
+        srs_error("rtsp: 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) {
+        return ret;
+    }
+    
+    return ret;
+}
+
+// TODO: FIXME: merge all client code.
+int SrsRtspConn::connect()
+{
+    int ret = ERROR_SUCCESS;
+
+    // when ok, ignore.
+    if (io || client) {
+        return ret;
+    }
+    
+    // parse uri
+    if (!req) {
+        req = new SrsRequest();
+
+        std::string schema, host, vhost, app, port, param;
+        srs_discovery_tc_url(rtsp_tcUrl, schema, host, vhost, app, port, param);
+
+        // generate output by template.
+        std::string output = output_template;
+        output = srs_string_replace(output, "[app]", app);
+        output = srs_string_replace(output, "[stream]", rtsp_stream);
+
+        size_t pos = string::npos;
+        string uri = req->tcUrl = output;
+
+        // tcUrl, stream
+        if ((pos = uri.rfind("/")) != string::npos) {
+            req->stream = uri.substr(pos + 1);
+            req->tcUrl = uri = uri.substr(0, pos);
+        }
+    
+        srs_discovery_tc_url(req->tcUrl, 
+            req->schema, req->host, req->vhost, req->app, req->port,
+            req->param);
+    }
+
+    // connect host.
+    if ((ret = srs_socket_connect(req->host, ::atoi(req->port.c_str()), ST_UTIME_NO_TIMEOUT, &stfd)) != ERROR_SUCCESS) {
+        srs_error("rtsp: connect server %s:%s failed. ret=%d", req->host.c_str(), req->port.c_str(), ret);
+        return ret;
+    }
+    io = new SrsStSocket(stfd);
+    client = new SrsRtmpClient(io);
+
+    client->set_recv_timeout(SRS_CONSTS_RTMP_RECV_TIMEOUT_US);
+    client->set_send_timeout(SRS_CONSTS_RTMP_SEND_TIMEOUT_US);
+    
+    // connect to vhost/app
+    if ((ret = client->handshake()) != ERROR_SUCCESS) {
+        srs_error("rtsp: handshake with server failed. ret=%d", ret);
+        return ret;
+    }
+    if ((ret = connect_app(req->host, req->port)) != ERROR_SUCCESS) {
+        srs_error("rtsp: connect with server failed. ret=%d", ret);
+        return ret;
+    }
+    if ((ret = client->create_stream(stream_id)) != ERROR_SUCCESS) {
+        srs_error("rtsp: connect with server failed, stream_id=%d. ret=%d", stream_id, ret);
+        return ret;
+    }
+    
+    // publish.
+    if ((ret = client->publish(req->stream, stream_id)) != ERROR_SUCCESS) {
+        srs_error("rtsp: publish failed, stream=%s, stream_id=%d. ret=%d", 
+            req->stream.c_str(), stream_id, ret);
+        return ret;
+    }
+
+    return write_sequence_header();
+}
+
+// TODO: FIXME: refine the connect_app.
+int SrsRtspConn::connect_app(string ep_server, string ep_port)
+{
+    int ret = ERROR_SUCCESS;
+    
+    // args of request takes the srs info.
+    if (req->args == NULL) {
+        req->args = SrsAmf0Any::object();
+    }
+    
+    // notify server the edge identity,
+    // @see https://github.com/winlinvip/simple-rtmp-server/issues/147
+    SrsAmf0Object* data = req->args;
+    data->set("srs_sig", SrsAmf0Any::str(RTMP_SIG_SRS_KEY));
+    data->set("srs_server", SrsAmf0Any::str(RTMP_SIG_SRS_KEY" "RTMP_SIG_SRS_VERSION" ("RTMP_SIG_SRS_URL_SHORT")"));
+    data->set("srs_license", SrsAmf0Any::str(RTMP_SIG_SRS_LICENSE));
+    data->set("srs_role", SrsAmf0Any::str(RTMP_SIG_SRS_ROLE));
+    data->set("srs_url", SrsAmf0Any::str(RTMP_SIG_SRS_URL));
+    data->set("srs_version", SrsAmf0Any::str(RTMP_SIG_SRS_VERSION));
+    data->set("srs_site", SrsAmf0Any::str(RTMP_SIG_SRS_WEB));
+    data->set("srs_email", SrsAmf0Any::str(RTMP_SIG_SRS_EMAIL));
+    data->set("srs_copyright", SrsAmf0Any::str(RTMP_SIG_SRS_COPYRIGHT));
+    data->set("srs_primary", SrsAmf0Any::str(RTMP_SIG_SRS_PRIMARY));
+    data->set("srs_authors", SrsAmf0Any::str(RTMP_SIG_SRS_AUTHROS));
+    // for edge to directly get the id of client.
+    data->set("srs_pid", SrsAmf0Any::number(getpid()));
+    data->set("srs_id", SrsAmf0Any::number(_srs_context->get_id()));
+    
+    // local ip of edge
+    std::vector<std::string> ips = srs_get_local_ipv4_ips();
+    assert(_srs_config->get_stats_network() < (int)ips.size());
+    std::string local_ip = ips[_srs_config->get_stats_network()];
+    data->set("srs_server_ip", SrsAmf0Any::str(local_ip.c_str()));
+    
+    // generate the tcUrl
+    std::string param = "";
+    std::string tc_url = srs_generate_tc_url(ep_server, req->vhost, req->app, ep_port, param);
+    
+    // upnode server identity will show in the connect_app of client.
+    // @see https://github.com/winlinvip/simple-rtmp-server/issues/160
+    // the debug_srs_upnode is config in vhost and default to true.
+    bool debug_srs_upnode = _srs_config->get_debug_srs_upnode(req->vhost);
+    if ((ret = client->connect_app(req->app, tc_url, req, debug_srs_upnode)) != ERROR_SUCCESS) {
+        srs_error("rtsp: connect with server failed, tcUrl=%s, dsu=%d. ret=%d", 
+            tc_url.c_str(), debug_srs_upnode, ret);
+        return ret;
+    }
+    
+    return ret;
+}
+
+void SrsRtspConn::close()
+{
+    srs_freep(client);
+    srs_freep(io);
+    srs_freep(req);
+    srs_close_stfd(stfd);
+}
+
 SrsRtspCaster::SrsRtspCaster(SrsConfDirective* c)
 {
     // TODO: FIXME: support reload.

trunk/src/app/srs_app_rtsp.hpp

@@ -46,6 +46,15 @@ class SrsRtspStack;
 class SrsRtspCaster;
 class SrsConfDirective;
 class SrsRtpPacket;
+class SrsRequest;
+class SrsStSocket;
+class SrsRtmpClient;
+class SrsRawH264Stream;
+class SrsRawAacStream;
+class SrsRawAacStreamCodec;
+class SrsSharedPtrMessage;
+class SrsCodecSample;
+class SrsSimpleBuffer;
 
 /**
 * a rtp connection which transport a stream.
@@ -69,13 +78,46 @@ public:
     virtual int on_udp_packet(sockaddr_in* from, char* buf, int nb_buf);
 };
 
+/**
+* audio is group by frames.
+*/
+struct SrsRtspAudioCache
+{
+    int64_t dts;
+    SrsCodecSample* audio_samples;
+    SrsSimpleBuffer* payload;
+
+    SrsRtspAudioCache();
+    virtual ~SrsRtspAudioCache();
+};
+
+/**
+* the time jitter correct for rtsp.
+*/
+class SrsRtspJitter
+{
+private:
+    int64_t previous_timestamp;
+    int64_t pts;
+    int delta;
+public:
+    SrsRtspJitter();
+    virtual ~SrsRtspJitter();
+public:
+    virtual int64_t timestamp();
+    virtual int correct(int64_t& ts);
+};
+
 /**
 * the rtsp connection serve the fd.
 */
 class SrsRtspConn : public ISrsThreadHandler
 {
 private:
-    std::string output;
+    std::string output_template;
+    std::string rtsp_tcUrl;
+    std::string rtsp_stream;
+
 private:
     std::string session;
     // video stream.
@@ -88,17 +130,28 @@ private:
     int audio_sample_rate;
     int audio_channel;
     SrsRtpConn* audio_rtp;
-    // video sequence header.
-    std::string sps;
-    std::string pps;
-    // audio sequence header.
-    std::string asc;
 private:
     st_netfd_t stfd;
     SrsStSocket* skt;
     SrsRtspStack* rtsp;
     SrsRtspCaster* caster;
     SrsThread* trd;
+private:
+    SrsRequest* req;
+    SrsStSocket* io;
+    SrsRtmpClient* client;
+    SrsRtspJitter* vjitter;
+    SrsRtspJitter* ajitter;
+    int stream_id;
+private:
+    SrsRawH264Stream* avc;
+    std::string h264_sps;
+    std::string h264_pps;
+private:
+    SrsRawAacStream* aac;
+    SrsRawAacStreamCodec* acodec;
+    std::string aac_specific_config;
+    SrsRtspAudioCache* acache;
 public:
     SrsRtspConn(SrsRtspCaster* c, st_netfd_t fd, std::string o);
     virtual ~SrsRtspConn();
@@ -108,11 +161,28 @@ private:
     virtual int do_cycle();
 // internal methods
 public:
-    virtual int on_rtp_packet(SrsRtpPacket* pkt);
+    virtual int on_rtp_packet(SrsRtpPacket* pkt, int stream_id);
 // interface ISrsThreadHandler
 public:
     virtual int cycle();
     virtual void on_thread_stop();
+private:
+    virtual int on_rtp_video(SrsRtpPacket* pkt, int64_t dts, int64_t pts);
+    virtual int on_rtp_audio(SrsRtpPacket* pkt, int64_t dts);
+    virtual int kickoff_audio_cache(SrsRtpPacket* pkt, int64_t dts);
+private:
+    virtual int write_sequence_header();
+    virtual int write_h264_sps_pps(u_int32_t dts, u_int32_t pts);
+    virtual int write_h264_ipb_frame(char* frame, int frame_size, u_int32_t dts, u_int32_t pts);
+    virtual int write_audio_raw_frame(char* frame, int frame_size, SrsRawAacStreamCodec* codec, u_int32_t dts);
+    virtual int rtmp_write_packet(char type, u_int32_t timestamp, char* data, int size);
+private:
+    // connect to rtmp output url. 
+    // @remark ignore when not connected, reconnect when disconnected.
+    virtual int connect();
+    virtual int connect_app(std::string ep_server, std::string ep_port);
+    // close the connected io and rtmp to ready to be re-connect.
+    virtual void close();
 };
 
 /**

trunk/src/core/srs_core.hpp

@@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 // current release version
 #define VERSION_MAJOR       2
 #define VERSION_MINOR       0
-#define VERSION_REVISION    119
+#define VERSION_REVISION    120
 
 // server info.
 #define RTMP_SIG_SRS_KEY "SRS"

trunk/src/kernel/srs_kernel_codec.cpp

@@ -279,58 +279,12 @@ int SrsAvcAacCodec::audio_aac_demux(char* data, int size, SrsCodecSample* sample
             srs_freep(aac_extra_data);
             aac_extra_data = new char[aac_extra_size];
             memcpy(aac_extra_data, stream->data() + stream->pos(), aac_extra_size);
-        }
-        
-        // only need to decode the first 2bytes:
-        //      audioObjectType, aac_profile, 5bits.
-        //      samplingFrequencyIndex, aac_sample_rate, 4bits.
-        //      channelConfiguration, aac_channels, 4bits
-        if (!stream->require(2)) {
-            ret = ERROR_HLS_DECODE_ERROR;
-            srs_error("audio codec decode aac sequence header failed. ret=%d", ret);
-            return ret;
-        }
-        u_int8_t profile_ObjectType = stream->read_1bytes();
-        u_int8_t samplingFrequencyIndex = stream->read_1bytes();
-        
-        aac_channels = (samplingFrequencyIndex >> 3) & 0x0f;
-        samplingFrequencyIndex = ((profile_ObjectType << 1) & 0x0e) | ((samplingFrequencyIndex >> 7) & 0x01);
-        profile_ObjectType = (profile_ObjectType >> 3) & 0x1f;
-
-        // set the aac sample rate.
-        aac_sample_rate = samplingFrequencyIndex;
 
-        // the profile = object_id + 1
-        // @see aac-mp4a-format-ISO_IEC_14496-3+2001.pdf, page 78,
-        //      Table 1. A.9 ¨C MPEG-2 Audio profiles and MPEG-4 Audio object types
-        aac_profile = profile_ObjectType + 1;
-        
-        // the valid aac profile:
-        //      MPEG-2 profile
-        //      Main profile (ID == 1)
-        //      Low Complexity profile (LC) (ID == 2)
-        //      Scalable Sampling Rate profile (SSR) (ID == 3)
-        //      (reserved) (ID == 4)
-        // @see aac-mp4a-format-ISO_IEC_14496-3+2001.pdf, page 78,
-        //      Table 1. A.9 ¨C MPEG-2 Audio profiles and MPEG-4 Audio object types
-        if (aac_profile > 4) {
-            ret = ERROR_HLS_DECODE_ERROR;
-            srs_error("audio codec decode aac sequence header failed, "
-                "adts object=%d invalid. ret=%d", profile_ObjectType, ret);
-            return ret;
+            // demux the sequence header.
+            if ((ret = audio_aac_sequence_header_demux(aac_extra_data, aac_extra_size)) != ERROR_SUCCESS) {
+                return ret;
+            }
         }
-        
-        // TODO: FIXME: to support aac he/he-v2, see: ngx_rtmp_codec_parse_aac_header
-        // @see: https://github.com/winlinvip/nginx-rtmp-module/commit/3a5f9eea78fc8d11e8be922aea9ac349b9dcbfc2
-        // 
-        // donot force to LC, @see: https://github.com/winlinvip/simple-rtmp-server/issues/81
-        // the source will print the sequence header info.
-        //if (aac_profile > 3) {
-            // Mark all extended profiles as LC
-            // to make Android as happy as possible.
-            // @see: ngx_rtmp_hls_parse_aac_header
-            //aac_profile = 1;
-        //}
     } else if (aac_packet_type == SrsCodecAudioTypeRawData) {
         // ensure the sequence header demuxed
         if (aac_extra_size <= 0 || !aac_extra_data) {
@@ -403,6 +357,68 @@ int SrsAvcAacCodec::audio_mp3_demux(char* data, int size, SrsCodecSample* sample
     return ret;
 }
 
+int SrsAvcAacCodec::audio_aac_sequence_header_demux(char* data, int size)
+{
+    int ret = ERROR_SUCCESS;
+    
+    if ((ret = stream->initialize(data, size)) != ERROR_SUCCESS) {
+        return ret;
+    }
+        
+    // only need to decode the first 2bytes:
+    //      audioObjectType, aac_profile, 5bits.
+    //      samplingFrequencyIndex, aac_sample_rate, 4bits.
+    //      channelConfiguration, aac_channels, 4bits
+    if (!stream->require(2)) {
+        ret = ERROR_HLS_DECODE_ERROR;
+        srs_error("audio codec decode aac sequence header failed. ret=%d", ret);
+        return ret;
+    }
+    u_int8_t profile_ObjectType = stream->read_1bytes();
+    u_int8_t samplingFrequencyIndex = stream->read_1bytes();
+        
+    aac_channels = (samplingFrequencyIndex >> 3) & 0x0f;
+    samplingFrequencyIndex = ((profile_ObjectType << 1) & 0x0e) | ((samplingFrequencyIndex >> 7) & 0x01);
+    profile_ObjectType = (profile_ObjectType >> 3) & 0x1f;
+
+    // set the aac sample rate.
+    aac_sample_rate = samplingFrequencyIndex;
+
+    // the profile = object_id + 1
+    // @see aac-mp4a-format-ISO_IEC_14496-3+2001.pdf, page 78,
+    //      Table 1. A.9 ¨C MPEG-2 Audio profiles and MPEG-4 Audio object types
+    aac_profile = profile_ObjectType + 1;
+        
+    // the valid aac profile:
+    //      MPEG-2 profile
+    //      Main profile (ID == 1)
+    //      Low Complexity profile (LC) (ID == 2)
+    //      Scalable Sampling Rate profile (SSR) (ID == 3)
+    //      (reserved) (ID == 4)
+    // @see aac-mp4a-format-ISO_IEC_14496-3+2001.pdf, page 78,
+    //      Table 1. A.9 ¨C MPEG-2 Audio profiles and MPEG-4 Audio object types
+    if (aac_profile > 4) {
+        ret = ERROR_HLS_DECODE_ERROR;
+        srs_error("audio codec decode aac sequence header failed, "
+            "adts object=%d invalid. ret=%d", profile_ObjectType, ret);
+        return ret;
+    }
+        
+    // TODO: FIXME: to support aac he/he-v2, see: ngx_rtmp_codec_parse_aac_header
+    // @see: https://github.com/winlinvip/nginx-rtmp-module/commit/3a5f9eea78fc8d11e8be922aea9ac349b9dcbfc2
+    // 
+    // donot force to LC, @see: https://github.com/winlinvip/simple-rtmp-server/issues/81
+    // the source will print the sequence header info.
+    //if (aac_profile > 3) {
+        // Mark all extended profiles as LC
+        // to make Android as happy as possible.
+        // @see: ngx_rtmp_hls_parse_aac_header
+        //aac_profile = 1;
+    //}
+
+    return ret;
+}
+
 int SrsAvcAacCodec::video_avc_demux(char* data, int size, SrsCodecSample* sample)
 {
     int ret = ERROR_SUCCESS;

trunk/src/kernel/srs_kernel_codec.hpp

@@ -475,6 +475,11 @@ public:
     * demux the h.264 NALUs to sampe units.
     */
     virtual int video_avc_demux(char* data, int size, SrsCodecSample* sample);
+public:
+    /**
+    * directly demux the sequence header, without RTMP packet header.
+    */
+    virtual int audio_aac_sequence_header_demux(char* data, int size);
 private:
     /**
     * when avc packet type is SrsCodecVideoAVCTypeSequenceHeader,

trunk/src/kernel/srs_kernel_error.hpp

@@ -147,6 +147,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #define ERROR_RTP_HEADER_CORRUPT            2044
 #define ERROR_RTP_TYPE96_CORRUPT            2045
 #define ERROR_RTP_TYPE97_CORRUPT            2046
+#define ERROR_RTSP_AUDIO_CONFIG             2047
 //                                           
 // system control message, 
 // not an error, but special control logic.

trunk/src/kernel/srs_kernel_utility.cpp

@@ -621,3 +621,41 @@ char* srs_av_base64_encode(char* out, int out_size, const u_int8_t* in, int in_s
     return ret;
 }
 
+#define SPACE_CHARS " \t\r\n"
+
+int av_toupper(int c)
+{
+    if (c >= 'a' && c <= 'z') {
+        c ^= 0x20;
+    }
+    return c;
+}
+
+int ff_hex_to_data(u_int8_t* data, const char* p)
+{
+    int c, len, v;
+
+    len = 0;
+    v = 1;
+    for (;;) {
+        p += strspn(p, SPACE_CHARS);
+        if (*p == '\0')
+            break;
+        c = av_toupper((unsigned char) *p++);
+        if (c >= '0' && c <= '9')
+            c = c - '0';
+        else if (c >= 'A' && c <= 'F')
+            c = c - 'A' + 10;
+        else
+            break;
+        v = (v << 4) | c;
+        if (v & 0x100) {
+            if (data)
+                data[len] = v;
+            len++;
+            v = 1;
+        }
+    }
+    return len;
+}
+

trunk/src/kernel/srs_kernel_utility.hpp

@@ -115,5 +115,12 @@ extern char* srs_av_base64_encode(char* out, int out_size, const u_int8_t* in, i
  */
 #define SRS_AV_BASE64_SIZE(x)  (((x)+2) / 3 * 4 + 1)
 
+/**
+* convert hex string to data.
+* for example, p=config='139056E5A0'
+* output hex to data={0x13, 0x90, 0x56, 0xe5, 0xa0}
+*/
+extern int ff_hex_to_data(u_int8_t* data, const char* p);
+
 #endif
 

trunk/src/protocol/srs_rtsp_stack.cpp

@@ -200,6 +200,8 @@ int SrsRtpPacket::decode(SrsStream* stream)
     timestamp = stream->read_4bytes();
     ssrc = stream->read_4bytes();
 
+    // TODO: FIXME: check sequence number.
+
     // video codec.
     if (payload_type == 96) {
         return decode_96(stream);
@@ -232,7 +234,6 @@ int SrsRtpPacket::decode_97(SrsStream* stream)
     }
 
     int nb_samples = au_size / 2;
-    int guess_sample_size = (stream->size() - stream->pos() - au_size) / nb_samples;
     int required_size = 0;
 
     // append left bytes to payload.
@@ -247,11 +248,9 @@ int SrsRtpPacket::decode_97(SrsStream* stream)
         lasv = stream->read_1bytes();
 
         u_int16_t sample_size = ((hasv << 5) & 0xE0) | ((lasv >> 3) & 0x1f);
-        if (sample_size != guess_sample_size) {
-            // guess the size lost 0x100.
-            if (guess_sample_size == (sample_size | 0x100)) {
-                sample_size = guess_sample_size;
-            }
+        // TODO: FIXME: finger out how to parse the size of sample.
+        if (sample_size < 0x100 && stream->require(required_size + sample_size + 0x100)) {
+            sample_size = sample_size | 0x100;
         }
 
         char* sample = p + required_size;
@@ -541,7 +540,17 @@ int SrsRtspSdp::parse_fmtp_attribute(string attr)
             } else if (item_key == "indexdeltalength") {
                 audio_index_delta_length = item_value;
             } else if (item_key == "config") {
-                audio_sh = base64_decode(item_value);
+                if (item_value.length() <= 0) {
+                    ret = ERROR_RTSP_AUDIO_CONFIG;
+                    srs_error("rtsp: audio config failed. ret=%d", ret);
+                    return ret;
+                }
+
+                char* tmp_sh = new char[item_value.length()];
+                SrsAutoFree(char, tmp_sh);
+                int nb_tmp_sh = ff_hex_to_data((u_int8_t*)tmp_sh, item_value.c_str());
+                srs_assert(nb_tmp_sh > 0);
+                audio_sh.append(tmp_sh, nb_tmp_sh);
             }
         }
     }