Support NACK for RTC player

trunk/research/players/rtc_player.html

@@ -46,7 +46,7 @@
     </div>
 
     <label></label>
-    <video id="rtc_media_player" width="320" controls autoplay></video>
+    <video id="rtc_media_player" controls autoplay></video>
 
     <footer>
         <p></p>

trunk/src/app/srs_app_http_api.cpp

@@ -203,7 +203,19 @@ srs_error_t SrsGoApiRoot::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage*
     obj->set("urls", urls);
     
     urls->set("api", SrsJsonAny::str("the api root"));
-    
+
+    if (true) {
+        SrsJsonObject* rtc = SrsJsonAny::object();
+        urls->set("rtc", rtc);
+
+        SrsJsonObject* v1 = SrsJsonAny::object();
+        rtc->set("v1", v1);
+
+        v1->set("play", SrsJsonAny::str("Play stream"));
+        v1->set("publish", SrsJsonAny::str("Publish stream"));
+        v1->set("nack", SrsJsonAny::str("Simulate the NACK"));
+    }
+
     return srs_api_response(w, r, obj->dumps());
 }
 
@@ -957,7 +969,8 @@ srs_error_t SrsGoApiRtcPlay::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMe
     res->set("sdp", SrsJsonAny::str(local_sdp_str.c_str()));
     res->set("sessionid", SrsJsonAny::str(session->id().c_str()));
 
-    srs_trace("RTC sid=%s, offer=%dB, answer=%dB", session->id().c_str(), remote_sdp_str.length(), local_sdp_str.length());
+    srs_trace("RTC username=%s, offer=%dB, answer=%dB", session->username().c_str(),
+        remote_sdp_str.length(), local_sdp_str.length());
 
     return err;
 }
@@ -1298,7 +1311,8 @@ srs_error_t SrsGoApiRtcPublish::do_serve_http(ISrsHttpResponseWriter* w, ISrsHtt
     res->set("sdp", SrsJsonAny::str(local_sdp_str.c_str()));
     res->set("sessionid", SrsJsonAny::str(session->id().c_str()));
 
-    srs_trace("RTC sid=%s, offer=%dB, answer=%dB", session->id().c_str(), remote_sdp_str.length(), local_sdp_str.length());
+    srs_trace("RTC username=%s, offer=%dB, answer=%dB", session->username().c_str(),
+        remote_sdp_str.length(), local_sdp_str.length());
 
     return err;
 }
@@ -1465,6 +1479,63 @@ srs_error_t SrsGoApiRtcPublish::exchange_sdp(const std::string& app, const std::
 
     return err;
 }
+
+SrsGoApiRtcNACK::SrsGoApiRtcNACK(SrsRtcServer* server)
+{
+    server_ = server;
+}
+
+SrsGoApiRtcNACK::~SrsGoApiRtcNACK()
+{
+}
+
+srs_error_t SrsGoApiRtcNACK::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
+{
+    srs_error_t err = srs_success;
+
+    SrsJsonObject* res = SrsJsonAny::object();
+    SrsAutoFree(SrsJsonObject, res);
+
+    res->set("code", SrsJsonAny::integer(ERROR_SUCCESS));
+
+    if ((err = do_serve_http(w, r, res)) != srs_success) {
+        srs_warn("RTC NACK err %s", srs_error_desc(err).c_str());
+        res->set("code", SrsJsonAny::integer(srs_error_code(err)));
+        srs_freep(err);
+    }
+
+    return srs_api_response(w, r, res->dumps());
+}
+
+srs_error_t SrsGoApiRtcNACK::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, SrsJsonObject* res)
+{
+    string username = r->query_get("username");
+    string dropv = r->query_get("drop");
+
+    SrsJsonObject* query = SrsJsonAny::object();
+    res->set("query", query);
+
+    query->set("username", SrsJsonAny::str(username.c_str()));
+    query->set("drop", SrsJsonAny::str(dropv.c_str()));
+    query->set("help", SrsJsonAny::str("?username=string&drop=int"));
+
+    int drop = ::atoi(dropv.c_str());
+    if (drop <= 0) {
+        return srs_error_new(ERROR_RTC_INVALID_PARAMS, "invalid drop=%s/%d", dropv.c_str(), drop);
+    }
+
+    SrsRtcSession* session = server_->find_session_by_username(username);
+    if (!session) {
+        return srs_error_new(ERROR_RTC_NO_SESSION, "no session username=%s", username.c_str());
+    }
+
+    session->simulate_nack_drop(drop);
+
+    srs_trace("RTC NACK session peer_id=%s, username=%s, drop=%s/%d", session->peer_id().c_str(),
+        username.c_str(), dropv.c_str(), drop);
+
+    return srs_success;
+}
 #endif
 
 SrsGoApiClients::SrsGoApiClients()

