FLV: Refine source and http handler. v6.0.155 v7.0.14

Do not create a source when mounting FLV because it may not unmount FLV when freeing the source. If you access the FLV stream without any publisher, then wait for source cleanup and review the FLV stream again, there is an annoying warning message.

# View HTTP FLV stream by curl, wait for stream to be ready.
# curl http://localhost:8080/live/livestream.flv -v >/dev/null
HTTP #0 127.0.0.1:58026 GET http://localhost:8080/live/livestream.flv, content-length=-1
new live source, stream_url=/live/livestream
http: mount flv stream for sid=/live/livestream, mount=/live/livestream.flv

# Cancel the curl and trigger source cleanup without http unmount.
client disconnect peer. ret=1007
Live: cleanup die source, id=[], total=1

# View the stream again, it fails.
# curl http://localhost:8080/live/livestream.flv -v >/dev/null
HTTP #0 127.0.0.1:58040 GET http://localhost:8080/live/livestream.flv, content-length=-1
serve error code=1097(NoSource)(No source found) : process request=0 : cors serve : serve http : no source for /live/livestream
serve_http() [srs_app_http_stream.cpp:641]

Note: There is an inconsistency. The first time, you can access the FLV stream and wait for the publisher, but the next time, you cannot.

Create a source when starting to serve the FLV client. We do not need to create the source when creating the HTTP handler. Instead, we should try to create the source in the cache or stream. Because the source cleanup does not unmount the HTTP handler, the handler remains after the source is destroyed. The next time you access the FLV stream, the source is not found.

srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph) {
    SrsSharedPtr<SrsLiveSource> live_source;
    if ((err = _srs_sources->fetch_or_create(r.get(), server, live_source)) != srs_success) { }
    if ((err = http_mount(r.get())) != srs_success) { }

srs_error_t SrsBufferCache::cycle() {
    SrsSharedPtr<SrsLiveSource> live_source = _srs_sources->fetch(req);
    if (!live_source.get()) {
        return srs_error_new(ERROR_NO_SOURCE, "no source for %s", req->get_stream_url().c_str());
    }

srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) {
    SrsSharedPtr<SrsLiveSource> live_source = _srs_sources->fetch(req);
    if (!live_source.get()) {
        return srs_error_new(ERROR_NO_SOURCE, "no source for %s", req->get_stream_url().c_str());
    }

Note: We should not create the source in hijack, instead, we create it in cache or stream:

srs_error_t SrsHttpStreamServer::hijack(ISrsHttpMessage* request, ISrsHttpHandler** ph) {
    if ((err = http_mount(r.get())) != srs_success) { }

srs_error_t SrsBufferCache::cycle() {
    SrsSharedPtr<SrsLiveSource> live_source;
    if ((err = _srs_sources->fetch_or_create(req, server_, live_source)) != srs_success) { }

srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) {
    SrsSharedPtr<SrsLiveSource> live_source;
    if ((err = _srs_sources->fetch_or_create(req, server_, live_source)) != srs_success) { }

Note: This fixes the failure and annoying warning message, and maintains consistency by always waiting for the stream to be ready if there is no publisher.

Fail the http request if the HTTP handler is disposing, and also keep the handler entry when disposing the stream, because we should dispose the handler entry and stream at the same time.

srs_error_t SrsHttpStreamServer::http_mount(SrsRequest* r) {
        entry = streamHandlers[sid];
        if (entry->disposing) {
            return srs_error_new(ERROR_STREAM_DISPOSING, "stream is disposing");
        }

void SrsHttpStreamServer::http_unmount(SrsRequest* r) {
    std::map<std::string, SrsLiveEntry*>::iterator it = streamHandlers.find(sid);
    SrsUniquePtr<SrsLiveEntry> entry(it->second);
    entry->disposing = true;

Note: If the disposal process takes a long time, this will prevent unexpected behavior or access to the resource that is being disposed of.

In edge mode, the edge ingester will unpublish the source when the last consumer quits, which is actually triggered by the HTTP stream. While it also waits for the stream to quit when the HTTP unmounts, there is a self-destruction risk: the HTTP live stream object destroys itself.

srs_error_t SrsLiveStream::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r) {
    SrsUniquePtr<SrsLiveConsumer> consumer(consumer_raw); // Trigger destroy.

void SrsHttpStreamServer::http_unmount(SrsRequest* r) {
    for (;;) { if (!cache->alive() && !stream->alive()) { break; } // A circle reference.
    mux.unhandle(entry->mount, stream.get()); // Free the SrsLiveStream itself.

Note: It also introduces a circular reference in the object relationships, the stream reference to itself when unmount:

SrsLiveStream::serve_http 
    -> SrsLiveConsumer::~SrsLiveConsumer -> SrsEdgeIngester::stop 
    -> SrsLiveSource::on_unpublish -> SrsHttpStreamServer::http_unmount 
        -> SrsLiveStream::alive

Note: We should use an asynchronous worker to perform the cleanup to avoid the stream destroying itself and to prevent self-referencing.

void SrsHttpStreamServer::http_unmount(SrsRequest* r) {
    entry->disposing = true;
    if ((err = async_->execute(new SrsHttpStreamDestroy(&mux, &streamHandlers, sid))) != srs_success) { }

Note: This also ensures there are no circular references and no self-destruction.

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

ossrs / srs

FLV: Refine source and http handler. v6.0.155 v7.0.14 #4165