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

Merge GB28181

Browse source
This commit is contained in:
winlin 2020-03-31 22:45:20 +08:00
commit e19b927f92
18 changed files with 4428 additions and 52 deletions
Download patch file Download diff file Expand all files Collapse all files

336
trunk/src/app/srs_app_config.cpp
View file

@ -275,6 +275,11 @@ bool srs_stream_caster_is_flv(string caster)
return caster == "flv";
}
bool srs_stream_caster_is_gb28181(string caster)
{
return caster == "gb28181";
}
bool srs_config_apply_filter(SrsConfDirective* dvr_apply, SrsRequest* req)
{
static bool DEFAULT = true;
@ -2137,7 +2142,30 @@ srs_error_t SrsConfig::global_to_json(SrsJsonObject* obj)
sobj->set(sdir->name, sdir->dumps_arg0_to_integer());
} else if (sdir->name == "rtp_port_max") {
sobj->set(sdir->name, sdir->dumps_arg0_to_integer());
} else if (sdir->name == "rtp_idle_timeout") {
sobj->set(sdir->name, sdir->dumps_arg0_to_integer());
} else if (sdir->name == "ack_timeout") {
sobj->set(sdir->name, sdir->dumps_arg0_to_integer());
} else if (sdir->name == "keepalive_timeout") {
sobj->set(sdir->name, sdir->dumps_arg0_to_integer());
} else if (sdir->name == "audio_enable") {
sobj->set(sdir->name, sdir->dumps_arg0_to_boolean());
} else if (sdir->name == "host") {
sobj->set(sdir->name, sdir->dumps_arg0_to_str());
} else if (sdir->name == "serial") {
sobj->set(sdir->name, sdir->dumps_arg0_to_str());
} else if (sdir->name == "realm") {
sobj->set(sdir->name, sdir->dumps_arg0_to_str());
} else if (sdir->name == "wait_keyframe") {
sobj->set(sdir->name, sdir->dumps_arg0_to_str());
} else if (sdir->name == "print_sip_message") {
sobj->set(sdir->name, sdir->dumps_arg0_to_str());
} else if (sdir->name == "invite_port_fixed") {
sobj->set(sdir->name, sdir->dumps_arg0_to_str());
} else if (sdir->name == "auto_play") {
sobj->set(sdir->name, sdir->dumps_arg0_to_str());
}
}
obj->set(dir->name, sobj);
} else {
@ -3650,9 +3678,25 @@ srs_error_t SrsConfig::check_normal_config()
SrsConfDirective* conf = stream_caster->at(i);
string n = conf->name;
if (n != "enabled" && n != "caster" && n != "output"
&& n != "listen" && n != "rtp_port_min" && n != "rtp_port_max") {
&& n != "listen" && n != "rtp_port_min" && n != "rtp_port_max"
&& n != "rtp_idle_timeout" && n != "sip"
&& n != "audio_enable" && n != "wait_keyframe"
&& n != "host") {
return srs_error_new(ERROR_SYSTEM_CONFIG_INVALID, "illegal stream_caster.%s", n.c_str());
}
if (n == "sip") {
for (int j = 0; j < (int)conf->directives.size(); j++) {
string m = conf->at(j)->name;
if (m != "enabled" && m != "listen"
&& m != "ack_timeout" && m != "keepalive_timeout"
&& m != "host" && m != "serial" && m != "realm"
&& m != "print_sip_message" && m != "auto_play"
&& m != "invite_port_fixed") {
return srs_error_new(ERROR_SYSTEM_CONFIG_INVALID, "illegal stream_caster.%s", m.c_str());
}
}
}
}
}
@ -4273,6 +4317,262 @@ int SrsConfig::get_stream_caster_rtp_port_max(SrsConfDirective* conf)
return ::atoi(conf->arg0().c_str());
}
srs_utime_t SrsConfig::get_stream_caster_gb28181_rtp_idle_timeout(SrsConfDirective* conf)
{
static srs_utime_t DEFAULT = 30 * SRS_UTIME_SECONDS;
if (!conf) {
return DEFAULT;
}
conf = conf->get("rtp_idle_timeout");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return (srs_utime_t)(::atoi(conf->arg0().c_str()) * SRS_UTIME_SECONDS);
}
int SrsConfig::get_stream_caster_gb28181_ack_timeout(SrsConfDirective* conf)
{
static int DEFAULT = 30;
if (!conf) {
return DEFAULT;
}
conf = conf->get("sip");
if (!conf) {
return DEFAULT;
}
conf = conf->get("ack_timeout");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return ::atoi(conf->arg0().c_str());
}
int SrsConfig::get_stream_caster_gb28181_keepalive_timeout(SrsConfDirective* conf)
{
static int DEFAULT = 120;
if (!conf) {
return DEFAULT;
}
conf = conf->get("sip");
if (!conf) {
return DEFAULT;
}
conf = conf->get("keepalive_timeout");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return ::atoi(conf->arg0().c_str());
}
string SrsConfig::get_stream_caster_gb28181_host(SrsConfDirective* conf)
{
static string DEFAULT = "";
if (!conf) {
return DEFAULT;
}
conf = conf->get("host");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return conf->arg0();
}
string SrsConfig::get_stream_caster_gb28181_serial(SrsConfDirective* conf)
{
static string DEFAULT = "34020000002000000001";
if (!conf) {
return DEFAULT;
}
conf = conf->get("sip");
if (!conf) {
return DEFAULT;
}
conf = conf->get("serial");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return conf->arg0();
}
string SrsConfig::get_stream_caster_gb28181_realm(SrsConfDirective* conf)
{
static string DEFAULT = "3402000000";
if (!conf) {
return DEFAULT;
}
conf = conf->get("sip");
if (!conf) {
return DEFAULT;
}
conf = conf->get("realm");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return conf->arg0();
}
bool SrsConfig::get_stream_caster_gb28181_audio_enable(SrsConfDirective* conf)
{
static bool DEFAULT = false;
if (!conf) {
return DEFAULT;
}
conf = conf->get("audio_enable");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return SRS_CONF_PERFER_FALSE(conf->arg0());
}
bool SrsConfig::get_stream_caster_gb28181_print_sip_message(SrsConfDirective* conf)
{
static bool DEFAULT = false;
if (!conf) {
return DEFAULT;
}
conf = conf->get("sip");
if (!conf) {
return DEFAULT;
}
conf = conf->get("print_sip_message");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return SRS_CONF_PERFER_FALSE(conf->arg0());
}
bool SrsConfig::get_stream_caster_gb28181_wait_keyframe(SrsConfDirective* conf)
{
static bool DEFAULT = false;
if (!conf) {
return DEFAULT;
}
conf = conf->get("wait_keyframe");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return SRS_CONF_PERFER_FALSE(conf->arg0());
}
bool SrsConfig::get_stream_caster_gb28181_sip_enable(SrsConfDirective* conf)
{
static bool DEFAULT = true;
if (!conf) {
return DEFAULT;
}
conf = conf->get("sip");
if (!conf) {
return DEFAULT;
}
conf = conf->get("enabled");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return SRS_CONF_PERFER_FALSE(conf->arg0());
}
bool SrsConfig::get_stream_caster_gb28181_sip_auto_play(SrsConfDirective* conf)
{
static bool DEFAULT = true;
if (!conf) {
return DEFAULT;
}
conf = conf->get("sip");
if (!conf) {
return DEFAULT;
}
conf = conf->get("auto_play");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return SRS_CONF_PERFER_FALSE(conf->arg0());
}
int SrsConfig::get_stream_caster_gb28181_sip_listen(SrsConfDirective* conf)
{
static int DEFAULT = 5060;
if (!conf) {
return DEFAULT;
}
conf = conf->get("sip");
if (!conf) {
return DEFAULT;
}
conf = conf->get("listen");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return ::atoi(conf->arg0().c_str());
}
bool SrsConfig::get_stream_caster_gb28181_sip_invite_port_fixed(SrsConfDirective* conf)
{
static bool DEFAULT = true;
if (!conf) {
return DEFAULT;
}
conf = conf->get("sip");
if (!conf) {
return DEFAULT;
}
conf = conf->get("invite_port_fixed");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return SRS_CONF_PERFER_FALSE(conf->arg0());
}
int SrsConfig::get_rtc_server_enabled()
{
SrsConfDirective* conf = root->get("rtc_server");
@ -4282,45 +4582,45 @@ int SrsConfig::get_rtc_server_enabled()
bool SrsConfig::get_rtc_server_enabled(SrsConfDirective* conf)
{
static bool DEFAULT = false;
if (!conf) {
return DEFAULT;
}
conf = conf->get("enabled");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return SRS_CONF_PERFER_FALSE(conf->arg0());
}
int SrsConfig::get_rtc_server_listen()
{
static int DEFAULT = 8000;
SrsConfDirective* conf = root->get("rtc_server");
if (!conf) {
return DEFAULT;
}
conf = conf->get("listen");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return ::atoi(conf->arg0().c_str());
}
std::string SrsConfig::get_rtc_server_candidates()
{
static string DEFAULT = "*";
SrsConfDirective* conf = root->get("rtc_server");
if (!conf) {
return DEFAULT;
}
conf = conf->get("candidate");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
@ -4335,7 +4635,7 @@ std::string SrsConfig::get_rtc_server_candidates()
if (srs_string_starts_with(conf->arg0(), "$")) {
return DEFAULT;
}
return (conf->arg0().c_str());
}
@ -4348,36 +4648,36 @@ SrsConfDirective* SrsConfig::get_rtc(string vhost)
bool SrsConfig::get_rtc_enabled(string vhost)
{
static bool DEFAULT = false;
SrsConfDirective* conf = get_rtc(vhost);
if (!conf) {
return DEFAULT;
}
conf = conf->get("enabled");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return SRS_CONF_PERFER_FALSE(conf->arg0());
}
bool SrsConfig::get_rtc_bframe_discard(string vhost)
{
static bool DEFAULT = false;
SrsConfDirective* conf = get_rtc(vhost);
if (!conf) {
return DEFAULT;
}
conf = conf->get("bframe");
if (!conf || conf->arg0().empty()) {
return DEFAULT;
}
return conf->arg0() == "discard";
}

15
trunk/src/app/srs_app_config.hpp
View file

@ -119,6 +119,7 @@ extern bool srs_config_dvr_is_plan_session(std::string plan);
extern bool srs_stream_caster_is_udp(std::string caster);
extern bool srs_stream_caster_is_rtsp(std::string caster);
extern bool srs_stream_caster_is_flv(std::string caster);
extern bool srs_stream_caster_is_gb28181(std::string caster);
// Whether the dvr_apply active the stream specified by req.
extern bool srs_config_apply_filter(SrsConfDirective* dvr_apply, SrsRequest* req);
@ -499,6 +500,20 @@ public:
// Get the max udp port for rtp of stream caster rtsp.
virtual int get_stream_caster_rtp_port_max(SrsConfDirective* conf);
virtual srs_utime_t get_stream_caster_gb28181_rtp_idle_timeout(SrsConfDirective* conf);
virtual int get_stream_caster_gb28181_ack_timeout(SrsConfDirective* conf);
virtual int get_stream_caster_gb28181_keepalive_timeout(SrsConfDirective* conf);
virtual bool get_stream_caster_gb28181_audio_enable(SrsConfDirective* conf);
virtual std::string get_stream_caster_gb28181_host(SrsConfDirective* conf);
virtual std::string get_stream_caster_gb28181_serial(SrsConfDirective* conf);
virtual std::string get_stream_caster_gb28181_realm(SrsConfDirective* conf);
virtual bool get_stream_caster_gb28181_print_sip_message(SrsConfDirective* conf);
virtual bool get_stream_caster_gb28181_wait_keyframe(SrsConfDirective* conf);
virtual bool get_stream_caster_gb28181_sip_enable(SrsConfDirective* conf);
virtual bool get_stream_caster_gb28181_sip_auto_play(SrsConfDirective* conf);
virtual int get_stream_caster_gb28181_sip_listen(SrsConfDirective* conf);
virtual bool get_stream_caster_gb28181_sip_invite_port_fixed(SrsConfDirective* conf);
// rtc section
public:
virtual int get_rtc_server_enabled();

1647
trunk/src/app/srs_app_gb28181.cpp Normal file
View file

File diff suppressed because it is too large Load diff

438
trunk/src/app/srs_app_gb28181.hpp Normal file
View file

@ -0,0 +1,438 @@
/**
* The MIT License (MIT)
*
* Copyright (c) 2013-2020 Lixin
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to
* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
* the Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
* IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_APP_GB28181_HPP
#define SRS_APP_GB28181_HPP
#include <srs_core.hpp>
#include <arpa/inet.h>
#include <string>
#include <vector>
#include <queue>
#include <map>
#include <srs_app_st.hpp>
#include <srs_app_thread.hpp>
#include <srs_app_listener.hpp>
#include <srs_rtsp_stack.hpp>
#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 SrsConfDirective;
class SrsRtpPacket;
class SrsRtmpClient;
class SrsRawH264Stream;
class SrsRawAacStream;
struct SrsRawAacStreamCodec;
class SrsSharedPtrMessage;
class SrsAudioFrame;
class SrsSimpleStream;
class SrsPithyPrint;
class SrsSimpleRtmpClient;
class SrsSipStack;
class SrsGb28181Manger;
class SrsRtspJitter;
class SrsSipRequest;
class SrsGb28181RtmpMuxer;
class SrsGb28181Config;
class SrsGb28181PsRtpProcessor;
class SrsGb28181SipService;
class SrsGb28181StreamChannel;
//ps rtp header packet parse
class SrsPsRtpPacket: public SrsRtpPacket
{
public:
SrsPsRtpPacket();
virtual ~SrsPsRtpPacket();
public:
virtual srs_error_t decode(SrsBuffer* stream);
};
//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;
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;
bool first_keyframe_flag;
bool wait_first_keyframe;
bool audio_enable;
std::string channel_id;
ISrsPsStreamHander *hander;
public:
SrsPsStreamDemixer(ISrsPsStreamHander *h, std::string sid, bool a, bool k);
virtual ~SrsPsStreamDemixer();
private:
bool can_send_ps_av_packet();
public:
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);
};
//RTMP multiplexer, which processes the raw H264 / AAC,
//then publish it to RTMP server
class SrsGb28181RtmpMuxer : public ISrsCoroutineHandler,
public ISrsConnection, public ISrsPsStreamHander
{
private:
SrsPithyPrint* pprint;
SrsGb28181StreamChannel *channel;
int stream_idle_timeout;
srs_utime_t recv_stream_time;
private:
std::string channel_id;
std::string _rtmp_url;
std::string video_ssrc;
std::string audio_ssrc;
int audio_sample_rate;
int audio_channel;
SrsGb28181Manger* gb28181_manger;
SrsCoroutine* trd;
SrsPsStreamDemixer* ps_demixer;
srs_cond_t wait_ps_queue;
SrsSimpleRtmpClient* sdk;
SrsRtspJitter* vjitter;
SrsRtspJitter* ajitter;
SrsRawH264Stream* avc;
std::string h264_sps;
std::string h264_pps;
bool h264_sps_changed;
bool h264_pps_changed;
bool h264_sps_pps_sent;
SrsRawAacStream* aac;
std::string aac_specific_config;
public:
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 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();
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);
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);
virtual srs_error_t write_h264_ipb_frame(char* frame, int frame_size, uint32_t dts, uint32_t pts);
virtual srs_error_t write_audio_raw_frame(char* frame, int frame_size, SrsRawAacStreamCodec* codec, uint32_t dts);
virtual srs_error_t rtmp_write_packet(char type, uint32_t timestamp, char* data, int size);
private:
// Connect to RTMP server.
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 host;
srs_utime_t rtp_idle_timeout;
bool audio_enable;
bool wait_keyframe;
std::string output;
int rtp_port_min;
int rtp_port_max;
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 sip_auto_play;
bool sip_invite_port_fixed;
public:
SrsGb28181Config(SrsConfDirective* c);
virtual ~SrsGb28181Config();
};
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;
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:
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:
void destroy();
public:
// Alloc a rtp port from local ports pool.
// @param pport output the rtp port.
void alloc_port(int* pport);
// Free the alloced rtp port.
void free_port(int lpmin, int lpmax);
srs_error_t initialize();
SrsGb28181Config get_gb28181_config();
srs_error_t start_ps_rtp_listen(std::string id, int port);
void stop_rtp_listen(std::string id);
public:
void remove(SrsGb28181RtmpMuxer* conn);
};
#endif

471
trunk/src/app/srs_app_gb28181_sip.cpp Normal file
View file

@ -0,0 +1,471 @@
/**
* The MIT License (MIT)
*
* Copyright (c) 2013-2020 Lixin
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to
* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
* the Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
* IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include <srs_app_gb28181_sip.hpp>
#include <algorithm>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
using namespace std;
#include <srs_app_config.hpp>
#include <srs_kernel_error.hpp>
#include <srs_kernel_log.hpp>
#include <srs_app_utility.hpp>
#include <srs_core_autofree.hpp>
#include <srs_kernel_buffer.hpp>
#include <srs_kernel_stream.hpp>
#include <srs_protocol_utility.hpp>
#include <srs_kernel_utility.hpp>
#include <srs_kernel_codec.hpp>
#include <srs_app_pithy_print.hpp>
#include <srs_sip_stack.hpp>
#include <srs_app_gb28181.hpp>
SrsGb28181SipSession::SrsGb28181SipSession(SrsGb28181SipService *c, SrsSipRequest* r)
{
caster = c;
req = new SrsSipRequest();
req->copy(r);
_register_status = SrsGb28181SipSessionUnkonw;
_alive_status = SrsGb28181SipSessionUnkonw;
_invite_status = SrsGb28181SipSessionUnkonw;
_register_time = 0;
_alive_time = 0;
_invite_time = 0;
_recv_rtp_time = 0;
_reg_expires = 0;
_peer_ip = "";
_peer_port = 0;
_from = NULL;
_fromlen = 0;
}
SrsGb28181SipSession::~SrsGb28181SipSession()
{
srs_freep(req);
}
//gb28181 sip Service
SrsGb28181SipService::SrsGb28181SipService(SrsConfDirective* c)
{
// TODO: FIXME: support reload.
config = new SrsGb28181Config(c);
sip = new SrsSipStack();
if (_srs_gb28181){
_srs_gb28181->set_sip_service(this);
}
}
SrsGb28181SipService::~SrsGb28181SipService()
{
destroy();
srs_freep(sip);
srs_freep(config);
}
void SrsGb28181SipService::set_stfd(srs_netfd_t fd)
{
lfd = fd;
}
srs_error_t SrsGb28181SipService::on_udp_packet(const sockaddr* from, const int fromlen, char* buf, int nb_buf)
{
char address_string[64];
char port_string[16];
if(getnameinfo(from, fromlen,
(char*)&address_string, sizeof(address_string),
(char*)&port_string, sizeof(port_string),
NI_NUMERICHOST|NI_NUMERICSERV)) {
return srs_error_new(ERROR_SYSTEM_IP_INVALID, "bad address");
}
std::string peer_ip = std::string(address_string);
int peer_port = atoi(port_string);
srs_error_t err = on_udp_sip(peer_ip, peer_port, buf, nb_buf, (sockaddr*)from, fromlen);
if (err != srs_success) {
return srs_error_wrap(err, "process udp");
}
return err;
}
srs_error_t SrsGb28181SipService::on_udp_sip(string peer_ip, int peer_port,
char* buf, int nb_buf, sockaddr* from, const int fromlen)
{
srs_error_t err = srs_success;
if (config->print_sip_message)
{
srs_trace("gb28181: request peer_ip=%s, peer_port=%d nbbuf=%d", peer_ip.c_str(), peer_port, nb_buf);
srs_trace("gb28181: request recv message=%s", buf);
}
if (nb_buf < 10) {
return err;
}
SrsSipRequest* req = NULL;
if ((err = sip->parse_request(&req, buf, nb_buf)) != srs_success) {
return srs_error_wrap(err, "parse sip request");
}
if (config->print_sip_message)
{
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());
}
req->peer_ip = peer_ip;
req->peer_port = peer_port;
SrsAutoFree(SrsSipRequest, req);
std::string session_id = req->sip_auth_id;
if (req->is_register()) {
std::vector<std::string> serial = srs_string_split(srs_string_replace(req->uri,"sip:", ""), "@");
if (serial.at(0) != config->sip_serial){
srs_trace("gb28181: client:%s request serial and server serial inconformity(%s:%s)",
req->sip_auth_id.c_str(), serial.at(0).c_str(), config->sip_serial.c_str());
return err;
}
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());
SrsGb28181SipSession* sip_session = create_sip_session(req);
if (!sip_session) {
srs_trace("gb28181: create sip session faild:%s", req->uri.c_str());
return err;
}
send_status(req, from, fromlen);
sip_session->set_register_status(SrsGb28181SipSessionRegisterOk);
sip_session->set_register_time(srs_get_system_time());
sip_session->set_reg_expires(req->expires);
sip_session->set_sockaddr(from);
sip_session->set_sockaddr_len(fromlen);
sip_session->set_peer_ip(peer_ip);
sip_session->set_peer_port(peer_port);
}else if (req->is_message()) {
SrsGb28181SipSession* sip_session = fetch(session_id);
if (!sip_session || sip_session->register_status() == SrsGb28181SipSessionUnkonw){
srs_trace("gb28181: %s client not registered", req->sip_auth_id.c_str());
return err;
}
//reponse status
send_status(req, from, fromlen);
sip_session->set_register_status(SrsGb28181SipSessionRegisterOk);
sip_session->set_register_time(srs_get_system_time());
sip_session->set_alive_status(SrsGb28181SipSessionAliveOk);
sip_session->set_alive_time(srs_get_system_time());
sip_session->set_sockaddr(from);
sip_session->set_sockaddr_len(fromlen);
sip_session->set_peer_port(peer_port);
sip_session->set_peer_ip(peer_ip);
//send invite, play client av
//start ps rtp listen, recv ps stream
if (config->sip_auto_play && sip_session->register_status() == SrsGb28181SipSessionRegisterOk &&
sip_session->alive_status() == SrsGb28181SipSessionAliveOk &&
sip_session->invite_status() == SrsGb28181SipSessionUnkonw)
{
//stop the possible stream and push a new stream
//send_bye(req, from, fromlen);
SrsGb28181StreamChannel ch;
ch.set_channel_id(session_id);
ch.set_ip(config->host);
ch.set_stream(session_id);
ch.set_app("live");
if (config->sip_invite_port_fixed){
ch.set_port_mode(RTP_PORT_MODE_FIXED);
}else {
ch.set_port_mode(RTP_PORT_MODE_RANDOM);
}
int code = _srs_gb28181->create_stream_channel(&ch);
if (code == ERROR_SUCCESS){
code = send_invite(req, ch.get_ip(),
ch.get_rtp_port(), ch.get_ssrc());
}
if (code == ERROR_SUCCESS){
sip_session->set_invite_status(SrsGb28181SipSessionTrying);
sip_session->set_invite_time(srs_get_system_time());
}
}
}else if (req->is_invite()) {
SrsGb28181SipSession* sip_session = fetch(session_id);
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());
if (!sip_session){
send_bye(req);
srs_trace("gb28181: %s client not registered", req->sip_auth_id.c_str());
return err;
}
if (sip_session->register_status() == SrsGb28181SipSessionUnkonw ||
sip_session->alive_status() == SrsGb28181SipSessionUnkonw) {
srs_trace("gb28181: %s client not registered or not alive", req->sip_auth_id.c_str());
return err;
}
if (req->cmdtype == SrsSipCmdRespone && req->status == "200") {
srs_trace("gb28181: INVITE response %s client status=%s", req->sip_auth_id.c_str(), req->status.c_str());
send_ack(req, from, fromlen);
sip_session->set_invite_status(SrsGb28181SipSessionInviteOk);
sip_session->set_invite_time(srs_get_system_time());
//Record tag and branch, which are required by the 'bye' command,
sip_session->set_request(req);
}else{
sip_session->set_invite_status(SrsGb28181SipSessionUnkonw);
sip_session->set_invite_time(0);
}
}else if (req->is_bye()) {
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());
SrsGb28181SipSession* sip_session = fetch(session_id);
send_status(req, from, fromlen);
if (!sip_session){
srs_trace("gb28181: %s client not registered", req->sip_auth_id.c_str());
return err;
}
sip_session->set_invite_status(SrsGb28181SipSessionBye);
sip_session->set_invite_time(0);
}else{
srs_trace("gb28181: ingor request method=%s", req->method.c_str());
}
return err;
}
int SrsGb28181SipService::send_message(sockaddr* from, int fromlen, std::stringstream& ss)
{
std::string str = ss.str();
if (config->print_sip_message)
srs_trace("gb28181: send_message:%s", str.c_str());
srs_assert(!str.empty());
int ret = srs_sendto(lfd, (char*)str.c_str(), (int)str.length(), from, fromlen, SRS_UTIME_NO_TIMEOUT);
if (ret <= 0){
srs_trace("gb28181: send_message falid (%d)", ret);
}
return ret;
}
int SrsGb28181SipService::send_ack(SrsSipRequest *req, sockaddr *f, int l)
{
srs_assert(req);
std::stringstream ss;
req->host = config->host;
req->host_port = config->sip_port;
req->realm = config->sip_realm;
req->serial = config->sip_serial;
sip->resp_ack(ss, req);
return send_message(f, l, ss);
}
int SrsGb28181SipService::send_status(SrsSipRequest *req, sockaddr *f, int l)
{
srs_assert(req);
std::stringstream ss;
req->host = config->host;
req->host_port = config->sip_port;
req->realm = config->sip_realm;
req->serial = config->sip_serial;
sip->resp_status(ss, req);
return send_message(f, l, ss);
}
int SrsGb28181SipService::send_invite(SrsSipRequest *req, string ip, int port, uint32_t ssrc)
{
srs_assert(req);
SrsGb28181SipSession *sip_session = fetch(req->sip_auth_id);
if (!sip_session){
return ERROR_GB28181_SESSION_IS_NOTEXIST;
}
//if you are inviting or succeed in invite,
//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;
}
req->host = config->host;
req->host_port = config->sip_port;
req->realm = config->sip_realm;
req->serial = config->sip_serial;
std::stringstream ss;
sip->req_invite(ss, req, ip, port, ssrc);
if (send_message(sip_session->sockaddr_from(), sip_session->sockaddr_fromlen(), ss) <= 0)
{
return ERROR_GB28281_SIP_INVITE_FAILED;
}
sip_session->set_invite_status(SrsGb28181SipSessionTrying);
return ERROR_SUCCESS;
}
int SrsGb28181SipService::send_bye(SrsSipRequest *req)
{
srs_assert(req);
SrsGb28181SipSession *sip_session = fetch(req->sip_auth_id);
if (!sip_session){
return ERROR_GB28181_SESSION_IS_NOTEXIST;
}
//prame branch, from_tag, to_tag, call_id,
//The parameter of 'bye' must be the same as 'invite'
SrsSipRequest r = sip_session->request();
req->copy(&r);
req->host = config->host;
req->host_port = config->sip_port;
req->realm = config->sip_realm;
req->serial = config->sip_serial;
//get protocol stack
std::stringstream ss;
sip->req_bye(ss, req);
if (send_message(sip_session->sockaddr_from(), sip_session->sockaddr_fromlen(), ss) <= 0)
{
return ERROR_GB28281_SIP_BYE_FAILED;
}
return ERROR_SUCCESS;
}
int SrsGb28181SipService::send_sip_raw_data(SrsSipRequest *req, std::string data)
{
srs_assert(req);
SrsGb28181SipSession *sip_session = fetch(req->sip_auth_id);
if (!sip_session){
return ERROR_GB28181_SESSION_IS_NOTEXIST;
}
std::stringstream ss;
ss << data;
if (send_message(sip_session->sockaddr_from(), sip_session->sockaddr_fromlen(), ss) <= 0)
{
return ERROR_GB28281_SIP_BYE_FAILED;
}
return ERROR_SUCCESS;
}
SrsGb28181SipSession* SrsGb28181SipService::create_sip_session(SrsSipRequest *req)
{
SrsGb28181SipSession *sess = NULL;
std::map<std::string, SrsGb28181SipSession*>::iterator it = sessions.find(req->sip_auth_id);
if (it == sessions.end()){
sess = new SrsGb28181SipSession(this, req);
}else{
return it->second;
}
sessions[req->sip_auth_id] = sess;
return sess;
}
SrsGb28181SipSession* SrsGb28181SipService::fetch(std::string sid)
{
std::map<std::string, SrsGb28181SipSession*>::iterator it = sessions.find(sid);
if (it == sessions.end()){
return NULL;
}else{
return it->second;
}
}
void SrsGb28181SipService::remove_session(std::string sid)
{
std::map<std::string, SrsGb28181SipSession*>::iterator it = sessions.find(sid);
if (it != sessions.end()){
srs_freep(it->second);
sessions.erase(it);
}
}
void SrsGb28181SipService::destroy()
{
//destory all sip session
std::map<std::string, SrsGb28181SipSession*>::iterator it;
for (it = sessions.begin(); it != sessions.end(); ++it) {
srs_freep(it->second);
}
sessions.clear();
}

161
trunk/src/app/srs_app_gb28181_sip.hpp Normal file
View file

@ -0,0 +1,161 @@
/**
* The MIT License (MIT)
*
* Copyright (c) 2013-2020 Lixin
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to
* use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
* the Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
* IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
* CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#ifndef SRS_APP_GB28181_SIP_HPP
#define SRS_APP_GB28181_SIP_HPP
#include <srs_core.hpp>
#include <string>
#include <vector>
#include <map>
#include <srs_app_log.hpp>
#include <srs_sip_stack.hpp>
#include <srs_app_gb28181.hpp>
class SrsConfDirective;
class SrsSipRequest;
class SrsGb28181Config;
class SrsSipStack;
class SrsGb28181SipService;
enum SrsGb28281SipSessionStatusType{
SrsGb28181SipSessionUnkonw = 0,
SrsGb28181SipSessionRegisterOk = 1,
SrsGb28181SipSessionAliveOk = 2,
SrsGb28181SipSessionInviteOk = 3,
SrsGb28181SipSessionTrying = 4,
SrsGb28181SipSessionBye = 5,
};
class SrsGb28181SipSession
{
private:
//SrsSipRequest *req;
SrsGb28181SipService *caster;
std::string session_id;
private:
SrsGb28281SipSessionStatusType _register_status;
SrsGb28281SipSessionStatusType _alive_status;
SrsGb28281SipSessionStatusType _invite_status;
srs_utime_t _register_time;
srs_utime_t _alive_time;
srs_utime_t _invite_time;
srs_utime_t _recv_rtp_time;
int _reg_expires;
std::string _peer_ip;
int _peer_port;
sockaddr *_from;
int _fromlen;
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_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;}
void set_recv_rtp_time(srs_utime_t t) { _recv_rtp_time = t;}
void set_reg_expires(int e) { _reg_expires = e;}
void set_peer_ip(std::string i) { _peer_ip = i;}
void set_peer_port(int o) { _peer_port = o;}
void set_sockaddr(sockaddr *f) { _from = f;}
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;}
srs_utime_t register_time() { return _register_time;}
srs_utime_t alive_time() { return _alive_time;}
srs_utime_t invite_time() { return _invite_time;}
srs_utime_t recv_rtp_time() { return _recv_rtp_time;}
int reg_expires() { return _reg_expires;}
std::string peer_ip() { return _peer_ip;}
int peer_port() { return _peer_port;}
sockaddr* sockaddr_from() { return _from;}
int sockaddr_fromlen() { return _fromlen;}
SrsSipRequest request() { return *req;}
public:
SrsGb28181SipSession(SrsGb28181SipService *c, SrsSipRequest* r);
virtual ~SrsGb28181SipSession();
};
class SrsGb28181SipService : public ISrsUdpHandler
{
private:
SrsSipStack *sip;
SrsGb28181Config *config;
srs_netfd_t lfd;
std::map<std::string, SrsGb28181SipSession*> sessions;
public:
SrsGb28181SipService(SrsConfDirective* c);
virtual ~SrsGb28181SipService();
// Interface ISrsUdpHandler
public:
virtual srs_error_t on_udp_packet(const sockaddr* from, const int fromlen, char* buf, int nb_buf);
virtual void set_stfd(srs_netfd_t fd);
private:
void destroy();
srs_error_t on_udp_sip(std::string host, int port, char* buf, int nb_buf, sockaddr* from, int fromlen);
public:
int send_message(sockaddr* f, int l, std::stringstream& ss);
int send_ack(SrsSipRequest *req, sockaddr *f, int l);
int send_status(SrsSipRequest *req, sockaddr *f, int l);
int send_invite(SrsSipRequest *req, std::string ip, int port, uint32_t ssrc);
int send_bye(SrsSipRequest *req);
// The SIP command is transmitted through HTTP API,
// and the body content is transmitted to the device,
// mainly for testing and debugging, For example, here is HTTP body:
// BYE sip:34020000001320000003@3402000000 SIP/2.0
// Via: SIP/2.0/UDP 39.100.155.146:15063;rport;branch=z9hG4bK34205410
// From: <sip:34020000002000000001@3402000000>;tag=512355410
// To: <sip:34020000001320000003@3402000000>;tag=680367414
// Call-ID: 200003304
// CSeq: 21 BYE
// Max-Forwards: 70
// User-Agent: SRS/4.0.4(Leo)
// Content-Length: 0
//
//
int send_sip_raw_data(SrsSipRequest *req, std::string data);
SrsGb28181SipSession* create_sip_session(SrsSipRequest *req);
SrsGb28181SipSession* fetch(std::string id);
void remove_session(std::string id);
};
#endif

125
trunk/src/app/srs_app_http_api.cpp
View file

@ -1580,6 +1580,131 @@ srs_error_t SrsGoApiError::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage
return srs_api_response_code(w, r, 100);
}
SrsGoApiGb28181::SrsGoApiGb28181()
{
}
SrsGoApiGb28181::~SrsGoApiGb28181()
{
}
srs_error_t SrsGoApiGb28181::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{
SrsJsonObject* obj = SrsJsonAny::object();
SrsAutoFree(SrsJsonObject, obj);
obj->set("code", SrsJsonAny::integer(ERROR_SUCCESS));
SrsJsonObject* data = SrsJsonAny::object();
obj->set("data", data);
string id = r->query_get("id");
string action = r->query_get("action");
string vhost = r->query_get("vhost");
string app = r->query_get("app");
string stream = r->query_get("stream");
//fixed, random
string port_mode = r->query_get("port_mode");
if (_srs_gb28181) {
if(action == "create_channel"){
if (id.empty()){
return srs_api_response_code(w, r, ERROR_GB28181_VALUE_EMPTY);
}
SrsGb28181StreamChannel channel;
channel.set_channel_id(id);
channel.set_app(app);
channel.set_stream(stream);
channel.set_port_mode(port_mode);
uint32_t code = _srs_gb28181->create_stream_channel(&channel);
if (code != ERROR_SUCCESS) {
return srs_api_response_code(w, r, code);
}
data->set("query", SrsJsonAny::object()
->set("id", SrsJsonAny::str(channel.get_channel_id().c_str()))
->set("ip", SrsJsonAny::str(channel.get_ip().c_str()))
->set("rtmp_port", SrsJsonAny::integer(channel.get_rtmp_port()))
->set("app", SrsJsonAny::str(channel.get_app().c_str()))
->set("stream", SrsJsonAny::str(channel.get_stream().c_str()))
->set("rtp_port", SrsJsonAny::integer(channel.get_rtp_port()))
->set("ssrc", SrsJsonAny::integer(channel.get_ssrc())));
return srs_api_response(w, r, obj->dumps());
}
else if(action == "delete_channel"){
if (id.empty()){
return srs_api_response_code(w, r, ERROR_GB28181_VALUE_EMPTY);
}
uint32_t code = _srs_gb28181->delete_stream_channel(id);
return srs_api_response_code(w, r, code);
}
else if(action == "query_channel") {
SrsJsonArray* arr = SrsJsonAny::array();
data->set("channels", arr);
uint32_t code = _srs_gb28181->queue_stream_channel(id, arr);
if (code != ERROR_SUCCESS) {
return srs_api_response_code(w, r, code);
}
return srs_api_response(w, r, obj->dumps());
}
else if(action == "sip_invite"){
if (id.empty()){
return srs_api_response_code(w, r, ERROR_GB28181_VALUE_EMPTY);
}
string ssrc = r->query_get("ssrc");
string rtp_port = r->query_get("rtp_port");
string ip = r->query_get("ip");
int _port = strtoul(rtp_port.c_str(), NULL, 10);
uint32_t _ssrc = (uint32_t)(strtoul(ssrc.c_str(), NULL, 10));
int code = _srs_gb28181->notify_sip_invite(id, ip, _port, _ssrc);
return srs_api_response_code(w, r, code);
}
else if(action == "sip_bye"){
if (id.empty()){
return srs_api_response_code(w, r, ERROR_GB28181_VALUE_EMPTY);
}
int code = _srs_gb28181->notify_sip_bye(id);
return srs_api_response_code(w, r, code);
}
else if(action == "sip_raw_data"){
if (id.empty()){
return srs_api_response_code(w, r, ERROR_GB28181_VALUE_EMPTY);
}
std::string body;
r->body_read_all(body);
int code = _srs_gb28181->notify_sip_raw_data(id, body);
return srs_api_response_code(w, r, code);
}
else if(action == "sip_unregister"){
if (id.empty()){
return srs_api_response_code(w, r, ERROR_GB28181_VALUE_EMPTY);
}
int code = _srs_gb28181->notify_sip_unregister(id);
return srs_api_response_code(w, r, code);
}
else
{
return srs_api_response_code(w, r, ERROR_GB28181_ACTION_INVALID);
}
}else {
return srs_api_response_code(w, r, ERROR_GB28181_SERVER_NOT_RUN);
}
}
SrsHttpApi::SrsHttpApi(IConnectionManager* cm, srs_netfd_t fd, SrsHttpServeMux* m, string cip)
: SrsConnection(cm, fd, cip)
{

10
trunk/src/app/srs_app_http_api.hpp
View file

@ -232,6 +232,16 @@ public:
virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
};
class SrsGoApiGb28181 : public ISrsHttpHandler
{
public:
SrsGoApiGb28181();
virtual ~SrsGoApiGb28181();
public:
virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
};
class SrsHttpApi : virtual public SrsConnection, virtual public ISrsReloadHandler
{
private:

7
trunk/src/app/srs_app_listener.cpp
View file

@ -73,6 +73,10 @@ srs_error_t ISrsUdpMuxHandler::on_stfd_change(srs_netfd_t /*fd*/)
return srs_success;
}
void ISrsUdpHandler::set_stfd(srs_netfd_t /*fd*/)
{
}
ISrsTcpHandler::ISrsTcpHandler()
{
}
@ -118,6 +122,8 @@ srs_error_t SrsUdpListener::listen()
if ((err = srs_udp_listen(ip, port, &lfd)) != srs_success) {
return srs_error_wrap(err, "listen %s:%d", ip.c_str(), port);
}
handler->set_stfd(lfd);
srs_freep(trd);
trd = new SrsSTCoroutine("udp", this);
@ -431,3 +437,4 @@ srs_error_t SrsUdpMuxListener::cycle()
return err;
}

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

