[apps] srt-live-transmit stops listening after SRT disconnect during …

…no active input (#2997)
Haivision · Jan 24, 2025 · dfa9a99 · dfa9a99
1 parent a6b9959
commit dfa9a99
Show file tree

Hide file tree

Showing 2 changed files with 58 additions and 16 deletions.
diff --git a/apps/srt-live-transmit.cpp b/apps/srt-live-transmit.cpp
@@ -487,6 +487,7 @@ int main(int argc, char** argv)
     bool srcConnected = false;
     unique_ptr<Target> tar;
     bool tarConnected = false;
+    bool srcMayBlock = false;
 
     int pollid = srt_epoll_create();
     if (pollid < 0)
@@ -514,6 +515,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 +539,19 @@ 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;
+                        int con = src->GetSysSocket();
+                        // try to make the standard input non blocking
+                        srcMayBlock = fcntl(con, F_SETFL, fcntl(con, F_GETFL) | O_NONBLOCK) < 0;
+                        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 +596,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 +779,46 @@ int main(int argc, char** argv)
                     break;
                 }
 
+
+                bool srcReady = false;
+
+                if (src.get() && src->IsOpen())
+                {
+                    if (srtrfdslen > 0)
+                    {
+                        SRTSOCKET sock = src->GetSRTSocket();
+                        if (sock != SRT_INVALID_SOCK)
+                        {
+                            for (int n = 0; n < srtrfdslen; n ++)
+                                if (sock == srtrwfds[n])
+                                {
+                                    srcReady = true;
+                                    break;
+                                }
+
+                        }
+                    }
+                    if (!srcReady && sysrfdslen > 0)
+                    {
+                        int sock = src->GetSysSocket();
+                        if (sock != -1)
+                        {
+                            for (int n = 0; n < sysrfdslen; n++)
+                                if (sock == sysrfds[n])
+                                {
+                                    srcReady = true;
+                                    break;
+                                }
+
+                        }
+                    } 
+                }
                 // 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 +842,10 @@ int main(int argc, char** argv)
 
                         dataqueue.push_back(pkt);
                         receivedBytes += pkt->payload.size();
+                        if (srcMayBlock) 
+                            break;
                     }
                 }
-
                 // if there is no target, let the received data be lost
                 while (!dataqueue.empty())
                 {

diff --git a/apps/transmitmedia.cpp b/apps/transmitmedia.cpp
@@ -720,9 +720,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)
+        int ret = ::read(GetSysSocket(), pkt.payload.data(), chunk);
+        if (ret <= 0)
         {
             pkt.payload.clear();
             return 0;
@@ -731,14 +730,14 @@ 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 IsOpen() override { return !cin.eof(); }
     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 +766,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;