HLS: Stat the HLS streaming clients bandwidth. v5.0.49

trunk/conf/clion.conf

@@ -41,5 +41,8 @@ vhost __defaultVhost__ {
         enabled     on;
         mount       [vhost]/[app]/[stream].flv;
     }
+    hls {
+        enabled         on;
+    }
 }
 

trunk/doc/CHANGELOG.md

@@ -7,6 +7,7 @@ The changelog for SRS.
 
 ## SRS 5.0 Changelog
 
+* v5.0, 2022-08-29, HLS: Stat the HLS streaming clients bandwidth. v5.0.49
 * v5.0, 2022-08-28, URL: Use SrsHttpUri to parse URL and query. v5.0.48
 * v5.0, 2022-08-28, Fix [#2881](https://github.com/ossrs/srs/issues/2881): HTTP: Support merging api to server. v5.0.47
 * v5.0, 2022-08-27, Fix [#3108](https://github.com/ossrs/srs/issues/3108): STAT: Update stat for SRT. v5.0.46

trunk/src/app/srs_app_http_static.cpp

@@ -68,7 +68,7 @@ SrsHlsStream::~SrsHlsStream()
     map_ctx_info_.clear();
 }
 
-srs_error_t SrsHlsStream::serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, SrsRequest* req)
+srs_error_t SrsHlsStream::serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, ISrsFileReaderFactory* factory, string fullpath, SrsRequest* req)
 {
     srs_error_t err = srs_success;
 
@@ -79,18 +79,40 @@ srs_error_t SrsHlsStream::serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMess
 
     // Already exists context, response with rebuilt m3u8 content.
     if (!ctx.empty() && ctx_is_exist(ctx)) {
-        return serve_exists_session(w, r, fullpath);
+        return serve_exists_session(w, r, factory, fullpath);
     }
 
     // Create a m3u8 in memory, contains the session id(ctx).
     return serve_new_session(w, r, req);
 }
 
