mirror of
https://github.com/ossrs/srs.git
synced 2025-02-15 04:42:04 +00:00
Merge branch 'feature/rtc' into develop
This commit is contained in:
winlin
commit
8e5d4ccab7
14 changed files with 300 additions and 84 deletions
|
|
@ -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>
|
||||
|
|
|
|||
|
|
@ -81,8 +81,8 @@ srs_error_t SrsDtls::init(SrsRequest* r)
|
|||
#if OPENSSL_VERSION_NUMBER < 0x10002000L // v1.0.2
|
||||
dtls_ctx = SSL_CTX_new(DTLSv1_method());
|
||||
#else
|
||||
//dtls_ctx = SSL_CTX_new(DTLS_method());
|
||||
dtls_ctx = SSL_CTX_new(DTLSv1_method());
|
||||
dtls_ctx = SSL_CTX_new(DTLS_method());
|
||||
//dtls_ctx = SSL_CTX_new(DTLSv1_method());
|
||||
//dtls_ctx = SSL_CTX_new(DTLSv1_2_method());
|
||||
#endif
|
||||
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
@ -1033,6 +1046,11 @@ srs_error_t SrsGoApiRtcPlay::exchange_sdp(const std::string& app, const std::str
|
|||
// TODO: FIXME: Only support some transport algorithms.
|
||||
vector<string> rtcp_fb;
|
||||
payload_type.rtcp_fb_.swap(rtcp_fb);
|
||||
for (int j = 0; j < (int)rtcp_fb.size(); j++) {
|
||||
if (rtcp_fb.at(j) == "nack" || rtcp_fb.at(j) == "nack pli") {
|
||||
payload_type.rtcp_fb_.push_back(rtcp_fb.at(j));
|
||||
}
|
||||
}
|
||||
|
||||
// Only choose one match opus codec.
|
||||
break;
|
||||
|
|
@ -1062,6 +1080,11 @@ srs_error_t SrsGoApiRtcPlay::exchange_sdp(const std::string& app, const std::str
|
|||
// TODO: FIXME: Only support some transport algorithms.
|
||||
vector<string> rtcp_fb;
|
||||
payload_type.rtcp_fb_.swap(rtcp_fb);
|
||||
for (int j = 0; j < (int)rtcp_fb.size(); j++) {
|
||||
if (rtcp_fb.at(j) == "nack" || rtcp_fb.at(j) == "nack pli") {
|
||||
payload_type.rtcp_fb_.push_back(rtcp_fb.at(j));
|
||||
}
|
||||
}
|
||||
|
||||
// Only choose first match H.264 payload type.
|
||||
break;
|
||||
|
|
@ -1288,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;
|
||||
}
|
||||
|
|
@ -1364,6 +1388,11 @@ srs_error_t SrsGoApiRtcPublish::exchange_sdp(const std::string& app, const std::
|
|||
// TODO: FIXME: Only support some transport algorithms.
|
||||
vector<string> rtcp_fb;
|
||||
payload_type.rtcp_fb_.swap(rtcp_fb);
|
||||
for (int j = 0; j < (int)rtcp_fb.size(); j++) {
|
||||
if (rtcp_fb.at(j) == "nack" || rtcp_fb.at(j) == "nack pli") {
|
||||
payload_type.rtcp_fb_.push_back(rtcp_fb.at(j));
|
||||
}
|
||||
}
|
||||
|
||||
// Only choose one match opus codec.
|
||||
break;
|
||||
|
|
@ -1394,6 +1423,11 @@ srs_error_t SrsGoApiRtcPublish::exchange_sdp(const std::string& app, const std::
|
|||
// TODO: FIXME: Only support some transport algorithms.
|
||||
vector<string> rtcp_fb;
|
||||
payload_type.rtcp_fb_.swap(rtcp_fb);
|
||||
for (int j = 0; j < (int)rtcp_fb.size(); j++) {
|
||||
if (rtcp_fb.at(j) == "nack" || rtcp_fb.at(j) == "nack pli") {
|
||||
payload_type.rtcp_fb_.push_back(rtcp_fb.at(j));
|
||||
}
|
||||
}
|
||||
|
||||
// Only choose first match H.264 payload type.
|
||||
break;
|
||||
|
|
@ -1445,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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
};
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
@ -1882,7 +1960,7 @@ srs_error_t SrsRtcPublisher::on_rtp(char* buf, int nb_buf)
|
|||
SrsRtpPacket2* pkt = new SrsRtpPacket2();
|
||||
|
||||
pkt->set_decode_handler(this);
|
||||
pkt->set_original_bytes(buf, nb_buf);
|
||||
pkt->original_bytes = buf;
|
||||
|
||||
SrsBuffer b(buf, nb_buf);
|
||||
if ((err = pkt->decode(&b)) != srs_success) {
|
||||
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -372,6 +372,7 @@ void SrsRtpAudioQueue::collect_frames(SrsRtpNackForReceiver* nack, vector<SrsRtp
|
|||
for (; next != queue_->end; ++next) {
|
||||
SrsRtpPacket2* pkt = queue_->at(next);
|
||||
|
||||
// TODO: FIXME: Should not wait for NACK packets.
|
||||
// Not found or in NACK, stop collecting frame.
|
||||
if (!pkt || nack->find(next) != NULL) {
|
||||
srs_trace("wait for nack seq=%u", next);
|
||||
|
|
@ -514,6 +515,7 @@ void SrsRtpVideoQueue::collect_frame(SrsRtpNackForReceiver* nack, SrsRtpPacket2*
|
|||
for (; next != queue_->end; ++next) {
|
||||
SrsRtpPacket2* pkt = queue_->at(next);
|
||||
|
||||
// TODO: FIXME: Should not wait for NACK packets.
|
||||
// Not found or in NACK, stop collecting frame.
|
||||
if (!pkt || nack->find(next) != NULL) {
|
||||
srs_trace("wait for nack seq=%u", next);
|
||||
|
|
@ -590,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;
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
}
|
||||
|
|
@ -287,7 +287,6 @@ SrsRtpPacket2::SrsRtpPacket2()
|
|||
cache_payload = 0;
|
||||
|
||||
original_bytes = NULL;
|
||||
nn_original_bytes = 0;
|
||||
nn_original_payload = 0;
|
||||
}
|
||||
|
||||
|
|
@ -355,12 +354,6 @@ void SrsRtpPacket2::set_decode_handler(ISrsRtpPacketDecodeHandler* h)
|
|||
decode_handler = h;
|
||||
}
|
||||
|
||||
void SrsRtpPacket2::set_original_bytes(char* buf, int nn_buf)
|
||||
{
|
||||
original_bytes = buf;
|
||||
nn_original_bytes = nn_buf;
|
||||
}
|
||||
|
||||
int SrsRtpPacket2::nb_bytes()
|
||||
{
|
||||
if (!cache_payload) {
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
@ -114,6 +114,7 @@ public:
|
|||
int padding;
|
||||
// Decoder helper.
|
||||
public:
|
||||
// TODO: FIXME: Move to video decoder queue SrsRtpVideoQueue.
|
||||
// Helper information for video decoder only.
|
||||
bool video_is_first_packet;
|
||||
bool video_is_last_packet;
|
||||
|
|
@ -122,11 +123,8 @@ public:
|
|||
SrsAvcNaluType nalu_type;
|
||||
// The original payload bytes length.
|
||||
int nn_original_payload;
|
||||
// Decoder helper.
|
||||
private:
|
||||
// The original bytes for decoder only, we will free it.
|
||||
char* original_bytes;
|
||||
int nn_original_bytes;
|
||||
// Fast cache for performance.
|
||||
private:
|
||||
// Cache frequently used payload for performance.
|
||||
|
|
@ -151,8 +149,6 @@ public:
|
|||
SrsRtpFUAPayload2* reuse_fua();
|
||||
// Set the decode handler.
|
||||
void set_decode_handler(ISrsRtpPacketDecodeHandler* h);
|
||||
// Set the original bytes.
|
||||
void set_original_bytes(char* buf, int nn_buf);
|
||||
// interface ISrsEncoder
|
||||
public:
|
||||
virtual int nb_bytes();
|
||||
|
|
|
|||
Loading…
Reference in a new issue