TSSRT Support

src/input/input.cpp

@@ -679,12 +679,16 @@ namespace Mist{
       // if not shutting down, wait 1 second before looping
       if (config->is_active){Util::wait(INPUT_USER_INTERVAL);}
     }
-    if (streamStatus){streamStatus.mapped[0] = STRMSTAT_SHUTDOWN;}
-    config->is_active = false;
+    if (!isThread()){
+      if (streamStatus){streamStatus.mapped[0] = STRMSTAT_SHUTDOWN;}
+      config->is_active = false;
+    }
     finish();
     INFO_MSG("Input closing clean, reason: %s", Util::exitReason);
     userSelect.clear();
-    if (streamStatus){streamStatus.mapped[0] = STRMSTAT_OFF;}
+    if (!isThread()){
+      if (streamStatus){streamStatus.mapped[0] = STRMSTAT_OFF;}
+    }
   }
 
   /// This function checks if an input in serve mode should keep running or not.
@@ -729,7 +733,6 @@ namespace Mist{
   /// - if there are tracks, register as a non-viewer on the user page of the buffer
   /// - call getNext() in a loop, buffering packets
   void Input::stream(){
-
     std::map<std::string, std::string> overrides;
     overrides["throughboot"] = "";
     if (config->getBool("realtime") ||
@@ -895,6 +898,7 @@ namespace Mist{
           statComm.setTime(now - startTime);
           statComm.setLastSecond(0);
           statComm.setHost(getConnectedBinHost());
+          handleLossyStats(statComm);
         }
 
         statTimer = Util::bootSecs();

src/input/input.h

@@ -41,6 +41,7 @@ namespace Mist{
     virtual bool readHeader();
     virtual bool needHeader(){return !readExistingHeader();}
     virtual bool preRun(){return true;}
+    virtual bool isThread(){return false;}
     virtual bool isSingular(){return !config->getBool("realtime");}
     virtual bool readExistingHeader();
     virtual bool atKeyFrame();
@@ -69,6 +70,10 @@ namespace Mist{
     virtual void userOnDisconnect(size_t id);
     virtual void userLeadOut();
 
+    virtual void handleLossyStats(Comms::Statistics & statComm){}
+
+    virtual bool preventBufferStart() {return false;}
+
     virtual void parseHeader();
     bool bufferFrame(size_t track, uint32_t keyNum);
 

src/input/input_tssrt.cpp

@@ -0,0 +1,217 @@
+#include "input_tssrt.h"
+#include <cerrno>
+#include <cstdio>
+#include <cstdlib>
+#include <cstring>
+#include <fstream>
+#include <iomanip>
+#include <iostream>
+#include <mist/defines.h>
+#include <mist/downloader.h>
+#include <mist/flv_tag.h>
+#include <mist/http_parser.h>
+#include <mist/mp4_generic.h>
+#include <mist/socket_srt.h>
+#include <mist/stream.h>
+#include <mist/timing.h>
+#include <mist/ts_packet.h>
+#include <mist/util.h>
+#include <string>
+
+#include <mist/procs.h>
+#include <mist/tinythread.h>
+#include <sys/stat.h>
+
+Util::Config *cfgPointer = NULL;
+std::string baseStreamName;
+
+/// Global, so that all tracks stay in sync
+int64_t timeStampOffset = 0;
+
+// We use threads here for multiple input pushes, because of the internals of the SRT Library
+static void callThreadCallbackSRT(void *socknum){
+  SRTSOCKET sock = *((SRTSOCKET *)socknum);
+  // use the accepted socket as the second parameter
+  Mist::inputTSSRT inp(cfgPointer, sock);
+  inp.setSingular(false);
+  inp.run();
+}
+
+namespace Mist{
+  /// Constructor of TS Input
+  /// \arg cfg Util::Config that contains all current configurations.
+  inputTSSRT::inputTSSRT(Util::Config *cfg, SRTSOCKET s) : Input(cfg){
+    capa["name"] = "TSSRT";
+    capa["desc"] = "This input allows for processing MPEG2-TS-based SRT streams. Use mode=listener "
+                   "for push input.";
+    capa["source_match"].append("srt://*");
+    // These can/may be set to always-on mode
+    capa["always_match"].append("srt://*");
+    capa["incoming_push_url"] = "srt://$host:$port";
+    capa["incoming_push_url_match"] = "srt://*";
+    capa["priority"] = 9;
+    capa["codecs"][0u][0u].append("H264");
+    capa["codecs"][0u][0u].append("HEVC");
+    capa["codecs"][0u][0u].append("MPEG2");
+    capa["codecs"][0u][1u].append("AAC");
+    capa["codecs"][0u][1u].append("AC3");
+    capa["codecs"][0u][1u].append("MP2");
+
+    JSON::Value option;
+    option["arg"] = "integer";
+    option["long"] = "buffer";
+    option["short"] = "b";
+    option["help"] = "DVR buffer time in ms";
+    option["value"].append(50000);
+    config->addOption("bufferTime", option);
+    capa["optional"]["DVR"]["name"] = "Buffer time (ms)";
+    capa["optional"]["DVR"]["help"] =
+        "The target available buffer time for this live stream, in milliseconds. This is the time "
+        "available to seek around in, and will automatically be extended to fit whole keyframes as "
+        "well as the minimum duration needed for stable playback.";
+    capa["optional"]["DVR"]["option"] = "--buffer";
+    capa["optional"]["DVR"]["type"] = "uint";
+    capa["optional"]["DVR"]["default"] = 50000;
+
+    // Setup if we are called form with a thread for push-based input.
+    if (s != -1){
+      srtConn = Socket::SRTConnection(s);
+      streamName = baseStreamName;
+      if (srtConn.getStreamName() != ""){streamName += "+" + srtConn.getStreamName();}
+    }
+    lastTimeStamp = 0;
+    singularFlag = true;
+  }
+
+  inputTSSRT::~inputTSSRT(){}
+
+  bool inputTSSRT::checkArguments(){return true;}
+
+  /// Live Setup of SRT Input. Runs only if we are the "main" thread
+  bool inputTSSRT::preRun(){
+    if (srtConn.getSocket() == -1){
+      std::string source = config->getString("input");
+      standAlone = false;
+      HTTP::URL u(source);
+      INFO_MSG("Parsed url: %s", u.getUrl().c_str());
+      if (Socket::interpretSRTMode(u) == "listener"){
+        sSock = Socket::SRTServer(u.getPort(), u.host, false);
+        config->registerSRTSockPtr(&sSock);
+      }else{
+        INFO_MSG("A");
+        std::map<std::string, std::string> arguments;
+        HTTP::parseVars(u.args, arguments);
+        size_t connectCnt = 0;
+        do{
+          srtConn.connect(u.host, u.getPort(), "input", arguments);
+          if (!srtConn){Util::sleep(1000);}
+          ++connectCnt;
+        }while (!srtConn && connectCnt < 10);
+        if (!srtConn){WARN_MSG("Connecting to %s timed out", u.getUrl().c_str());}
+      }
+    }
+    return true;
+  }
+
+  // Retrieve the next packet to be played from the srt connection.
+  void inputTSSRT::getNext(size_t idx){
+    thisPacket.null();
+    bool hasPacket = tsStream.hasPacket();
+    bool firstloop = true;
+    while (!hasPacket && srtConn.connected() && config->is_active){
+      firstloop = false;
+      // Receive data from the socket. SRT Sockets handle some internal timing as well, based on the provided settings.
+      leftBuffer.append(srtConn.RecvNow());
+      if (leftBuffer.size()){
+        size_t offset = 0;
+        size_t garbage = 0;
+        while ((offset + 188) < leftBuffer.size()){
+          if (leftBuffer[offset] != 0x47){
+            ++garbage;
+            if (garbage % 100 == 0){INFO_MSG("Accumulated %zu bytes of garbage", garbage);}
+            ++offset;
+            continue;
+          }
+          if (garbage != 0){
+            WARN_MSG("Thrown away %zu bytes of garbage data", garbage);
+            garbage = 0;
+          }
+          if (offset + 188 <= leftBuffer.size()){
+            tsBuf.FromPointer(leftBuffer.data() + offset);
+            tsStream.parse(tsBuf, 0);
+            offset += 188;
+          }
+        }
+        leftBuffer.erase(0, offset);
+        hasPacket = tsStream.hasPacket();
+      }else if (srtConn.connected()){
+        // This should not happen as the SRT socket is read blocking and won't return until there is
+        // data. But if it does, wait before retry
+        Util::sleep(10);
+      }
+    }
+    if (hasPacket){tsStream.getEarliestPacket(thisPacket);}
+
+    if (!thisPacket){
+      INFO_MSG("Could not getNext TS packet!");
+      return;
+    }
+
+    tsStream.initializeMetadata(meta);
+    size_t thisIdx = M.trackIDToIndex(thisPacket.getTrackId(), getpid());
+    if (thisIdx == INVALID_TRACK_ID){getNext(idx);}
+
+    uint64_t adjustTime = thisPacket.getTime() + timeStampOffset;
+    if (lastTimeStamp || timeStampOffset){
+      if (lastTimeStamp + 5000 < adjustTime || lastTimeStamp > adjustTime + 5000){
+        INFO_MSG("Timestamp jump " PRETTY_PRINT_MSTIME " -> " PRETTY_PRINT_MSTIME ", compensating.",
+                 PRETTY_ARG_MSTIME(lastTimeStamp), PRETTY_ARG_MSTIME(adjustTime));
+        timeStampOffset += (lastTimeStamp - adjustTime);
+        adjustTime = thisPacket.getTime() + timeStampOffset;
+      }
+    }
+    lastTimeStamp = adjustTime;
+    thisPacket.setTime(adjustTime);
+  }
+
+  bool inputTSSRT::openStreamSource(){return true;}
+
+  void inputTSSRT::parseStreamHeader(){
+    // Placeholder empty track to force normal code to continue despite no tracks available
+    tmpIdx = meta.addTrack(0, 0, 0, 0);
+  }
+
+  void inputTSSRT::streamMainLoop(){
+    meta.removeTrack(tmpIdx);
+    // If we do not have a srtConn here, we are the main thread and should start accepting pushes.
+    if (srtConn.getSocket() == -1){
+      cfgPointer = config;
+      baseStreamName = streamName;
+      while (config->is_active && sSock.connected()){
+        Socket::SRTConnection S = sSock.accept();
+        if (S.connected()){// check if the new connection is valid
+          SRTSOCKET sock = S.getSocket();
+          // spawn a new thread for this connection
+          tthread::thread T(callThreadCallbackSRT, (void *)&sock);
+          // detach it, no need to keep track of it anymore
+          T.detach();
+          HIGH_MSG("Spawned new thread for socket %i", S.getSocket());
+        }
+      }
+      return;
+    }
+    // If we are here: we have a proper connection (either accepted or pull input) and should start parsing it as such
+    Input::streamMainLoop();
+  }
+
+  bool inputTSSRT::needsLock(){return false;}
+
+  void inputTSSRT::setSingular(bool newSingular){singularFlag = newSingular;}
+
+  void inputTSSRT::handleLossyStats(Comms::Statistics &statComm){
+    statComm.setPacketCount(srtConn.packetCount());
+    statComm.setPacketLostCount(srtConn.packetLostCount());
+    statComm.setPacketRetransmitCount(srtConn.packetRetransmitCount());
+  }
+
+}// namespace Mist

src/input/input_tssrt.h

@@ -0,0 +1,48 @@
+#include "input.h"
+#include <mist/dtsc.h>
+#include <mist/nal.h>
+#include <mist/socket_srt.h>
+#include <mist/ts_packet.h>
+#include <mist/ts_stream.h>
+#include <set>
+#include <string>
+
+namespace Mist{
+  /// This class contains all functions needed to implement TS Input
+  class inputTSSRT : public Input{
+  public:
+    inputTSSRT(Util::Config *cfg, SRTSOCKET s = -1);
+    ~inputTSSRT();
+    void setSingular(bool newSingular);
+    virtual bool needsLock();
+
+  protected:
+    // Private Functions
+    bool checkArguments();
+    bool preRun();
+    virtual void getNext(size_t idx = INVALID_TRACK_ID);
+    virtual bool needHeader(){return false;}
+    virtual bool preventBufferStart(){return srtConn.getSocket() == -1;}
+    virtual bool isSingular(){return singularFlag;}
+    virtual bool isThread(){return !singularFlag;}
+
+    bool openStreamSource();
+    void parseStreamHeader();
+    void streamMainLoop();
+    TS::Stream tsStream; ///< Used for parsing the incoming ts stream
+    TS::Packet tsBuf;
+    std::string leftBuffer;
+    uint64_t lastTimeStamp;
+
+    Socket::SRTServer sSock;
+    Socket::SRTConnection srtConn;
+    bool singularFlag;
+    size_t tmpIdx;
+    virtual size_t streamByteCount(){
+      return srtConn.dataDown();
+    }; // For live streams: to update the stats with correct values.
+    virtual void handleLossyStats(Comms::Statistics &statComm);
+  };
+}// namespace Mist
+
+typedef Mist::inputTSSRT mistIn;