+void SrsHlsStream::on_serve_ts_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
+{
+    string ctx = r->query_get(SRS_CONTEXT_IN_HLS);
+    if (ctx.empty() || !ctx_is_exist(ctx)) {
+        return;
+    }
+
+    SrsHttpMessage* hr = dynamic_cast<SrsHttpMessage*>(r);
+    srs_assert(hr);
+
+    SrsHttpConn* hc = dynamic_cast<SrsHttpConn*>(hr->connection());
+    srs_assert(hc);
+
+    ISrsKbpsDelta* conn = dynamic_cast<ISrsKbpsDelta*>(hc);
+    srs_assert(conn);
+
+    // Only update the delta, because SrsServer will sample it. Note that SrsServer also does the stat for all clients
+    // including this one, but it should be ignored because the id is not matched, and instead we use the hls_ctx as
+    // session id to match the client.
+    SrsStatistic::instance()->kbps_add_delta(ctx, conn);
+}
+
 srs_error_t SrsHlsStream::serve_new_session(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, SrsRequest* req)
 {
     srs_error_t err = srs_success;
 
-    SrsHttpMessage *hr = dynamic_cast<SrsHttpMessage *>(r);
+    SrsHttpMessage* hr = dynamic_cast<SrsHttpMessage*>(r);
     srs_assert(hr);
 
     string ctx;
@@ -140,18 +162,20 @@ srs_error_t SrsHlsStream::serve_new_session(ISrsHttpResponseWriter* w, ISrsHttpM
     return err;
 }
 
-srs_error_t SrsHlsStream::serve_exists_session(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath)
+srs_error_t SrsHlsStream::serve_exists_session(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, ISrsFileReaderFactory* factory, std::string fullpath)
 {
     srs_error_t err = srs_success;
 
     // Read m3u8 content.
-    SrsFileReader fs;
+    SrsFileReader* fs = factory->create_file_reader();
-    if ((err = fs.open(fullpath)) != srs_success) {
+    SrsAutoFree(SrsFileReader, fs);
+
+    if ((err = fs->open(fullpath)) != srs_success) {
         return srs_error_wrap(err, "open %s", fullpath.c_str());
     }
 
     string content;
-    if ((err = srs_ioutil_read_all(&fs, content)) != srs_success) {
+    if ((err = srs_ioutil_read_all(fs, content)) != srs_success) {
         return srs_error_wrap(err, "read %s", fullpath.c_str());
     }
 
@@ -445,14 +469,14 @@ srs_error_t SrsVodStream::serve_m3u8_ctx(ISrsHttpResponseWriter * w, ISrsHttpMes
 {
     srs_error_t err = srs_success;
 
-    SrsHttpMessage *hr = dynamic_cast<SrsHttpMessage *>(r);
+    SrsHttpMessage* hr = dynamic_cast<SrsHttpMessage*>(r);
     srs_assert(hr);
 
-    SrsRequest *req = hr->to_request(hr->host())->as_http();
+    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);
+    SrsConfDirective* parsed_vhost = _srs_config->get_vhost(req->vhost);
     if (parsed_vhost) {
         req->vhost = parsed_vhost->arg0();
     }
@@ -464,7 +488,23 @@ srs_error_t SrsVodStream::serve_m3u8_ctx(ISrsHttpResponseWriter * w, ISrsHttpMes
     }
 
     // Try to serve by HLS streaming.
-    return hls_.serve_m3u8_ctx(w, r, fullpath, req);
+    return hls_.serve_m3u8_ctx(w, r, fs_factory, fullpath, req);
+}
+
+srs_error_t SrsVodStream::serve_ts_ctx(ISrsHttpResponseWriter * w, ISrsHttpMessage * r, std::string fullpath)
+{
+    srs_error_t err = srs_success;
+
+    // SrsServer also stat all HTTP connections including this one, but it should be ignored because the id is not
+    // matched to any exists client. And we will do stat for the HLS streaming by session in hls_ctx.
+
+    // Serve by default HLS handler.
+    err = SrsHttpFileServer::serve_ts_ctx(w, r, fullpath);
+
+    // Notify the HLS to stat the ts after serving.
+    hls_.on_serve_ts_ctx(w, r);
+
+    return err;
 }
 
 SrsHttpStaticServer::SrsHttpStaticServer(SrsServer* svr)

trunk/src/app/srs_app_http_static.hpp

@@ -11,6 +11,8 @@
 
 #include <srs_app_http_conn.hpp>
 
+class ISrsFileReaderFactory;
+
 struct SrsM3u8CtxInfo
 {
     srs_utime_t request_time;
@@ -29,10 +31,11 @@ public:
     SrsHlsStream();
     virtual ~SrsHlsStream();
 public:
-    virtual srs_error_t serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, SrsRequest* req);
+    virtual srs_error_t serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, ISrsFileReaderFactory* factory, std::string fullpath, SrsRequest* req);
+    virtual void on_serve_ts_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
 private:
     srs_error_t serve_new_session(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, SrsRequest *req);
-    srs_error_t serve_exists_session(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
+    srs_error_t serve_exists_session(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, ISrsFileReaderFactory* factory, std::string fullpath);
     bool ctx_is_exist(std::string ctx);
     void alive(std::string ctx, SrsRequest* req);
     srs_error_t http_hooks_on_play(SrsRequest* req);
@@ -60,6 +63,7 @@ protected:
     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);
+    virtual srs_error_t serve_ts_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
 };
 
 // The http static server instance,

trunk/src/core/srs_core_version5.hpp

@@ -9,6 +9,6 @@
 
 #define VERSION_MAJOR       5
 #define VERSION_MINOR       0
-#define VERSION_REVISION    48
+#define VERSION_REVISION    49
 
 #endif

trunk/src/protocol/srs_protocol_http_stack.cpp

@@ -382,6 +382,8 @@ srs_error_t SrsHttpFileServer::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMes
         return serve_mp4_file(w, r, fullpath);
     } else if (srs_string_ends_with(upath, ".m3u8")) {
         return serve_m3u8_file(w, r, fullpath);
+    } else if (srs_string_ends_with(upath, ".ts")) {
+        return serve_ts_file(w, r, fullpath);
     }
     
     // serve common static file.
@@ -531,6 +533,11 @@ srs_error_t SrsHttpFileServer::serve_m3u8_file(ISrsHttpResponseWriter * w, ISrsH
     return serve_m3u8_ctx(w, r, fullpath);
 }
 
+srs_error_t SrsHttpFileServer::serve_ts_file(ISrsHttpResponseWriter * w, ISrsHttpMessage * r, std::string fullpath)
+{
+    return serve_ts_ctx(w, r, fullpath);
+}
+
 srs_error_t SrsHttpFileServer::serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, string fullpath, int64_t offset)
 {
     // @remark For common http file server, we don't support stream request, please use SrsVodStream instead.
@@ -548,7 +555,12 @@ srs_error_t SrsHttpFileServer::serve_mp4_stream(ISrsHttpResponseWriter* w, ISrsH
 srs_error_t SrsHttpFileServer::serve_m3u8_ctx(ISrsHttpResponseWriter * w, ISrsHttpMessage * r, std::string fullpath)
 {
     // @remark For common http file server, we don't support stream request, please use SrsVodStream instead.
-    // TODO: FIXME: Support range in header https://developer.mozilla.org/zh-CN/docs/Web/HTTP/Range_requests
+    return serve_file(w, r, fullpath);
+}
+
+srs_error_t SrsHttpFileServer::serve_ts_ctx(ISrsHttpResponseWriter * w, ISrsHttpMessage * r, std::string fullpath)
+{
+    // @remark For common http file server, we don't support stream request, please use SrsVodStream instead.
     return serve_file(w, r, fullpath);
 }
 

trunk/src/protocol/srs_protocol_http_stack.hpp

@@ -286,6 +286,7 @@ private:
     virtual srs_error_t serve_flv_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
     virtual srs_error_t serve_mp4_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
     virtual srs_error_t serve_m3u8_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
+    virtual srs_error_t serve_ts_file(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
 protected:
     // When access flv file with x.flv?start=xxx
     virtual srs_error_t serve_flv_stream(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath, int64_t offset);
@@ -304,6 +305,7 @@ protected:
     // Remark 2:
     //           If use two same "hls_ctx" in different requests, SRS cannot detect so that they will be treated as one.
     virtual srs_error_t serve_m3u8_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
+    virtual srs_error_t serve_ts_ctx(ISrsHttpResponseWriter* w, ISrsHttpMessage* r, std::string fullpath);
 protected:
     // Copy the fs to response writer in size bytes.
     virtual srs_error_t copy(ISrsHttpResponseWriter* w, SrsFileReader* fs, ISrsHttpMessage* r, int64_t size);

trunk/src/utest/srs_utest_http.cpp

@@ -1225,7 +1225,7 @@ VOID TEST(ProtocolHTTPTest, VodStreamHandlers)
         HELPER_ASSERT_SUCCESS(r.set_url("/index.m3u8", false));
 
         HELPER_ASSERT_SUCCESS(h.serve_http(&w, &r));
-        __MOCK_HTTP_EXPECT_STRCT(200, "index.m3u8?hls_ctx=", w);
+        __MOCK_HTTP_EXPECT_STRHAS(200, "index.m3u8?hls_ctx=", w);
     }
 
     // Should return "hls_ctx"
@@ -1243,7 +1243,25 @@ VOID TEST(ProtocolHTTPTest, VodStreamHandlers)
         HELPER_ASSERT_SUCCESS(r.set_url("/index.m3u8?hls_ctx=123456", false));
 
         HELPER_ASSERT_SUCCESS(h.serve_http(&w, &r));
-        __MOCK_HTTP_EXPECT_STRCT(200, "index.m3u8?hls_ctx=123456", w);
+        __MOCK_HTTP_EXPECT_STREQ(200, "Hello, world!", w);
+    }
+
+    // Should return "hls_ctx"
+    if (true) {
+        SrsHttpMuxEntry e;
+        e.pattern = "/";
+
+        SrsVodStream h("/tmp");
+        h.set_fs_factory(new MockFileReaderFactory("livestream-13.ts"));
+        h.set_path_check(_mock_srs_path_always_exists);
+        h.entry = &e;
+
+        MockResponseWriter w;
+        SrsHttpMessage r(NULL, NULL);
+        HELPER_ASSERT_SUCCESS(r.set_url("/index.m3u8?hls_ctx=123456", false));
+
+        HELPER_ASSERT_SUCCESS(h.serve_http(&w, &r));
+        __MOCK_HTTP_EXPECT_STREQ(200, "livestream-13.ts?hls_ctx=123456", w);
     }
 }
 

trunk/src/utest/srs_utest_http.hpp

@@ -47,7 +47,7 @@ bool is_string_contain(string substr, string str);
 #define __MOCK_HTTP_EXPECT_STREQ2(status, text, w) \
         EXPECT_STREQ(mock_http_response2(status, text).c_str(), HELPER_BUFFER2STR(&w.io.out_buffer).c_str())
 
-#define __MOCK_HTTP_EXPECT_STRCT(status, text, w) \
+#define __MOCK_HTTP_EXPECT_STRHAS(status, text, w) \
         EXPECT_PRED2(is_string_contain, text, HELPER_BUFFER2STR(&w.io.out_buffer).c_str())
 
 #endif