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

refactor gb28181, supporting SIP server enable, multiplex, API interface

Browse source
This commit is contained in:
xialixin 2020-03-31 00:39:10 +08:00
parent 8b4f84e336
commit c99fb99ab6
15 changed files with 2524 additions and 980 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -29,6 +29,7 @@
#include <arpa/inet.h>
#include <string>
#include <vector>
#include <queue>
#include <map>
#include <srs_app_st.hpp>
@ -38,16 +39,14 @@
#include <srs_kernel_stream.hpp>
#include <srs_app_log.hpp>
#include <srs_kernel_file.hpp>
#include <srs_protocol_json.hpp>
#include <srs_app_gb28181_sip.hpp>
#define RTP_PORT_MODE_FIXED "fixed"
#define RTP_PORT_MODE_RANDOM "random"
class SrsStSocket;
class SrsRtpConn;
class SrsRtspConn;
class SrsRtspStack;
class SrsRtspCaster;
class SrsConfDirective;
class SrsRtpPacket;
class SrsRequest;
class SrsStSocket;
class SrsRtmpClient;
class SrsRawH264Stream;
class SrsRawAacStream;
@ -58,52 +57,16 @@ class SrsSimpleStream;
class SrsPithyPrint;
class SrsSimpleRtmpClient;
class SrsSipStack;
class SrsGb28181Caster;
class SrsGb28181Manger;
class SrsRtspJitter;
class SrsRtspAudioCache;
class SrsSipRequest;
class SrsGb28181Conn;
class SrsGb28281ClientInfo;
/* gb28181 program stream struct define
*/
struct SrsPsPacketStartCode
{
uint8_t start_code[3];
uint8_t stream_id[1];
};
struct SrsPsPacketHeader
{
SrsPsPacketStartCode start;// 4
uint8_t info[9];
uint8_t stuffing_length;
};
struct SrsPsPacketBBHeader
{
SrsPsPacketStartCode start;
uint16_t length;
};
struct SrsPsePacket
{
SrsPsPacketStartCode start;
uint16_t length;
uint8_t info[2];
uint8_t stuffing_length;
};
struct SrsPsMapPacket
{
SrsPsPacketStartCode start;
uint16_t length;
};
class SrsGb28181RtmpMuxer;
class SrsGb28181Config;
class SrsGb28181PsRtpProcessor;
class SrsGb28181SipService;
class SrsGb28181StreamChannel;
//ps rtp header packet parse
class SrsPsRtpPacket: public SrsRtpPacket
{
public:
@ -113,19 +76,115 @@ public:
virtual srs_error_t decode(SrsBuffer* stream);
};
// A rtp connection which transport a stream.
class SrsPsRtpConn: public ISrsUdpHandler
//randomly assigned ports receive gb28281 device streams
class SrsPsRtpListener: public ISrsUdpHandler
{
private:
SrsUdpListener* listener;
SrsGb28181PsRtpProcessor* rtp_processor;
int _port;
public:
SrsPsRtpListener(SrsGb28181Config* c, int p, std::string s);
virtual ~SrsPsRtpListener();
public:
virtual int port();
virtual srs_error_t listen();
// Interface ISrsUdpHandler
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
class SrsGb28181RtpMuxService : public ISrsUdpHandler
{
private:
SrsGb28181Config *config;
SrsGb28181PsRtpProcessor *rtp_processor;
public:
SrsGb28181RtpMuxService(SrsConfDirective* c);
virtual ~SrsGb28181RtpMuxService();
// Interface ISrsUdpHandler
public:
virtual srs_error_t on_udp_packet(const sockaddr* from, const int fromlen, char* buf, int nb_buf);
};
//process gb28281 RTP package, generate a completed PS stream data,
//call the PS stream parser, parse the original video and audio
class SrsGb28181PsRtpProcessor: public ISrsUdpHandler
{
private:
SrsPithyPrint* pprint;
SrsUdpListener* listener;
SrsGb28181Conn* gb28181;
SrsPsRtpPacket* cache;
std::map<uint32_t, SrsSimpleStream*> cache_payload;
std::string session_id;
int _port;
uint32_t pre_timestamp;
SrsGb28181Config* config;
std::map<std::string, SrsPsRtpPacket*> cache_ps_rtp_packet;
std::map<std::string, SrsPsRtpPacket*> pre_packet;
std::string channel_id;
bool auto_create_channel;
public:
SrsGb28181PsRtpProcessor(SrsGb28181Config* c, std::string sid);
virtual ~SrsGb28181PsRtpProcessor();
private:
bool can_send_ps_av_packet();
void dispose();
// Interface ISrsUdpHandler
public:
virtual srs_error_t on_udp_packet(const sockaddr* from, const int fromlen, char* buf, int nb_buf);
};
//ps stream processing parsing interface
class ISrsPsStreamHander
{
public:
ISrsPsStreamHander();
virtual ~ISrsPsStreamHander();
public:
virtual srs_error_t on_rtp_video(SrsSimpleStream* stream, int64_t dts)=0;
virtual srs_error_t on_rtp_audio(SrsSimpleStream* stream, int64_t dts)=0;
};
//analysis of PS stream and
//extraction of H264 raw data and audio data
//then process the flow through PS stream hander,
//such as RTMP multiplexer, and composited into RTMP av stream
class SrsPsStreamDemixer
{
public:
// gb28181 program stream struct define
struct SrsPsPacketStartCode
{
uint8_t start_code[3];
uint8_t stream_id[1];
};
struct SrsPsPacketHeader
{
SrsPsPacketStartCode start;// 4
uint8_t info[9];
uint8_t stuffing_length;
};
struct SrsPsPacketBBHeader
{
SrsPsPacketStartCode start;
uint16_t length;
};
struct SrsPsePacket
{
SrsPsPacketStartCode start;
uint16_t length;
uint8_t info[2];
uint8_t stuffing_length;
};
struct SrsPsMapPacket
{
SrsPsPacketStartCode start;
uint16_t length;
};
private:
SrsFileWriter ps_fw;
SrsFileWriter video_fw;
SrsFileWriter audio_fw;
@ -133,117 +192,89 @@ private:
bool first_keyframe_flag;
bool wait_first_keyframe;
bool audio_enable;
std::string channel_id;
ISrsPsStreamHander *hander;
public:
SrsPsRtpConn(SrsGb28181Conn* r, int p, std::string sid, bool a, bool k);
virtual ~SrsPsRtpConn();
private:
int64_t parse_ps_timestamp(const uint8_t* p);
SrsPsStreamDemixer(ISrsPsStreamHander *h, std::string sid, bool a, bool k);
virtual ~SrsPsStreamDemixer();
private:
bool can_send_ps_av_packet();
void dispose();
public:
virtual int port();
virtual srs_error_t listen();
// Interface ISrsUdpHandler
public:
virtual srs_error_t on_udp_packet(const sockaddr* from, const int fromlen, char* buf, int nb_buf);
virtual srs_error_t on_ps_stream(char* ps_data, int ps_size, uint32_t timestamp);
int64_t parse_ps_timestamp(const uint8_t* p);
virtual srs_error_t on_ps_stream(char* ps_data, int ps_size, uint32_t timestamp, uint32_t ssrc);
};
class SrsGb28281ClientInfo {
public:
SrsGb28281ClientInfo();
virtual ~SrsGb28281ClientInfo();
public:
sockaddr* sock_from;
int sock_fromlen;
srs_netfd_t stfd;
SrsSipRequest *req;
};
enum Srs28181CtrlStatusType{
Srs28181Unkonw = 0,
Srs28181RegisterOk = 1,
Srs28181AliveOk = 2,
Srs28181InviteOk = 3,
Srs28181Trying = 4,
Srs28181Bye = 5,
};
class SrsGb28181Conn : public ISrsCoroutineHandler, public ISrsConnection
//RTMP multiplexer, which processes the raw H264 / AAC,
//then publish it to RTMP server
class SrsGb28181RtmpMuxer : public ISrsCoroutineHandler,
public ISrsConnection, public ISrsPsStreamHander
{
private:
std::string output_template;
SrsPithyPrint* pprint;
public:
Srs28181CtrlStatusType register_status;
Srs28181CtrlStatusType alive_status;
Srs28181CtrlStatusType invite_status;
srs_utime_t register_time;
srs_utime_t alive_time;
srs_utime_t invite_time;
srs_utime_t recv_rtp_time;
std::string rtmp_url;
int reg_expires;
SrsGb28181StreamChannel *channel;
int stream_idle_timeout;
srs_utime_t recv_stream_time;
private:
std::string session_id;
// video stream.
int video_id;
std::string video_codec;
SrsPsRtpConn* video_rtp;
// audio stream.
int audio_id;
std::string audio_codec;
std::string channel_id;
std::string _rtmp_url;
std::string video_ssrc;
std::string audio_ssrc;
int audio_sample_rate;
int audio_channel;
SrsPsRtpConn* audio_rtp;
public:
SrsGb28281ClientInfo* info;
private:
SrsStSocket* skt;
SrsSipStack* sip;
SrsGb28181Caster* caster;
SrsGb28181Manger* gb28181_manger;
SrsCoroutine* trd;
private:
SrsSipRequest* req;
SrsPsStreamDemixer* ps_demixer;
srs_cond_t wait_ps_queue;
SrsSimpleRtmpClient* sdk;
SrsRtspJitter* vjitter;
SrsRtspJitter* ajitter;
private:
SrsRawH264Stream* avc;
std::string h264_sps;
std::string h264_pps;
bool h264_sps_changed;
bool h264_pps_changed;
bool h264_sps_pps_sent;
private:
SrsRawAacStream* aac;
std::string aac_specific_config;
public:
SrsGb28181Conn(SrsGb28181Caster* c, std::string id);
virtual ~SrsGb28181Conn();
std::queue<SrsPsRtpPacket*> ps_queue;
public:
SrsGb28181RtmpMuxer(SrsGb28181Manger* m, std::string id, bool a, bool k);
virtual ~SrsGb28181RtmpMuxer();
public:
virtual srs_error_t serve();
virtual std::string remote_ip();
virtual void set_request_info(SrsSipRequest *req);
virtual std::string get_session_id();
virtual void stop();
virtual std::string get_channel_id();
virtual void ps_packet_enqueue(SrsPsRtpPacket *pkt);
virtual void copy_channel(SrsGb28181StreamChannel *s);
virtual void set_channel_peer_ip(std::string ip);
virtual void set_channel_peer_port(int port);
virtual int channel_peer_port();
virtual std::string channel_peer_ip();
virtual void set_rtmp_url(std::string url);
virtual std::string rtmp_url();
virtual SrsGb28181StreamChannel get_channel();
private:
virtual srs_error_t do_cycle();
// internal methods
public:
virtual srs_error_t start_rtp_listen(int port);
virtual srs_error_t stop_rtp_listen();
virtual void destroy();
// Interface ISrsOneCycleThreadHandler
public:
virtual srs_error_t cycle();
virtual std::string remote_ip();
public:
virtual srs_error_t on_rtp_video(SrsSimpleStream* stream, int64_t dts, int keyframe);
virtual srs_error_t on_rtp_video(SrsSimpleStream* stream, int64_t dts);
virtual srs_error_t on_rtp_audio(SrsSimpleStream* stream, int64_t dts);
private:
virtual srs_error_t write_h264_sps_pps(uint32_t dts, uint32_t pts);
@ -255,75 +286,152 @@ private:
virtual srs_error_t connect();
// Close the connection to RTMP server.
virtual void close();
public:
virtual void rtmp_close();
};
//system parameter configuration of gb28281 module,
//read file from configuration file to generate
class SrsGb28181Config
{
public:
std::string sip_host;
std::string sip_port;
std::string sip_serial;
std::string sip_realm;
int sip_ack_timeout;
int sip_keepalive_timeout;
std::string host;
int rtp_idle_timeout;
bool audio_enable;
bool wait_keyframe;
std::string output;
int rtp_port_min;
int rtp_port_max;
int listen_port;
int rtp_mux_port;
//sip config
int sip_port;
std::string sip_serial;
std::string sip_realm;
bool sip_enable;
int sip_ack_timeout;
int sip_keepalive_timeout;
bool print_sip_message;
bool wait_keyframe;
bool sip_auto_play;
bool sip_invite_port_fixed;
public:
SrsGb28181Config(SrsConfDirective* c);
virtual ~SrsGb28181Config();
};
//gb28181 conn manager
class SrsGb28181Caster : public ISrsUdpHandler
class SrsGb28181StreamChannel
{
private:
std::string channel_id;
std::string port_mode;
std::string app;
std::string stream;
std::string ip;
int rtp_port;
int rtmp_port;
uint32_t ssrc;
//send rtp stream client local port
int rtp_peer_port;
//send rtp stream client local ip
std::string rtp_peer_ip;
public:
SrsGb28181StreamChannel();
virtual ~SrsGb28181StreamChannel();
std::string get_channel_id() const { return channel_id; }
std::string get_port_mode() const { return port_mode; }
std::string get_app() const { return app; }
std::string get_stream() const { return stream; }
std::string get_ip() const { return ip; }
int get_rtp_port() const { return rtp_port; }
int get_rtmp_port() const { return rtmp_port; }
uint32_t get_ssrc() const { return ssrc; }
uint32_t get_rtp_peer_port() const { return rtp_peer_port; }
std::string get_rtp_peer_ip() const { return rtp_peer_ip; }
void set_channel_id(const std::string &i) { channel_id = i; }
void set_port_mode(const std::string &p) { port_mode = p; }
void set_app(const std::string &a) { app = a; }
void set_stream(const std::string &s) { stream = s; }
void set_ip(const std::string &i) { ip = i; }
void set_rtp_port( const int &p) { rtp_port = p; }
void set_rtmp_port( const int &p) { rtmp_port = p; }
void set_ssrc( const int &s) { ssrc = s;}
void set_rtp_peer_ip( const std::string &p) { rtp_peer_ip = p; }
void set_rtp_peer_port( const int &s) { rtp_peer_port = s;}
void copy(const SrsGb28181StreamChannel *s);
void dumps(SrsJsonObject* obj);
};
// Global singleton instance.
extern SrsGb28181Manger* _srs_gb28181;
//gb28181 module management, management of all RTMP multiplexers,
//random assignment of RTP listeners, and external control interfaces
class SrsGb28181Manger
{
private:
SrsGb28181Config *config;
// The key: port, value: whether used.
std::map<int, bool> used_ports;
SrsSipStack *sip;
srs_netfd_t lfd;
private:
std::map<std::string, SrsGb28181Conn*> clients;
std::map<uint32_t, SrsPsRtpListener*> rtp_pool;
std::map<uint32_t, SrsGb28181RtmpMuxer*> rtmpmuxers_ssrc;
std::map<std::string, SrsGb28181RtmpMuxer*> rtmpmuxers;
SrsCoroutineManager* manager;
SrsGb28181SipService* sip_service;
public:
SrsGb28181Caster(SrsConfDirective* c);
virtual ~SrsGb28181Caster();
SrsGb28181Manger(SrsConfDirective* c);
virtual ~SrsGb28181Manger();
public:
srs_error_t fetch_or_create_rtmpmuxer(std::string id, SrsGb28181RtmpMuxer** gb28181);
SrsGb28181RtmpMuxer* fetch_rtmpmuxer(std::string id);
SrsGb28181RtmpMuxer* fetch_rtmpmuxer_by_ssrc(uint32_t ssrc);
void rtmpmuxer_map_by_ssrc(SrsGb28181RtmpMuxer*muxer, uint32_t ssrc);
void rtmpmuxer_unmap_by_ssrc(uint32_t ssrc);
uint32_t generate_ssrc(std::string id);
uint32_t hash_code(std::string str);
void set_sip_service(SrsGb28181SipService *s) { sip_service = s; }
SrsGb28181SipService* get_sip_service() { return sip_service; }
public:
//stream channel api
uint32_t create_stream_channel(SrsGb28181StreamChannel *channel);
uint32_t delete_stream_channel(std::string id);
uint32_t queue_stream_channel(std::string id, SrsJsonArray* arr);
//sip api
uint32_t notify_sip_invite(std::string id, std::string ip, int port, uint32_t ssrc);
uint32_t notify_sip_bye(std::string id);
uint32_t notify_sip_raw_data(std::string id, std::string data);
uint32_t notify_sip_unregister(std::string id);
private:
srs_error_t fetch_or_create(SrsSipRequest* r, SrsGb28181Conn** gb28181);
virtual SrsGb28181Conn* fetch(const SrsSipRequest* r);
virtual void destroy();
void destroy();
public:
// Alloc a rtp port from local ports pool.
// @param pport output the rtp port.
virtual srs_error_t alloc_port(int* pport);
void alloc_port(int* pport);
// Free the alloced rtp port.
virtual void free_port(int lpmin, int lpmax);
virtual srs_error_t initialize();
void free_port(int lpmin, int lpmax);
srs_error_t initialize();
virtual void set_stfd(srs_netfd_t fd);
virtual SrsGb28181Config GetGb28181Config();
SrsGb28181Config get_gb28181_config();
srs_error_t start_ps_rtp_listen(std::string id, int port);
void stop_rtp_listen(std::string id);
// Interface ISrsUdpHandler
public:
virtual srs_error_t on_udp_packet(const sockaddr* from, const int fromlen, char* buf, int nb_buf);
private:
virtual srs_error_t on_udp_bytes(std::string host, int port, char* buf, int nb_buf, sockaddr* from, int fromlen);
// internal methods.
public:
virtual srs_error_t send_message(sockaddr* f, int l, std::stringstream& ss);
virtual srs_error_t send_bye(SrsSipRequest *req, sockaddr *f, int l);
virtual srs_error_t send_ack(SrsSipRequest *req, sockaddr *f, int l);
virtual srs_error_t send_invite(SrsSipRequest *req, sockaddr *f, int l, int port);
virtual srs_error_t send_status(SrsSipRequest *req, sockaddr *f, int l);
virtual void remove(SrsGb28181Conn* conn);
void remove(SrsGb28181RtmpMuxer* conn);
};
#endif