MistInBuffer no longer boots if the process starting it has no data to fill it with. Optimized and simplified HTTP internal output.

lib/stream.cpp

@@ -115,11 +115,11 @@ bool Util::streamAlive(std::string & streamname){
 }
 
 /// Assures the input for the given stream name is active.
-/// Does stream name sanitizion first, followed by a stream name length check (<= 100 chars).
+/// Does stream name sanitation first, followed by a stream name length check (<= 100 chars).
-/// Then, checks if an input is already active by running streamAlive(). If yes, aborts.
+/// Then, checks if an input is already active by running streamAlive(). If yes, return true.
-/// If no, loads up the server configuration and attempts to start the given stream according to current config.
+/// If no, loads up the server configuration and attempts to start the given stream according to current configuration.
 /// At this point, fails and aborts if MistController isn't running.
-bool Util::startInput(std::string streamname, std::string filename, bool forkFirst) {
+bool Util::startInput(std::string streamname, std::string filename, bool forkFirst, bool isProvider) {
   sanitizeName(streamname);
   if (streamname.size() > 100){
     FAIL_MSG("Stream opening denied: %s is longer than 100 characters (%lu).", streamname.c_str(), streamname.size());
@@ -172,6 +172,7 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
   DTSC::Scan inputs = config.getMember("capabilities").getMember("inputs");
   DTSC::Scan input;
   unsigned int input_size = inputs.getSize();
+  bool noProviderNoPick = false;
   for (unsigned int i = 0; i < input_size; ++i){
     DTSC::Scan tmp_input = inputs.getIndice(i);
     
@@ -185,6 +186,10 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
           MEDIUM_MSG("Checking input %s: %s (%s)", inputs.getIndiceName(i).c_str(), tmp_input.getMember("name").asString().c_str(), source.c_str());
           
           if (filename.substr(0,front.size()) == front && filename.substr(filename.size()-back.size()) == back){
+            if (tmp_input.getMember("non-provider") && !isProvider){
+              noProviderNoPick = true;
+              continue;
+            }
             player_bin = Util::getMyPath() + "MistIn" + tmp_input.getMember("name").asString();
             curPrio = tmp_input.getMember("priority").asInt();
             selected = true;
@@ -198,6 +203,10 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
         MEDIUM_MSG("Checking input %s: %s (%s)", inputs.getIndiceName(i).c_str(), tmp_input.getMember("name").asString().c_str(), source.c_str());
         
         if (filename.substr(0,front.size()) == front && filename.substr(filename.size()-back.size()) == back){
+          if (tmp_input.getMember("non-provider") && !isProvider){
+            noProviderNoPick = true;
+            continue;
+          }
           player_bin = Util::getMyPath() + "MistIn" + tmp_input.getMember("name").asString();
           curPrio = tmp_input.getMember("priority").asInt();
           selected = true;
@@ -210,7 +219,11 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
   
   if (!selected){
     configLock.post();//unlock the config semaphore
+    if (noProviderNoPick){
+      INFO_MSG("Not a media provider for stream %s: %s", streamname.c_str(), filename.c_str());
+    }else{
       FAIL_MSG("No compatible input found for stream %s: %s", streamname.c_str(), filename.c_str());
+    }
     return false;
   }
 

lib/stream.h

@@ -10,7 +10,7 @@ namespace Util {
   std::string getTmpFolder();
   void sanitizeName(std::string & streamname);
   bool streamAlive(std::string & streamname);
-  bool startInput(std::string streamname, std::string filename = "", bool forkFirst = true);
+  bool startInput(std::string streamname, std::string filename = "", bool forkFirst = true, bool isProvider = false);
   JSON::Value getStreamConfig(std::string streamname);
 }
 

src/input/input.cpp

@@ -262,6 +262,7 @@ namespace Mist {
     }
     finish();
     DEBUG_MSG(DLVL_DEVEL, "Input for stream %s closing clean", streamName.c_str());
+    userPage.finishEach();
     //end player functionality
   }
 
@@ -301,7 +302,7 @@ namespace Mist {
       WARN_MSG("Stream already online, cancelling");
       return;
     }
-    if (!Util::startInput(streamName, "push://INTERNAL_ONLY:"+config->getString("input"))) {//manually override stream url to start the buffer
+    if (!Util::startInput(streamName, "push://INTERNAL_ONLY:"+config->getString("input"), true, true)) {//manually override stream url to start the buffer
       pullLock.post();
       pullLock.close();
       pullLock.unlink();

src/input/input_buffer.cpp

@@ -51,6 +51,7 @@ namespace Mist {
     option.null();
 
     capa["source_match"] = "push://*";
+    capa["non-provider"] = true;//Indicates we don't provide data, only collect it
     capa["priority"] = 9ll;
     capa["desc"] = "Provides buffered live input";
     capa["codecs"][0u][0u].append("*");

src/output/output.cpp

@@ -186,7 +186,7 @@ namespace Mist{
         return;
       }
     }else{
-      if (!Util::startInput(streamName)){
+      if (!Util::startInput(streamName, "", true, isPushing())){
         FAIL_MSG("Opening stream %s failed - aborting initialization", streamName.c_str());
         onFail();
         return;
@@ -224,8 +224,8 @@ namespace Mist{
     updateMeta();
     selectDefaultTracks();
     if (!myMeta.vod && !isReadyForPlay()){
-      unsigned long long waitUntil = Util::epoch() + 15;
+      unsigned long long waitUntil = Util::epoch() + 30;
-      while (!myMeta.vod && !isReadyForPlay()){
+      while (!myMeta.vod && !isReadyForPlay() && nProxy.userClient.isAlive()){
         if (Util::epoch() > waitUntil + 45 || (!selectedTracks.size() && Util::epoch() > waitUntil)){
           INFO_MSG("Giving up waiting for playable tracks. Stream: %s, IP: %s", streamName.c_str(), getConnectedHost().c_str());
           break;

src/output/output_http.cpp

@@ -224,6 +224,7 @@ namespace Mist {
       }
 
       INFO_MSG("Received request %s", H.getUrl().c_str());
+      if (H.GetVar("audio") != "" || H.GetVar("video") != ""){
         selectedTracks.clear();
         if (H.GetVar("audio") != ""){
           selectedTracks.insert(JSON::Value(H.GetVar("audio")).asInt());
@@ -251,6 +252,7 @@ namespace Mist {
         for (std::set<unsigned long>::iterator it = toRemove.begin(); it != toRemove.end(); it++){
           selectedTracks.erase(*it);
         }
+      }
 
       onHTTP();
       if (!H.bufferChunks){

src/output/output_http_internal.cpp

@@ -53,11 +53,6 @@ namespace Mist {
     Output::onFail();
   }
 
-  /// The HTTP output is always ready to play
-  bool OutHTTP::isReadyForPlay() {
-    return true;
-  }
-  
   void OutHTTP::init(Util::Config * cfg){
     HTTPOutput::init(cfg);
     capa.removeMember("deps");
@@ -65,6 +60,7 @@ namespace Mist {
     capa["desc"] = "Generic HTTP handler, required for all other HTTP-based outputs.";
     capa["provides"] = "HTTP";
     capa["protocol"] = "http://";
+    capa["codecs"][0u][0u].append("*");
     capa["url_match"].append("/crossdomain.xml");
     capa["url_match"].append("/clientaccesspolicy.xml");
     capa["url_match"].append("/$.html");
@@ -321,12 +317,11 @@ namespace Mist {
       }
       
       std::string trackSources;//this string contains all track sources for MBR smil
-      DTSC::Scan tracks = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("streams").getMember(streamName).getMember("meta").getMember("tracks");
+      initialize();
-      unsigned int track_ctr = tracks.getSize();
+      if (!myConn){return;}
-      for (unsigned int i = 0; i < track_ctr; ++i){//for all video tracks
+      for (std::map<unsigned int, DTSC::Track>::iterator trit = myMeta.tracks.begin(); trit != myMeta.tracks.end(); trit++){
-        DTSC::Scan trk = tracks.getIndice(i);
+        if (trit->second.type == "video"){
-        if (trk.getMember("type").asString() == "video"){
+          trackSources += "      <video src='"+ streamName + "?track=" + JSON::Value((long long)trit->first).asString() + "' height='" + JSON::Value((long long)trit->second.height).asString() + "' system-bitrate='" + JSON::Value((long long)trit->second.bps).asString() + "' width='" + JSON::Value((long long)trit->second.width).asString() + "' />\n";
-          trackSources += "      <video src='"+ streamName + "?track=" + trk.getMember("trackid").asString() + "' height='" + trk.getMember("height").asString() + "' system-bitrate='" + trk.getMember("bps").asString() + "' width='" + trk.getMember("width").asString() + "' />\n";
         }
       }
       configLock.post();
@@ -370,59 +365,41 @@ namespace Mist {
       }
       response = "// Generating info code for stream " + streamName + "\n\nif (!mistvideo){var mistvideo = {};}\n";
       JSON::Value json_resp;
+      initialize();
+      if (!myConn){
+        return;
+      }
       IPC::semaphore configLock(SEM_CONF, O_CREAT | O_RDWR, ACCESSPERMS, 1);
-      static char liveSemName[NAME_BUFFER_SIZE];
-      snprintf(liveSemName, NAME_BUFFER_SIZE, SEM_LIVE, streamName.c_str());
-      IPC::semaphore metaLocker(liveSemName, O_CREAT | O_RDWR, ACCESSPERMS, 1);
-      bool metaLock = false;
       configLock.wait();
       IPC::sharedPage serverCfg(SHM_CONF, DEFAULT_CONF_PAGE_SIZE);
-      DTSC::Scan strm = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("streams").getMember(streamName).getMember("meta");
-      IPC::sharedPage streamIndex;
-      if (!strm){
-        configLock.post();
-        //Stream metadata not found - attempt to start it
-        if (Util::startInput(streamName)){
-          char pageId[NAME_BUFFER_SIZE];
-          snprintf(pageId, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, streamName.c_str());
-          streamIndex.init(pageId, DEFAULT_STRM_PAGE_SIZE);
-          if (streamIndex.mapped){
-            metaLock = true;
-            metaLocker.wait();
-            strm = DTSC::Packet(streamIndex.mapped, streamIndex.len, true).getScan();
-          }
-        }
-        if (!strm){
-          //stream failed to start or isn't configured
-          response += "// Stream isn't configured and/or couldn't be started. Sorry.\n";
-        }
-        configLock.wait();
-      }
       DTSC::Scan prots = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("config").getMember("protocols");
-      if (strm && prots){
+      if (prots){
-        DTSC::Scan trcks = strm.getMember("tracks");
+        bool hasVideo = false;
-        unsigned int trcks_ctr = trcks.getSize();
+        for (std::map<unsigned int, DTSC::Track>::iterator trit = myMeta.tracks.begin(); trit != myMeta.tracks.end(); trit++){
-        for (unsigned int i = 0; i < trcks_ctr; ++i){
+          if (trit->second.type == "video"){
-          if (trcks.getIndice(i).getMember("width").asInt() > json_resp["width"].asInt()){
+            hasVideo = true;
-            json_resp["width"] = trcks.getIndice(i).getMember("width").asInt();
+            if (trit->second.width > json_resp["width"].asInt()){
+              json_resp["width"] = trit->second.width;
+            }
+            if (trit->second.height > json_resp["height"].asInt()){
+              json_resp["height"] = trit->second.height;
             }
-          if (trcks.getIndice(i).getMember("height").asInt() > json_resp["height"].asInt()){
-            json_resp["height"] = trcks.getIndice(i).getMember("height").asInt();
           }
         }
         if (json_resp["width"].asInt() < 1 || json_resp["height"].asInt() < 1){
           json_resp["width"] = 640ll;
           json_resp["height"] = 480ll;
+          if (!hasVideo){json_resp["height"] = 20ll;}
         }
-        if (strm.getMember("vod")){
+        if (myMeta.vod){
           json_resp["type"] = "vod";
         }
-        if (strm.getMember("live")){
+        if (myMeta.live){
           json_resp["type"] = "live";
         }
         
         // show ALL the meta datas!
-        json_resp["meta"] = strm.asJSON();
+        json_resp["meta"] = myMeta.toJSON();
         jsonForEach(json_resp["meta"]["tracks"], it) {
           if (it->isMember("lang")){
             (*it)["language"] = Encodings::ISO639::decode((*it)["lang"].asStringRef());
@@ -485,10 +462,6 @@ namespace Mist {
       }else{
         json_resp["error"] = "The specified stream is not available on this server.";
       }
-      if (metaLock){
-        metaLocker.post();
-      }
-      
       configLock.post();
       configLock.close();
       if (rURL.substr(0, 6) != "/json_"){

src/output/output_http_internal.h

@@ -10,7 +10,6 @@ namespace Mist {
       static bool listenMode();
       virtual void onFail();
       void onHTTP();
-      bool isReadyForPlay();
       void sendIcon();
   };
 }