R3944Realms/srs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

AI: Add utest to cover srt module.

Browse Source
This commit is contained in:
OSSRS-AI 2025-10-10 20:05:56 -04:00 committed by winlin
parent af655c53c5
commit 604f9450fc
22 changed files with 1808 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
trunk/src/app/srs_app_caster_flv.cpp
View File

@ -171,6 +171,11 @@ void SrsAppCasterFlv::add_with_fast_id(uint64_t id, ISrsResource *conn)
manager_->add_with_fast_id(id, conn); manager_->add_with_fast_id(id, conn);
} }
void SrsAppCasterFlv::add_with_name(const std::string &name, ISrsResource *conn)
{
manager_->add_with_name(name, conn);
}
ISrsResource *SrsAppCasterFlv::at(int index) ISrsResource *SrsAppCasterFlv::at(int index)
{ {
return manager_->at(index); return manager_->at(index);

1
trunk/src/app/srs_app_caster_flv.hpp
View File

@ -76,6 +76,7 @@ public:
virtual void add(ISrsResource *conn, bool *exists = NULL); virtual void add(ISrsResource *conn, bool *exists = NULL);
virtual void add_with_id(const std::string &id, ISrsResource *conn); virtual void add_with_id(const std::string &id, ISrsResource *conn);
virtual void add_with_fast_id(uint64_t id, ISrsResource *conn); virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
virtual void add_with_name(const std::string &name, ISrsResource *conn);
virtual ISrsResource *at(int index); virtual ISrsResource *at(int index);
virtual ISrsResource *find_by_id(std::string id); virtual ISrsResource *find_by_id(std::string id);
virtual ISrsResource *find_by_fast_id(uint64_t id); virtual ISrsResource *find_by_fast_id(uint64_t id);

8
trunk/src/app/srs_app_config.hpp
View File

@ -347,6 +347,12 @@ public:
virtual std::vector<std::string> get_rtc_server_listens() = 0; virtual std::vector<std::string> get_rtc_server_listens() = 0;
virtual int get_rtc_server_reuseport() = 0; virtual int get_rtc_server_reuseport() = 0;
virtual bool get_rtc_server_encrypt() = 0; virtual bool get_rtc_server_encrypt() = 0;
virtual bool get_api_as_candidates() = 0;
virtual bool get_resolve_api_domain() = 0;
virtual bool get_keep_api_domain() = 0;
virtual std::string get_rtc_server_candidates() = 0;
virtual bool get_use_auto_detect_network_ip() = 0;
virtual std::string get_rtc_server_ip_family() = 0;
public: public:
// RTSP config // RTSP config
@ -453,6 +459,8 @@ public:
virtual bool get_rtc_twcc_enabled(std::string vhost) = 0; virtual bool get_rtc_twcc_enabled(std::string vhost) = 0;
virtual bool get_srt_enabled() = 0; virtual bool get_srt_enabled() = 0;
virtual bool get_srt_enabled(std::string vhost) = 0; virtual bool get_srt_enabled(std::string vhost) = 0;
virtual std::string get_srt_default_streamid() = 0;
virtual bool get_srt_to_rtmp(std::string vhost) = 0;
virtual bool get_rtc_to_rtmp(std::string vhost) = 0; virtual bool get_rtc_to_rtmp(std::string vhost) = 0;
virtual srs_utime_t get_rtc_stun_timeout(std::string vhost) = 0; virtual srs_utime_t get_rtc_stun_timeout(std::string vhost) = 0;
virtual bool get_rtc_stun_strict_check(std::string vhost) = 0; virtual bool get_rtc_stun_strict_check(std::string vhost) = 0;

44
trunk/src/app/srs_app_rtc_server.cpp
View File

@ -177,11 +177,11 @@ bool srs_is_rtcp(const uint8_t *data, size_t len)
return (len >= 12) && (data[0] & 0x80) && (data[1] >= 192 && data[1] <= 223); return (len >= 12) && (data[0] & 0x80) && (data[1] >= 192 && data[1] <= 223);
} }
srs_error_t api_server_as_candidates(string api, set<string> &candidate_ips) srs_error_t api_server_as_candidates(ISrsAppConfig *config, string api, set<string> &candidate_ips)
{ {
srs_error_t err = srs_success; srs_error_t err = srs_success;
if (api.empty() || !_srs_config->get_api_as_candidates()) { if (api.empty() || !config->get_api_as_candidates()) {
return err; return err;
} }
@ -194,12 +194,12 @@ srs_error_t api_server_as_candidates(string api, set<string> &candidate_ips)
} }
// Whether add domain name. // Whether add domain name.
if (!srs_net_is_ipv4(hostname) && _srs_config->get_keep_api_domain()) { if (!srs_net_is_ipv4(hostname) && config->get_keep_api_domain()) {
candidate_ips.insert(hostname); candidate_ips.insert(hostname);
} }
// Try to parse the domain name if not IP. // Try to parse the domain name if not IP.
if (!srs_net_is_ipv4(hostname) && _srs_config->get_resolve_api_domain()) { if (!srs_net_is_ipv4(hostname) && config->get_resolve_api_domain()) {
int family = 0; int family = 0;
string ip = srs_dns_resolve(hostname, family); string ip = srs_dns_resolve(hostname, family);
if (ip.empty() || ip == SRS_CONSTS_LOCALHOST || ip == SRS_CONSTS_LOOPBACK || ip == SRS_CONSTS_LOOPBACK6) { if (ip.empty() || ip == SRS_CONSTS_LOCALHOST || ip == SRS_CONSTS_LOOPBACK || ip == SRS_CONSTS_LOOPBACK6) {
@ -218,7 +218,7 @@ srs_error_t api_server_as_candidates(string api, set<string> &candidate_ips)
return err; return err;
} }
set<string> discover_candidates(SrsRtcUserConfig *ruc) set<string> discover_candidates(SrsProtocolUtility *utility, ISrsAppConfig *config, SrsRtcUserConfig *ruc)
{ {
srs_error_t err = srs_success; srs_error_t err = srs_success;
@ -229,29 +229,28 @@ set<string> discover_candidates(SrsRtcUserConfig *ruc)
} }
// Try to discover from api of request, if api_as_candidates enabled. // Try to discover from api of request, if api_as_candidates enabled.
if ((err = api_server_as_candidates(ruc->req_->host_, candidate_ips)) != srs_success) { if ((err = api_server_as_candidates(config, ruc->req_->host_, candidate_ips)) != srs_success) {
srs_warn("ignore discovering ip from api %s, err %s", ruc->req_->host_.c_str(), srs_error_summary(err).c_str()); srs_warn("ignore discovering ip from api %s, err %s", ruc->req_->host_.c_str(), srs_error_summary(err).c_str());
srs_freep(err); srs_freep(err);
} }
// If not * or 0.0.0.0, use the candidate as exposed IP. // If not * or 0.0.0.0, use the candidate as exposed IP.
string candidate = _srs_config->get_rtc_server_candidates(); string candidate = config->get_rtc_server_candidates();
if (candidate != "*" && candidate != "0.0.0.0") { if (candidate != "*" && candidate != "0.0.0.0") {
candidate_ips.insert(candidate); candidate_ips.insert(candidate);
return candidate_ips; return candidate_ips;
} }
// All automatically detected IP list. // All automatically detected IP list.
SrsProtocolUtility utility; vector<SrsIPAddress *> &ips = utility->local_ips();
vector<SrsIPAddress *> &ips = utility.local_ips();
if (ips.empty()) { if (ips.empty()) {
return candidate_ips; return candidate_ips;
} }
// Discover from local network interface addresses. // Discover from local network interface addresses.
if (_srs_config->get_use_auto_detect_network_ip()) { if (config->get_use_auto_detect_network_ip()) {
// We try to find the best match candidates, no loopback. // We try to find the best match candidates, no loopback.
string family = _srs_config->get_rtc_server_ip_family(); string family = config->get_rtc_server_ip_family();
for (int i = 0; i < (int)ips.size(); ++i) { for (int i = 0; i < (int)ips.size(); ++i) {
SrsIPAddress *ip = ips[i]; SrsIPAddress *ip = ips[i];
if (ip->is_loopback_) { if (ip->is_loopback_) {
@ -313,12 +312,16 @@ SrsRtcUserConfig::~SrsRtcUserConfig()
SrsRtcSessionManager::SrsRtcSessionManager() SrsRtcSessionManager::SrsRtcSessionManager()
{ {
rtc_async_ = new SrsAsyncCallWorker(); rtc_async_ = new SrsAsyncCallWorker();
conn_manager_ = _srs_conn_manager;
} }
SrsRtcSessionManager::~SrsRtcSessionManager() SrsRtcSessionManager::~SrsRtcSessionManager()
{ {
rtc_async_->stop(); rtc_async_->stop();
srs_freep(rtc_async_); srs_freep(rtc_async_);
conn_manager_ = NULL;
} }
srs_error_t SrsRtcSessionManager::initialize() srs_error_t SrsRtcSessionManager::initialize()
@ -334,7 +337,7 @@ srs_error_t SrsRtcSessionManager::initialize()
SrsRtcConnection *SrsRtcSessionManager::find_rtc_session_by_username(const std::string &username) SrsRtcConnection *SrsRtcSessionManager::find_rtc_session_by_username(const std::string &username)
{ {
ISrsResource *conn = _srs_conn_manager->find_by_name(username); ISrsResource *conn = conn_manager_->find_by_name(username);
return dynamic_cast<SrsRtcConnection *>(conn); return dynamic_cast<SrsRtcConnection *>(conn);
} }
@ -410,7 +413,7 @@ srs_error_t SrsRtcSessionManager::do_create_rtc_session(SrsRtcUserConfig *ruc, S
std::string username = ""; std::string username = "";
while (true) { while (true) {
username = local_ufrag + ":" + ruc->remote_sdp_.get_ice_ufrag(); username = local_ufrag + ":" + ruc->remote_sdp_.get_ice_ufrag();
if (!_srs_conn_manager->find_by_name(username)) { if (!conn_manager_->find_by_name(username)) {
break; break;
} }
@ -442,7 +445,8 @@ srs_error_t SrsRtcSessionManager::do_create_rtc_session(SrsRtcUserConfig *ruc, S
string protocol = _srs_config->get_rtc_server_protocol(); string protocol = _srs_config->get_rtc_server_protocol();
set<string> candidates = discover_candidates(ruc); SrsProtocolUtility utility;
set<string> candidates = discover_candidates(&utility, _srs_config, ruc);
for (set<string>::iterator it = candidates.begin(); it != candidates.end(); ++it) { for (set<string>::iterator it = candidates.begin(); it != candidates.end(); ++it) {
string hostname; string hostname;
int uport = udp_port; int uport = udp_port;
@ -494,7 +498,7 @@ srs_error_t SrsRtcSessionManager::do_create_rtc_session(SrsRtcUserConfig *ruc, S
} }
// We allows username is optional, but it never empty here. // We allows username is optional, but it never empty here.
_srs_conn_manager->add_with_name(username, session); conn_manager_->add_with_name(username, session);
return err; return err;
} }
@ -505,8 +509,8 @@ void SrsRtcSessionManager::srs_update_rtc_sessions()
int nn_rtc_conns = 0; int nn_rtc_conns = 0;
// Check all sessions and dispose the dead sessions. // Check all sessions and dispose the dead sessions.
for (int i = 0; i < (int)_srs_conn_manager->size(); i++) { for (int i = 0; i < (int)conn_manager_->size(); i++) {
SrsRtcConnection *session = dynamic_cast<SrsRtcConnection *>(_srs_conn_manager->at(i)); SrsRtcConnection *session = dynamic_cast<SrsRtcConnection *>(conn_manager_->at(i));
// Ignore not session, or already disposing. // Ignore not session, or already disposing.
if (!session || session->disposing_) { if (!session || session->disposing_) {
continue; continue;
@ -525,7 +529,7 @@ void SrsRtcSessionManager::srs_update_rtc_sessions()
srs_trace("RTC: session destroy by timeout, username=%s", username.c_str()); srs_trace("RTC: session destroy by timeout, username=%s", username.c_str());
// Use manager to free session and notify other objects. // Use manager to free session and notify other objects.
_srs_conn_manager->remove(session); conn_manager_->remove(session);
} }
// Ignore stats if no RTC connections. // Ignore stats if no RTC connections.
@ -569,11 +573,11 @@ srs_error_t SrsRtcSessionManager::on_udp_packet(ISrsUdpMuxSocket *skt)
uint64_t fast_id = skt->fast_id(); uint64_t fast_id = skt->fast_id();
// Try fast id first, if not found, search by long peer id. // Try fast id first, if not found, search by long peer id.
if (fast_id) { if (fast_id) {
session = (SrsRtcConnection *)_srs_conn_manager->find_by_fast_id(fast_id); session = (SrsRtcConnection *)conn_manager_->find_by_fast_id(fast_id);
} }
if (!session) { if (!session) {
string peer_id = skt->peer_id(); string peer_id = skt->peer_id();
session = (SrsRtcConnection *)_srs_conn_manager->find_by_id(peer_id); session = (SrsRtcConnection *)conn_manager_->find_by_id(peer_id);
} }
if (session) { if (session) {

10
trunk/src/app/srs_app_rtc_server.hpp
View File

@ -29,6 +29,7 @@ class SrsRtcSource;
class SrsResourceManager; class SrsResourceManager;
class SrsAsyncCallWorker; class SrsAsyncCallWorker;
class ISrsUdpMuxSocket; class ISrsUdpMuxSocket;
class ISrsResourceManager;
// The UDP black hole, for developer to use wireshark to catch plaintext packets. // The UDP black hole, for developer to use wireshark to catch plaintext packets.
// For example, server receive UDP packets at udp://8000, and forward the plaintext packet to black hole, // For example, server receive UDP packets at udp://8000, and forward the plaintext packet to black hole,
@ -85,7 +86,7 @@ public:
}; };
// Discover the candidates for RTC server. // Discover the candidates for RTC server.
extern std::set<std::string> discover_candidates(SrsRtcUserConfig *ruc); extern std::set<std::string> discover_candidates(SrsProtocolUtility *utility, ISrsAppConfig *config, SrsRtcUserConfig *ruc);
// The dns resolve utility, return the resolved ip address. // The dns resolve utility, return the resolved ip address.
extern std::string srs_dns_resolve(std::string host, int &family); extern std::string srs_dns_resolve(std::string host, int &family);
@ -93,6 +94,9 @@ extern std::string srs_dns_resolve(std::string host, int &family);
// RTC session manager to handle WebRTC session lifecycle and management. // RTC session manager to handle WebRTC session lifecycle and management.
class SrsRtcSessionManager : public ISrsExecRtcAsyncTask class SrsRtcSessionManager : public ISrsExecRtcAsyncTask
{ {
private:
ISrsResourceManager *conn_manager_;
private: private:
// WebRTC async call worker for non-blocking operations. // WebRTC async call worker for non-blocking operations.
SrsAsyncCallWorker *rtc_async_; SrsAsyncCallWorker *rtc_async_;
@ -122,4 +126,8 @@ public:
virtual srs_error_t on_udp_packet(ISrsUdpMuxSocket *skt); virtual srs_error_t on_udp_packet(ISrsUdpMuxSocket *skt);
}; };
// Helper function to discover candidate IPs from API server hostname
// Used by WebRTC ICE candidate discovery
extern srs_error_t api_server_as_candidates(ISrsAppConfig *config, std::string api, std::set<std::string> &candidate_ips);
#endif #endif

113
trunk/src/app/srs_app_srt_conn.cpp
View File

@ -92,7 +92,15 @@ srs_error_t SrsSrtConnection::writev(const iovec *iov, int iov_size, ssize_t *nw
return srs_error_new(ERROR_SRT_CONN, "unsupport method"); return srs_error_new(ERROR_SRT_CONN, "unsupport method");
} }
SrsSrtRecvThread::SrsSrtRecvThread(SrsSrtConnection *srt_conn) ISrsSrtRecvThread::ISrsSrtRecvThread()
{
}
ISrsSrtRecvThread::~ISrsSrtRecvThread()
{
}
SrsSrtRecvThread::SrsSrtRecvThread(ISrsProtocolReadWriter *srt_conn)
{ {
srt_conn_ = srt_conn; srt_conn_ = srt_conn;
trd_ = new SrsSTCoroutine("srt-recv", this, _srs_context->get_id()); trd_ = new SrsSTCoroutine("srt-recv", this, _srs_context->get_id());
@ -153,6 +161,14 @@ srs_error_t SrsSrtRecvThread::get_recv_err()
return srs_error_copy(recv_err_); return srs_error_copy(recv_err_);
} }
ISrsMpegtsSrtConnection::ISrsMpegtsSrtConnection()
{
}
ISrsMpegtsSrtConnection::~ISrsMpegtsSrtConnection()
{
}
SrsMpegtsSrtConn::SrsMpegtsSrtConn(ISrsResourceManager *resource_manager, srs_srt_t srt_fd, std::string ip, int port) : srt_source_(new SrsSrtSource()) SrsMpegtsSrtConn::SrsMpegtsSrtConn(ISrsResourceManager *resource_manager, srs_srt_t srt_fd, std::string ip, int port) : srt_source_(new SrsSrtSource())
{ {
// Create a identify for this client. // Create a identify for this client.
@ -176,6 +192,14 @@ SrsMpegtsSrtConn::SrsMpegtsSrtConn(ISrsResourceManager *resource_manager, srs_sr
req_->ip_ = ip; req_->ip_ = ip;
security_ = new SrsSecurity(); security_ = new SrsSecurity();
stat_ = _srs_stat;
config_ = _srs_config;
stream_publish_tokens_ = _srs_stream_publish_tokens;
srt_sources_ = _srs_srt_sources;
live_sources_ = _srs_sources;
rtc_sources_ = _srs_rtc_sources;
hooks_ = _srs_hooks;
} }
SrsMpegtsSrtConn::~SrsMpegtsSrtConn() SrsMpegtsSrtConn::~SrsMpegtsSrtConn()
@ -187,6 +211,14 @@ SrsMpegtsSrtConn::~SrsMpegtsSrtConn()
srs_freep(srt_conn_); srs_freep(srt_conn_);
srs_freep(req_); srs_freep(req_);
srs_freep(security_); srs_freep(security_);
stat_ = NULL;
config_ = NULL;
stream_publish_tokens_ = NULL;
srt_sources_ = NULL;
live_sources_ = NULL;
rtc_sources_ = NULL;
hooks_ = NULL;
} }
std::string SrsMpegtsSrtConn::desc() std::string SrsMpegtsSrtConn::desc()
@ -230,9 +262,8 @@ srs_error_t SrsMpegtsSrtConn::cycle()
srs_error_t err = do_cycle(); srs_error_t err = do_cycle();
// Update statistic when done. // Update statistic when done.
SrsStatistic *stat = _srs_stat; stat_->kbps_add_delta(get_id().c_str(), delta_);
stat->kbps_add_delta(get_id().c_str(), delta_); stat_->on_disconnect(get_id().c_str(), err);
stat->on_disconnect(get_id().c_str(), err);
// Notify manager to remove it. // Notify manager to remove it.
// Note that we create this object, so we use manager to remove it. // Note that we create this object, so we use manager to remove it.
@ -262,7 +293,7 @@ srs_error_t SrsMpegtsSrtConn::do_cycle()
// If streamid empty, using default streamid instead. // If streamid empty, using default streamid instead.
if (streamid.empty()) { if (streamid.empty()) {
streamid = _srs_config->get_srt_default_streamid(); streamid = config_->get_srt_default_streamid();
srs_warn("srt get empty streamid, using default streamid %s instead", streamid.c_str()); srs_warn("srt get empty streamid, using default streamid %s instead", streamid.c_str());
} }
@ -273,13 +304,13 @@ srs_error_t SrsMpegtsSrtConn::do_cycle()
} }
// discovery vhost, resolve the vhost from config // discovery vhost, resolve the vhost from config
SrsConfDirective *parsed_vhost = _srs_config->get_vhost(req_->vhost_); SrsConfDirective *parsed_vhost = config_->get_vhost(req_->vhost_);
if (parsed_vhost) { if (parsed_vhost) {
req_->vhost_ = parsed_vhost->arg0(); req_->vhost_ = parsed_vhost->arg0();
} }
bool srt_enabled = _srs_config->get_srt_enabled(req_->vhost_); bool srt_enabled = config_->get_srt_enabled(req_->vhost_);
bool edge = _srs_config->get_vhost_is_edge(req_->vhost_); bool edge = config_->get_vhost_is_edge(req_->vhost_);
if (srt_enabled && edge) { if (srt_enabled && edge) {
srt_enabled = false; srt_enabled = false;
@ -295,7 +326,7 @@ srs_error_t SrsMpegtsSrtConn::do_cycle()
// Acquire stream publish token to prevent race conditions across all protocols. // Acquire stream publish token to prevent race conditions across all protocols.
SrsStreamPublishToken *publish_token_raw = NULL; SrsStreamPublishToken *publish_token_raw = NULL;
if (mode == SrtModePush && (err = _srs_stream_publish_tokens->acquire_token(req_, publish_token_raw)) != srs_success) { if (mode == SrtModePush && (err = stream_publish_tokens_->acquire_token(req_, publish_token_raw)) != srs_success) {
return srs_error_wrap(err, "acquire stream publish token"); return srs_error_wrap(err, "acquire stream publish token");
} }
SrsUniquePtr<SrsStreamPublishToken> publish_token(publish_token_raw); SrsUniquePtr<SrsStreamPublishToken> publish_token(publish_token_raw);
@ -303,7 +334,7 @@ srs_error_t SrsMpegtsSrtConn::do_cycle()
srs_trace("stream publish token acquired, type=srt, url=%s", req_->get_stream_url().c_str()); srs_trace("stream publish token acquired, type=srt, url=%s", req_->get_stream_url().c_str());
} }
if ((err = _srs_srt_sources->fetch_or_create(req_, srt_source_)) != srs_success) { if ((err = srt_sources_->fetch_or_create(req_, srt_source_)) != srs_success) {
return srs_error_wrap(err, "fetch srt source"); return srs_error_wrap(err, "fetch srt source");
} }
@ -327,8 +358,7 @@ srs_error_t SrsMpegtsSrtConn::publishing()
srs_error_t err = srs_success; srs_error_t err = srs_success;
// We must do stat the client before hooks, because hooks depends on it. // We must do stat the client before hooks, because hooks depends on it.
SrsStatistic *stat = _srs_stat; if ((err = stat_->on_client(_srs_context->get_id().c_str(), req_, this, SrsSrtConnPublish)) != srs_success) {
if ((err = stat->on_client(_srs_context->get_id().c_str(), req_, this, SrsSrtConnPublish)) != srs_success) {
return srs_error_wrap(err, "srt: stat client"); return srs_error_wrap(err, "srt: stat client");
} }
@ -356,8 +386,7 @@ srs_error_t SrsMpegtsSrtConn::playing()
srs_error_t err = srs_success; srs_error_t err = srs_success;
// We must do stat the client before hooks, because hooks depends on it. // We must do stat the client before hooks, because hooks depends on it.
SrsStatistic *stat = _srs_stat; if ((err = stat_->on_client(_srs_context->get_id().c_str(), req_, this, SrsSrtConnPlay)) != srs_success) {
if ((err = stat->on_client(_srs_context->get_id().c_str(), req_, this, SrsSrtConnPlay)) != srs_success) {
return srs_error_wrap(err, "srt: stat client"); return srs_error_wrap(err, "srt: stat client");
} }
@ -387,19 +416,19 @@ srs_error_t SrsMpegtsSrtConn::acquire_publish()
} }
// Check rtmp stream is busy. // Check rtmp stream is busy.
SrsSharedPtr<SrsLiveSource> live_source = _srs_sources->fetch(req_); SrsSharedPtr<SrsLiveSource> live_source = live_sources_->fetch(req_);
if (live_source.get() && !live_source->can_publish(false)) { if (live_source.get() && !live_source->can_publish(false)) {
return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "live_source stream %s busy", req_->get_stream_url().c_str()); return srs_error_new(ERROR_SYSTEM_STREAM_BUSY, "live_source stream %s busy", req_->get_stream_url().c_str());
} }
if ((err = _srs_sources->fetch_or_create(req_, live_source)) != srs_success) { if ((err = live_sources_->fetch_or_create(req_, live_source)) != srs_success) {
return srs_error_wrap(err, "create source"); return srs_error_wrap(err, "create source");
} }
srs_assert(live_source.get() != NULL); srs_assert(live_source.get() != NULL);
bool enabled_cache = _srs_config->get_gop_cache(req_->vhost_); bool enabled_cache = config_->get_gop_cache(req_->vhost_);
int gcmf = _srs_config->get_gop_cache_max_frames(req_->vhost_); int gcmf = config_->get_gop_cache_max_frames(req_->vhost_);
live_source->set_cache(enabled_cache); live_source->set_cache(enabled_cache);
live_source->set_gop_cache_max_frames(gcmf); live_source->set_gop_cache_max_frames(gcmf);
@ -408,9 +437,9 @@ srs_error_t SrsMpegtsSrtConn::acquire_publish()
// Check whether RTC stream is busy. // Check whether RTC stream is busy.
SrsSharedPtr<SrsRtcSource> rtc; SrsSharedPtr<SrsRtcSource> rtc;
bool rtc_server_enabled = _srs_config->get_rtc_server_enabled(); bool rtc_server_enabled = config_->get_rtc_server_enabled();
bool rtc_enabled = _srs_config->get_rtc_enabled(req_->vhost_); bool rtc_enabled = config_->get_rtc_enabled(req_->vhost_);
bool edge = _srs_config->get_vhost_is_edge(req_->vhost_); bool edge = config_->get_vhost_is_edge(req_->vhost_);
if (rtc_enabled && edge) { if (rtc_enabled && edge) {
rtc_enabled = false; rtc_enabled = false;
@ -418,7 +447,7 @@ srs_error_t SrsMpegtsSrtConn::acquire_publish()
} }
if (rtc_server_enabled && rtc_enabled) { if (rtc_server_enabled && rtc_enabled) {
if ((err = _srs_rtc_sources->fetch_or_create(req_, rtc)) != srs_success) { if ((err = rtc_sources_->fetch_or_create(req_, rtc)) != srs_success) {
return srs_error_wrap(err, "create source"); return srs_error_wrap(err, "create source");
} }
@ -430,7 +459,7 @@ srs_error_t SrsMpegtsSrtConn::acquire_publish()
// Bridge to RTMP and RTC streaming. // Bridge to RTMP and RTC streaming.
SrsSrtBridge *bridge = new SrsSrtBridge(); SrsSrtBridge *bridge = new SrsSrtBridge();
bool srt_to_rtmp = _srs_config->get_srt_to_rtmp(req_->vhost_); bool srt_to_rtmp = config_->get_srt_to_rtmp(req_->vhost_);
if (srt_to_rtmp && edge) { if (srt_to_rtmp && edge) {
srt_to_rtmp = false; srt_to_rtmp = false;
srs_warn("disable SRT to RTMP for edge vhost=%s", req_->vhost_.c_str()); srs_warn("disable SRT to RTMP for edge vhost=%s", req_->vhost_.c_str());
@ -440,7 +469,7 @@ srs_error_t SrsMpegtsSrtConn::acquire_publish()
bridge->enable_srt2rtmp(live_source); bridge->enable_srt2rtmp(live_source);
} }
bool rtmp_to_rtc = _srs_config->get_rtc_from_rtmp(req_->vhost_); bool rtmp_to_rtc = config_->get_rtc_from_rtmp(req_->vhost_);
if (rtmp_to_rtc && edge) { if (rtmp_to_rtc && edge) {
rtmp_to_rtc = false; rtmp_to_rtc = false;
srs_warn("disable RTMP to WebRTC for edge vhost=%s", req_->vhost_.c_str()); srs_warn("disable RTMP to WebRTC for edge vhost=%s", req_->vhost_.c_str());
@ -632,7 +661,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_connect()
{ {
srs_error_t err = srs_success; srs_error_t err = srs_success;
if (!_srs_config->get_vhost_http_hooks_enabled(req_->vhost_)) { if (!config_->get_vhost_http_hooks_enabled(req_->vhost_)) {
return err; return err;
} }
@ -642,7 +671,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_connect()
vector<string> hooks; vector<string> hooks;
if (true) { if (true) {
SrsConfDirective *conf = _srs_config->get_vhost_on_connect(req_->vhost_); SrsConfDirective *conf = config_->get_vhost_on_connect(req_->vhost_);
if (!conf) { if (!conf) {
return err; return err;
@ -653,7 +682,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_connect()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
if ((err = _srs_hooks->on_connect(url, req_)) != srs_success) { if ((err = hooks_->on_connect(url, req_)) != srs_success) {
return srs_error_wrap(err, "srt on_connect %s", url.c_str()); return srs_error_wrap(err, "srt on_connect %s", url.c_str());
} }
} }
@ -663,7 +692,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_connect()
void SrsMpegtsSrtConn::http_hooks_on_close() void SrsMpegtsSrtConn::http_hooks_on_close()
{ {
if (!_srs_config->get_vhost_http_hooks_enabled(req_->vhost_)) { if (!config_->get_vhost_http_hooks_enabled(req_->vhost_)) {
return; return;
} }
@ -673,7 +702,7 @@ void SrsMpegtsSrtConn::http_hooks_on_close()
vector<string> hooks; vector<string> hooks;
if (true) { if (true) {
SrsConfDirective *conf = _srs_config->get_vhost_on_close(req_->vhost_); SrsConfDirective *conf = config_->get_vhost_on_close(req_->vhost_);
if (!conf) { if (!conf) {
return; return;
@ -684,7 +713,7 @@ void SrsMpegtsSrtConn::http_hooks_on_close()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
_srs_hooks->on_close(url, req_, srt_conn_->get_send_bytes(), srt_conn_->get_recv_bytes()); hooks_->on_close(url, req_, srt_conn_->get_send_bytes(), srt_conn_->get_recv_bytes());
} }
} }
@ -692,7 +721,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_publish()
{ {
srs_error_t err = srs_success; srs_error_t err = srs_success;
if (!_srs_config->get_vhost_http_hooks_enabled(req_->vhost_)) { if (!config_->get_vhost_http_hooks_enabled(req_->vhost_)) {
return err; return err;
} }
@ -702,7 +731,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_publish()
vector<string> hooks; vector<string> hooks;
if (true) { if (true) {
SrsConfDirective *conf = _srs_config->get_vhost_on_publish(req_->vhost_); SrsConfDirective *conf = config_->get_vhost_on_publish(req_->vhost_);
if (!conf) { if (!conf) {
return err; return err;
@ -713,7 +742,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_publish()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
if ((err = _srs_hooks->on_publish(url, req_)) != srs_success) { if ((err = hooks_->on_publish(url, req_)) != srs_success) {
return srs_error_wrap(err, "srt on_publish %s", url.c_str()); return srs_error_wrap(err, "srt on_publish %s", url.c_str());
} }
} }
@ -723,7 +752,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_publish()
void SrsMpegtsSrtConn::http_hooks_on_unpublish() void SrsMpegtsSrtConn::http_hooks_on_unpublish()
{ {
if (!_srs_config->get_vhost_http_hooks_enabled(req_->vhost_)) { if (!config_->get_vhost_http_hooks_enabled(req_->vhost_)) {
return; return;
} }
@ -733,7 +762,7 @@ void SrsMpegtsSrtConn::http_hooks_on_unpublish()
vector<string> hooks; vector<string> hooks;
if (true) { if (true) {
SrsConfDirective *conf = _srs_config->get_vhost_on_unpublish(req_->vhost_); SrsConfDirective *conf = config_->get_vhost_on_unpublish(req_->vhost_);
if (!conf) { if (!conf) {
return; return;
@ -744,7 +773,7 @@ void SrsMpegtsSrtConn::http_hooks_on_unpublish()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
_srs_hooks->on_unpublish(url, req_); hooks_->on_unpublish(url, req_);
} }
} }
@ -752,7 +781,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_play()
{ {
srs_error_t err = srs_success; srs_error_t err = srs_success;
if (!_srs_config->get_vhost_http_hooks_enabled(req_->vhost_)) { if (!config_->get_vhost_http_hooks_enabled(req_->vhost_)) {
return err; return err;
} }
@ -762,7 +791,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_play()
vector<string> hooks; vector<string> hooks;
if (true) { if (true) {
SrsConfDirective *conf = _srs_config->get_vhost_on_play(req_->vhost_); SrsConfDirective *conf = config_->get_vhost_on_play(req_->vhost_);
if (!conf) { if (!conf) {
return err; return err;
@ -773,7 +802,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_play()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
if ((err = _srs_hooks->on_play(url, req_)) != srs_success) { if ((err = hooks_->on_play(url, req_)) != srs_success) {
return srs_error_wrap(err, "srt on_play %s", url.c_str()); return srs_error_wrap(err, "srt on_play %s", url.c_str());
} }
} }
@ -783,7 +812,7 @@ srs_error_t SrsMpegtsSrtConn::http_hooks_on_play()
void SrsMpegtsSrtConn::http_hooks_on_stop() void SrsMpegtsSrtConn::http_hooks_on_stop()
{ {
if (!_srs_config->get_vhost_http_hooks_enabled(req_->vhost_)) { if (!config_->get_vhost_http_hooks_enabled(req_->vhost_)) {
return; return;
} }
@ -793,7 +822,7 @@ void SrsMpegtsSrtConn::http_hooks_on_stop()
vector<string> hooks; vector<string> hooks;
if (true) { if (true) {
SrsConfDirective *conf = _srs_config->get_vhost_on_stop(req_->vhost_); SrsConfDirective *conf = config_->get_vhost_on_stop(req_->vhost_);
if (!conf) { if (!conf) {
return; return;
@ -804,7 +833,7 @@ void SrsMpegtsSrtConn::http_hooks_on_stop()
for (int i = 0; i < (int)hooks.size(); i++) { for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i); std::string url = hooks.at(i);
_srs_hooks->on_stop(url, req_); hooks_->on_stop(url, req_);
} }
return; return;

57
trunk/src/app/srs_app_srt_conn.hpp
View File

@ -23,6 +23,17 @@ class SrsLiveSource;
class SrsSrtSource; class SrsSrtSource;
class SrsSrtServer; class SrsSrtServer;
class SrsNetworkDelta; class SrsNetworkDelta;
class ISrsNetworkDelta;
class ISrsSrtSocket;
class SrsNetworkKbps;
class ISrsSecurity;
class ISrsStatistic;
class ISrsAppConfig;
class ISrsStreamPublishTokenManager;
class ISrsSrtSourceManager;
class ISrsLiveSourceManager;
class ISrsRtcSourceManager;
class ISrsHttpHooks;
// The basic connection of SRS, for SRT based protocols, // The basic connection of SRS, for SRT based protocols,
// all srt connections accept from srt listener must extends from this base class, // all srt connections accept from srt listener must extends from this base class,
@ -52,13 +63,24 @@ private:
// The underlayer srt fd handler. // The underlayer srt fd handler.
srs_srt_t srt_fd_; srs_srt_t srt_fd_;
// The underlayer srt socket. // The underlayer srt socket.
SrsSrtSocket *srt_skt_; ISrsSrtSocket *srt_skt_;
}; };
class SrsSrtRecvThread : public ISrsCoroutineHandler // The recv thread for SRT connection.
class ISrsSrtRecvThread : public ISrsCoroutineHandler
{ {
public: public:
SrsSrtRecvThread(SrsSrtConnection *srt_conn); ISrsSrtRecvThread();
virtual ~ISrsSrtRecvThread();
public:
};
// The recv thread for SRT connection.
class SrsSrtRecvThread : public ISrsSrtRecvThread
{
public:
SrsSrtRecvThread(ISrsProtocolReadWriter *srt_conn);
~SrsSrtRecvThread(); ~SrsSrtRecvThread();
// Interface ISrsCoroutineHandler // Interface ISrsCoroutineHandler
public: public:
@ -72,17 +94,36 @@ public:
srs_error_t get_recv_err(); srs_error_t get_recv_err();
private: private:
SrsSrtConnection *srt_conn_; ISrsProtocolReadWriter *srt_conn_;
ISrsCoroutine *trd_; ISrsCoroutine *trd_;
srs_error_t recv_err_; srs_error_t recv_err_;
}; };
// The SRT connection, for client to publish or play stream. // The SRT connection, for client to publish or play stream.
class SrsMpegtsSrtConn : public ISrsConnection, // It's a resource. class ISrsMpegtsSrtConnection : public ISrsConnection, // It's a resource.
public ISrsStartable, public ISrsStartable,
public ISrsCoroutineHandler, public ISrsCoroutineHandler,
public ISrsExpire public ISrsExpire
{ {
public:
ISrsMpegtsSrtConnection();
virtual ~ISrsMpegtsSrtConnection();
public:
};
// The SRT connection, for client to publish or play stream.
class SrsMpegtsSrtConn : public ISrsMpegtsSrtConnection
{
private:
ISrsStatistic *stat_;
ISrsAppConfig *config_;
ISrsStreamPublishTokenManager *stream_publish_tokens_;
ISrsSrtSourceManager *srt_sources_;
ISrsLiveSourceManager *live_sources_;
ISrsRtcSourceManager *rtc_sources_;
ISrsHttpHooks *hooks_;
public: public:
SrsMpegtsSrtConn(ISrsResourceManager *resource_manager, srs_srt_t srt_fd, std::string ip, int port); SrsMpegtsSrtConn(ISrsResourceManager *resource_manager, srs_srt_t srt_fd, std::string ip, int port);
virtual ~SrsMpegtsSrtConn(); virtual ~SrsMpegtsSrtConn();
@ -131,8 +172,8 @@ private:
private: private:
ISrsResourceManager *resource_manager_; ISrsResourceManager *resource_manager_;
srs_srt_t srt_fd_; srs_srt_t srt_fd_;
SrsSrtConnection *srt_conn_; ISrsProtocolReadWriter *srt_conn_;
SrsNetworkDelta *delta_; ISrsNetworkDelta *delta_;
SrsNetworkKbps *kbps_; SrsNetworkKbps *kbps_;
std::string ip_; std::string ip_;
int port_; int port_;
@ -140,7 +181,7 @@ private:
ISrsRequest *req_; ISrsRequest *req_;
SrsSharedPtr<SrsSrtSource> srt_source_; SrsSharedPtr<SrsSrtSource> srt_source_;
SrsSecurity *security_; ISrsSecurity *security_;
}; };
#endif #endif

2
trunk/src/kernel/srs_kernel_resource.hpp
View File

@ -105,6 +105,8 @@ public:
virtual ISrsResource *find_by_fast_id(uint64_t id) = 0; virtual ISrsResource *find_by_fast_id(uint64_t id) = 0;
// Find resource by name. // Find resource by name.
virtual ISrsResource *find_by_name(std::string name) = 0; virtual ISrsResource *find_by_name(std::string name) = 0;
// Add a resource with name to the manager.
virtual void add_with_name(const std::string &name, ISrsResource *conn) = 0;
public: public:
// Remove then free the specified connection. Note that the manager always free c resource, // Remove then free the specified connection. Note that the manager always free c resource,

8
trunk/src/protocol/srs_protocol_srt.cpp
View File

@ -698,6 +698,14 @@ ISrsSrtPoller *srs_srt_poller_new()
return new SrsSrtPoller(); return new SrsSrtPoller();
} }
ISrsSrtSocket::ISrsSrtSocket()
{
}
ISrsSrtSocket::~ISrsSrtSocket()
{
}
SrsSrtSocket::SrsSrtSocket(ISrsSrtPoller *srt_poller, srs_srt_t srt_fd) SrsSrtSocket::SrsSrtSocket(ISrsSrtPoller *srt_poller, srs_srt_t srt_fd)
{ {
srt_poller_ = srt_poller; srt_poller_ = srt_poller;

20
trunk/src/protocol/srs_protocol_srt.hpp
View File

@ -115,8 +115,26 @@ public:
}; };
ISrsSrtPoller *srs_srt_poller_new(); ISrsSrtPoller *srs_srt_poller_new();
// Srt socket interface.
class ISrsSrtSocket
{
public:
ISrsSrtSocket();
virtual ~ISrsSrtSocket();
public:
virtual srs_error_t recvmsg(void *buf, size_t size, ssize_t *nread) = 0;
virtual srs_error_t sendmsg(void *buf, size_t size, ssize_t *nwrite) = 0;
virtual void set_recv_timeout(srs_utime_t tm) = 0;
virtual void set_send_timeout(srs_utime_t tm) = 0;
virtual srs_utime_t get_send_timeout() = 0;
virtual srs_utime_t get_recv_timeout() = 0;
virtual int64_t get_send_bytes() = 0;
virtual int64_t get_recv_bytes() = 0;
};
// Srt ST socket, wrap SRT io and make it adapt to ST-thread. // Srt ST socket, wrap SRT io and make it adapt to ST-thread.
class SrsSrtSocket class SrsSrtSocket : public ISrsSrtSocket
{ {
public: public:
SrsSrtSocket(ISrsSrtPoller *srt_poller, srs_srt_t srt_fd); SrsSrtSocket(ISrsSrtPoller *srt_poller, srs_srt_t srt_fd);

8
trunk/src/utest/srs_utest_app10.cpp
View File

@ -707,6 +707,10 @@ void MockConnectionManagerForResampleKbps::add_with_fast_id(uint64_t /*id*/, ISr
{ {
} }
void MockConnectionManagerForResampleKbps::add_with_name(const std::string & /*name*/, ISrsResource * /*conn*/)
{
}
ISrsResource *MockConnectionManagerForResampleKbps::at(int index) ISrsResource *MockConnectionManagerForResampleKbps::at(int index)
{ {
if (index < 0 || index >= (int)connections_.size()) { if (index < 0 || index >= (int)connections_.size()) {
@ -1002,6 +1006,10 @@ void MockConnectionManagerForConnectionLimit::add_with_fast_id(uint64_t /*id*/,
{ {
} }
void MockConnectionManagerForConnectionLimit::add_with_name(const std::string & /*name*/, ISrsResource * /*conn*/)
{
}
ISrsResource *MockConnectionManagerForConnectionLimit::at(int index) ISrsResource *MockConnectionManagerForConnectionLimit::at(int index)
{ {
return NULL; return NULL;

2
trunk/src/utest/srs_utest_app10.hpp
View File

@ -233,6 +233,7 @@ public:
virtual void add(ISrsResource *conn, bool *exists = NULL); virtual void add(ISrsResource *conn, bool *exists = NULL);
virtual void add_with_id(const std::string &id, ISrsResource *conn); virtual void add_with_id(const std::string &id, ISrsResource *conn);
virtual void add_with_fast_id(uint64_t id, ISrsResource *conn); virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
virtual void add_with_name(const std::string &name, ISrsResource *conn);
virtual ISrsResource *at(int index); virtual ISrsResource *at(int index);
virtual ISrsResource *find_by_id(std::string id); virtual ISrsResource *find_by_id(std::string id);
virtual ISrsResource *find_by_fast_id(uint64_t id); virtual ISrsResource *find_by_fast_id(uint64_t id);
@ -308,6 +309,7 @@ public:
virtual void add(ISrsResource *conn, bool *exists = NULL); virtual void add(ISrsResource *conn, bool *exists = NULL);
virtual void add_with_id(const std::string &id, ISrsResource *conn); virtual void add_with_id(const std::string &id, ISrsResource *conn);
virtual void add_with_fast_id(uint64_t id, ISrsResource *conn); virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
virtual void add_with_name(const std::string &name, ISrsResource *conn);
virtual ISrsResource *at(int index); virtual ISrsResource *at(int index);
virtual ISrsResource *find_by_id(std::string id); virtual ISrsResource *find_by_id(std::string id);
virtual ISrsResource *find_by_fast_id(uint64_t id); virtual ISrsResource *find_by_fast_id(uint64_t id);

16
trunk/src/utest/srs_utest_app14.cpp
View File

@ -859,6 +859,10 @@ void MockResourceManagerForBindSession::add_with_fast_id(uint64_t id, ISrsResour
{ {
} }
void MockResourceManagerForBindSession::add_with_name(const std::string & /*name*/, ISrsResource * /*conn*/)
{
}
ISrsResource *MockResourceManagerForBindSession::at(int index) ISrsResource *MockResourceManagerForBindSession::at(int index)
{ {
return NULL; return NULL;
@ -2135,6 +2139,10 @@ void MockResourceManagerForGbPublish::add_with_fast_id(uint64_t id, ISrsResource
fast_id_map_[id] = conn; fast_id_map_[id] = conn;
} }
void MockResourceManagerForGbPublish::add_with_name(const std::string & /*name*/, ISrsResource * /*conn*/)
{
}
ISrsResource *MockResourceManagerForGbPublish::at(int index) ISrsResource *MockResourceManagerForGbPublish::at(int index)
{ {
return NULL; return NULL;
@ -3245,6 +3253,10 @@ void MockResourceManagerForUdpNetwork::add_with_fast_id(uint64_t id, ISrsResourc
fast_id_map_[id] = conn; fast_id_map_[id] = conn;
} }
void MockResourceManagerForUdpNetwork::add_with_name(const std::string & /*name*/, ISrsResource * /*conn*/)
{
}
ISrsResource *MockResourceManagerForUdpNetwork::at(int index) ISrsResource *MockResourceManagerForUdpNetwork::at(int index)
{ {
return NULL; return NULL;
@ -3918,6 +3930,10 @@ void MockResourceManagerForTcpConnHandshake::add_with_fast_id(uint64_t id, ISrsR
{ {
} }
void MockResourceManagerForTcpConnHandshake::add_with_name(const std::string & /*name*/, ISrsResource * /*conn*/)
{
}
ISrsResource *MockResourceManagerForTcpConnHandshake::at(int index) ISrsResource *MockResourceManagerForTcpConnHandshake::at(int index)
{ {
return NULL; return NULL;

4
trunk/src/utest/srs_utest_app14.hpp
View File

@ -309,6 +309,7 @@ public:
virtual void add(ISrsResource *conn, bool *exists = NULL); virtual void add(ISrsResource *conn, bool *exists = NULL);
virtual void add_with_id(const std::string &id, ISrsResource *conn); virtual void add_with_id(const std::string &id, ISrsResource *conn);
virtual void add_with_fast_id(uint64_t id, ISrsResource *conn); virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
virtual void add_with_name(const std::string &name, ISrsResource *conn);
virtual ISrsResource *at(int index); virtual ISrsResource *at(int index);
virtual ISrsResource *find_by_id(std::string id); virtual ISrsResource *find_by_id(std::string id);
virtual ISrsResource *find_by_fast_id(uint64_t id); virtual ISrsResource *find_by_fast_id(uint64_t id);
@ -531,6 +532,7 @@ public:
virtual void add(ISrsResource *conn, bool *exists = NULL); virtual void add(ISrsResource *conn, bool *exists = NULL);
virtual void add_with_id(const std::string &id, ISrsResource *conn); virtual void add_with_id(const std::string &id, ISrsResource *conn);
virtual void add_with_fast_id(uint64_t id, ISrsResource *conn); virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
virtual void add_with_name(const std::string &name, ISrsResource *conn);
virtual ISrsResource *at(int index); virtual ISrsResource *at(int index);
virtual ISrsResource *find_by_id(std::string id); virtual ISrsResource *find_by_id(std::string id);
virtual ISrsResource *find_by_fast_id(uint64_t id); virtual ISrsResource *find_by_fast_id(uint64_t id);
@ -700,6 +702,7 @@ public:
virtual void add(ISrsResource *conn, bool *exists = NULL); virtual void add(ISrsResource *conn, bool *exists = NULL);
virtual void add_with_id(const std::string &id, ISrsResource *conn); virtual void add_with_id(const std::string &id, ISrsResource *conn);
virtual void add_with_fast_id(uint64_t id, ISrsResource *conn); virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
virtual void add_with_name(const std::string &name, ISrsResource *conn);
virtual ISrsResource *at(int index); virtual ISrsResource *at(int index);
virtual ISrsResource *find_by_id(std::string id); virtual ISrsResource *find_by_id(std::string id);
virtual ISrsResource *find_by_fast_id(uint64_t id); virtual ISrsResource *find_by_fast_id(uint64_t id);
@ -796,6 +799,7 @@ public:
virtual void add(ISrsResource *conn, bool *exists = NULL); virtual void add(ISrsResource *conn, bool *exists = NULL);
virtual void add_with_id(const std::string &id, ISrsResource *conn); virtual void add_with_id(const std::string &id, ISrsResource *conn);
virtual void add_with_fast_id(uint64_t id, ISrsResource *conn); virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
virtual void add_with_name(const std::string &name, ISrsResource *conn);
virtual ISrsResource *at(int index); virtual ISrsResource *at(int index);
virtual ISrsResource *find_by_id(std::string id); virtual ISrsResource *find_by_id(std::string id);
virtual ISrsResource *find_by_fast_id(uint64_t id); virtual ISrsResource *find_by_fast_id(uint64_t id);

1348
trunk/src/utest/srs_utest_app16.cpp
View File

File diff suppressed because it is too large Load Diff

199
trunk/src/utest/srs_utest_app16.hpp
View File

@ -13,10 +13,41 @@
#include <srs_utest.hpp> #include <srs_utest.hpp>
#include <srs_app_listener.hpp> #include <srs_app_listener.hpp>
#include <srs_protocol_srt.hpp>
#include <srs_utest_app10.hpp> #include <srs_utest_app10.hpp>
#include <srs_utest_app11.hpp> #include <srs_utest_app11.hpp>
#include <srs_utest_app6.hpp> #include <srs_utest_app6.hpp>
// Mock ISrsSrtSocket for testing SrsSrtConnection
class MockSrtSocket : public ISrsSrtSocket
{
public:
srs_utime_t recv_timeout_;
srs_utime_t send_timeout_;
int64_t recv_bytes_;
int64_t send_bytes_;
srs_error_t recvmsg_error_;
srs_error_t sendmsg_error_;
int recvmsg_called_count_;
int sendmsg_called_count_;
std::string last_recv_data_;
std::string last_send_data_;
public:
MockSrtSocket();
virtual ~MockSrtSocket();
public:
virtual srs_error_t recvmsg(void *buf, size_t size, ssize_t *nread);
virtual srs_error_t sendmsg(void *buf, size_t size, ssize_t *nwrite);
virtual void set_recv_timeout(srs_utime_t tm);
virtual void set_send_timeout(srs_utime_t tm);
virtual srs_utime_t get_send_timeout();
virtual srs_utime_t get_recv_timeout();
virtual int64_t get_send_bytes();
virtual int64_t get_recv_bytes();
};
// Mock ISrsUdpHandler for testing SrsUdpListener // Mock ISrsUdpHandler for testing SrsUdpListener
class MockUdpHandler : public ISrsUdpHandler class MockUdpHandler : public ISrsUdpHandler
{ {
@ -53,4 +84,172 @@ public:
virtual srs_error_t on_udp_packet(ISrsUdpMuxSocket *skt); virtual srs_error_t on_udp_packet(ISrsUdpMuxSocket *skt);
}; };
// Mock ISrsProtocolReadWriter for testing SrsSrtRecvThread
class MockSrtProtocolReadWriter : public ISrsProtocolReadWriter
{
public:
srs_error_t read_error_;
int read_count_;
bool simulate_timeout_;
std::string test_data_;
srs_utime_t recv_timeout_;
srs_utime_t send_timeout_;
int64_t recv_bytes_;
int64_t send_bytes_;
public:
MockSrtProtocolReadWriter();
virtual ~MockSrtProtocolReadWriter();
public:
virtual srs_error_t read(void *buf, size_t size, ssize_t *nread);
virtual srs_error_t read_fully(void *buf, size_t size, ssize_t *nread);
virtual srs_error_t write(void *buf, size_t size, ssize_t *nwrite);
virtual srs_error_t writev(const iovec *iov, int iov_size, ssize_t *nwrite);
virtual void set_recv_timeout(srs_utime_t tm);
virtual srs_utime_t get_recv_timeout();
virtual int64_t get_recv_bytes();
virtual void set_send_timeout(srs_utime_t tm);
virtual srs_utime_t get_send_timeout();
virtual int64_t get_send_bytes();
};
// Mock ISrsCoroutine for testing SrsSrtRecvThread
class MockSrtCoroutine : public ISrsCoroutine
{
public:
srs_error_t pull_error_;
int pull_count_;
bool started_;
SrsContextId cid_;
public:
MockSrtCoroutine();
virtual ~MockSrtCoroutine();
public:
virtual srs_error_t start();
virtual void stop();
virtual void interrupt();
virtual srs_error_t pull();
virtual const SrsContextId &cid();
virtual void set_cid(const SrsContextId &cid);
};
// Mock SrsSrtSource for testing SrsMpegtsSrtConn::on_srt_packet
class MockSrtSourceForPacket : public SrsSrtSource
{
public:
int on_packet_called_count_;
srs_error_t on_packet_error_;
SrsSrtPacket *last_packet_;
public:
MockSrtSourceForPacket();
virtual ~MockSrtSourceForPacket();
virtual srs_error_t on_packet(SrsSrtPacket *packet);
};
// Mock ISrsAppConfig for testing SrsMpegtsSrtConn HTTP hooks
class MockAppConfigForSrtHooks : public MockAppConfig
{
public:
SrsConfDirective *on_connect_directive_;
SrsConfDirective *on_close_directive_;
SrsConfDirective *on_publish_directive_;
SrsConfDirective *on_unpublish_directive_;
SrsConfDirective *on_play_directive_;
SrsConfDirective *on_stop_directive_;
public:
MockAppConfigForSrtHooks();
virtual ~MockAppConfigForSrtHooks();
virtual SrsConfDirective *get_vhost_on_connect(std::string vhost);
virtual SrsConfDirective *get_vhost_on_close(std::string vhost);
virtual SrsConfDirective *get_vhost_on_publish(std::string vhost);
virtual SrsConfDirective *get_vhost_on_unpublish(std::string vhost);
virtual SrsConfDirective *get_vhost_on_play(std::string vhost);
virtual SrsConfDirective *get_vhost_on_stop(std::string vhost);
void set_on_connect_urls(const std::vector<std::string> &urls);
void set_on_close_urls(const std::vector<std::string> &urls);
void set_on_publish_urls(const std::vector<std::string> &urls);
void set_on_unpublish_urls(const std::vector<std::string> &urls);
void set_on_play_urls(const std::vector<std::string> &urls);
void set_on_stop_urls(const std::vector<std::string> &urls);
void clear_on_connect_directive();
void clear_on_close_directive();
void clear_on_publish_directive();
void clear_on_unpublish_directive();
void clear_on_play_directive();
void clear_on_stop_directive();
};
// Mock ISrsHttpHooks for testing SrsMpegtsSrtConn HTTP hooks
class MockHttpHooksForSrt : public ISrsHttpHooks
{
public:
std::vector<std::pair<std::string, ISrsRequest *> > on_connect_calls_;
int on_connect_count_;
srs_error_t on_connect_error_;
std::vector<std::tuple<std::string, ISrsRequest *, int64_t, int64_t> > on_close_calls_;
int on_close_count_;
std::vector<std::pair<std::string, ISrsRequest *> > on_publish_calls_;
int on_publish_count_;
srs_error_t on_publish_error_;
std::vector<std::pair<std::string, ISrsRequest *> > on_unpublish_calls_;
int on_unpublish_count_;
std::vector<std::pair<std::string, ISrsRequest *> > on_play_calls_;
int on_play_count_;
srs_error_t on_play_error_;
std::vector<std::pair<std::string, ISrsRequest *> > on_stop_calls_;
int on_stop_count_;
public:
MockHttpHooksForSrt();
virtual ~MockHttpHooksForSrt();
virtual srs_error_t on_connect(std::string url, ISrsRequest *req);
virtual void on_close(std::string url, ISrsRequest *req, int64_t send_bytes, int64_t recv_bytes);
virtual srs_error_t on_publish(std::string url, ISrsRequest *req);
virtual void on_unpublish(std::string url, ISrsRequest *req);
virtual srs_error_t on_play(std::string url, ISrsRequest *req);
virtual void on_stop(std::string url, ISrsRequest *req);
virtual srs_error_t on_dvr(SrsContextId cid, std::string url, ISrsRequest *req, std::string file);
virtual srs_error_t on_hls(SrsContextId cid, std::string url, ISrsRequest *req, std::string file, std::string ts_url,
std::string m3u8, std::string m3u8_url, int sn, srs_utime_t duration);
virtual srs_error_t on_hls_notify(SrsContextId cid, std::string url, ISrsRequest *req, std::string ts_url, int nb_notify);
virtual srs_error_t discover_co_workers(std::string url, std::string &host, int &port);
virtual srs_error_t on_forward_backend(std::string url, ISrsRequest *req, std::vector<std::string> &rtmp_urls);
void clear_calls();
};
// Mock SrsProtocolUtility for testing discover_candidates
class MockProtocolUtility : public SrsProtocolUtility
{
public:
std::vector<SrsIPAddress *> mock_ips_;
public:
MockProtocolUtility();
virtual ~MockProtocolUtility();
virtual std::vector<SrsIPAddress *> &local_ips();
void add_ip(std::string ip, std::string ifname, bool is_ipv4, bool is_loopback, bool is_internet);
void clear_ips();
};
// Mock ISrsAppConfig for testing discover_candidates
class MockAppConfigForDiscoverCandidates : public MockAppConfig
{
public:
std::string rtc_server_candidates_;
bool use_auto_detect_network_ip_;
std::string rtc_server_ip_family_;
public:
MockAppConfigForDiscoverCandidates();
virtual ~MockAppConfigForDiscoverCandidates();
virtual std::string get_rtc_server_candidates();
virtual bool get_use_auto_detect_network_ip();
virtual std::string get_rtc_server_ip_family();
};
#endif #endif

28
trunk/src/utest/srs_utest_app6.cpp
View File

@ -2183,6 +2183,9 @@ MockAppConfig::MockAppConfig()
rtc_to_rtmp_ = false; rtc_to_rtmp_ = false;
dash_dispose_ = 0; dash_dispose_ = 0;
dash_enabled_ = false; dash_enabled_ = false;
api_as_candidates_ = true;
resolve_api_domain_ = true;
keep_api_domain_ = false;
} }
MockAppConfig::~MockAppConfig() MockAppConfig::~MockAppConfig()
@ -2271,6 +2274,16 @@ bool MockAppConfig::get_srt_enabled(std::string vhost)
return srt_enabled_; return srt_enabled_;
} }
std::string MockAppConfig::get_srt_default_streamid()
{
return "#!::r=live/livestream,m=request";
}
bool MockAppConfig::get_srt_to_rtmp(std::string vhost)
{
return true;
}
bool MockAppConfig::get_rtc_to_rtmp(std::string vhost) bool MockAppConfig::get_rtc_to_rtmp(std::string vhost)
{ {
return rtc_to_rtmp_; return rtc_to_rtmp_;
@ -2571,6 +2584,21 @@ void MockAppConfig::set_rtc_to_rtmp(bool enabled)
rtc_to_rtmp_ = enabled; rtc_to_rtmp_ = enabled;
} }
void MockAppConfig::set_api_as_candidates(bool enabled)
{
api_as_candidates_ = enabled;
}
void MockAppConfig::set_resolve_api_domain(bool enabled)
{
resolve_api_domain_ = enabled;
}
void MockAppConfig::set_keep_api_domain(bool enabled)
{
keep_api_domain_ = enabled;
}
// Mock request implementation // Mock request implementation
MockRtcAsyncCallRequest::MockRtcAsyncCallRequest(std::string vhost, std::string app, std::string stream) MockRtcAsyncCallRequest::MockRtcAsyncCallRequest(std::string vhost, std::string app, std::string stream)
{ {

14
trunk/src/utest/srs_utest_app6.hpp
View File

@ -244,6 +244,9 @@ public:
bool rtc_to_rtmp_; bool rtc_to_rtmp_;
srs_utime_t dash_dispose_; srs_utime_t dash_dispose_;
bool dash_enabled_; bool dash_enabled_;
bool api_as_candidates_;
bool resolve_api_domain_;
bool keep_api_domain_;
public: public:
MockAppConfig(); MockAppConfig();
@ -295,6 +298,12 @@ public:
virtual std::vector<std::string> get_rtc_server_listens() { return std::vector<std::string>(); } virtual std::vector<std::string> get_rtc_server_listens() { return std::vector<std::string>(); }
virtual int get_rtc_server_reuseport() { return 1; } virtual int get_rtc_server_reuseport() { return 1; }
virtual bool get_rtc_server_encrypt() { return false; } virtual bool get_rtc_server_encrypt() { return false; }
virtual bool get_api_as_candidates() { return api_as_candidates_; }
virtual bool get_resolve_api_domain() { return resolve_api_domain_; }
virtual bool get_keep_api_domain() { return keep_api_domain_; }
virtual std::string get_rtc_server_candidates() { return "*"; }
virtual bool get_use_auto_detect_network_ip() { return true; }
virtual std::string get_rtc_server_ip_family() { return "ipv4"; }
virtual bool get_rtsp_server_enabled() { return false; } virtual bool get_rtsp_server_enabled() { return false; }
virtual std::vector<std::string> get_rtsp_server_listens() { return std::vector<std::string>(); } virtual std::vector<std::string> get_rtsp_server_listens() { return std::vector<std::string>(); }
virtual std::vector<std::string> get_srt_listens() { return std::vector<std::string>(); } virtual std::vector<std::string> get_srt_listens() { return std::vector<std::string>(); }
@ -358,6 +367,8 @@ public:
virtual bool get_rtc_twcc_enabled(std::string vhost); virtual bool get_rtc_twcc_enabled(std::string vhost);
virtual bool get_srt_enabled(); virtual bool get_srt_enabled();
virtual bool get_srt_enabled(std::string vhost); virtual bool get_srt_enabled(std::string vhost);
virtual std::string get_srt_default_streamid();
virtual bool get_srt_to_rtmp(std::string vhost);
virtual bool get_rtc_to_rtmp(std::string vhost); virtual bool get_rtc_to_rtmp(std::string vhost);
virtual srs_utime_t get_rtc_stun_timeout(std::string vhost); virtual srs_utime_t get_rtc_stun_timeout(std::string vhost);
virtual bool get_rtc_stun_strict_check(std::string vhost); virtual bool get_rtc_stun_strict_check(std::string vhost);
@ -448,6 +459,9 @@ public:
void set_rtc_twcc_enabled(bool enabled); void set_rtc_twcc_enabled(bool enabled);
void set_srt_enabled(bool enabled); void set_srt_enabled(bool enabled);
void set_rtc_to_rtmp(bool enabled); void set_rtc_to_rtmp(bool enabled);
void set_api_as_candidates(bool enabled);
void set_resolve_api_domain(bool enabled);
void set_keep_api_domain(bool enabled);
}; };
// Mock request for testing SrsRtcAsyncCallOnStop // Mock request for testing SrsRtcAsyncCallOnStop

4
trunk/src/utest/srs_utest_app7.cpp
View File

@ -1029,6 +1029,10 @@ void MockConnectionManagerForExpire::add_with_fast_id(uint64_t /*id*/, ISrsResou
{ {
} }
void MockConnectionManagerForExpire::add_with_name(const std::string & /*name*/, ISrsResource * /*conn*/)
{
}
ISrsResource *MockConnectionManagerForExpire::at(int /*index*/) ISrsResource *MockConnectionManagerForExpire::at(int /*index*/)
{ {
return NULL; return NULL;

1
trunk/src/utest/srs_utest_app7.hpp
View File

@ -107,6 +107,7 @@ public:
virtual void add(ISrsResource *conn, bool *exists = NULL); virtual void add(ISrsResource *conn, bool *exists = NULL);
virtual void add_with_id(const std::string &id, ISrsResource *conn); virtual void add_with_id(const std::string &id, ISrsResource *conn);
virtual void add_with_fast_id(uint64_t id, ISrsResource *conn); virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
virtual void add_with_name(const std::string &name, ISrsResource *conn);
virtual ISrsResource *at(int index); virtual ISrsResource *at(int index);
virtual ISrsResource *find_by_id(std::string id); virtual ISrsResource *find_by_id(std::string id);
virtual ISrsResource *find_by_fast_id(uint64_t id); virtual ISrsResource *find_by_fast_id(uint64_t id);

4
trunk/src/utest/srs_utest_service.cpp
View File

@ -1379,6 +1379,10 @@ void MockConnectionManager::add_with_fast_id(uint64_t /*id*/, ISrsResource * /*c
{ {
} }
void MockConnectionManager::add_with_name(const std::string & /*name*/, ISrsResource * /*conn*/)
{
}
ISrsResource *MockConnectionManager::at(int /*index*/) ISrsResource *MockConnectionManager::at(int /*index*/)
{ {
return NULL; return NULL;

1
trunk/src/utest/srs_utest_service.hpp
View File

@ -94,6 +94,7 @@ public:
virtual void add(ISrsResource *conn, bool *exists = NULL); virtual void add(ISrsResource *conn, bool *exists = NULL);
virtual void add_with_id(const std::string &id, ISrsResource *conn); virtual void add_with_id(const std::string &id, ISrsResource *conn);
virtual void add_with_fast_id(uint64_t id, ISrsResource *conn); virtual void add_with_fast_id(uint64_t id, ISrsResource *conn);
virtual void add_with_name(const std::string &name, ISrsResource *conn);
virtual ISrsResource *at(int index); virtual ISrsResource *at(int index);
virtual ISrsResource *find_by_id(std::string id); virtual ISrsResource *find_by_id(std::string id);
virtual ISrsResource *find_by_fast_id(uint64_t id); virtual ISrsResource *find_by_fast_id(uint64_t id);