hydrar/mistserver
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Added lots of DLVL_INSANE level debug messages to controller metadata checks. Simplified internal stream handling.

Browse source
This commit is contained in:
Thulinma 2014-07-29 13:36:43 +02:00
parent e0d6a18f70
commit 848f075ff4
3 changed files with 105 additions and 90 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
src/controller/controller.cpp
View file

@ -588,7 +588,6 @@ int main(int argc, char ** argv){
} }
if (Request.isMember("streams")){ if (Request.isMember("streams")){
Controller::CheckStreams(Request["streams"], Controller::Storage["streams"]); Controller::CheckStreams(Request["streams"], Controller::Storage["streams"]);
Controller::CheckAllStreams(Controller::Storage["streams"]);
} }
if (Request.isMember("clearstatlogs")){ if (Request.isMember("clearstatlogs")){
Controller::Storage["log"].null(); Controller::Storage["log"].null();
@ -615,7 +614,6 @@ int main(int argc, char ** argv){
} }
if (Request.isMember("streams")){ if (Request.isMember("streams")){
Controller::CheckStreams(Request["streams"], Controller::Storage["streams"]); Controller::CheckStreams(Request["streams"], Controller::Storage["streams"]);
Controller::CheckAllStreams(Controller::Storage["streams"]);
} }
if (Request.isMember("capabilities")){ if (Request.isMember("capabilities")){
Controller::checkCapable(capabilities); Controller::checkCapable(capabilities);
@ -661,7 +659,6 @@ int main(int argc, char ** argv){
} }
} }
//sent current configuration, no matter if it was changed or not //sent current configuration, no matter if it was changed or not
//Response["streams"] = Storage["streams"];
Response["config"] = Controller::Storage["config"]; Response["config"] = Controller::Storage["config"];
Response["config"]["version"] = PACKAGE_VERSION "/" + Util::Config::libver + "/" RELEASE; Response["config"]["version"] = PACKAGE_VERSION "/" + Util::Config::libver + "/" RELEASE;
Response["streams"] = Controller::Storage["streams"]; Response["streams"] = Controller::Storage["streams"];

58
src/controller/controller_streams.cpp
View file