trunk/src/app/srs_app_http_api.hpp

@@ -201,6 +201,19 @@ private:
     srs_error_t exchange_sdp(const std::string& app, const std::string& stream, const SrsSdp& remote_sdp, SrsSdp& local_sdp);
     srs_error_t check_remote_sdp(const SrsSdp& remote_sdp);
 };
+
+class SrsGoApiRtcNACK : public ISrsHttpHandler
+{
+private:
+    SrsRtcServer* server_;
+public:
+    SrsGoApiRtcNACK(SrsRtcServer* server);
+    virtual ~SrsGoApiRtcNACK();
+public:
+    virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
+private:
+    virtual srs_error_t do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, SrsJsonObject* res);
+};
 #endif
 
 class SrsGoApiClients : public ISrsHttpHandler

trunk/src/app/srs_app_listener.cpp

@@ -344,7 +344,7 @@ int SrsUdpMuxSocket::get_peer_port() const
     return peer_port;
 }
 
-std::string SrsUdpMuxSocket::get_peer_id()
+std::string SrsUdpMuxSocket::peer_id()
 {
     char id_buf[1024];
     int len = snprintf(id_buf, sizeof(id_buf), "%s:%d", peer_ip.c_str(), peer_port);

trunk/src/app/srs_app_listener.hpp

@@ -173,7 +173,7 @@ public:
     int size();
     std::string get_peer_ip() const;
     int get_peer_port() const;
-    std::string get_peer_id();
+    std::string peer_id();
     SrsUdpMuxSocket* copy_sendonly();
     ISrsUdpSender* sender();
 };

trunk/src/app/srs_app_rtc_conn.cpp

@@ -615,6 +615,12 @@ SrsRtcPlayer::SrsRtcPlayer(SrsRtcSession* s, int parent_cid)
     mw_msgs = 0;
     realtime = true;
 
+    // TODO: FIXME: Config the capacity?
+    audio_queue_ = new SrsRtpRingBuffer(100);
+    video_queue_ = new SrsRtpRingBuffer(1000);
+
+    nn_simulate_nack_drop = 0;
+
     _srs_config->subscribe(this);
 }
 
@@ -623,6 +629,8 @@ SrsRtcPlayer::~SrsRtcPlayer()
     _srs_config->unsubscribe(this);
 
     srs_freep(trd);
+    srs_freep(audio_queue_);
+    srs_freep(video_queue_);
 }
 
 srs_error_t SrsRtcPlayer::initialize(const uint32_t& vssrc, const uint32_t& assrc, const uint16_t& v_pt, const uint16_t& a_pt)
