[apps] FIX srt-live-transmit stops listening after SRT disconnect #2997…

…... (#3108) * [apps] Fix srt-live-transmit stops listening after SRT disconnects during no active input on console #2997 (#3108) --------- Co-authored-by: Clement Gerouville <[email protected]> Co-authored-by: Maxim Sharabayko <[email protected]>
Haivision · Feb 13, 2025 · ce0a888 · ce0a888
1 parent bc8a084
commit ce0a888
Show file tree

Hide file tree

Showing 3 changed files with 51 additions and 15 deletions.
diff --git a/apps/srt-live-transmit.cpp b/apps/srt-live-transmit.cpp
@@ -514,6 +514,7 @@ int main(int argc, char** argv)
                     return 1;
                 }
                 int events = SRT_EPOLL_IN | SRT_EPOLL_ERR;
+
                 switch (src->uri.type())
                 {
                 case UriParser::SRT:
@@ -537,14 +538,18 @@ int main(int argc, char** argv)
                     }
                     break;
                 case UriParser::FILE:
-                    if (srt_epoll_add_ssock(pollid,
-                        src->GetSysSocket(), &events))
                     {
-                        cerr << "Failed to add FILE source to poll, "
-                            << src->GetSysSocket() << endl;
-                        return 1;
+                        const int con = src->GetSysSocket();
+                        // try to make the standard input non blocking
+                        if (srt_epoll_add_ssock(pollid, con, &events))
+                        {
+                            cerr << "Failed to add FILE source to poll, "
+                                << src->GetSysSocket() << endl;
+                            return 1;
+                        }
+                        break;
+
                     }
-                    break;
                 default:
                     break;
                 }
@@ -589,6 +594,7 @@ int main(int argc, char** argv)
             SRTSOCKET srtrwfds[4] = {SRT_INVALID_SOCK, SRT_INVALID_SOCK , SRT_INVALID_SOCK , SRT_INVALID_SOCK };
             int sysrfdslen = 2;
             SYSSOCKET sysrfds[2];
+
             if (srt_epoll_wait(pollid,
                 &srtrwfds[0], &srtrfdslen, &srtrwfds[2], &srtwfdslen,
                 100,
@@ -771,12 +777,34 @@ int main(int argc, char** argv)
                     break;
                 }
 
+
+                bool srcReady = false;
+
+                if (src.get() && src->IsOpen() && !src->End())
+                {
+                    if (srtrfdslen > 0)
+                    {
+                        SRTSOCKET sock = src->GetSRTSocket();
+                        if (sock != SRT_INVALID_SOCK)
+                        {
+                            for (int n = 0; n < srtrfdslen && !(srcReady = (sock == srtrwfds[n])); n ++);
+                        }
+                    }
+                    if (!srcReady && sysrfdslen > 0)
+                    {
+                        int sock = src->GetSysSocket();
+                        if (sock != -1)
+                        {
+                            for (int n = 0; n < sysrfdslen && !(srcReady = (sock == sysrfds[n])); n++);
+                        }
+                    } 
+                }
                 // read a few chunks at a time in attempt to deplete
                 // read buffers as much as possible on each read event
                 // note that this implies live streams and does not
                 // work for cached/file sources
                 std::list<std::shared_ptr<MediaPacket>> dataqueue;
-                if (src.get() && src->IsOpen() && (srtrfdslen || sysrfdslen))
+                if (srcReady)
                 {
                     while (dataqueue.size() < cfg.buffering)
                     {
@@ -800,9 +828,10 @@ int main(int argc, char** argv)
 
                         dataqueue.push_back(pkt);
                         receivedBytes += pkt->payload.size();
+                        if (src->MayBlock())
+                            break;
                     }
                 }
-
                 // if there is no target, let the received data be lost
                 while (!dataqueue.empty())
                 {

diff --git a/apps/transmitbase.hpp b/apps/transmitbase.hpp
@@ -71,6 +71,7 @@ class Source: public Location
 
     virtual SRTSOCKET GetSRTSocket() const { return SRT_INVALID_SOCK; }
     virtual int GetSysSocket() const { return -1; }
+    virtual bool MayBlock() const { return false; }
     virtual bool AcceptNewClient() { return false; }
 };
 

diff --git a/apps/transmitmedia.cpp b/apps/transmitmedia.cpp
@@ -704,9 +704,15 @@ Iface* CreateSrt(const string& host, int port, const map<string,string>& par) {
 
 class ConsoleSource: public Source
 {
+    const bool may_block = true;
 public:
 
     ConsoleSource()
+#ifdef _WIN32
+    : may_block(true)
+#else
+    : may_block(fcntl(fileno(stdin), F_SETFL, fcntl(fileno(stdin), F_GETFL) | O_NONBLOCK) < 0)
+#endif
     {
 #ifdef _WIN32
         // The default stdin mode on windows is text.
@@ -720,9 +726,8 @@ class ConsoleSource: public Source
         if (pkt.payload.size() < chunk)
             pkt.payload.resize(chunk);
 
-        bool st = cin.read(pkt.payload.data(), chunk).good();
-        chunk = cin.gcount();
-        if (chunk == 0 || !st)
+        const int ret = ::read(GetSysSocket(), pkt.payload.data(), chunk);
+        if (ret <= 0)
         {
             pkt.payload.clear();
             return 0;
@@ -731,14 +736,15 @@ class ConsoleSource: public Source
         // Save this time to potentially use it for SRT target.
         pkt.time = srt_time_now();
         if (chunk < pkt.payload.size())
-            pkt.payload.resize(chunk);
+            pkt.payload.resize(ret);
 
-        return (int) chunk;
+        return ret;
     }
 
     bool IsOpen() override { return cin.good(); }
+    bool MayBlock() const final { return may_block; }
     bool End() override { return cin.eof(); }
-    int GetSysSocket() const override { return 0; };
+    int GetSysSocket() const override { return fileno(stdin); };
 };
 
 class ConsoleTarget: public Target
@@ -767,7 +773,7 @@ class ConsoleTarget: public Target
 
     bool IsOpen() override { return cout.good(); }
     bool Broken() override { return cout.eof(); }
-    int GetSysSocket() const override { return 0; };
+    int GetSysSocket() const override { return fileno(stdout); };
 };
 
 template <class Iface> struct Console;