Converted MP4 input to use URIReader

src/input/input_mp4.cpp

@@ -159,11 +159,14 @@ namespace Mist{
   }
 
   inputMP4::inputMP4(Util::Config *cfg) : Input(cfg){
-    malSize = 4; // initialise data read buffer to 0;
-    data = (char *)malloc(malSize);
     capa["name"] = "MP4";
     capa["desc"] = "This input allows streaming of MP4 files as Video on Demand.";
-    capa["source_match"] = "/*.mp4";
+    capa["source_match"].append("/*.mp4");
+    capa["source_match"].append("http://*.mp4");
+    capa["source_match"].append("https://*.mp4");
+    capa["source_match"].append("s3+http://*.mp4");
+    capa["source_match"].append("s3+https://*.mp4");
+    capa["source_match"].append("mp4:*");
     capa["source_file"] = "$source";
     capa["priority"] = 9;
     capa["codecs"][0u][0u].append("HEVC");
@@ -173,10 +176,9 @@ namespace Mist{
     capa["codecs"][0u][1u].append("AAC");
     capa["codecs"][0u][1u].append("AC3");
     capa["codecs"][0u][1u].append("MP3");
+    readPos = 0;
   }
 
-  inputMP4::~inputMP4(){free(data);}
-
   bool inputMP4::checkArguments(){
     if (config->getString("input") == "-"){
       std::cerr << "Input from stdin not yet supported" << std::endl;
@@ -199,57 +201,93 @@ namespace Mist{
 
   bool inputMP4::preRun(){
     // open File
-    inFile = fopen(config->getString("input").c_str(), "r");
+    std::string inUrl = config->getString("input");
+    if (inUrl.size() > 4 && inUrl.substr(0, 4) == "mp4:"){inUrl.erase(0, 4);}
+    inFile.open(inUrl);
     if (!inFile){return false;}
+    if (!inFile.isSeekable()){
+      FAIL_MSG("MP4 input only supports seekable data sources, for now, and this source is not seekable: %s", config->getString("input").c_str());
+      return false;
+    }
     return true;
   }
 
+  void inputMP4::dataCallback(const char *ptr, size_t size){readBuffer.append(ptr, size);}
+
   bool inputMP4::readHeader(){
     if (!inFile){
       INFO_MSG("inFile failed!");
       return false;
     }
+    bool hasMoov = false;
+    readBuffer.truncate(0);
+    readPos = 0;
 
     // first we get the necessary header parts
     size_t tNumber = 0;
-    while (!feof(inFile)){
+    activityCounter = Util::bootSecs();
-      std::string boxType = MP4::readBoxType(inFile);
+    while (inFile && keepRunning()){
-      if (boxType == "erro"){break;}
+      //Read box header if needed
+      while (readBuffer.size() < 16 && inFile && keepRunning()){inFile.readSome(16, *this);}
+      //Failed? Abort.
+      if (readBuffer.size() < 16){
+        FAIL_MSG("Could not read box header from input!");
+        break;
+      }
+      //Box type is always on bytes 5-8 from the start of a box
+      std::string boxType = std::string(readBuffer+4, 4);
+      uint64_t boxSize = MP4::calcBoxSize(readBuffer);
       if (boxType == "moov"){
-        MP4::MOOV moovBox;
+        while (readBuffer.size() < boxSize && inFile && keepRunning()){inFile.readSome(boxSize-readBuffer.size(), *this);}
-        moovBox.read(inFile);
+        if (readBuffer.size() < boxSize){
-        // for all box in moov
+          FAIL_MSG("Could not read entire MOOV box into memory");
+          break;
+        }
+        MP4::Box moovBox(readBuffer, false);
 
-        std::deque<MP4::TRAK> trak = moovBox.getChildren<MP4::TRAK>();
+        // for all box in moov
+        std::deque<MP4::TRAK> trak = ((MP4::MOOV*)&moovBox)->getChildren<MP4::TRAK>();
         for (std::deque<MP4::TRAK>::iterator trakIt = trak.begin(); trakIt != trak.end(); trakIt++){
           trackHeaders.push_back(mp4TrackHeader());
           trackHeaders.rbegin()->read(*trakIt);
         }
-        continue;
+        hasMoov = true;
+        break;
       }
-      if (!MP4::skipBox(inFile)){// moving on to next box
+      activityCounter = Util::bootSecs();
-        FAIL_MSG("Error in skipping box, exiting");
+      //Skip to next box
-        return false;
+      if (readBuffer.size() > boxSize){
+        readBuffer.shift(boxSize);
+        readPos += boxSize;
+      }else{
+        readBuffer.truncate(0);
+        if (!inFile.seek(readPos + boxSize)){
+          FAIL_MSG("Seek to %" PRIu64 " failed! Aborting load", readPos+boxSize);
+        }
+        readPos = inFile.getPos();
       }
     }
-    fseeko(inFile, 0, SEEK_SET);
 
     // See whether a separate header file exists.
-    if (readExistingHeader()){return true;}
+    if (readExistingHeader()){
-    HIGH_MSG("Not read existing header");
+      bps = 0;
+      std::set<size_t> tracks = M.getValidTracks();
+      for (std::set<size_t>::iterator it = tracks.begin(); it != tracks.end(); it++){bps += M.getBps(*it);}
+      return true;
+    }
+    INFO_MSG("Not read existing header");
 
     meta.reInit(isSingular() ? streamName : "");
+    if (!hasMoov){
+      FAIL_MSG("No MOOV box found; aborting header creation!");
+      return false;
+    }
 
     tNumber = 0;
     // Create header file from MP4 data
-    while (!feof(inFile)){
+    MP4::Box moovBox(readBuffer, false);
-      std::string boxType = MP4::readBoxType(inFile);
-      if (boxType == "erro"){break;}
-      if (boxType == "moov"){
-        MP4::MOOV moovBox;
-        moovBox.read(inFile);
 
-        std::deque<MP4::TRAK> trak = moovBox.getChildren<MP4::TRAK>();
+    std::deque<MP4::TRAK> trak = ((MP4::MOOV*)&moovBox)->getChildren<MP4::TRAK>();
     HIGH_MSG("Obtained %zu trak Boxes", trak.size());
 
     for (std::deque<MP4::TRAK>::iterator trakIt = trak.begin(); trakIt != trak.end(); trakIt++){
@@ -437,17 +475,18 @@ namespace Mist{
         }
       }
     }
-        continue;
-      }
-      if (!MP4::skipBox(inFile)){// moving on to next box
-        FAIL_MSG("Error in Skipping box, exiting");
-        return false;
-      }
-    }
-    clearerr(inFile);
 
     // outputting dtsh file
-    M.toFile(config->getString("input") + ".dtsh");
+    std::string inUrl = config->getString("input");
+    if (inUrl.size() > 4 && inUrl.substr(0, 4) == "mp4:"){inUrl.erase(0, 4);}
+    if (inUrl != "-" && HTTP::URL(inUrl).isLocalPath()){
+      M.toFile(inUrl + ".dtsh");
+    }else{
+      INFO_MSG("Skipping header write, as the source is not a local file");
+    }
+    bps = 0;
+    std::set<size_t> tracks = M.getValidTracks();
+    for (std::set<size_t>::iterator it = tracks.begin(); it != tracks.end(); it++){bps += M.getBps(*it);}
     return true;
   }
 
@@ -474,23 +513,48 @@ namespace Mist{
         ++nextKeyNum;
       }
     }
-    if (fseeko(inFile, curPart.bpos, SEEK_SET)){
+    if (curPart.bpos < readPos || curPart.bpos > readPos + readBuffer.size() + 512*1024 + bps){
+      INFO_MSG("Buffer contains %" PRIu64 "-%" PRIu64 ", but we need %" PRIu64 "; seeking!", readPos, readPos + readBuffer.size(), curPart.bpos);
+      readBuffer.truncate(0);
+      if (!inFile.seek(curPart.bpos)){
         FAIL_MSG("seek unsuccessful @bpos %" PRIu64 ": %s", curPart.bpos, strerror(errno));
         thisPacket.null();
         return;
       }
-    if (curPart.size > malSize){
+      readPos = inFile.getPos();
-      data = (char *)realloc(data, curPart.size);
+    }else{
-      malSize = curPart.size;
+      //If we have more than 5MiB buffered and are more than 5MiB into the buffer, shift the first 4MiB off the buffer.
+      //This prevents infinite growth of the read buffer for large files
+      if (readBuffer.size() >= 5*1024*1024 && curPart.bpos > readPos + 5*1024*1024 + bps){
+        readBuffer.shift(4*1024*1024);
+        readPos += 4*1024*1024;
       }
-    if (fread(data, curPart.size, 1, inFile) != 1){
+    }
-      FAIL_MSG("read unsuccessful at %ld", ftell(inFile));
+
+    while (readPos+readBuffer.size() < curPart.bpos+curPart.size && inFile && keepRunning()){
+      inFile.readSome((curPart.bpos+curPart.size) - (readPos+readBuffer.size()), *this);
+    }
+    if (readPos+readBuffer.size() < curPart.bpos+curPart.size){
+      FAIL_MSG("Read unsuccessful at %" PRIu64 ", seeking to retry...", readPos+readBuffer.size());
+      readBuffer.truncate(0);
+      if (!inFile.seek(curPart.bpos)){
+        FAIL_MSG("seek unsuccessful @bpos %" PRIu64 ": %s", curPart.bpos, strerror(errno));
         thisPacket.null();
         return;
       }
+      readPos = inFile.getPos();
+      while (readPos+readBuffer.size() < curPart.bpos+curPart.size && inFile && keepRunning()){
+        inFile.readSome((curPart.bpos+curPart.size) - (readPos+readBuffer.size()), *this);
+      }
+      if (readPos+readBuffer.size() < curPart.bpos+curPart.size){
+        FAIL_MSG("Read retry unsuccessful at %" PRIu64 ", aborting", readPos+readBuffer.size());
+        thisPacket.null();
+        return;
+      }
+    }
 
     if (M.getCodec(curPart.trackID) == "subtitle"){
-      unsigned int txtLen = Bit::btohs(data);
+      unsigned int txtLen = Bit::btohs(readBuffer + (curPart.bpos-readPos));
       if (!txtLen && false){
         curPart.index++;
         return getNext(idx);
@@ -499,14 +563,14 @@ namespace Mist{
       thisPack.null();
       thisPack["trackid"] = curPart.trackID;
       thisPack["bpos"] = curPart.bpos; //(long long)fileSource.tellg();
-      thisPack["data"] = std::string(data + 2, txtLen);
+      thisPack["data"] = std::string(readBuffer + (curPart.bpos-readPos) + 2, txtLen);
       thisPack["time"] = curPart.time;
       if (curPart.duration){thisPack["duration"] = curPart.duration;}
       thisPack["keyframe"] = true;
       std::string tmpStr = thisPack.toNetPacked();
       thisPacket.reInit(tmpStr.data(), tmpStr.size());
     }else{
-      thisPacket.genericFill(curPart.time, curPart.offset, curPart.trackID, data, curPart.size, 0, isKeyframe);
+      thisPacket.genericFill(curPart.time, curPart.offset, curPart.trackID, readBuffer + (curPart.bpos-readPos), curPart.size, 0, isKeyframe);
     }
     thisTime = curPart.time;
     thisIdx = curPart.trackID;

src/input/input_mp4.h

@@ -1,5 +1,6 @@
 #include "input.h"
 #include <mist/dtsc.h>
+#include <mist/urireader.h>
 #include <mist/mp4.h>
 #include <mist/mp4_generic.h>
 namespace Mist{
@@ -70,10 +71,10 @@ namespace Mist{
     bool stco64;
   };
 
-  class inputMP4 : public Input{
+  class inputMP4 : public Input, public Util::DataCallback {
   public:
     inputMP4(Util::Config *cfg);
-    ~inputMP4();
+    void dataCallback(const char *ptr, size_t size);
 
   protected:
     // Private Functions
@@ -85,7 +86,10 @@ namespace Mist{
     void seek(uint64_t seekTime, size_t idx = INVALID_TRACK_ID);
     void handleSeek(uint64_t seekTime, size_t idx);
 
-    FILE *inFile;
+    HTTP::URIReader inFile;
+    Util::ResizeablePointer readBuffer;
+    uint64_t readPos;
+    uint64_t bps;
 
     mp4TrackHeader &headerData(size_t trackID);
 
@@ -94,11 +98,6 @@ namespace Mist{
 
     // remember last seeked keyframe;
     std::map<size_t, uint32_t> nextKeyframe;
-
-    // these next two variables keep a buffer for reading from filepointer inFile;
-    uint64_t malSize;
-    char *data; ///\todo rename this variable to a more sensible name, it is a temporary piece of
-                /// memory to read from files
   };
 }// namespace Mist