external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-03-09 15:49:59 +00:00
Code Issues Releases Wiki Activity

fix 28281 to 28181, sip heart message, timeout channel peer ip and port reset

Browse source
This commit is contained in:
xialixin 2020-04-02 16:56:03 +08:00
parent c5c0df6536
commit ac8acc0b22
10 changed files with 136 additions and 81 deletions
Download patch file Download diff file Expand all files Collapse all files

33
trunk/src/app/srs_app_gb28181.cpp
View file

@ -236,7 +236,7 @@ srs_error_t SrsGb28181PsRtpProcessor::on_udp_packet(const sockaddr* from, const
//TODO: check sequence number out of order
//it may be out of order, or multiple streaming ssrc are the same
// if (pre_sequence_number > pkt.sequence_number){
// srs_info("gb28281: ps sequence_number out of order, ssrc=%#x, pre=%u, cur=%u, addr=%s,port=%s",
// srs_info("gb28181: ps sequence_number out of order, ssrc=%#x, pre=%u, cur=%u, addr=%s,port=%s",
// pkt.ssrc, pre_sequence_number, pkt.sequence_number, address_string, port_string);
// //return err;
// }
@ -305,11 +305,11 @@ srs_error_t SrsGb28181PsRtpProcessor::on_udp_packet(const sockaddr* from, const
muxer->get_channel_id().c_str(), pkt.ssrc, muxer->channel_peer_port(), peer_port);
srs_freep(key->second);
}else {
//put it in queue, wait for conn to process, and then free
//put it in queue, wait for consumer to process, and then free
muxer->ps_packet_enqueue(key->second);
}
}else{
//no connection process it, discarded
//no consumer process it, discarded
srs_freep(key->second);
}
cache_ps_rtp_packet.erase(pkt_key);
@ -461,7 +461,7 @@ srs_error_t SrsPsStreamDemixer::on_ps_stream(char* ps_data, int ps_size, uint32_
//in a frame of data, pts is obtained from the first PSE packet
if (pse_index == 0 && pts_dts_flags > 0) {
video_pts = parse_ps_timestamp((unsigned char*)next_ps_pack + 9);
srs_info("gb28181: ps stream video ts=%u pkt_ts=%u", pts, timestamp);
srs_info("gb28181: ps stream video ts=%u pkt_ts=%u", video_pts, timestamp);
}
pse_index +=1;
@ -704,6 +704,11 @@ std::string SrsGb28181RtmpMuxer::rtmp_url()
return _rtmp_url;
}
srs_utime_t SrsGb28181RtmpMuxer::get_recv_stream_time()
{
return recv_stream_time;
}
void SrsGb28181RtmpMuxer::destroy()
{
@ -750,6 +755,16 @@ srs_error_t SrsGb28181RtmpMuxer::do_cycle()
srs_utime_t now = srs_get_system_time();
srs_utime_t duration = now - recv_stream_time;
//if no RTP data is received within 2 seconds,
//the peer-port and peer-ip will be cleared and
//other port data will be received again
if (duration > (2 * SRS_UTIME_SECONDS) && channel->get_rtp_peer_port() != 0){
srs_warn("gb28181: client id=%s ssrc=%#x, peer(ip=%s port=%d), no rtp data %d in seconds, clean it, wait other port!",
channel_id.c_str(), channel->get_ssrc(), channel->get_rtp_peer_ip().c_str(), channel->get_rtp_peer_port(), duration/SRS_UTIME_SECONDS);
channel->set_rtp_peer_port(0);
channel->set_rtp_peer_ip("");
}
SrsGb28181Config config = gb28181_manger->get_gb28181_config();
if (duration > config.rtp_idle_timeout){
@ -777,6 +792,8 @@ void SrsGb28181RtmpMuxer::stop()
void SrsGb28181RtmpMuxer::ps_packet_enqueue(SrsPsRtpPacket *pkt)
{
srs_assert(pkt);
recv_stream_time = srs_update_system_time();
//prevent consumers from being unable to process data
//and accumulating in the queue
@ -1292,7 +1309,7 @@ srs_error_t SrsGb28181Manger::fetch_or_create_rtmpmuxer(std::string id, SrsGb28
muxer = new SrsGb28181RtmpMuxer(this, id, config->audio_enable, config->wait_keyframe);
if ((err = muxer->serve()) != srs_success) {
return srs_error_wrap(err, "gb28281: rtmp muxer serve %s", id.c_str());
return srs_error_wrap(err, "gb28181: rtmp muxer serve %s", id.c_str());
}
rtmpmuxers[id] = muxer;
*gb28181 = muxer;
@ -1453,12 +1470,12 @@ uint32_t SrsGb28181Manger::create_stream_channel(SrsGb28181StreamChannel *channe
channel->set_port_mode(RTP_PORT_MODE_FIXED);
}
//create on rtmp muxer, gb28281 stream to rtmp
//create on rtmp muxer, gb28181 stream to rtmp
srs_error_t err = srs_success;
if ((err = fetch_or_create_rtmpmuxer(id, &muxer)) != srs_success){
srs_warn("gb28181: create rtmp muxer error, %s", srs_error_desc(err).c_str());
srs_freep(err);
return ERROR_GB28281_CREATER_RTMPMUXER_FAILED;
return ERROR_GB28181_CREATER_RTMPMUXER_FAILED;
}
//Start RTP listening port, receive gb28181 stream,
@ -1477,7 +1494,7 @@ uint32_t SrsGb28181Manger::create_stream_channel(SrsGb28181StreamChannel *channe
srs_warn("gb28181: start ps rtp listen error, %s", srs_error_desc(err).c_str());
srs_freep(err);
free_port(rtp_port, rtp_port + 1);
return ERROR_GB28281_CREATER_RTMPMUXER_FAILED;
return ERROR_GB28181_CREATER_RTMPMUXER_FAILED;
}
}
else if(port_mode == RTP_PORT_MODE_FIXED) {

9
trunk/src/app/srs_app_gb28181.hpp
View file

@ -76,7 +76,7 @@ public:
virtual srs_error_t decode(SrsBuffer* stream);
};
//randomly assigned ports receive gb28281 device streams
//randomly assigned ports receive gb28181 device streams
class SrsPsRtpListener: public ISrsUdpHandler
{
private:
@ -94,7 +94,7 @@ public:
virtual srs_error_t on_udp_packet(const sockaddr* from, const int fromlen, char* buf, int nb_buf);
};
//multiplexing service, single port receiving all gb28281 device streams
//multiplexing service, single port receiving all gb28181 device streams
class SrsGb28181RtpMuxService : public ISrsUdpHandler
{
private:
@ -110,7 +110,7 @@ public:
};
//process gb28281 RTP package, generate a completed PS stream data,
//process gb28181 RTP package, generate a completed PS stream data,
//call the PS stream parser, parse the original video and audio
class SrsGb28181PsRtpProcessor: public ISrsUdpHandler
{
@ -262,6 +262,7 @@ public:
virtual void set_rtmp_url(std::string url);
virtual std::string rtmp_url();
virtual SrsGb28181StreamChannel get_channel();
srs_utime_t get_recv_stream_time();
private:
virtual srs_error_t do_cycle();
@ -288,7 +289,7 @@ public:
virtual void rtmp_close();
};
//system parameter configuration of gb28281 module,
//system parameter configuration of gb28181 module,
//read file from configuration file to generate
class SrsGb28181Config
{

18
trunk/src/app/srs_app_gb28181_sip.cpp
View file

@ -143,7 +143,7 @@ srs_error_t SrsGb28181SipService::on_udp_sip(string peer_ip, int peer_port,
{
srs_trace("gb28181: %s method=%s, uri=%s, version=%s ",
req->get_cmdtype_str().c_str(), req->method.c_str(), req->uri.c_str(), req->version.c_str());
srs_trace("gb28281: request client id=%s", req->sip_auth_id.c_str());
srs_trace("gb28181: request client id=%s", req->sip_auth_id.c_str());
}
req->peer_ip = peer_ip;
@ -163,7 +163,7 @@ srs_error_t SrsGb28181SipService::on_udp_sip(string peer_ip, int peer_port,
srs_trace("gb28181: request peer_ip=%s, peer_port=%d", peer_ip.c_str(), peer_port, nb_buf);
srs_trace("gb28181: request %s method=%s, uri=%s, version=%s ",
req->get_cmdtype_str().c_str(), req->method.c_str(), req->uri.c_str(), req->version.c_str());
srs_trace("gb28281: request client id=%s", req->sip_auth_id.c_str());
srs_trace("gb28181: request client id=%s", req->sip_auth_id.c_str());
SrsGb28181SipSession* sip_session = create_sip_session(req);
if (!sip_session) {
@ -235,7 +235,7 @@ srs_error_t SrsGb28181SipService::on_udp_sip(string peer_ip, int peer_port,
srs_trace("gb28181: request peer_ip=%s, peer_port=%d", peer_ip.c_str(), peer_port, nb_buf);
srs_trace("gb28181: request %s method=%s, uri=%s, version=%s ",
req->get_cmdtype_str().c_str(), req->method.c_str(), req->uri.c_str(), req->version.c_str());
srs_trace("gb28281: request client id=%s", req->sip_auth_id.c_str());
srs_trace("gb28181: request client id=%s", req->sip_auth_id.c_str());
if (!sip_session){
send_bye(req);
@ -264,7 +264,7 @@ srs_error_t SrsGb28181SipService::on_udp_sip(string peer_ip, int peer_port,
srs_trace("gb28181: request peer_ip=%s, peer_port=%d", peer_ip.c_str(), peer_port, nb_buf);
srs_trace("gb28181: request %s method=%s, uri=%s, version=%s ",
req->get_cmdtype_str().c_str(), req->method.c_str(), req->uri.c_str(), req->version.c_str());
srs_trace("gb28281: request client id=%s", req->sip_auth_id.c_str());
srs_trace("gb28181: request client id=%s", req->sip_auth_id.c_str());
SrsGb28181SipSession* sip_session = fetch(session_id);
send_status(req, from, fromlen);
@ -311,7 +311,7 @@ int SrsGb28181SipService::send_ack(SrsSipRequest *req, sockaddr *f, int l)
req->realm = config->sip_realm;
req->serial = config->sip_serial;
sip->resp_ack(ss, req);
sip->req_ack(ss, req);
return send_message(f, l, ss);
}
@ -345,7 +345,7 @@ int SrsGb28181SipService::send_invite(SrsSipRequest *req, string ip, int port,
//you cannot invite again. you need to 'bye' and try again
if (sip_session->invite_status() == SrsGb28181SipSessionTrying ||
sip_session->invite_status() == SrsGb28181SipSessionInviteOk){
return ERROR_GB28281_SIP_IS_INVITING;
return ERROR_GB28181_SIP_IS_INVITING;
}
req->host = config->host;
@ -358,7 +358,7 @@ int SrsGb28181SipService::send_invite(SrsSipRequest *req, string ip, int port,
if (send_message(sip_session->sockaddr_from(), sip_session->sockaddr_fromlen(), ss) <= 0)
{
return ERROR_GB28281_SIP_INVITE_FAILED;
return ERROR_GB28181_SIP_INVITE_FAILED;
}
sip_session->set_invite_status(SrsGb28181SipSessionTrying);
@ -393,7 +393,7 @@ int SrsGb28181SipService::send_bye(SrsSipRequest *req)
if (send_message(sip_session->sockaddr_from(), sip_session->sockaddr_fromlen(), ss) <= 0)
{
return ERROR_GB28281_SIP_BYE_FAILED;
return ERROR_GB28181_SIP_BYE_FAILED;
}
return ERROR_SUCCESS;
@ -417,7 +417,7 @@ int SrsGb28181SipService::send_sip_raw_data(SrsSipRequest *req, std::string dat
if (send_message(sip_session->sockaddr_from(), sip_session->sockaddr_fromlen(), ss) <= 0)
{
return ERROR_GB28281_SIP_BYE_FAILED;
return ERROR_GB28181_SIP_BYE_FAILED;
}
return ERROR_SUCCESS;

20
trunk/src/app/srs_app_gb28181_sip.hpp
View file

@ -41,7 +41,7 @@ class SrsGb28181Config;
class SrsSipStack;
class SrsGb28181SipService;
enum SrsGb28281SipSessionStatusType{
enum SrsGb28181SipSessionStatusType{
SrsGb28181SipSessionUnkonw = 0,
SrsGb28181SipSessionRegisterOk = 1,
SrsGb28181SipSessionAliveOk = 2,
@ -57,9 +57,9 @@ private:
SrsGb28181SipService *caster;
std::string session_id;
private:
SrsGb28281SipSessionStatusType _register_status;
SrsGb28281SipSessionStatusType _alive_status;
SrsGb28281SipSessionStatusType _invite_status;
SrsGb28181SipSessionStatusType _register_status;
SrsGb28181SipSessionStatusType _alive_status;
SrsGb28181SipSessionStatusType _invite_status;
srs_utime_t _register_time;
srs_utime_t _alive_time;
srs_utime_t _invite_time;
@ -74,9 +74,9 @@ private:
SrsSipRequest *req;
public:
void set_register_status(SrsGb28281SipSessionStatusType s) { _register_status = s;}
void set_alive_status(SrsGb28281SipSessionStatusType s) { _alive_status = s;}
void set_invite_status(SrsGb28281SipSessionStatusType s) { _invite_status = s;}
void set_register_status(SrsGb28181SipSessionStatusType s) { _register_status = s;}
void set_alive_status(SrsGb28181SipSessionStatusType s) { _alive_status = s;}
void set_invite_status(SrsGb28181SipSessionStatusType s) { _invite_status = s;}
void set_register_time(srs_utime_t t) { _register_time = t;}
void set_alive_time(srs_utime_t t) { _alive_time = t;}
void set_invite_time(srs_utime_t t) { _invite_time = t;}
@ -88,9 +88,9 @@ public:
void set_sockaddr_len(int l) { _fromlen = l;}
void set_request(SrsSipRequest *r) { req->copy(r);}
SrsGb28281SipSessionStatusType register_status() { return _register_status;}
SrsGb28281SipSessionStatusType alive_status() { return _alive_status;}
SrsGb28281SipSessionStatusType invite_status() { return _invite_status;}
SrsGb28181SipSessionStatusType register_status() { return _register_status;}
SrsGb28181SipSessionStatusType alive_status() { return _alive_status;}
SrsGb28181SipSessionStatusType invite_status() { return _invite_status;}
srs_utime_t register_time() { return _register_time;}
srs_utime_t alive_time() { return _alive_time;}
srs_utime_t invite_time() { return _invite_time;}

10
trunk/src/app/srs_app_server.cpp
View file

@ -696,7 +696,7 @@ void SrsServer::destroy()
srs_freep(conn_manager);
#ifdef SRS_AUTO_GB28181
//free global gb28281 manager
//free global gb28181 manager
srs_freep(_srs_gb28181);
#endif
}
@ -1349,7 +1349,7 @@ srs_error_t SrsServer::listen_http_stream()
}
#ifdef SRS_AUTO_GB28181
srs_error_t SrsServer::listen_gb28281_sip(SrsConfDirective* stream_caster)
srs_error_t SrsServer::listen_gb28181_sip(SrsConfDirective* stream_caster)
{
srs_error_t err = srs_success;
@ -1400,7 +1400,7 @@ srs_error_t SrsServer::listen_stream_caster()
listener = new SrsHttpFlvListener(this, SrsListenerFlv, stream_caster);
} else if (srs_stream_caster_is_gb28181(caster)) {
#ifdef SRS_AUTO_GB28181
//init global gb28281 manger
//init global gb28181 manger
if (_srs_gb28181 == NULL){
_srs_gb28181 = new SrsGb28181Manger(stream_caster);
if ((err = _srs_gb28181->initialize()) != srs_success){
@ -1410,12 +1410,12 @@ srs_error_t SrsServer::listen_stream_caster()
//sip listener
if (_srs_config->get_stream_caster_gb28181_sip_enable(stream_caster)){
if ((err = listen_gb28281_sip(stream_caster)) != srs_success){
if ((err = listen_gb28181_sip(stream_caster)) != srs_success){
return err;
}
}
//gb28281 stream listener
//gb28181 stream listener
listener = new SrsGb28181Listener(this, SrsListenerGb28181RtpMux, stream_caster);
#else
srs_warn("gb28181 is disabled, please enable it by: ./configure --with-gb28181");

2
trunk/src/app/srs_app_server.hpp
View file

@ -324,7 +324,7 @@ private:
virtual srs_error_t listen_http_stream();
virtual srs_error_t listen_stream_caster();
#ifdef SRS_AUTO_GB28181
virtual srs_error_t listen_gb28281_sip(SrsConfDirective* c);
virtual srs_error_t listen_gb28181_sip(SrsConfDirective* c);
#endif
// Close the listeners for specified type,
// Remove the listen object from manager.