Merge branch 'development' into LTS_development

# Conflicts:
#	CMakeLists.txt
#	src/input/input_buffer.cpp

CMakeLists.txt

@@ -142,6 +142,7 @@ set(libHeaders
   ${SOURCE_DIR}/lib/h264.h
   ${SOURCE_DIR}/lib/h265.h
   ${SOURCE_DIR}/lib/http_parser.h
+  ${SOURCE_DIR}/lib/downloader.h
   ${SOURCE_DIR}/lib/json.h
   ${SOURCE_DIR}/lib/langcodes.h
   ${SOURCE_DIR}/lib/mp4_adobe.h
@@ -190,6 +191,7 @@ set(libSources
   ${SOURCE_DIR}/lib/h264.cpp
   ${SOURCE_DIR}/lib/h265.cpp
   ${SOURCE_DIR}/lib/http_parser.cpp
+  ${SOURCE_DIR}/lib/downloader.cpp
   ${SOURCE_DIR}/lib/json.cpp
   ${SOURCE_DIR}/lib/langcodes.cpp
   ${SOURCE_DIR}/lib/mp4_adobe.cpp
@@ -350,6 +352,7 @@ if (DEFINED WITH_AV )
 endif()
 makeInput(OGG ogg)
 makeInput(Buffer buffer)
+makeInput(H264 h264)
 makeInput(ISMV ismv)#LTS
 makeInput(MP4 mp4)#LTS
 makeInput(TS ts)#LTS

lib/downloader.cpp

@@ -0,0 +1,117 @@
+#include "downloader.h"
+#include "defines.h"
+#include "timing.h"
+
+namespace HTTP{
+
+  /// Returns a reference to the internal HTTP::Parser body element
+  std::string &Downloader::data(){return H.body;}
+
+  /// Returns the status text of the HTTP Request.
+  std::string &Downloader::getStatusText(){return H.method;}
+
+  /// Returns the status code of the HTTP Request.
+  uint32_t Downloader::getStatusCode(){return atoi(H.url.c_str());}
+
+  /// Returns true if the HTTP Request is OK
+  bool Downloader::isOk(){return (getStatusCode() == 200);}
+
+  /// Returns the given header from the response, or empty string if it does not exist.
+  std::string Downloader::getHeader(const std::string &headerName){
+    return H.GetHeader(headerName);
+  }
+
+  /// Simply turns link into a HTTP::URL and calls get(const HTTP::URL&)
+  bool Downloader::get(const std::string &link){
+    HTTP::URL uri(link);
+    return get(uri);
+  }
+
+  /// Sets an extra (or overridden) header to be sent with outgoing requests.
+  void Downloader::setHeader(const std::string &name, const std::string &val){
+    extraHeaders[name] = val;
+  }
+
+  /// Clears all extra/override headers for outgoing requests.
+  void Downloader::clearHeaders(){extraHeaders.clear();}
+
+  /// Downloads the given URL into 'H', returns true on success.
+  /// Makes at most 5 attempts, and will wait no longer than 5 seconds without receiving data.
+  bool Downloader::get(const HTTP::URL &link, uint8_t maxRecursiveDepth){
+    if (!link.host.size()){return false;}
+    if (link.protocol != "http"){
+      FAIL_MSG("Protocol not supported: %s", link.protocol.c_str());
+      return false;
+    }
+    INFO_MSG("Retrieving %s", link.getUrl().c_str());
+    unsigned int loop = 6; // max 5 attempts
+
+    while (--loop){// loop while we are unsuccessful
+      H.Clean();
+      // Reconnect if needed
+      if (!S || link.host != connectedHost || link.getPort() != connectedPort){
+        S.close();
+        connectedHost = link.host;
+        connectedPort = link.getPort();
+        S = Socket::Connection(connectedHost, connectedPort, true);
+      }
+      H.url = "/" + link.path;
+      if (link.args.size()){H.url += "?" + link.args;}
+      if (link.port.size()){
+        H.SetHeader("Host", link.host + ":" + link.port);
+      }else{
+        H.SetHeader("Host", link.host);
+      }
+      H.SetHeader("User-Agent", "MistServer " PACKAGE_VERSION);
+      H.SetHeader("X-Version", PACKAGE_VERSION);
+      H.SetHeader("Accept", "*/*");
+      if (extraHeaders.size()){
+        for (std::map<std::string, std::string>::iterator it = extraHeaders.begin();
+             it != extraHeaders.end(); ++it){
+          H.SetHeader(it->first, it->second);
+        }
+      }
+      H.SendRequest(S);
+      H.Clean();
+      uint64_t reqTime = Util::bootSecs();
+      while (S && Util::bootSecs() < reqTime + 5){
+        // No data? Wait for a second or so.
+        if (!S.spool()){
+          if (progressCallback != 0){
+            if (!progressCallback()){
+              WARN_MSG("Download aborted by callback");
+              return false;
+            }
+          }
+          Util::sleep(250);
+          continue;
+        }
+        // Data! Check if we can parse it...
+        if (H.Read(S)){
+          if (getStatusCode() >= 300 && getStatusCode() < 400){
+            // follow redirect
+            std::string location = getHeader("Location");
+            if (maxRecursiveDepth == 0){
+              FAIL_MSG("Maximum redirect depth reached: %s", location.c_str());
+              return false;
+            }else{
+              FAIL_MSG("Following redirect to %s", location.c_str());
+              return get(link.link(location), maxRecursiveDepth--);
+            }
+          }
+          return true; // Success!
+        }
+        // reset the 5 second timeout
+        reqTime = Util::bootSecs();
+      }
+      if (S){
+        FAIL_MSG("Timeout while retrieving %s", link.getUrl().c_str());
+        return false;
+      }
+      Util::sleep(500); // wait a bit before retrying
+    }
+    FAIL_MSG("Could not retrieve %s", link.getUrl().c_str());
+    return false;
+  }
+}
+

lib/downloader.h

@@ -0,0 +1,27 @@
+#include "http_parser.h"
+#include "socket.h"
+
+namespace HTTP{
+  class Downloader{
+  public:
+    Downloader(){progressCallback = 0;}
+    std::string &data();
+    bool get(const std::string &link);
+    bool get(const HTTP::URL &link, uint8_t maxRecursiveDepth = 6);
+    std::string getHeader(const std::string &headerName);
+    std::string &getStatusText();
+    uint32_t getStatusCode();
+    bool isOk();
+    bool (*progressCallback)(); ///< Called every time the socket stalls, up to 4X per second.
+    void setHeader(const std::string &name, const std::string &val);
+    void clearHeaders();
+
+  private:
+    std::map<std::string, std::string> extraHeaders; ///< Holds extra headers to sent with request
+    std::string connectedHost;                       ///< Currently connected host name
+    uint32_t connectedPort;                          ///< Currently connected port number
+    Parser H;                                        ///< HTTP parser for downloader
+    Socket::Connection S;                            ///< TCP socket for downloader
+  };
+}
+

lib/http_parser.cpp

@@ -154,7 +154,7 @@ std::string HTTP::URL::getBareUrl() const{
 }
 
 ///Returns a URL object for the given link, resolved relative to the current URL object.
-HTTP::URL HTTP::URL::link(const std::string &l){
+HTTP::URL HTTP::URL::link(const std::string &l) const{
   //Full link
   if (l.find("://") < l.find('/') && l.find('/' != std::string::npos)){
     DONTEVEN_MSG("Full link: %s", l.c_str());

lib/http_parser.h

@@ -83,7 +83,7 @@ namespace HTTP {
       std::string path;///<Path after the first slash (not inclusive) but before any question mark
       std::string args;///<Everything after the question mark in the path, if it was present
       std::string frag;///<Everything after the # in the path, if it was present
-      URL link(const std::string &l);
+      URL link(const std::string &l) const;
   };
 
 }//HTTP namespace

lib/mp4_generic.cpp

@@ -574,7 +574,12 @@ namespace MP4 {
   }
 
   uint32_t AVCC::getSPSLen() {
-    return getInt16(6);
+    uint16_t len = getInt16(6);
+    if (len > payloadSize() - 8){
+      WARN_MSG("SPS length of %u is more than AVCC box size %lu", len, payloadSize());
+      return 0;
+    }
+    return len;
   }
 
   char * AVCC::getSPS() {
@@ -621,7 +626,16 @@ namespace MP4 {
 
   uint32_t AVCC::getPPSLen() {
     int offset = 8 + getSPSLen() + 1;
-    return getInt16(offset);
+    if (offset > payloadSize() - 2){
+      WARN_MSG("Invalid PPS length offset! Aborting PPS read.");
+      return 0;
+    }
+    uint16_t len = getInt16(offset);
+    if (len > payloadSize() - offset - 2){
+      WARN_MSG("PPS length of %u is more than AVCC box size %lu", len, payloadSize());
+      return 0;
+    }
+    return len;
   }
 
   char * AVCC::getPPS() {

lib/procs.h

@@ -15,7 +15,6 @@ namespace Util {
   /// Deals with spawning, monitoring and stopping child processes
   class Procs {
     private:
-      static bool childRunning(pid_t p);
       static tthread::mutex plistMutex;
       static std::set<pid_t> plist; ///< Holds active process list.
       static bool thread_handler;///< True while thread handler should be running.
@@ -25,6 +24,7 @@ namespace Util {
       static char* const* dequeToArgv(std::deque<std::string> & argDeq);
       static void grim_reaper(void * n);
     public:
+      static bool childRunning(pid_t p);
       static tthread::thread * reaper_thread;
       static bool handler_set; ///< If true, the sigchld handler has been setup.
       static void setHandler();

lib/socket.cpp

@@ -548,6 +548,9 @@ Socket::Connection::Connection(std::string host, int port, bool nonblock){
       flags |= O_NONBLOCK;
       fcntl(sock, F_SETFL, flags);
     }
+    int optval = 1;
+    int optlen = sizeof(optval);
+    setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen);
   }
 }// Socket::Connection TCP Contructor
 
@@ -985,6 +988,11 @@ Socket::Connection Socket::Server::accept(bool nonblock){
     flags |= O_NONBLOCK;
     fcntl(r, F_SETFL, flags);
   }
+  if (r >= 0){
+    int optval = 1;
+    int optlen = sizeof(optval);
+    setsockopt(r, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen);
+  }
   Socket::Connection tmp(r);
   tmp.remoteaddr = tmpaddr;
   if (r < 0){

src/analysers/analyser_dtsc.cpp

@@ -1,5 +1,6 @@
 #include "analyser_dtsc.h"
 #include <mist/h264.h>
+#include <iomanip>
 
 void AnalyserDTSC::init(Util::Config &conf){
   Analyser::init(conf);
@@ -34,6 +35,15 @@ bool AnalyserDTSC::parsePacket(){
       std::cout << "DTSCv2 packet (Track " << P.getTrackId() << ", time " << P.getTime()
                 << "): " << P.getScan().toPrettyString() << std::endl;
     }
+    if (detail >= 8){
+      char * payDat;
+      unsigned int payLen;
+      P.getString("data", payDat, payLen);
+      for (uint64_t i = 0; i < payLen; ++i){
+        if ((i % 32) == 0){std::cout << std::endl;}
+        std::cout << std::hex << std::setw(2) << std::setfill('0') << (unsigned int)payDat[i];
+      }
+    }
     break;
   }
   case DTSC::DTSC_HEAD:{

src/input/input.cpp

@@ -116,7 +116,7 @@ namespace Mist {
       return 0;
     }
 
-    if (!setup()) {
+    if (!checkArguments()) {
       FAIL_MSG("Setup failed - exiting");
       return 0;
     }
@@ -141,6 +141,7 @@ namespace Mist {
       pid_t pid = fork();
       if (pid == 0){
         if (needsLock()){playerLock.close();}
+        if (!preRun()){return 0;}
         return run();
       }
       if (pid == -1){

src/input/input.h

@@ -27,17 +27,18 @@ namespace Mist {
       virtual bool needsLock(){return true;}
     protected:
       static void callbackWrapper(char * data, size_t len, unsigned int id);
-      virtual bool setup() = 0;
+      virtual bool checkArguments() = 0;
       virtual bool readHeader() = 0;
+      virtual bool preRun(){return true;}
       virtual bool readExistingHeader();
       virtual bool atKeyFrame();
-      virtual void getNext(bool smart = true) {};
+      virtual void getNext(bool smart = true) {}
       virtual void seek(int seekTime){};
       virtual void finish();
       virtual bool keepRunning();
-      virtual bool openStreamSource() { return false; };
-      virtual void closeStreamSource() {};
-      virtual void parseStreamHeader() {};
+      virtual bool openStreamSource() { return false; }
+      virtual void closeStreamSource() {}
+      virtual void parseStreamHeader() {}
       void play(int until = 0);
       void playOnce();
       void quitPlay();

src/input/input_buffer.cpp

@@ -935,7 +935,7 @@ namespace Mist {
     }
   }
 
-  bool inputBuffer::setup() {
+  bool inputBuffer::preRun() {
     lastReTime = Util::epoch(); /*LTS*/
     std::string strName = config->getString("streamname");
     Util::sanitizeName(strName);

src/input/input_buffer.h

@@ -21,7 +21,8 @@ namespace Mist {
       IPC::semaphore * liveMeta;
     protected:
       //Private Functions
-      bool setup();
+      bool preRun();
+      bool checkArguments(){return true;}
       void updateMeta();
       bool readHeader();
       void getNext(bool smart = true);

src/input/input_dtsc.cpp

@@ -58,7 +58,7 @@ namespace Mist {
   }
 
   bool inputDTSC::needsLock(){
-    return config->getString("input").substr(0, 7) != "dtsc://";
+    return config->getString("input").substr(0, 7) != "dtsc://" && config->getString("input") != "-";
   }
 
   void parseDTSCURI(const std::string & src, std::string & host, uint16_t & port, std::string & password, std::string & streamName) {
@@ -121,7 +121,7 @@ namespace Mist {
   }
 
   void inputDTSC::parseStreamHeader() {
-    while (srcConn.connected()){
+    while (srcConn.connected() && config->is_active){
       srcConn.spool();
       if (srcConn.Received().available(8)){
         if (srcConn.Received().copy(4) == "DTCM" || srcConn.Received().copy(4) == "DTSC") {
@@ -158,6 +158,10 @@ namespace Mist {
 
   bool inputDTSC::openStreamSource() {
     std::string source = config->getString("input");
+    if (source == "-"){
+      srcConn = Socket::Connection(fileno(stdout),fileno(stdin));
+      return true;
+    }
     if (source.find("dtsc://") == 0) {
       source.erase(0, 7);
     }
@@ -194,14 +198,10 @@ namespace Mist {
     srcConn.close();
   }
 
-  bool inputDTSC::setup() {
+  bool inputDTSC::checkArguments() {
     if (!needsLock()) {
       return true;
     } else {
-      if (config->getString("input") == "-") {
-        std::cerr << "Input from stdin not yet supported" << std::endl;
-        return false;
-      }
       if (!config->getString("streamname").size()) {
         if (config->getString("output") == "-") {
           std::cerr << "Output to stdout not yet supported" << std::endl;

src/input/input_dtsc.h

@@ -11,7 +11,7 @@ namespace Mist {
       bool openStreamSource();
       void closeStreamSource();
       void parseStreamHeader();
-      bool setup();
+      bool checkArguments();
       bool readHeader();
       void getNext(bool smart = true);
       void seek(int seekTime);

src/input/input_flv.cpp

@@ -27,7 +27,7 @@ namespace Mist {
     capa["codecs"][0u][1u].append("MP3");
   }
 
-  bool inputFLV::setup() {
+  bool inputFLV::checkArguments() {
     if (config->getString("input") == "-") {
       std::cerr << "Input from stdin not yet supported" << std::endl;
       return false;
@@ -43,7 +43,10 @@ namespace Mist {
         return false;
       }
     }
+    return true;
+  }
     
+  bool inputFLV::preRun() {
     //open File
     inFile = fopen(config->getString("input").c_str(), "r");
     if (!inFile) {

src/input/input_flv.h

@@ -8,7 +8,8 @@ namespace Mist {
       inputFLV(Util::Config * cfg);
     protected:
       //Private Functions
-      bool setup();
+      bool checkArguments();
+      bool preRun();
       bool readHeader();
       void getNext(bool smart = true);
       void seek(int seekTime);

src/input/input_h264.cpp

@@ -0,0 +1,127 @@
+#include "input_h264.h"
+#include <mist/h264.h>
+#include <mist/mp4_generic.h>
+
+namespace Mist{
+  InputH264::InputH264(Util::Config *cfg) : Input(cfg){
+    capa["name"] = "H264";
+    capa["desc"] = "H264 Annex B input";
+    capa["source_match"] = "h264-exec:*";
+    //May be set to always-on mode
+    capa["always_match"].append("h264-exec:*");
+    capa["priority"] = 0ll;
+    capa["codecs"][0u][0u].append("H264");
+    frameCount = 0;
+    startTime = Util::bootMS();
+    inputProcess = 0;
+  }
+
+  bool InputH264::preRun(){
+    if (config->getString("input") != "-"){
+      std::string input = config->getString("input");
+      const char *argv[2];
+      input = input.substr(10);
+
+      char *args[128];
+      uint8_t argCnt = 0;
+      char *startCh = 0;
+      for (char *i = (char*)input.c_str(); i <= input.data() + input.size(); ++i){
+        if (!*i){
+          if (startCh){args[argCnt++] = startCh;}
+          break;
+        }
+        if (*i == ' '){
+          if (startCh){
+            args[argCnt++] = startCh;
+            startCh = 0;
+            *i = 0;
+          }
+        }else{
+          if (!startCh){startCh = i;}
+        }
+      }
+      args[argCnt] = 0;
+
+      int fin = -1, fout = -1, ferr = -1;
+      inputProcess = Util::Procs::StartPiped(args, &fin, &fout, &ferr);
+      myConn = Socket::Connection(-1, fout);
+    }else{
+      myConn = Socket::Connection(fileno(stdout), fileno(stdin));
+    }
+    myConn.Received().splitter.assign("\000\000\001", 3);
+    myMeta.vod = false;
+    myMeta.live = true;
+    myMeta.tracks[1].type = "video";
+    myMeta.tracks[1].codec = "H264";
+    myMeta.tracks[1].trackID = 1;
+    waitsSinceData = 0;
+    return true;
+  }
+
+  bool InputH264::checkArguments(){
+    std::string input = config->getString("input");
+    if (input != "-" && input.substr(0, 10) != "h264-exec:"){
+      FAIL_MSG("Unsupported input type: %s", input.c_str());
+      return false;
+    }
+    return true;
+  }
+
+  void InputH264::getNext(bool smart){
+    do{
+      if (!myConn.spool()){
+        Util::sleep(25);
+        ++waitsSinceData;
+        if (waitsSinceData > 5000 / 25 && (waitsSinceData % 40) == 0){
+          WARN_MSG("No H264 data received for > 5s, killing source process");
+          Util::Procs::Stop(inputProcess);
+        }
+        continue;
+      }
+      waitsSinceData = 0;
+      uint32_t bytesToRead = myConn.Received().bytesToSplit();
+      if (!bytesToRead){continue;}
+      std::string NAL = myConn.Received().remove(bytesToRead);
+      uint32_t nalSize = NAL.size() - 3;
+      while (nalSize && NAL.data()[nalSize - 1] == 0){--nalSize;}
+      if (!nalSize){continue;}
+      uint8_t nalType = NAL.data()[0] & 0x1F;
+      INSANE_MSG("NAL unit, type %u, size %lu", nalType, nalSize);
+      if (nalType == 7 || nalType == 8){
+        if (nalType == 7){spsInfo = NAL.substr(0, nalSize);}
+        if (nalType == 8){ppsInfo = NAL.substr(0, nalSize);}
+        if (!myMeta.tracks[1].init.size() && spsInfo.size() && ppsInfo.size()){
+          h264::sequenceParameterSet sps(spsInfo.data(), spsInfo.size());
+          h264::SPSMeta spsChar = sps.getCharacteristics();
+          myMeta.tracks[1].width = spsChar.width;
+          myMeta.tracks[1].height = spsChar.height;
+          myMeta.tracks[1].fpks = spsChar.fps * 1000;
+          if (myMeta.tracks[1].fpks < 100 || myMeta.tracks[1].fpks > 1000000){
+            myMeta.tracks[1].fpks = 0;
+          }
+          MP4::AVCC avccBox;
+          avccBox.setVersion(1);
+          avccBox.setProfile(spsInfo[1]);
+          avccBox.setCompatibleProfiles(spsInfo[2]);
+          avccBox.setLevel(spsInfo[3]);
+          avccBox.setSPSNumber(1);
+          avccBox.setSPS(spsInfo);
+          avccBox.setPPSNumber(1);
+          avccBox.setPPS(ppsInfo);
+          myMeta.tracks[1].init = std::string(avccBox.payload(), avccBox.payloadSize());
+        }
+        continue;
+      }
+      if (myMeta.tracks[1].init.size()){
+        uint64_t ts = Util::bootMS() - startTime;
+        if (myMeta.tracks[1].fpks){ts = frameCount * (1000000 / myMeta.tracks[1].fpks);}
+        thisPacket.genericFill(ts, 0, 1, 0, 0, 0, h264::isKeyframe(NAL.data(), nalSize));
+        thisPacket.appendNal(NAL.data(), nalSize, nalSize);
+        ++frameCount;
+        return;
+      }
+    }while (myConn && (inputProcess == 0 || Util::Procs::childRunning(inputProcess)));
+    if (inputProcess){myConn.close();}
+  }
+}
+

src/input/input_h264.h

@@ -0,0 +1,33 @@
+#include "input.h"
+#include <mist/dtsc.h>
+#include <mist/procs.h>
+
+namespace Mist{
+  class InputH264 : public Input{
+  public:
+    InputH264(Util::Config *cfg);
+
+  protected:
+    bool checkArguments();
+    bool preRun();
+    void getNext(bool smart = true);
+    Socket::Connection myConn;
+    std::string ppsInfo;
+    std::string spsInfo;
+    uint64_t frameCount;
+    // Empty defaults
+    bool readHeader(){return true;}
+    bool openStreamSource(){return true;}
+    void closeStreamSource(){}
+    void parseStreamHeader(){}
+    void seek(int seekTime){}
+    void trackSelect(std::string trackSpec){}
+    bool needsLock(){return false;}
+    uint64_t startTime;
+    pid_t inputProcess;
+    uint32_t waitsSinceData;
+  };
+}
+
+typedef Mist::InputH264 mistIn;
+

src/input/input_mp3.cpp

@@ -21,7 +21,7 @@ namespace Mist {
     timestamp = 0;
   }
 
-  bool inputMP3::setup() {
+  bool inputMP3::checkArguments() {
     if (config->getString("input") == "-") {
       std::cerr << "Input from stdin not yet supported" << std::endl;
       return false;
@@ -37,7 +37,10 @@ namespace Mist {
         return false;
       }
     }
+    return true;
+  }
     
+  bool inputMP3::preRun() {
     //open File
     inFile = fopen(config->getString("input").c_str(), "r");
     if (!inFile) {

src/input/input_mp3.h

@@ -16,7 +16,8 @@ namespace Mist {
       inputMP3(Util::Config * cfg);
     protected:
       //Private Functions
-      bool setup();
+      bool checkArguments();
+      bool preRun();
       bool readHeader();
       void getNext(bool smart = true);
       void seek(int seekTime);

src/input/input_ogg.cpp

@@ -53,12 +53,15 @@ namespace Mist {
     capa["codecs"][0u][1u].append("opus");
   }
 
-  bool inputOGG::setup(){
+  bool inputOGG::checkArguments(){
     if (config->getString("input") == "-"){
       std::cerr << "Input from stream not yet supported" << std::endl;
       return false;
     }
+    return true;
+  }
 
+  bool inputOGG::preRun(){
     //open File
     inFile = fopen(config->getString("input").c_str(), "r");
     if (!inFile){

src/input/input_ogg.h

@@ -70,7 +70,8 @@ namespace Mist {
       inputOGG(Util::Config * cfg);
     protected:
       //Private Functions
-      bool setup();
+      bool checkArguments();
+      bool preRun();
       bool readHeader();
       position seekFirstData(long long unsigned int tid);      
       void getNext(bool smart = true);