@@ -850,7 +858,8 @@ srs_error_t SrsRtcPlayer::send_messages(
 
 #ifndef SRS_OSX
     // If enabled GSO, send out some packets in a msghdr.
-    if (packets.use_gso) {
+    // @remark When NACK simulator is on, we don't use GSO.
+    if (packets.use_gso && !nn_simulate_nack_drop) {
         if ((err = send_packets_gso(packets)) != srs_success) {
             return srs_error_wrap(err, "gso send");
         }
@@ -997,6 +1006,25 @@ srs_error_t SrsRtcPlayer::send_packets(SrsRtcOutgoingPackets& packets)
             iov->iov_len = (size_t)nn_encrypt;
         }
 
+        // Put final RTP packet to NACK/ARQ queue.
+        if (true) {
+            SrsRtpPacket2* nack = new SrsRtpPacket2();
+            nack->rtp_header = packet->rtp_header;
+            nack->padding = packet->padding;
+
+            // TODO: FIXME: Should avoid memory copying.
+            SrsRtpRawPayload* payload = nack->reuse_raw();
+            payload->nn_payload = (int)iov->iov_len;
+            payload->payload = new char[payload->nn_payload];
+            memcpy((void*)payload->payload, iov->iov_base, iov->iov_len);
+
+            if (nack->rtp_header.get_ssrc() == video_ssrc) {
+                video_queue_->set(nack->rtp_header.get_sequence(), nack);
+            } else {
+                audio_queue_->set(nack->rtp_header.get_sequence(), nack);
+            }
+        }
+
         packets.nn_rtp_bytes += (int)iov->iov_len;
 
         // Set the address and control information.
@@ -1007,9 +1035,20 @@ srs_error_t SrsRtcPlayer::send_packets(SrsRtcOutgoingPackets& packets)
         mhdr->msg_hdr.msg_namelen = (socklen_t)addrlen;
         mhdr->msg_hdr.msg_controllen = 0;
 
-        // When we send out a packet, we commit a RTP packet.
+        // When we send out a packet, increase the stat counter.
         packets.nn_rtp_pkts++;
 
+        // For NACK simulator, drop packet.
+        if (nn_simulate_nack_drop) {
+            SrsRtpHeader* h = &packet->rtp_header;
+            srs_warn("RTC NACK simulator #%d drop seq=%u, ssrc=%u/%s, ts=%u, %d bytes", nn_simulate_nack_drop,
+                h->get_sequence(), h->get_ssrc(), (h->get_ssrc()==video_ssrc? "Video":"Audio"), h->get_timestamp(),
+                (int)iov->iov_len);
+            nn_simulate_nack_drop--;
+            iov->iov_len = 0;
+            continue;
+        }
+
         if ((err = sender->sendmmsg(mhdr)) != srs_success) {
             return srs_error_wrap(err, "send msghdr");
         }
@@ -1144,6 +1183,25 @@ srs_error_t SrsRtcPlayer::send_packets_gso(SrsRtcOutgoingPackets& packets)
             iov->iov_len = (size_t)nn_encrypt;
         }
 
+        // Put final RTP packet to NACK/ARQ queue.
+        if (true) {
+            SrsRtpPacket2* nack = new SrsRtpPacket2();
+            nack->rtp_header = packet->rtp_header;
+            nack->padding = packet->padding;
+
+            // TODO: FIXME: Should avoid memory copying.
+            SrsRtpRawPayload* payload = nack->reuse_raw();
+            payload->nn_payload = (int)iov->iov_len;
+            payload->payload = new char[payload->nn_payload];
+            memcpy((void*)payload->payload, iov->iov_base, iov->iov_len);
+
+            if (nack->rtp_header.get_ssrc() == video_ssrc) {
+                video_queue_->set(nack->rtp_header.get_sequence(), nack);
+            } else {
+                audio_queue_->set(nack->rtp_header.get_sequence(), nack);
+            }
+        }
+
         packets.nn_rtp_bytes += (int)iov->iov_len;
 
         // If GSO, they must has same size, except the final one.
@@ -1464,6 +1522,26 @@ srs_error_t SrsRtcPlayer::package_stap_a(SrsSource* source, SrsSharedPtrMessage*
     return err;
 }
 
+void SrsRtcPlayer::nack_fetch(vector<SrsRtpPacket2*>& pkts, uint32_t ssrc, uint16_t seq)
+{
+    SrsRtpPacket2* pkt = NULL;
+
+    if (ssrc == video_ssrc) {
+        pkt = video_queue_->at(seq);
+    } else if (ssrc == audio_ssrc) {
+        pkt = audio_queue_->at(seq);
+    }
+
+    if (pkt) {
+        pkts.push_back(pkt);
+    }
+}
+
+void SrsRtcPlayer::simulate_nack_drop(int nn)
+{
+    nn_simulate_nack_drop = nn;
+}
+
 SrsRtcPublisher::SrsRtcPublisher(SrsRtcSession* session)
 {
     report_timer = new SrsHourGlass(this, 200 * SRS_UTIME_MILLISECONDS);
@@ -2137,6 +2215,11 @@ srs_error_t SrsRtcPublisher::notify(int type, srs_utime_t interval, srs_utime_t
     return err;
 }
 
+void SrsRtcPublisher::simulate_nack_drop(int nn)
+{
+    // TODO: FIXME: Implements it.
+}
+
 SrsRtcSession::SrsRtcSession(SrsRtcServer* s)
 {
     req = NULL;
@@ -2201,20 +2284,25 @@ void SrsRtcSession::set_session_state(SrsRtcSessionStateType state)
     session_state = state;
 }
 
-std::string SrsRtcSession::id() const
+string SrsRtcSession::id()
 {
-    return peer_id + "_" + username;
+    return peer_id_ + "/" + username_;
 }
 
 
-std::string SrsRtcSession::get_peer_id() const
+string SrsRtcSession::peer_id()
 {
-    return peer_id;
+    return peer_id_;
 }
 
-void SrsRtcSession::set_peer_id(const std::string& id)
+void SrsRtcSession::set_peer_id(string v)
 {
-    peer_id = id;
+    peer_id_ = v;
+}
+
+string SrsRtcSession::username()
+{
+    return username_;
 }
 
 void SrsRtcSession::set_encrypt(bool v)
@@ -2232,11 +2320,11 @@ int SrsRtcSession::context_id()
     return cid;
 }
 
-srs_error_t SrsRtcSession::initialize(SrsSource* source, SrsRequest* r, bool is_publisher, const std::string& un, int context_id)
+srs_error_t SrsRtcSession::initialize(SrsSource* source, SrsRequest* r, bool is_publisher, string username, int context_id)
 {
     srs_error_t err = srs_success;
 
-    username = un;
+    username_ = username;
     req = r->copy();
     cid = context_id;
     is_publisher_ = is_publisher;
@@ -2289,7 +2377,7 @@ srs_error_t SrsRtcSession::on_stun(SrsUdpMuxSocket* skt, SrsStunPacket* r)
 
     // We are running in the ice-lite(server) mode. If client have multi network interface,
     // we only choose one candidate pair which is determined by client.
-    if (!sendonly_skt || sendonly_skt->get_peer_id() != skt->get_peer_id()) {
+    if (!sendonly_skt || sendonly_skt->peer_id() != skt->peer_id()) {
         update_sendonly_socket(skt);
     }
 
@@ -2515,13 +2603,24 @@ void SrsRtcSession::update_sendonly_socket(SrsUdpMuxSocket* skt)
 {
     if (sendonly_skt) {
         srs_trace("session %s address changed, update %s -> %s",
-            id().c_str(), sendonly_skt->get_peer_id().c_str(), skt->get_peer_id().c_str());
+            id().c_str(), sendonly_skt->peer_id().c_str(), skt->peer_id().c_str());
     }
 
     srs_freep(sendonly_skt);
     sendonly_skt = skt->copy_sendonly();
 }
 
+void SrsRtcSession::simulate_nack_drop(int nn)
+{
+    if (player_) {
+        player_->simulate_nack_drop(nn);
+    }
+
+    if (publisher_) {
+        publisher_->simulate_nack_drop(nn);
+    }
+}
+
 #ifdef SRS_OSX
 // These functions are similar to the older byteorder(3) family of functions.
 // For example, be32toh() is identical to ntohl().
@@ -2564,8 +2663,8 @@ srs_error_t SrsRtcSession::on_binding_request(SrsStunPacket* r)
     if (get_session_state() == WAITING_STUN) {
         set_session_state(DOING_DTLS_HANDSHAKE);
 
-        peer_id = sendonly_skt->get_peer_id();
-        server_->insert_into_id_sessions(peer_id, this);
+        peer_id_ = sendonly_skt->peer_id();
+        server_->insert_into_id_sessions(peer_id_, this);
 
         set_session_state(DOING_DTLS_HANDSHAKE);
         srs_trace("rtc session=%s, STUN done, waitting DTLS handshake.", id().c_str());
@@ -2617,7 +2716,7 @@ srs_error_t SrsRtcSession::on_rtcp_feedback(char* buf, int nb_buf)
     /*uint8_t payload_type = */stream->read_1bytes();
     /*uint16_t length = */stream->read_2bytes();
     /*uint32_t ssrc_of_sender = */stream->read_4bytes();
-    /*uint32_t ssrc_of_media_source = */stream->read_4bytes();
+    uint32_t ssrc_of_media_source = stream->read_4bytes();
 
     /*
          0                   1                   2                   3
@@ -2630,10 +2729,11 @@ srs_error_t SrsRtcSession::on_rtcp_feedback(char* buf, int nb_buf)
     uint16_t pid = stream->read_2bytes();
     int blp = stream->read_2bytes();
 
-    srs_verbose("pid=%u, blp=%d", pid, blp);
-
     // TODO: FIXME: Support ARQ.
     vector<SrsRtpPacket2*> resend_pkts;
+    if (player_) {
+        player_->nack_fetch(resend_pkts, ssrc_of_media_source, pid);
+    }
 
     uint16_t mask = 0x01;
     for (int i = 1; i < 16 && blp; ++i, mask <<= 1) {
@@ -2642,24 +2742,26 @@ srs_error_t SrsRtcSession::on_rtcp_feedback(char* buf, int nb_buf)
         }
 
         uint32_t loss_seq = pid + i;
-
-        // TODO: FIXME: Support ARQ.
-        (void)loss_seq;
+        if (player_) {
+            player_->nack_fetch(resend_pkts, ssrc_of_media_source, loss_seq);
+        }
     }
 
     for (int i = 0; i < (int)resend_pkts.size(); ++i) {
         SrsRtpPacket2* pkt = resend_pkts[i];
 
-        char* protected_buf = new char[kRtpPacketSize];
-        SrsAutoFreeA(char, protected_buf);
+        char* data = new char[pkt->nb_bytes()];
+        SrsAutoFreeA(char, data);
 
-        int nb_protected_buf = resend_pkts[i]->nb_bytes();
-        SrsBuffer buf(protected_buf, nb_protected_buf);
+        SrsBuffer buf(data, pkt->nb_bytes());
 
         // TODO: FIXME: Check error.
         pkt->encode(&buf);
-        dtls_->protect_rtp(protected_buf, protected_buf, nb_protected_buf);
-        sendonly_skt->sendto(protected_buf, nb_protected_buf, 0);
+        sendonly_skt->sendto(data, pkt->nb_bytes(), 0);
+
+        SrsRtpHeader* h = &pkt->rtp_header;
+        srs_trace("RTC NACK ARQ seq=%u, ssrc=%u, ts=%u, %d bytes", h->get_sequence(),
+            h->get_ssrc(), h->get_timestamp(), pkt->nb_bytes());
     }
 
     return err;
@@ -3169,7 +3271,7 @@ srs_error_t SrsRtcServer::on_udp_packet(SrsUdpMuxSocket* skt)
     srs_error_t err = srs_success;
 
     char* data = skt->data(); int size = skt->size();
-    SrsRtcSession* session = find_session_by_peer_id(skt->get_peer_id());
+    SrsRtcSession* session = find_session_by_peer_id(skt->peer_id());
 
     if (session) {
         // Now, we got the RTC session to handle the packet, switch to its context
@@ -3179,30 +3281,30 @@ srs_error_t SrsRtcServer::on_udp_packet(SrsUdpMuxSocket* skt)
 
     // For STUN, the peer address may change.
     if (is_stun((uint8_t*)data, size)) {
-        SrsStunPacket sr;
-        if ((err = sr.decode(data, size)) != srs_success) {
+        SrsStunPacket ping;
+        if ((err = ping.decode(data, size)) != srs_success) {
             return srs_error_wrap(err, "decode stun packet failed");
         }
         srs_verbose("recv stun packet from %s, use-candidate=%d, ice-controlled=%d, ice-controlling=%d",
-            skt->get_peer_id().c_str(), sr.get_use_candidate(), sr.get_ice_controlled(), sr.get_ice_controlling());
+            skt->peer_id().c_str(), ping.get_use_candidate(), ping.get_ice_controlled(), ping.get_ice_controlling());
 
         if (!session) {
-            session = find_session_by_username(sr.get_username());
+            session = find_session_by_username(ping.get_username());
             if (session) {
                 session->switch_to_context();
             }
         }
         if (session == NULL) {
             return srs_error_new(ERROR_RTC_STUN, "can not find session, stun username=%s, peer_id=%s",
-                sr.get_username().c_str(), skt->get_peer_id().c_str());
+                ping.get_username().c_str(), skt->peer_id().c_str());
         }
 
-        return session->on_stun(skt, &sr);
+        return session->on_stun(skt, &ping);
     }
 
     // For DTLS, RTCP or RTP, which does not support peer address changing.
     if (session == NULL) {
-        return srs_error_new(ERROR_RTC_STUN, "can not find session, peer_id=%s", skt->get_peer_id().c_str());
+        return srs_error_new(ERROR_RTC_STUN, "can not find session, peer_id=%s", skt->peer_id().c_str());
     }
 
     if (is_dtls((uint8_t*)data, size)) {
@@ -3231,6 +3333,10 @@ srs_error_t SrsRtcServer::listen_api()
         return srs_error_wrap(err, "handle publish");
     }
 
+    if ((err = http_api_mux->handle("/rtc/v1/nack/", new SrsGoApiRtcNACK(this))) != srs_success) {
+        return srs_error_wrap(err, "handle nack");
+    }
+
     return err;
 }
 
@@ -3253,6 +3359,7 @@ srs_error_t SrsRtcServer::create_session(
         return srs_error_new(ERROR_RTC_SOURCE_BUSY, "stream %s busy", req->get_stream_url().c_str());
     }
 
+    // TODO: FIXME: Seems not random, please check it.
     std::string local_pwd = gen_random_str(32);
     std::string local_ufrag = "";
     std::string username = "";
@@ -3320,7 +3427,7 @@ void SrsRtcServer::check_and_clean_timeout_session()
 
             srs_trace("rtc session=%s, STUN timeout", session->id().c_str());
             map_username_session.erase(iter++);
-            map_id_session.erase(session->get_peer_id());
+            map_id_session.erase(session->peer_id());
             delete session;
             continue;
         }

trunk/src/app/srs_app_rtc_conn.hpp

@@ -59,6 +59,7 @@ class SrsRtpPacket2;
 class ISrsCodec;
 class SrsRtpNackForReceiver;
 class SrsRtpIncommingVideoFrame;
+class SrsRtpRingBuffer;
 
 const uint8_t kSR   = 200;
 const uint8_t kRR   = 201;
@@ -198,25 +199,31 @@ public:
 class SrsRtcPlayer : virtual public ISrsCoroutineHandler, virtual public ISrsReloadHandler
 {
 protected:
-    SrsCoroutine* trd;
     int _parent_cid;
-private:
+    SrsCoroutine* trd;
     SrsRtcSession* session_;
-    uint32_t video_ssrc;
-    uint32_t audio_ssrc;
-    uint16_t video_payload_type;
-    uint16_t audio_payload_type;
 private:
     // TODO: FIXME: How to handle timestamp overflow?
+    // Information for audio.
     uint32_t audio_timestamp;
     uint16_t audio_sequence;
-private:
+    uint32_t audio_ssrc;
+    uint16_t audio_payload_type;
+    // Information for video.
     uint16_t video_sequence;
+    uint16_t video_payload_type;
+    uint32_t video_ssrc;
+    // NACK ARQ ring buffer.
+    SrsRtpRingBuffer* audio_queue_;
+    SrsRtpRingBuffer* video_queue_;
+    // Simulators.
+    int nn_simulate_nack_drop;
 private:
+    // For merged-write and GSO.
     bool merge_nalus;
     bool gso;
     int max_padding;
-private:
+    // For merged-write messages.
     srs_utime_t mw_sleep;
     int mw_msgs;
     bool realtime;
@@ -250,6 +257,9 @@ private:
     srs_error_t package_nalus(SrsSharedPtrMessage* msg, SrsRtcOutgoingPackets& packets);
     srs_error_t package_single_nalu(SrsSharedPtrMessage* msg, SrsSample* sample, SrsRtcOutgoingPackets& packets);
     srs_error_t package_stap_a(SrsSource* source, SrsSharedPtrMessage* msg, SrsRtcOutgoingPackets& packets);
+public:
+    void nack_fetch(std::vector<SrsRtpPacket2*>& pkts, uint32_t ssrc, uint16_t seq);
+    void simulate_nack_drop(int nn);
 };
 
 class SrsRtcPublisher : virtual public ISrsHourGlass, virtual public ISrsRtpPacketDecodeHandler
@@ -296,6 +306,8 @@ public:
 // interface ISrsHourGlass
 public:
     virtual srs_error_t notify(int type, srs_utime_t interval, srs_utime_t tick);
+public:
+    void simulate_nack_drop(int nn);
 };
 
 class SrsRtcSession
@@ -312,8 +324,8 @@ private:
     bool is_publisher_;
 private:
     SrsUdpMuxSocket* sendonly_skt;
-    std::string username;
-    std::string peer_id;
+    std::string username_;
+    std::string peer_id_;
 private:
     // The timeout of session, keep alive by STUN ping pong.
     srs_utime_t sessionStunTimeout;
@@ -344,14 +356,15 @@ public:
     void set_remote_sdp(const SrsSdp& sdp);
     SrsRtcSessionStateType get_session_state();
     void set_session_state(SrsRtcSessionStateType state);
-    std::string id() const;
-    std::string get_peer_id() const;
-    void set_peer_id(const std::string& id);
+    std::string id();
+    std::string peer_id();
+    void set_peer_id(std::string v);
+    std::string username();
     void set_encrypt(bool v);
     void switch_to_context();
     int context_id();
 public:
-    srs_error_t initialize(SrsSource* source, SrsRequest* r, bool is_publisher, const std::string& un, int context_id);
+    srs_error_t initialize(SrsSource* source, SrsRequest* r, bool is_publisher, std::string username, int context_id);
     // The peer address may change, we can identify that by STUN messages.
     srs_error_t on_stun(SrsUdpMuxSocket* skt, SrsStunPacket* r);
     srs_error_t on_dtls(char* data, int nb_data);
@@ -363,6 +376,9 @@ public:
     srs_error_t start_publish();
     bool is_stun_timeout();
     void update_sendonly_socket(SrsUdpMuxSocket* skt);
+public:
+    // Simulate the NACK to drop nn packets.
+    void simulate_nack_drop(int nn);
 private:
     srs_error_t on_binding_request(SrsStunPacket* r);
     srs_error_t on_rtcp_feedback(char* data, int nb_data);
@@ -444,8 +460,8 @@ public:
     bool insert_into_id_sessions(const std::string& peer_id, SrsRtcSession* session);
     void check_and_clean_timeout_session();
     int nn_sessions();
-private:
     SrsRtcSession* find_session_by_username(const std::string& ufrag);
+private:
     SrsRtcSession* find_session_by_peer_id(const std::string& peer_id);
 // interface ISrsHourGlass
 public:

trunk/src/app/srs_app_rtp_queue.cpp

@@ -592,6 +592,7 @@ void SrsRtpVideoQueue::covert_frame(std::vector<SrsRtpPacket2*>& frame, SrsRtpPa
     // TODO: FIXME: Should covert to multiple NALU RTP packet to avoid copying.
     SrsRtpPacket2* pkt = new SrsRtpPacket2();
     pkt->rtp_header = head->rtp_header;
+    pkt->padding = head->padding;
 
     SrsRtpFUAPayload2* head_payload = dynamic_cast<SrsRtpFUAPayload2*>(head->payload);
     pkt->nalu_type = head_payload->nalu_type;

trunk/src/app/srs_app_server.cpp

@@ -941,8 +941,8 @@ srs_error_t SrsServer::http_handle()
 {
     srs_error_t err = srs_success;
     
-    if ((err = http_api_mux->handle("/", new SrsHttpNotFoundHandler())) != srs_success) {
-        return srs_error_wrap(err, "handle not found");
+    if ((err = http_api_mux->handle("/", new SrsGoApiRoot())) != srs_success) {
+        return srs_error_wrap(err, "handle /");
     }
     if ((err = http_api_mux->handle("/api/", new SrsGoApiApi())) != srs_success) {
         return srs_error_wrap(err, "handle api");

trunk/src/app/srs_app_source.cpp

@@ -1121,11 +1121,6 @@ srs_error_t SrsOriginHub::on_video(SrsSharedPtrMessage* shared_video, bool is_se
         srs_error_reset(err);
         rtc->on_unpublish();
     }
-
-    // TODO: FIXME: Refactor to move to rtp?
-    // Save the RTP packets for find_rtp_packet() to rtx or restore it.
-    // TODO: FIXME: Remove dead code.
-    //source->rtp_queue->push(msg->rtp_packets);
 #endif
     
     if ((err = hls->on_video(msg, format)) != srs_success) {

trunk/src/kernel/srs_kernel_error.hpp

@@ -350,6 +350,8 @@
 #define ERROR_RTC_API_BODY                  5019
 #define ERROR_RTC_SOURCE_BUSY               5020
 #define ERROR_RTC_DISABLED                  5021
+#define ERROR_RTC_NO_SESSION                5022
+#define ERROR_RTC_INVALID_PARAMS            5023
 
 ///////////////////////////////////////////////////////
 // GB28181 API error.

trunk/src/kernel/srs_kernel_rtp.cpp

@@ -228,12 +228,12 @@ uint16_t SrsRtpHeader::get_sequence() const
     return sequence;
 }
 
-void SrsRtpHeader::set_timestamp(int64_t v)
+void SrsRtpHeader::set_timestamp(uint32_t v)
 {
-    timestamp = (uint32_t)v;
+    timestamp = v;
 }
 
-int64_t SrsRtpHeader::get_timestamp() const
+uint32_t SrsRtpHeader::get_timestamp() const
 {
     return timestamp;
 }

trunk/src/kernel/srs_kernel_rtp.hpp

@@ -64,7 +64,7 @@ private:
     bool marker;
     uint8_t payload_type;
     uint16_t sequence;
-    int32_t timestamp;
+    uint32_t timestamp;
     uint32_t ssrc;
     uint32_t csrc[15];
     uint16_t extension_length;
@@ -84,8 +84,8 @@ public:
     uint8_t get_payload_type() const;
     void set_sequence(uint16_t v);
     uint16_t get_sequence() const;
-    void set_timestamp(int64_t v);
-    int64_t get_timestamp() const;
+    void set_timestamp(uint32_t v);
+    uint32_t get_timestamp() const;
     void set_ssrc(uint32_t v);
     uint32_t get_ssrc() const;
     void set_padding(bool v);