external/srs
1
0
Fork 0
mirror of https://github.com/ossrs/srs.git synced 2025-02-15 04:42:04 +00:00
Code Issues Releases Wiki Activity

HLS: Extract a HLS streaming to serve pseudo session.

Browse source
This commit is contained in:
winlin 2022-08-27 20:32:38 +08:00
parent d39995cc31
commit ccf83a3f68
3 changed files with 235 additions and 186 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
trunk/conf/hls.origin.conf
View file

@ -10,6 +10,8 @@ http_server {
vhost __defaultVhost__ {
hls {
enabled on;
# Note that it will make NGINX edge cache always missed, so never enable HLS streaming if use NGINX edges.
hls_ctx off;
}
}

382
trunk/src/app/srs_app_http_static.cpp
View file

@ -41,14 +41,15 @@ using namespace std;
#define SRS_CONTEXT_IN_HLS "hls_ctx"
SrsVodStream::SrsVodStream(string root_dir) : SrsHttpFileServer(root_dir)
SrsHlsStream::SrsHlsStream()
{
_srs_hybrid->timer5s()->subscribe(this);
}
SrsVodStream::~SrsVodStream()
SrsHlsStream::~SrsHlsStream()
{
_srs_hybrid->timer5s()->unsubscribe(this);
std::map<std::string, SrsM3u8CtxInfo>::iterator it;
for (it = map_ctx_info_.begin(); it != map_ctx_info_.end(); ++it) {
srs_freep(it->second.req);
@ -56,6 +57,202 @@ SrsVodStream::~SrsVodStream()
map_ctx_info_.clear();
}
srs_error_t SrsHlsStream::serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, bool* served)
{
srs_error_t err = srs_success;
SrsHttpMessage* hr = dynamic_cast<SrsHttpMessage*>(r);
srs_assert(hr);
SrsRequest* req = hr->to_request(hr->host())->as_http();
SrsAutoFree(SrsRequest, req);
// discovery vhost, resolve the vhost from config
SrsConfDirective* parsed_vhost = _srs_config->get_vhost(req->vhost);
if (parsed_vhost) {
req->vhost = parsed_vhost->arg0();
}
// If HLS stream is disabled, use SrsHttpFileServer to serve HLS, which is normal file server.
if (!_srs_config->get_hls_ctx_enabled(req->vhost)) {
return err;
}
// Serve as HLS stream, create a HLS session to serve it.
string ctx = hr->query_get(SRS_CONTEXT_IN_HLS);
if (!ctx.empty() && ctx_is_exist(ctx)) {
alive(ctx, NULL);
return err;
}
// Create a m3u8 in memory, contains the session id(ctx).
if (ctx.empty()) {
// make sure unique
do {
ctx = srs_random_str(8); // the same as cid
} while (ctx_is_exist(ctx));
}
SrsContextRestore(_srs_context->get_id());
_srs_context->set_id(SrsContextId().set_value(ctx));
if ((err = http_hooks_on_play(req)) != srs_success) {
return srs_error_wrap(err, "HLS: http_hooks_on_play");
}
std::stringstream ss;
ss << "#EXTM3U" << SRS_CONSTS_LF;
ss << "#EXT-X-STREAM-INF:BANDWIDTH=1,AVERAGE-BANDWIDTH=1" << SRS_CONSTS_LF;
ss << hr->path() << "?" << SRS_CONTEXT_IN_HLS << "=" << ctx;
if (!hr->query().empty() && hr->query_get(SRS_CONTEXT_IN_HLS).empty()) {
ss << "&" << hr->query();
}
std::string res = ss.str();
int length = res.length();
w->header()->set_content_length(length);
w->header()->set_content_type("application/vnd.apple.mpegurl");
w->write_header(SRS_CONSTS_HTTP_OK);
if ((err = w->write((char*)res.c_str(), length)) != srs_success) {
return srs_error_wrap(err, "write bytes=%d", length);
}
if ((err = w->final_request()) != srs_success) {
return srs_error_wrap(err, "final request");
}
alive(ctx, req->copy());
// update the statistic when source disconveried.
SrsStatistic* stat = SrsStatistic::instance();
if ((err = stat->on_client(ctx, req, NULL, SrsRtmpConnPlay)) != srs_success) {
return srs_error_wrap(err, "stat on client");
}
// The request has been served by HLS streaming handler.
*served = true;
return err;
}
bool SrsHlsStream::ctx_is_exist(std::string ctx)
{
return (map_ctx_info_.find(ctx) != map_ctx_info_.end());
}
void SrsHlsStream::alive(std::string ctx, SrsRequest* req)
{
std::map<std::string, SrsM3u8CtxInfo>::iterator it;
if ((it = map_ctx_info_.find(ctx)) != map_ctx_info_.end()) {
it->second.request_time = srs_get_system_time();
} else {
SrsM3u8CtxInfo info;
info.req = req;
info.request_time = srs_get_system_time();
map_ctx_info_.insert(make_pair(ctx, info));
}
}
srs_error_t SrsHlsStream::http_hooks_on_play(SrsRequest* req)
{
srs_error_t err = srs_success;
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return err;
}
// the http hooks will cause context switch,
// so we must copy all hooks for the on_connect may freed.
// @see https://github.com/ossrs/srs/issues/475
vector<string> hooks;
if (true) {
SrsConfDirective* conf = _srs_config->get_vhost_on_play(req->vhost);
if (!conf) {
return err;
}
hooks = conf->args;
}
for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i);
if ((err = SrsHttpHooks::on_play(url, req)) != srs_success) {
return srs_error_wrap(err, "http on_play %s", url.c_str());
}
}
return err;
}
void SrsHlsStream::http_hooks_on_stop(SrsRequest* req)
{
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return;
}
// the http hooks will cause context switch,
// so we must copy all hooks for the on_connect may freed.
// @see https://github.com/ossrs/srs/issues/475
vector<string> hooks;
if (true) {
SrsConfDirective* conf = _srs_config->get_vhost_on_stop(req->vhost);
if (!conf) {
srs_info("ignore the empty http callback: on_stop");
return;
}
hooks = conf->args;
}
for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i);
SrsHttpHooks::on_stop(url, req);
}
return;
}
srs_error_t SrsHlsStream::on_timer(srs_utime_t interval)
{
srs_error_t err = srs_success;
std::map<std::string, SrsM3u8CtxInfo>::iterator it;
for (it = map_ctx_info_.begin(); it != map_ctx_info_.end(); ++it) {
string ctx = it->first;
SrsRequest* req = it->second.req;
srs_utime_t hls_window = _srs_config->get_hls_window(req->vhost);
if (it->second.request_time + (2 * hls_window) < srs_get_system_time()) {
SrsContextRestore(_srs_context->get_id());
_srs_context->set_id(SrsContextId().set_value(ctx));
http_hooks_on_stop(req);
srs_freep(req);
SrsStatistic* stat = SrsStatistic::instance();
stat->on_disconnect(ctx);
map_ctx_info_.erase(it);
break;
}
}
return err;
}
SrsVodStream::SrsVodStream(string root_dir) : SrsHttpFileServer(root_dir)
{
}
SrsVodStream::~SrsVodStream()
{
}
srs_error_t SrsVodStream::serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, int64_t offset)
{
srs_error_t err = srs_success;
@ -191,184 +388,19 @@ srs_error_t SrsVodStream::serve_m3u8_ctx(ISrsHttpResponseWriter * w, ISrsHttpMes
{
srs_error_t err = srs_success;
SrsHttpMessage* hr = dynamic_cast<SrsHttpMessage*>(r);
srs_assert(hr);
SrsRequest* req = hr->to_request(hr->host())->as_http();
SrsAutoFree(SrsRequest, req);
// discovery vhost, resolve the vhost from config
SrsConfDirective* parsed_vhost = _srs_config->get_vhost(req->vhost);
if (parsed_vhost) {
req->vhost = parsed_vhost->arg0();
// Try to serve by HLS streaming.
bool served = false;
if ((err = hls_.serve_m3u8_ctx(w, r, &served)) != srs_success) {
return srs_error_wrap(err, "hls stream");
}
// If HLS stream is disabled, use SrsHttpFileServer to serve HLS, which is normal file server.
if (!_srs_config->get_hls_ctx_enabled(req->vhost)) {
return SrsHttpFileServer::serve_m3u8_ctx(w, r, fullpath);
}
// Serve as HLS stream, create a HLS session to serve it.
string ctx = hr->query_get(SRS_CONTEXT_IN_HLS);
if (!ctx.empty() && ctx_is_exist(ctx)) {
alive(ctx, NULL);
return SrsHttpFileServer::serve_m3u8_ctx(w, r, fullpath);
}
if (ctx.empty()) {
// make sure unique
do {
ctx = srs_random_str(8); // the same as cid
} while (ctx_is_exist(ctx));
}
SrsContextRestore(_srs_context->get_id());
_srs_context->set_id(SrsContextId().set_value(ctx));
if ((err = http_hooks_on_play(req)) != srs_success) {
return srs_error_wrap(err, "HLS: http_hooks_on_play");
}
std::stringstream ss;
ss << "#EXTM3U" << SRS_CONSTS_LF;
ss << "#EXT-X-STREAM-INF:BANDWIDTH=1,AVERAGE-BANDWIDTH=1" << SRS_CONSTS_LF;
ss << hr->path() << "?" << SRS_CONTEXT_IN_HLS << "=" << ctx;
if (!hr->query().empty() && hr->query_get(SRS_CONTEXT_IN_HLS).empty()) {
ss << "&" << hr->query();
}
std::string res = ss.str();
int length = res.length();
w->header()->set_content_length(length);
w->header()->set_content_type("application/vnd.apple.mpegurl");
w->write_header(SRS_CONSTS_HTTP_OK);
if ((err = w->write((char*)res.c_str(), length)) != srs_success) {
return srs_error_wrap(err, "write bytes=%d", length);
}
if ((err = w->final_request()) != srs_success) {
return srs_error_wrap(err, "final request");
}
alive(ctx, req->copy());
// update the statistic when source disconveried.
SrsStatistic* stat = SrsStatistic::instance();
if ((err = stat->on_client(ctx, req, NULL, SrsRtmpConnPlay)) != srs_success) {
return srs_error_wrap(err, "stat on client");
}
return err;
}
bool SrsVodStream::ctx_is_exist(std::string ctx)
{
return (map_ctx_info_.find(ctx) != map_ctx_info_.end());
}
void SrsVodStream::alive(std::string ctx, SrsRequest* req)
{
std::map<std::string, SrsM3u8CtxInfo>::iterator it;
if ((it = map_ctx_info_.find(ctx)) != map_ctx_info_.end()) {
it->second.request_time = srs_get_system_time();
} else {
SrsM3u8CtxInfo info;
info.req = req;
info.request_time = srs_get_system_time();
map_ctx_info_.insert(make_pair(ctx, info));
}
}
srs_error_t SrsVodStream::http_hooks_on_play(SrsRequest* req)
{
srs_error_t err = srs_success;
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
// Done if already served.
if (served) {
return err;
}
// the http hooks will cause context switch,
// so we must copy all hooks for the on_connect may freed.
// @see https://github.com/ossrs/srs/issues/475
vector<string> hooks;
if (true) {
SrsConfDirective* conf = _srs_config->get_vhost_on_play(req->vhost);
if (!conf) {
return err;
}
hooks = conf->args;
}
for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i);
if ((err = SrsHttpHooks::on_play(url, req)) != srs_success) {
return srs_error_wrap(err, "http on_play %s", url.c_str());
}
}
return err;
}
void SrsVodStream::http_hooks_on_stop(SrsRequest* req)
{
if (!_srs_config->get_vhost_http_hooks_enabled(req->vhost)) {
return;
}
// the http hooks will cause context switch,
// so we must copy all hooks for the on_connect may freed.
// @see https://github.com/ossrs/srs/issues/475
vector<string> hooks;
if (true) {
SrsConfDirective* conf = _srs_config->get_vhost_on_stop(req->vhost);
if (!conf) {
srs_info("ignore the empty http callback: on_stop");
return;
}
hooks = conf->args;
}
for (int i = 0; i < (int)hooks.size(); i++) {
std::string url = hooks.at(i);
SrsHttpHooks::on_stop(url, req);
}
return;
}
srs_error_t SrsVodStream::on_timer(srs_utime_t interval)
{
srs_error_t err = srs_success;
std::map<std::string, SrsM3u8CtxInfo>::iterator it;
for (it = map_ctx_info_.begin(); it != map_ctx_info_.end(); ++it) {
string ctx = it->first;
SrsRequest* req = it->second.req;
srs_utime_t hls_window = _srs_config->get_hls_window(req->vhost);
if (it->second.request_time + (2 * hls_window) < srs_get_system_time()) {
SrsContextRestore(_srs_context->get_id());
_srs_context->set_id(SrsContextId().set_value(ctx));
http_hooks_on_stop(req);
srs_freep(req);
SrsStatistic* stat = SrsStatistic::instance();
stat->on_disconnect(ctx);
map_ctx_info_.erase(it);
break;
}
}
return err;
// Serve by default HLS handler.
return SrsHttpFileServer::serve_m3u8_ctx(w, r, fullpath);
}
SrsHttpStaticServer::SrsHttpStaticServer(SrsServer* svr)

37
trunk/src/app/srs_app_http_static.hpp
View file

@ -17,22 +17,17 @@ struct SrsM3u8CtxInfo
SrsRequest* req;
};
// The flv vod stream supports flv?start=offset-bytes.
// For example, http://server/file.flv?start=10240
// server will write flv header and sequence header,
// then seek(10240) and response flv tag data.
class SrsVodStream : public SrsHttpFileServer, public ISrsFastTimer
// Server HLS streaming.
class SrsHlsStream : public ISrsFastTimer
{
private:
// The period of validity of the ctx
std::map<std::string, SrsM3u8CtxInfo> map_ctx_info_;
public:
SrsVodStream(std::string root_dir);
virtual ~SrsVodStream();
protected:
virtual srs_error_t serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int64_t offset);
virtual srs_error_t serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int64_t start, int64_t end);
virtual srs_error_t serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
SrsHlsStream();
virtual ~SrsHlsStream();
public:
virtual srs_error_t serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, bool* served);
private:
virtual bool ctx_is_exist(std::string ctx);
virtual void alive(std::string ctx, SrsRequest* req);
@ -43,6 +38,26 @@ private:
srs_error_t on_timer(srs_utime_t interval);
};
// The Vod streaming, like FLV, MP4 or HLS streaming.
class SrsVodStream : public SrsHttpFileServer
{
private:
SrsHlsStream hls_;
public:
SrsVodStream(std::string root_dir);
virtual ~SrsVodStream();
protected:
// The flv vod stream supports flv?start=offset-bytes.
// For example, http://server/file.flv?start=10240
// server will write flv header and sequence header,
// then seek(10240) and response flv tag data.
virtual srs_error_t serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int64_t offset);
// Support mp4 with start and offset in query string.
virtual srs_error_t serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int64_t start, int64_t end);
// Support HLS streaming with pseudo session id.
virtual srs_error_t serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
};
// The http static server instance,
// serve http static file and flv/mp4 vod stream.
class SrsHttpStaticServer : public ISrsReloadHandler