HTTP-FLV: Notify connection to expire when unpublishing. v6.0.152 (#4164

)

When stopping the stream, it will wait for the HTTP Streaming to exit.
If the HTTP Streaming goroutine hangs, it will not exit automatically.

```cpp
void SrsHttpStreamServer::http_unmount(SrsRequest* r)
{
    SrsUniquePtr<SrsLiveStream> stream(entry->stream);
    if (stream->entry) stream->entry->enabled = false;
    srs_usleep(...); // Wait for about 120s.
    mux.unhandle(entry->mount, stream.get()); // Free stream.
}

srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
{
    err = do_serve_http(w, r); // If stuck in here for 120s+
    alive_viewers_--; // Crash at here, because stream has been deleted.
```

We should notify http stream connection to interrupt(expire):

```cpp
void SrsHttpStreamServer::http_unmount(SrsRequest* r)
{
    SrsUniquePtr<SrsLiveStream> stream(entry->stream);
    if (stream->entry) stream->entry->enabled = false;
    stream->expire(); // Notify http stream to interrupt.
```

Note that we should notify all viewers pulling stream from this http
stream.

Note that we have tried to fix this issue, but only try to wait for all
viewers to quit, without interrupting the viewers, see
#4144

---------

Co-authored-by: Jacob Su <suzp1984@gmail.com>

trunk/doc/CHANGELOG.md

@@ -7,6 +7,7 @@ The changelog for SRS.
 <a name="v6-changes"></a>
 
 ## SRS 6.0 Changelog
+* v6.0, 2024-08-31, Merge [#4164](https://github.com/ossrs/srs/pull/4164): HTTP-FLV: Notify connection to expire when unpublishing. v6.0.152 (#4164)
 * v6.0, 2024-08-24, Merge [#4157](https://github.com/ossrs/srs/pull/4157): Fix crash when quiting. v6.0.151 (#4157)
 * v6.0, 2024-08-24, Merge [#4156](https://github.com/ossrs/srs/pull/4156): Build: Fix srs_mp4_parser compiling error. v6.0.150 (#4156)
 * v6.0, 2024-08-21, Merge [#4150](https://github.com/ossrs/srs/pull/4150): API: Support new HTTP API for VALGRIND. v6.0.149 (#4150)

trunk/src/app/srs_app_http_api.cpp

@@ -1206,8 +1206,6 @@ SrsGoApiSignal::~SrsGoApiSignal()
 
 srs_error_t SrsGoApiSignal::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
 {
-    srs_error_t err = srs_success;
-
     std::string signal = r->query_get("signo");
     srs_trace("query signo=%s", signal.c_str());
 

trunk/src/app/srs_app_http_stream.cpp

@@ -583,13 +583,15 @@ SrsLiveStream::SrsLiveStream(SrsRequest* r, SrsBufferCache* c)
     cache = c;
     req = r->copy()->as_http();
     security_ = new SrsSecurity();
-    alive_viewers_ = 0;
 }
 
 SrsLiveStream::~SrsLiveStream()
 {
     srs_freep(req);
     srs_freep(security_);
+
+    // The live stream should never be destroyed when it's serving any viewers.
+    srs_assert(viewers_.empty());
 }
 
 srs_error_t SrsLiveStream::update_auth(SrsRequest* r)
@@ -634,18 +636,35 @@ srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage
         return srs_error_wrap(err, "http hook");
     }
 
-    alive_viewers_++;
+    // Add the viewer to the viewers list.
+    viewers_.push_back(hc);
+
+    // Serve the viewer connection.
     err = do_serve_http(w, r);
-    alive_viewers_--;
-
+
+    // Remove viewer from the viewers list.
+    vector<ISrsExpire*>::iterator it = std::find(viewers_.begin(), viewers_.end(), hc);
+    srs_assert (it != viewers_.end());
+    viewers_.erase(it);
+
+    // Do hook after serving.
     http_hooks_on_stop(r);
 
     return err;
 }
 
 bool SrsLiveStream::alive()
 {
-    return alive_viewers_ > 0;
+    return !viewers_.empty();
+}
+
+void SrsLiveStream::expire()
+{
+    vector<ISrsExpire*>::iterator it;
+    for (it = viewers_.begin(); it != viewers_.end(); ++it) {
+        ISrsExpire* conn = *it;
+        conn->expire();
+    }
 }
 
 srs_error_t SrsLiveStream::do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
@@ -1075,6 +1094,7 @@ void SrsHttpStreamServer::http_unmount(SrsRequest* r)
 
     // Notify cache and stream to stop.
     if (stream->entry) stream->entry->enabled = false;
+    stream->expire();
     cache->stop();
 
     // Wait for cache and stream to stop.

trunk/src/app/srs_app_http_stream.hpp

@@ -11,6 +11,8 @@
 #include <srs_app_security.hpp>
 #include <srs_app_http_conn.hpp>
 
+#include <vector>
+
 class SrsAacTransmuxer;
 class SrsMp3Transmuxer;
 class SrsFlvTransmuxer;
@@ -176,7 +178,7 @@ class SrsBufferWriter : public SrsFileWriter
 
 // HTTP Live Streaming, to transmux RTMP to HTTP FLV or other format.
 // TODO: FIXME: Rename to SrsHttpLive
-class SrsLiveStream : public ISrsHttpHandler
+class SrsLiveStream : public ISrsHttpHandler, public ISrsExpire
 {
 private:
     SrsRequest* req;
@@ -185,14 +187,17 @@ class SrsLiveStream : public ISrsHttpHandler
     // For multiple viewers, which means there will more than one alive viewers for a live stream, so we must
     // use an int value to represent if there is any viewer is alive. We should never do cleanup unless all
     // viewers closed the connection.
-    int alive_viewers_;
+    std::vector<ISrsExpire*> viewers_;
 public:
     SrsLiveStream(SrsRequest* r, SrsBufferCache* c);
     virtual ~SrsLiveStream();
     virtual srs_error_t update_auth(SrsRequest* r);
 public:
     virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
     virtual bool alive();
+// Interface ISrsExpire
+public:
+    virtual void expire();
 private:
     virtual srs_error_t do_serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
     virtual srs_error_t http_hooks_on_play(ISrsHttpMessage* r);

trunk/src/core/srs_core_version6.hpp

@@ -9,6 +9,6 @@
 
 #define VERSION_MAJOR       6
 #define VERSION_MINOR       0
-#define VERSION_REVISION    151
+#define VERSION_REVISION    152
 
 #endif