@ -47,6 +47,8 @@ public:
// When fd changed, for instance, reload the listen port,
// notify the handler and user can do something.
virtual srs_error_t on_stfd_change(srs_netfd_t fd);
virtual void set_stfd(srs_netfd_t fd);
public:
// When udp listener got a udp packet, notice server to process it.
// @param type, the client type, used to create concrete connection,

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

@ -54,6 +54,8 @@ using namespace std;
#include <srs_kernel_consts.hpp>
#include <srs_app_thread.hpp>
#include <srs_app_coworkers.hpp>
#include <srs_app_gb28181.hpp>
#include <srs_app_gb28181_sip.hpp>
// system interval in srs_utime_t,
// all resolution times should be times togother,
@ -111,6 +113,10 @@ std::string srs_listener_type2string(SrsListenerType type)
return "RTSP";
case SrsListenerFlv:
return "HTTP-FLV";
case SrsListenerGb28181Sip:
return "GB28181-SIP over UDP";
case SrsListenerGb28181RtpMux:
return "GB28181-Stream over RTP";
default:
return "UNKONWN";
}
@ -303,7 +309,9 @@ srs_error_t SrsUdpStreamListener::listen(string i, int p)
// the caller already ensure the type is ok,
// we just assert here for unknown stream caster.
srs_assert(type == SrsListenerMpegTsOverUdp);
srs_assert(type == SrsListenerMpegTsOverUdp
|| type == SrsListenerGb28181Sip
|| type == SrsListenerGb28181RtpMux);
ip = i;
port = p;
@ -341,6 +349,26 @@ SrsUdpCasterListener::~SrsUdpCasterListener()
srs_freep(caster);
}
SrsGb28181Listener::SrsGb28181Listener(SrsServer* svr, SrsListenerType t, SrsConfDirective* c) : SrsUdpStreamListener(svr, t, NULL)
{
// the caller already ensure the type is ok,
// we just assert here for unknown stream caster.
srs_assert(type == SrsListenerGb28181Sip
||type == SrsListenerGb28181RtpMux);
if (type == SrsListenerGb28181Sip) {
caster = new SrsGb28181SipService(c);
}else if(type == SrsListenerGb28181RtpMux){
caster = new SrsGb28181RtpMuxService(c);
}
}
SrsGb28181Listener::~SrsGb28181Listener()
{
srs_freep(caster);
}
SrsSignalManager* SrsSignalManager::instance = NULL;
SrsSignalManager::SrsSignalManager(SrsServer* s)
@ -663,6 +691,9 @@ void SrsServer::destroy()
srs_freep(signal_manager);
srs_freep(conn_manager);
//free global gb28281 manager
srs_freep(_srs_gb28181);
}
void SrsServer::dispose()
@ -953,6 +984,10 @@ srs_error_t SrsServer::http_handle()
if ((err = http_api_mux->handle("/api/v1/clusters", new SrsGoApiClusters())) != srs_success) {
return srs_error_wrap(err, "handle raw");
}
if ((err = http_api_mux->handle("/api/v1/gb28181", new SrsGoApiGb28181())) != srs_success) {
return srs_error_wrap(err, "handle raw");
}
// test the request info.
if ((err = http_api_mux->handle("/api/v1/tests/requests", new SrsGoApiRequests())) != srs_success) {
@ -1307,6 +1342,30 @@ srs_error_t SrsServer::listen_http_stream()
return err;
}
srs_error_t SrsServer::listen_gb28281_sip(SrsConfDirective* stream_caster)
{
srs_error_t err = srs_success;
SrsListener* sip_listener = NULL;
sip_listener = new SrsGb28181Listener(this, SrsListenerGb28181Sip, stream_caster);
int port = _srs_config->get_stream_caster_gb28181_sip_listen(stream_caster);
if (port <= 0) {
return srs_error_new(ERROR_STREAM_CASTER_PORT, "invalid sip port=%d", port);
}
srs_assert(sip_listener != NULL);
listeners.push_back(sip_listener);
// TODO: support listen at <[ip:]port>
if ((err = sip_listener->listen(srs_any_address_for_listener(), port)) != srs_success) {
return srs_error_wrap(err, "listen at %d", port);
}
return err;
}
srs_error_t SrsServer::listen_stream_caster()
{
srs_error_t err = srs_success;
@ -1331,18 +1390,34 @@ srs_error_t SrsServer::listen_stream_caster()
listener = new SrsRtspListener(this, SrsListenerRtsp, stream_caster);
} else if (srs_stream_caster_is_flv(caster)) {
listener = new SrsHttpFlvListener(this, SrsListenerFlv, stream_caster);
} else if (srs_stream_caster_is_gb28181(caster)) {
//init global gb28281 manger
if (_srs_gb28181 == NULL){
_srs_gb28181 = new SrsGb28181Manger(stream_caster);
if ((err = _srs_gb28181->initialize()) != srs_success){
return err;
}
}
//sip listener
if (_srs_config->get_stream_caster_gb28181_sip_enable(stream_caster)){
if ((err = listen_gb28281_sip(stream_caster)) != srs_success){
return err;
}
}
//gb28281 stream listener
listener = new SrsGb28181Listener(this, SrsListenerGb28181RtpMux, stream_caster);
} else {
return srs_error_new(ERROR_STREAM_CASTER_ENGINE, "invalid caster %s", caster.c_str());
}
srs_assert(listener != NULL);
listeners.push_back(listener);
int port = _srs_config->get_stream_caster_listen(stream_caster);
if (port <= 0) {
return srs_error_new(ERROR_STREAM_CASTER_PORT, "invalid port=%d", port);
}
// TODO: support listen at <[ip:]port>
if ((err = listener->listen(srs_any_address_for_listener(), port)) != srs_success) {
return srs_error_wrap(err, "listen at %d", port);

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

@ -36,6 +36,8 @@
#include <srs_app_listener.hpp>
#include <srs_app_conn.hpp>
#include <srs_service_st.hpp>
#include <srs_app_gb28181.hpp>
#include <srs_app_gb28181_sip.hpp>
class SrsServer;
class SrsConnection;
@ -52,6 +54,8 @@ class SrsTcpListener;
class SrsAppCasterFlv;
class SrsRtspCaster;
class SrsCoroutineManager;
class SrsGb28181Caster;
// The listener type for server to identify the connection,
// that is, use different type to process the connection.
@ -69,6 +73,10 @@ enum SrsListenerType
SrsListenerRtsp = 4,
// TCP stream, FLV stream over HTTP.
SrsListenerFlv = 5,
// UDP stream, gb28181 ps stream over rtp,
SrsListenerGb28181RtpMux = 6,
// UDP gb28181 sip server
SrsListenerGb28181Sip = 7,
};
// A common tcp listener, for RTMP/HTTP server.
@ -156,6 +164,14 @@ public:
virtual ~SrsUdpCasterListener();
};
// A UDP gb28181 listener, for sip and rtp stream mux server.
class SrsGb28181Listener : public SrsUdpStreamListener
{
public:
SrsGb28181Listener(SrsServer* svr, SrsListenerType t, SrsConfDirective* c);
virtual ~SrsGb28181Listener();
};
// Convert signal to io,
// @see: st-1.9/docs/notes.html
class SrsSignalManager : public ISrsCoroutineHandler
@ -303,6 +319,7 @@ private:
virtual srs_error_t listen_http_api();
virtual srs_error_t listen_http_stream();
virtual srs_error_t listen_stream_caster();
virtual srs_error_t listen_gb28281_sip(SrsConfDirective* c);
// Close the listeners for specified type,
// Remove the listen object from manager.
virtual void close_listeners(SrsListenerType type);