@ -31,10 +31,10 @@ namespace Controller {
return true; return true;
} }
///\brief Starts a single stream ///\brief Checks the validity of a stream, updates internal stream status.
///\param name The name of the stream ///\param name The name of the stream
///\param data The corresponding configuration values. ///\param data The corresponding configuration values.
void startStream(std::string name, JSON::Value & data){ void checkStream(std::string name, JSON::Value & data){
std::string prevState = data["error"].asStringRef(); std::string prevState = data["error"].asStringRef();
data["online"] = (std::string)"Checking..."; data["online"] = (std::string)"Checking...";
data.removeMember("error"); data.removeMember("error");
@ -53,6 +53,7 @@ namespace Controller {
return; return;
} }
if (URL.substr(0, 4) == "push"){ if (URL.substr(0, 4) == "push"){
//push-style stream
if (hasViewers(name)){ if (hasViewers(name)){
data["meta"].null(); data["meta"].null();
IPC::sharedPage streamIndex(name,0,false,false); IPC::sharedPage streamIndex(name,0,false,false);
@ -69,8 +70,10 @@ namespace Controller {
} }
} }
} }
}else{ return;
}
if (URL.substr(0, 1) == "/"){ if (URL.substr(0, 1) == "/"){
//vod-style stream
data.removeMember("error"); data.removeMember("error");
struct stat fileinfo; struct stat fileinfo;
if (stat(URL.c_str(), &fileinfo) != 0 || S_ISDIR(fileinfo.st_mode)){ if (stat(URL.c_str(), &fileinfo) != 0 || S_ISDIR(fileinfo.st_mode)){
@ -83,17 +86,19 @@ namespace Controller {
} }
bool getMeta = false; bool getMeta = false;
if ( !data.isMember("l_meta") || fileinfo.st_mtime != data["l_meta"].asInt()){ if ( !data.isMember("l_meta") || fileinfo.st_mtime != data["l_meta"].asInt()){
DEBUG_MSG(DLVL_INSANE, "File for stream %s is newer than metadata - triggering reload", name.c_str());
getMeta = true; getMeta = true;
data["l_meta"] = (long long)fileinfo.st_mtime; data["l_meta"] = (long long)fileinfo.st_mtime;
} }
if (stat((URL+".dtsh").c_str(), &fileinfo) == 0 && !S_ISDIR(fileinfo.st_mode)){ if (stat((URL+".dtsh").c_str(), &fileinfo) == 0 && !S_ISDIR(fileinfo.st_mode)){
if ( !data.isMember("h_meta") || fileinfo.st_mtime != data["h_meta"].asInt()){ if ( !data.isMember("h_meta") || fileinfo.st_mtime != data["h_meta"].asInt()){
DEBUG_MSG(DLVL_INSANE, "DTSH for stream %s is newer than metadata - triggering reload", name.c_str());
getMeta = true; getMeta = true;
data["h_meta"] = (long long)fileinfo.st_mtime; data["h_meta"] = (long long)fileinfo.st_mtime;
} }
} }
if ( !getMeta && data.isMember("meta") && data["meta"].isMember("tracks")){ if ( !getMeta && data.isMember("meta") && data["meta"].isMember("tracks")){
if ( !data["meta"] || !data["meta"]["tracks"]){ if (!data["meta"]["tracks"]){
data["error"] = "Stream offline: Corrupt file?"; data["error"] = "Stream offline: Corrupt file?";
if (data["error"].asStringRef() != prevState){ if (data["error"].asStringRef() != prevState){
Log("WARN", "Source file " + URL + " seems to be corrupt."); Log("WARN", "Source file " + URL + " seems to be corrupt.");
@ -102,27 +107,44 @@ namespace Controller {
return; return;
} }
}else{ }else{
DEBUG_MSG(DLVL_INSANE, "Invalid metadata (no tracks object) for stream %s - triggering reload", name.c_str());
getMeta = true; getMeta = true;
} }
if (getMeta){ if (getMeta){
// if the file isn't dtsc and there's no dtsh file, run getStream on it // if the file isn't dtsc and there's no dtsh file, run getStream on it
// this guarantees that if the stream is playable, it now has a valid header. // this guarantees that if the stream is playable, it now has a valid header.
DEBUG_MSG(DLVL_INSANE, "(re)loading metadata for stream %s", name.c_str());
if ((URL.substr(URL.size() - 5) != ".dtsc") && (stat((URL+".dtsh").c_str(), &fileinfo) != 0)){ if ((URL.substr(URL.size() - 5) != ".dtsc") && (stat((URL+".dtsh").c_str(), &fileinfo) != 0)){
DEBUG_MSG(DLVL_INSANE, "Stream %s is non-DTSC file without DTSH. Opening stream to generate DTSH...", name.c_str());
Util::Stream::getVod(URL, name); Util::Stream::getVod(URL, name);
DEBUG_MSG(DLVL_INSANE, "Waiting for stream %s to open...", name.c_str());
//wait for the stream //wait for the stream
{ {
IPC::sharedPage streamIndex; IPC::sharedPage streamIndex;
streamIndex.init(name, 8 * 1024 * 1024); streamIndex.init(name, 8 * 1024 * 1024);
if (!streamIndex.mapped){
DEBUG_MSG(DLVL_INSANE, "Stream %s opening failed! Cancelling and marking as corrupt.", name.c_str());
data["meta"].null();
data["meta"]["tracks"].null();
data["error"] = "Stream offline: Corrupt file?";
if (data["error"].asStringRef() != prevState){
Log("WARN", "Source file " + URL + " seems to be corrupt.");
} }
data["online"] = 0;
}
}
DEBUG_MSG(DLVL_INSANE, "Stream %s opened", name.c_str());
} }
//now, run mistinfo on the source - or on the accompanying dtsh file, if it exists //now, run mistinfo on the source - or on the accompanying dtsh file, if it exists
if (stat((URL+".dtsh").c_str(), &fileinfo) == 0){ if (stat((URL+".dtsh").c_str(), &fileinfo) == 0){
DEBUG_MSG(DLVL_INSANE, "Stream %s has a DTSH - opening DTSH instead of main stream file", name.c_str());
URL += ".dtsh"; URL += ".dtsh";
} }
char * tmp_cmd[3] = {0, 0, 0}; char * tmp_cmd[3] = {0, 0, 0};
std::string mistinfo = Util::getMyPath() + "MistInfo"; std::string mistinfo = Util::getMyPath() + "MistInfo";
tmp_cmd[0] = (char*)mistinfo.c_str(); tmp_cmd[0] = (char*)mistinfo.c_str();
tmp_cmd[1] = (char*)URL.c_str(); tmp_cmd[1] = (char*)URL.c_str();
DEBUG_MSG(DLVL_INSANE, "Running MistInfo for stream %s on file %s", name.c_str(), tmp_cmd[1]);
data["meta"] = JSON::fromString(Util::Procs::getOutputOf(tmp_cmd)); data["meta"] = JSON::fromString(Util::Procs::getOutputOf(tmp_cmd));
if ( !data["meta"] || !data["meta"].isMember("tracks") || !data["meta"]["tracks"]){ if ( !data["meta"] || !data["meta"].isMember("tracks") || !data["meta"]["tracks"]){
data["error"] = "Stream offline: Corrupt file?"; data["error"] = "Stream offline: Corrupt file?";
@ -132,6 +154,7 @@ namespace Controller {
data["online"] = 0; data["online"] = 0;
return; return;
} }
DEBUG_MSG(DLVL_INSANE, "Metadata for stream %s succesfully (re)loaded", name.c_str());
} }
if (!hasViewers(name)){ if (!hasViewers(name)){
if ( !data.isMember("error")){ if ( !data.isMember("error")){
@ -141,13 +164,18 @@ namespace Controller {
}else{ }else{
data["online"] = 1; data["online"] = 1;
} }
return; //MistPlayer handles VoD return;
}else{ }
/// \todo Implement ffmpeg pulling again? /// \todo Implement ffmpeg pulling again?
//Util::Procs::Start(name, "ffmpeg -re -async 2 -i " + URL + " -f flv -", Util::getMyPath() + "MistFLV2DTSC", Util::getMyPath() + buffcmd); //Util::Procs::Start(name, "ffmpeg -re -async 2 -i " + URL + " -f flv -", Util::getMyPath() + "MistFLV2DTSC", Util::getMyPath() + buffcmd);
//Log("BUFF", "(re)starting stream buffer " + name + " for ffmpeg data: ffmpeg -re -async 2 -i " + URL + " -f flv -"); //Log("BUFF", "(re)starting stream buffer " + name + " for ffmpeg data: ffmpeg -re -async 2 -i " + URL + " -f flv -");
//not recognized
data["error"] = "Invalid source format";
if (data["error"].asStringRef() != prevState){
Log("STRM", "Invalid source format for stream " + name + "!");
} }
} return;
} }
///\brief Checks all streams, restoring if needed. ///\brief Checks all streams, restoring if needed.
@ -155,9 +183,7 @@ namespace Controller {
void CheckAllStreams(JSON::Value & data){ void CheckAllStreams(JSON::Value & data){
long long int currTime = Util::epoch(); long long int currTime = Util::epoch();
for (JSON::ObjIter jit = data.ObjBegin(); jit != data.ObjEnd(); jit++){ for (JSON::ObjIter jit = data.ObjBegin(); jit != data.ObjEnd(); jit++){
if ( !Util::Procs::isActive(jit->first)){ checkStream(jit->first, jit->second);
startStream(jit->first, jit->second);
}
if (!jit->second.isMember("name")){ if (!jit->second.isMember("name")){
jit->second["name"] = jit->first; jit->second["name"] = jit->first;
} }
@ -235,14 +261,7 @@ namespace Controller {
} }
out[jit->first]["updated"] = 1ll; out[jit->first]["updated"] = 1ll;
Log("STRM", std::string("Updated stream ") + jit->first); Log("STRM", std::string("Updated stream ") + jit->first);
if (out[jit->first]["source"].asStringRef().substr(0, 7) != "push://"){ checkStream(jit->first, out[jit->first]);
Util::Procs::Stop(jit->first);
startStream(jit->first, out[jit->first]);
}else{
if ( !Util::Procs::isActive(jit->first)){
startStream(jit->first, out[jit->first]);
}
}
} }
}else{ }else{
out[jit->first]["name"] = jit->first; out[jit->first]["name"] = jit->first;
@ -254,7 +273,7 @@ namespace Controller {
out[jit->first]["cut"] = jit->second["cut"].asInt(); out[jit->first]["cut"] = jit->second["cut"].asInt();
} }
Log("STRM", std::string("New stream ") + jit->first); Log("STRM", std::string("New stream ") + jit->first);
startStream(jit->first, out[jit->first]); checkStream(jit->first, out[jit->first]);
} }
} }
} }
@ -318,7 +337,6 @@ namespace Controller {
if ( !in.isMember(jit->first)){ if ( !in.isMember(jit->first)){
toDelete.insert(jit->first); toDelete.insert(jit->first);
Log("STRM", std::string("Deleted stream ") + jit->first); Log("STRM", std::string("Deleted stream ") + jit->first);
Util::Procs::Stop(jit->first);
} }
} }
//actually delete the streams //actually delete the streams

2
src/controller/controller_streams.h
View file

@ -2,7 +2,7 @@
namespace Controller { namespace Controller {
bool streamsEqual(JSON::Value & one, JSON::Value & two); bool streamsEqual(JSON::Value & one, JSON::Value & two);
void startStream(std::string name, JSON::Value & data); void checkStream(std::string name, JSON::Value & data);
void CheckAllStreams(JSON::Value & data); void CheckAllStreams(JSON::Value & data);
void CheckStreams(JSON::Value & in, JSON::Value & out); void CheckStreams(JSON::Value & in, JSON::Value & out);
void AddStreams(JSON::Value & in, JSON::Value & out); void AddStreams(JSON::Value & in, JSON::Value & out);