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

JSON-based input selection.

Browse source
This commit is contained in:
Thulinma 2014-10-02 15:41:50 +02:00
parent d85fe140ca
commit 8542281ac2
13 changed files with 279 additions and 215 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

273
src/connectors/conn_http.cpp
View file

@ -25,6 +25,8 @@
#include <mist/procs.h> #include <mist/procs.h>
#include <mist/tinythread.h> #include <mist/tinythread.h>
#include <mist/defines.h> #include <mist/defines.h>
#include <mist/dtsc.h>
#include <mist/shared_memory.h>
#include "embed.js.h" #include "embed.js.h"
@ -86,21 +88,7 @@ namespace Connector_HTTP {
tthread::mutex timeoutStartMutex; ///< Mutex for starting timeout thread. tthread::mutex timeoutStartMutex; ///< Mutex for starting timeout thread.
tthread::mutex timeoutMutex; ///< Mutex for timeout thread. tthread::mutex timeoutMutex; ///< Mutex for timeout thread.
tthread::thread * timeouter = 0; ///< Thread that times out connections to connectors. tthread::thread * timeouter = 0; ///< Thread that times out connections to connectors.
JSON::Value capabilities; ///< Holds a list of all HTTP connectors and their properties IPC::sharedPage serverCfg; ///< Contains server configuration and capabilities
JSON::Value ServConf; /// < holds configuration, loads from file in main
void updateConfig(){
static unsigned long long int confUpdateTime=0;
static tthread::mutex updateLock;
if( Util::bootSecs() -confUpdateTime > 10 ){
tthread::lock_guard<tthread::mutex> guard(updateLock);
if( Util::bootSecs() -confUpdateTime > 10 ){
ServConf = JSON::fromFile(Util::getTmpFolder() + "streamlist");
confUpdateTime=Util::bootSecs();
}
}
}
///\brief Function run as a thread to timeout requests on the proxy. ///\brief Function run as a thread to timeout requests on the proxy.
///\param n A NULL-pointer ///\param n A NULL-pointer
@ -273,7 +261,6 @@ namespace Connector_HTTP {
///\param conn The connection to the client that issued the request. ///\param conn The connection to the client that issued the request.
///\return A timestamp indicating when the request was parsed. ///\return A timestamp indicating when the request was parsed.
long long int proxyHandleInternal(HTTP::Parser & H, Socket::Connection & conn){ long long int proxyHandleInternal(HTTP::Parser & H, Socket::Connection & conn){
updateConfig();
std::string url = H.getUrl(); std::string url = H.getUrl();
if (url == "/crossdomain.xml"){ if (url == "/crossdomain.xml"){
@ -315,20 +302,20 @@ namespace Connector_HTTP {
// send logo icon // send logo icon
if (url.length() > 6 && url.substr(url.length() - 5, 5) == ".html"){ if (url.length() > 6 && url.substr(url.length() - 5, 5) == ".html"){
std::string streamname = url.substr(1, url.length() - 6); std::string streamname = url.substr(1, url.length() - 6);
Util::Stream::sanitizeName(streamname); Util::sanitizeName(streamname);
H.Clean(); H.Clean();
H.SetHeader("Content-Type", "text/html"); H.SetHeader("Content-Type", "text/html");
H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver); H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver);
H.SetBody("<!DOCTYPE html><html><head><title>Stream "+streamname+"</title><style>BODY{color:white;background:black;}</style></head><body><script src=\"embed_"+streamname+".js\"></script></body></html>"); H.SetBody("<!DOCTYPE html><html><head><title>Stream "+streamname+"</title><style>BODY{color:white;background:black;}</style></head><body><script src=\"embed_"+streamname+".js\"></script></body></html>");
long long int ret = Util::getMS(); long long int ret = Util::getMS();
conn.SendNow(H.BuildResponse("200", "OK")); H.SendResponse("200", "OK", conn);
return ret; return ret;
} }
// send smil MBR index // send smil MBR index
if (url.length() > 6 && url.substr(url.length() - 5, 5) == ".smil"){ if (url.length() > 6 && url.substr(url.length() - 5, 5) == ".smil"){
std::string streamname = url.substr(1, url.length() - 6); std::string streamname = url.substr(1, url.length() - 6);
Util::Stream::sanitizeName(streamname); Util::sanitizeName(streamname);
std::string host = H.GetHeader("Host"); std::string host = H.GetHeader("Host");
if (host.find(':')){ if (host.find(':')){
@ -337,38 +324,45 @@ namespace Connector_HTTP {
std::string port, url_rel; std::string port, url_rel;
for (JSON::ArrIter it = ServConf["config"]["protocols"].ArrBegin(); it != ServConf["config"]["protocols"].ArrEnd(); it++){ IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
const std::string & cName = ( *it)["connector"].asStringRef(); configLock.wait();
if (cName != "RTMP"){continue;} DTSC::Scan prtcls = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("config").getMember("protocols");
//if we have the RTMP port, DTSC::Scan capa = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("capabilities").getMember("connectors").getMember("RTMP");
if (capabilities.isMember(cName) && capabilities[cName].isMember("optional") && capabilities[cName]["optional"].isMember("port")){ unsigned int pro_cnt = prtcls.getSize();
//get the default port if none is set for (unsigned int i = 0; i < pro_cnt; ++i){
if (( *it)["port"].asInt() == 0){ if (prtcls.getIndice(i).getMember("connector").asString() != "RTMP"){
port = capabilities[cName]["optional"]["port"]["default"].asString(); continue;
} }
//extract url port = prtcls.getIndice(i).getMember("port").asString();
if (capabilities[cName].isMember("url_rel")){ //get the default port if none is set
url_rel = capabilities[cName]["url_rel"].asString(); if (!port.size()){
if (url_rel.find('$')){ port = capa.getMember("optional").getMember("port").getMember("default").asString();
url_rel.resize(url_rel.find('$')); }
} //extract url
} url_rel = capa.getMember("url_rel").asString();
if (url_rel.find('$')){
url_rel.resize(url_rel.find('$'));
} }
} }
std::string trackSources;//this string contains all track sources for MBR smil std::string trackSources;//this string contains all track sources for MBR smil
for (JSON::ObjIter it = ServConf["streams"][streamname]["meta"]["tracks"].ObjBegin(); it != ServConf["streams"][streamname]["meta"]["tracks"].ObjEnd(); it++){//for all tracks DTSC::Scan tracks = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("streams").getMember(streamname).getMember("meta").getMember("tracks");
if (it->second.isMember("type") && it->second["type"].asString() == "video"){ unsigned int track_ctr = tracks.getSize();
trackSources += " <video src='"+ streamname + "?track=" + it->second["trackid"].asString() + "' height='" + it->second["height"].asString() + "' system-bitrate='" + it->second["bps"].asString() + "' width='" + it->second["width"].asString() + "' />\n"; for (unsigned int i = 0; i < track_ctr; ++i){//for all video tracks
DTSC::Scan trk = tracks.getIndice(i);
if (trk.getMember("type").asString() == "video"){
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();
configLock.close();
H.Clean(); H.Clean();
H.SetHeader("Content-Type", "application/smil"); H.SetHeader("Content-Type", "application/smil");
H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver); H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver);
H.SetBody("<smil>\n <head>\n <meta base='rtmp://" + host + ":" + port + url_rel + "' />\n </head>\n <body>\n <switch>\n"+trackSources+" </switch>\n </body>\n</smil>"); H.SetBody("<smil>\n <head>\n <meta base='rtmp://" + host + ":" + port + url_rel + "' />\n </head>\n <body>\n <switch>\n"+trackSources+" </switch>\n </body>\n</smil>");
long long int ret = Util::getMS(); long long int ret = Util::getMS();
conn.SendNow(H.BuildResponse("200", "OK")); H.SendResponse("200", "OK", conn);
return ret; return ret;
} }
@ -380,7 +374,7 @@ namespace Connector_HTTP {
}else{ }else{
streamname = url.substr(7, url.length() - 10); streamname = url.substr(7, url.length() - 10);
} }
Util::Stream::sanitizeName(streamname); Util::sanitizeName(streamname);
std::string response; std::string response;
std::string host = H.GetHeader("Host"); std::string host = H.GetHeader("Host");
@ -392,57 +386,68 @@ namespace Connector_HTTP {
H.SetHeader("Content-Type", "application/javascript"); H.SetHeader("Content-Type", "application/javascript");
response = "// Generating info code for stream " + streamname + "\n\nif (!mistvideo){var mistvideo = {};}\n"; response = "// Generating info code for stream " + streamname + "\n\nif (!mistvideo){var mistvideo = {};}\n";
JSON::Value json_resp; JSON::Value json_resp;
if (ServConf["streams"].isMember(streamname) && ServConf["config"]["protocols"].size() > 0){ IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
if (ServConf["streams"][streamname]["meta"].isMember("tracks") && ServConf["streams"][streamname]["meta"]["tracks"].size() > 0){ configLock.wait();
for (JSON::ObjIter it = ServConf["streams"][streamname]["meta"]["tracks"].ObjBegin(); it != ServConf["streams"][streamname]["meta"]["tracks"].ObjEnd(); it++){ DTSC::Scan strm = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("streams").getMember(streamname).getMember("meta");
if (it->second.isMember("width") && it->second["width"].asInt() > json_resp["width"].asInt()){ DTSC::Scan prots = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("config").getMember("protocols");
json_resp["width"] = it->second["width"].asInt(); if (strm && prots){
} DTSC::Scan trcks = strm.getMember("tracks");
if (it->second.isMember("height") && it->second["height"].asInt() > json_resp["height"].asInt()){ unsigned int trcks_ctr = trcks.getSize();
json_resp["height"] = it->second["height"].asInt(); for (unsigned int i = 0; i < trcks_ctr; ++i){
} if (trcks.getIndice(i).getMember("width").asInt() > json_resp["width"].asInt()){
json_resp["width"] = trcks.getIndice(i).getMember("width").asInt();
}
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){ if (json_resp["width"].asInt() < 1 || json_resp["height"].asInt() < 1){
json_resp["width"] = 640ll; json_resp["width"] = 640ll;
json_resp["height"] = 480ll; json_resp["height"] = 480ll;
} }
if (ServConf["streams"][streamname]["meta"].isMember("vod")){ if (strm.getMember("vod")){
json_resp["type"] = "vod"; json_resp["type"] = "vod";
} }
if (ServConf["streams"][streamname]["meta"].isMember("live")){ if (strm.getMember("live")){
json_resp["type"] = "live"; json_resp["type"] = "live";
} }
// show ALL the meta datas! // show ALL the meta datas!
json_resp["meta"] = ServConf["streams"][streamname]["meta"]; json_resp["meta"] = strm.asJSON();
//create a set for storing source information //create a set for storing source information
std::set<JSON::Value, sourceCompare> sources; std::set<JSON::Value, sourceCompare> sources;
//find out which connectors are enabled //find out which connectors are enabled
std::set<std::string> conns; std::set<std::string> conns;
for (JSON::ArrIter it = ServConf["config"]["protocols"].ArrBegin(); it != ServConf["config"]["protocols"].ArrEnd(); it++){ unsigned int prots_ctr = prots.getSize();
conns.insert(( *it)["connector"].asStringRef()); for (unsigned int i = 0; i < prots_ctr; ++i){
conns.insert(prots.getIndice(i).getMember("connector").asString());
} }
//loop over the connectors. //loop over the connectors.
for (JSON::ArrIter it = ServConf["config"]["protocols"].ArrBegin(); it != ServConf["config"]["protocols"].ArrEnd(); it++){ for (unsigned int i = 0; i < prots_ctr; ++i){
const std::string & cName = ( *it)["connector"].asStringRef(); std::string cName = prots.getIndice(i).getMember("connector").asString();
DTSC::Scan capa = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("capabilities").getMember("connectors").getMember(cName);
//if the connector has a port, //if the connector has a port,
if (capabilities.isMember(cName) && capabilities[cName].isMember("optional") && capabilities[cName]["optional"].isMember("port")){ if (capa.getMember("optional").getMember("port")){
//get the default port if none is set //get the default port if none is set
if (( *it)["port"].asInt() == 0){ std::string port = prots.getIndice(i).getMember("port").asString();
( *it)["port"] = capabilities[cName]["optional"]["port"]["default"]; if (!port.size()){
port = capa.getMember("optional").getMember("port").getMember("default").asString();
} }
//and a URL - then list the URL //and a URL - then list the URL
if (capabilities[cName].isMember("url_rel")){ if (capa.getMember("url_rel")){
addSources(streamname, capabilities[cName]["url_rel"].asStringRef(), sources, host, ( *it)["port"].asString(), capabilities[cName], ServConf["streams"][streamname]["meta"]); JSON::Value capa_json = capa.asJSON();
addSources(streamname, capa.getMember("url_rel").asString(), sources, host, port, capa_json, json_resp["meta"]);
} }
//check each enabled protocol separately to see if it depends on this connector //check each enabled protocol separately to see if it depends on this connector
for (JSON::ObjIter oit = capabilities.ObjBegin(); oit != capabilities.ObjEnd(); oit++){ DTSC::Scan capa_lst = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("capabilities").getMember("connectors");
unsigned int capa_lst_ctr = capa_lst.getSize();
for (unsigned int j = 0; j < capa_lst_ctr; ++j){
//if it depends on this connector and has a URL, list it //if it depends on this connector and has a URL, list it
if (conns.count(oit->first) && (oit->second["deps"].asStringRef() == cName || oit->second["deps"].asStringRef() + ".exe" == cName) && oit->second.isMember("methods")){ if (conns.count(capa_lst.getIndiceName(j)) && (capa_lst.getIndice(j).getMember("deps").asString() == cName || capa_lst.getIndice(j).getMember("deps").asString() + ".exe" == cName) && capa_lst.getIndice(j).getMember("methods")){
addSources(streamname, oit->second["url_rel"].asStringRef(), sources, host, ( *it)["port"].asString(), oit->second, ServConf["streams"][streamname]["meta"]); JSON::Value capa_json = capa_lst.getIndice(j).asJSON();
addSources(streamname, capa_lst.getIndice(j).getMember("url_rel").asString(), sources, host, port, capa_json, json_resp["meta"]);
} }
} }
} }
@ -457,6 +462,8 @@ namespace Connector_HTTP {
}else{ }else{
json_resp["error"] = "The specified stream is not available on this server."; json_resp["error"] = "The specified stream is not available on this server.";
} }
configLock.post();
configLock.close();
response += "mistvideo['" + streamname + "'] = " + json_resp.toString() + ";\n"; response += "mistvideo['" + streamname + "'] = " + json_resp.toString() + ";\n";
if (url.substr(0, 6) != "/info_" && !json_resp.isMember("error")){ if (url.substr(0, 6) != "/info_" && !json_resp.isMember("error")){
response.append("\n("); response.append("\n(");
@ -469,7 +476,7 @@ namespace Connector_HTTP {
} }
H.SetBody(response); H.SetBody(response);
long long int ret = Util::getMS(); long long int ret = Util::getMS();
conn.SendNow(H.BuildResponse("200", "OK")); H.SendResponse("200", "OK", conn);
return ret; return ret;
} //embed code generator } //embed code generator
@ -483,8 +490,6 @@ namespace Connector_HTTP {
///\param connector The type of connector to be invoked. ///\param connector The type of connector to be invoked.
///\return -1 on failure, else 0. ///\return -1 on failure, else 0.
long long int proxyHandleThroughConnector(HTTP::Parser & H, Socket::Connection & conn, std::string & connector){ long long int proxyHandleThroughConnector(HTTP::Parser & H, Socket::Connection & conn, std::string & connector){
updateConfig();
//create a unique ID based on a hash of the user agent and host, followed by the stream name and connector //create a unique ID based on a hash of the user agent and host, followed by the stream name and connector
std::stringstream uidtemp; std::stringstream uidtemp;
/// \todo verify the correct formation of the uid /// \todo verify the correct formation of the uid
@ -500,14 +505,21 @@ namespace Connector_HTTP {
for (int i=0; i<20; i++){argarr[i] = 0;} for (int i=0; i<20; i++){argarr[i] = 0;}
int id = -1; int id = -1;
for (unsigned int i=0; i < ServConf["config"]["protocols"].size(); ++i){ IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
if ( ServConf["config"]["protocols"][i]["connector"].asStringRef() == connector ) { configLock.wait();
DTSC::Scan prots = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("config").getMember("protocols");
unsigned int prots_ctr = prots.getSize();
for (unsigned int i=0; i < prots_ctr; ++i){
if (prots.getIndice(i).getMember("connector").asString() == connector) {
id = i; id = i;
break; //pick the first protocol in the list that matches the connector break; //pick the first protocol in the list that matches the connector
} }
} }
if (id == -1) { if (id == -1) {
DEBUG_MSG(DLVL_ERROR, "No connector found for: %s", connector.c_str()); DEBUG_MSG(DLVL_ERROR, "No connector found for: %s", connector.c_str());
configLock.post();
configLock.close();
return -1; return -1;
} }
@ -519,16 +531,18 @@ namespace Connector_HTTP {
std::string debuglevel = JSON::Value((long long)Util::Config::printDebugLevel).asString(); std::string debuglevel = JSON::Value((long long)Util::Config::printDebugLevel).asString();
std::string tmparg; std::string tmparg;
tmparg = Util::getMyPath() + std::string("MistOut") + ServConf["config"]["protocols"][id]["connector"].asStringRef(); tmparg = Util::getMyPath() + std::string("MistOut") + connector;
struct stat buf; struct stat buf;
if (::stat(tmparg.c_str(), &buf) != 0){ if (::stat(tmparg.c_str(), &buf) != 0){
tmparg = Util::getMyPath() + std::string("MistConn") + ServConf["config"]["protocols"][id]["connector"].asStringRef(); tmparg = Util::getMyPath() + std::string("MistConn") + connector;
} }
int argnum = 0; int argnum = 0;
argarr[argnum++] = (char*)tmparg.c_str(); argarr[argnum++] = (char*)tmparg.c_str();
JSON::Value & p = ServConf["config"]["protocols"][id]; JSON::Value p = prots.getIndice(id).asJSON();
JSON::Value & pipedCapa = capabilities[p["connector"].asStringRef()]; JSON::Value pipedCapa = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("capabilities").getMember("connectors").getMember(connector).asJSON();
configLock.post();
configLock.close();
argarr[argnum++] = (char*)"-i"; argarr[argnum++] = (char*)"-i";
argarr[argnum++] = (char*)(temphost.c_str()); argarr[argnum++] = (char*)(temphost.c_str());
argarr[argnum++] = (char*)"-s"; argarr[argnum++] = (char*)"-s";
@ -557,41 +571,7 @@ namespace Connector_HTTP {
/// - anything else: The request should be dispatched to a connector on the named socket. /// - anything else: The request should be dispatched to a connector on the named socket.
std::string proxyGetHandleType(HTTP::Parser & H){ std::string proxyGetHandleType(HTTP::Parser & H){
std::string url = H.getUrl(); std::string url = H.getUrl();
//loop over the connectors
for (JSON::ObjIter oit = capabilities.ObjBegin(); oit != capabilities.ObjEnd(); oit++){
//if it depends on HTTP and has a match or prefix...
if (oit->second["deps"].asStringRef() == "HTTP" && oit->second.isMember("socket") && (oit->second.isMember("url_match") || oit->second.isMember("url_prefix"))){
//if there is a matcher, try to match
if (oit->second.isMember("url_match")){
size_t found = oit->second["url_match"].asStringRef().find('$');
if (found != std::string::npos){
if (oit->second["url_match"].asStringRef().substr(0, found) == url.substr(0, found) && oit->second["url_match"].asStringRef().substr(found+1) == url.substr(url.size() - (oit->second["url_match"].asStringRef().size() - found) + 1)){
//it matched - handle it now
std::string streamname = url.substr(found, url.size() - oit->second["url_match"].asStringRef().size() + 1);
Util::Stream::sanitizeName(streamname);
H.SetVar("stream", streamname);
return oit->first;
}
}
}
//if there is a prefix, try to match
if (oit->second.isMember("url_prefix")){
size_t found = oit->second["url_prefix"].asStringRef().find('$');
if (found != std::string::npos){
size_t found_suf = url.find(oit->second["url_prefix"].asStringRef().substr(found+1), found);
if (oit->second["url_prefix"].asStringRef().substr(0, found) == url.substr(0, found) && found_suf != std::string::npos){
//it matched - handle it now
std::string streamname = url.substr(found, found_suf - found);
Util::Stream::sanitizeName(streamname);
H.SetVar("stream", streamname);
return oit->first;
}
}
}
}
}
if (url.length() > 4){ if (url.length() > 4){
std::string ext = url.substr(url.length() - 4, 4); std::string ext = url.substr(url.length() - 4, 4);
if (ext == ".ico"){ if (ext == ".ico"){
@ -616,6 +596,54 @@ namespace Connector_HTTP {
if (url.length() > 9 && url.substr(0, 6) == "/info_" && url.substr(url.length() - 3, 3) == ".js"){ if (url.length() > 9 && url.substr(0, 6) == "/info_" && url.substr(url.length() - 3, 3) == ".js"){
return "internal"; return "internal";
} }
//loop over the connectors
IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
configLock.wait();
DTSC::Scan capa = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("capabilities").getMember("connectors");
unsigned int capa_ctr = capa.getSize();
for (unsigned int i = 0; i < capa_ctr; ++i){
DTSC::Scan c = capa.getIndice(i);
//if it depends on HTTP and has a match or prefix...
if (c.getMember("deps").asString() == "HTTP" && (c.getMember("url_match") || c.getMember("url_prefix"))){
//if there is a matcher, try to match
if (c.getMember("url_match")){
std::string m = c.getMember("url_match").asString();
size_t found = m.find('$');
if (found != std::string::npos){
if (m.substr(0, found) == url.substr(0, found) && m.substr(found+1) == url.substr(url.size() - (m.size() - found) + 1)){
//it matched - handle it now
std::string streamname = url.substr(found, url.size() - m.size() + 1);
Util::sanitizeName(streamname);
H.SetVar("stream", streamname);
configLock.post();
configLock.close();
return capa.getIndiceName(i);
}
}
}
//if there is a prefix, try to match
if (c.getMember("url_prefix")){
std::string m = c.getMember("url_prefix").asString();
size_t found = m.find('$');
if (found != std::string::npos){
size_t found_suf = url.find(m.substr(found+1), found);
if (m.substr(0, found) == url.substr(0, found) && found_suf != std::string::npos){
//it matched - handle it now
std::string streamname = url.substr(found, found_suf - found);
Util::sanitizeName(streamname);
H.SetVar("stream", streamname);
configLock.post();
configLock.close();
return capa.getIndiceName(i);
}
}
}
}
}
configLock.post();
configLock.close();
return "none"; return "none";
} }
@ -675,7 +703,6 @@ int main(int argc, char ** argv){
capa["optional"]["debug"]["help"] = "The debug level at which messages need to be printed."; capa["optional"]["debug"]["help"] = "The debug level at which messages need to be printed.";
capa["optional"]["debug"]["option"] = "--debug"; capa["optional"]["debug"]["option"] = "--debug";
capa["optional"]["debug"]["type"] = "uint"; capa["optional"]["debug"]["type"] = "uint";
Connector_HTTP::ServConf = JSON::fromFile(Util::getTmpFolder() + "streamlist");
capa["desc"] = "Enables the generic HTTP listener, required by all other HTTP protocols. Needs other HTTP protocols enabled to do much of anything."; capa["desc"] = "Enables the generic HTTP listener, required by all other HTTP protocols. Needs other HTTP protocols enabled to do much of anything.";
capa["deps"] = ""; capa["deps"] = "";
conf.addConnectorOptions(8080, capa); conf.addConnectorOptions(8080, capa);
@ -684,30 +711,6 @@ int main(int argc, char ** argv){
std::cout << capa.toString() << std::endl; std::cout << capa.toString() << std::endl;
return -1; return -1;
} }
Connector_HTTP::serverCfg.init("!mistConfig", 4*1024*1024);
//list available protocols and report about them
std::deque<std::string> execs;
Util::getMyExec(execs);
std::string arg_one;
char const * conn_args[] = {0, "-j", 0};
for (std::deque<std::string>::iterator it = execs.begin(); it != execs.end(); it++){
if ((*it).substr(0, 8) == "MistConn"){
arg_one = Util::getMyPath() + (*it);
conn_args[0] = arg_one.c_str();
Connector_HTTP::capabilities[(*it).substr(8)] = JSON::fromString(Util::Procs::getOutputOf((char**)conn_args));
if (Connector_HTTP::capabilities[(*it).substr(8)].size() < 1){
Connector_HTTP::capabilities.removeMember((*it).substr(8));
}
}
if ((*it).substr(0, 7) == "MistOut"){
arg_one = Util::getMyPath() + (*it);
conn_args[0] = arg_one.c_str();
Connector_HTTP::capabilities[(*it).substr(7)] = JSON::fromString(Util::Procs::getOutputOf((char**)conn_args));
if (Connector_HTTP::capabilities[(*it).substr(7)].size() < 1){
Connector_HTTP::capabilities.removeMember((*it).substr(7));
}
}
}
return conf.serveThreadedSocket(Connector_HTTP::proxyHandleHTTPConnection); return conf.serveThreadedSocket(Connector_HTTP::proxyHandleHTTPConnection);
} //main }

8
src/controller/controller_capabilities.cpp
View file

@ -40,6 +40,14 @@ namespace Controller {
capabilities["connectors"].removeMember((*it).substr(7)); capabilities["connectors"].removeMember((*it).substr(7));
} }
} }
if ((*it).substr(0, 6) == "MistIn" && (*it) != "MistInfo"){
arg_one = Util::getMyPath() + (*it);
conn_args[0] = arg_one.c_str();
capabilities["inputs"][(*it).substr(6)] = JSON::fromString(Util::Procs::getOutputOf((char**)conn_args));
if (capabilities["inputs"][(*it).substr(6)].size() < 1){
capabilities["inputs"].removeMember((*it).substr(6));
}
}
} }
} }

22
src/controller/controller_storage.cpp
View file

@ -1,7 +1,11 @@
#include <sys/stat.h>
#include <iostream> #include <iostream>
#include <fstream> #include <fstream>
#include <algorithm>
#include <mist/timing.h> #include <mist/timing.h>
#include <mist/shared_memory.h>
#include "controller_storage.h" #include "controller_storage.h"
#include "controller_capabilities.h"
///\brief Holds everything unique to the controller. ///\brief Holds everything unique to the controller.
namespace Controller { namespace Controller {
@ -68,5 +72,23 @@ namespace Controller {
fclose(output); fclose(output);
close((long long int)err); close((long long int)err);
} }
/// Writes the current config to shared memory to be used in other processes
void writeConfig(){
JSON::Value writeConf;
writeConf["config"] = Storage["config"];
writeConf["streams"] = Storage["streams"];
writeConf["capabilities"] = capabilities;
static IPC::sharedPage mistConfOut("!mistConfig", 4*1024*1024, true);
IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
//lock semaphore
configLock.wait();
//write config
std::string temp = writeConf.toPacked();
memcpy(mistConfOut.mapped, temp.data(), std::min(temp.size(), (unsigned long)mistConfOut.len));
//unlock semaphore
configLock.post();
}
} }

2
src/controller/controller_storage.h
View file

@ -17,4 +17,6 @@ namespace Controller {
void handleMsg(void * err); void handleMsg(void * err);
void writeConfig();
} }

31
src/controller/controller_streams.cpp
View file

@ -6,6 +6,7 @@
#include <mist/defines.h> #include <mist/defines.h>
#include <mist/shared_memory.h> #include <mist/shared_memory.h>
#include "controller_streams.h" #include "controller_streams.h"
#include "controller_capabilities.h"
#include "controller_storage.h" #include "controller_storage.h"
#include "controller_statistics.h" #include "controller_statistics.h"
#include <sys/stat.h> #include <sys/stat.h>
@ -22,13 +23,15 @@ namespace Controller {
if (one.isMember("source") != two.isMember("source") || one["source"] != two["source"]){ if (one.isMember("source") != two.isMember("source") || one["source"] != two["source"]){
return false; return false;
} }
if (one.isMember("DVR") != two.isMember("DVR") || (one.isMember("DVR") && one["DVR"] != two["DVR"])){
return false; /// \todo Change this to use capabilities["inputs"] and only compare required/optional parameters.
} /// \todo Maybe change this to check for correct source and/or required parameters.
if (one.isMember("cut") != two.isMember("cut") || (one.isMember("cut") && one["cut"] != two["cut"])){
return false; //temporary: compare the two JSON::Value objects.
} return one==two;
return true;
//nothing different? return true by default
//return true;
} }
///\brief Checks the validity of a stream, updates internal stream status. ///\brief Checks the validity of a stream, updates internal stream status.
@ -116,7 +119,7 @@ namespace Controller {
DEBUG_MSG(DLVL_INSANE, "(re)loading metadata for stream %s", name.c_str()); 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()); 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::startInput(name);
DEBUG_MSG(DLVL_INSANE, "Waiting for stream %s to open...", name.c_str()); DEBUG_MSG(DLVL_INSANE, "Waiting for stream %s to open...", name.c_str());
//wait for the stream //wait for the stream
{ {
@ -241,7 +244,7 @@ namespace Controller {
changed = true; changed = true;
} }
if (changed){ if (changed){
WriteFile(Util::getTmpFolder() + "streamlist", strlist.toString()); writeConfig();
} }
} }
@ -250,16 +253,8 @@ namespace Controller {
for (JSON::ObjIter jit = in.ObjBegin(); jit != in.ObjEnd(); jit++){ for (JSON::ObjIter jit = in.ObjBegin(); jit != in.ObjEnd(); jit++){
if (out.isMember(jit->first)){ if (out.isMember(jit->first)){
if ( !streamsEqual(jit->second, out[jit->first])){ if ( !streamsEqual(jit->second, out[jit->first])){
out[jit->first].null(); out[jit->first] = jit->second;
out[jit->first]["name"] = jit->first; out[jit->first]["name"] = jit->first;
out[jit->first]["source"] = jit->second["source"];
if (jit->second.isMember("DVR")){
out[jit->first]["DVR"] = jit->second["DVR"].asInt();
}
if (jit->second.isMember("cut")){
out[jit->first]["cut"] = jit->second["cut"].asInt();
}
out[jit->first]["updated"] = 1ll;
Log("STRM", std::string("Updated stream ") + jit->first); Log("STRM", std::string("Updated stream ") + jit->first);
checkStream(jit->first, out[jit->first]); checkStream(jit->first, out[jit->first]);
} }

18
src/input/input.cpp
View file

@ -31,7 +31,7 @@ namespace Mist {
JSON::Value option; JSON::Value option;
option["long"] = "json"; option["long"] = "json";
option["short"] = "j"; option["short"] = "j";
option["help"] = "Output MistIn info in JSON format, then exit."; option["help"] = "Output MistIn info in JSON format, then exit";
option["value"].append(0ll); option["value"].append(0ll);
config->addOption("json", option); config->addOption("json", option);
option.null(); option.null();
@ -50,13 +50,13 @@ namespace Mist {
option["arg"] = "string"; option["arg"] = "string";
option["short"] = "s"; option["short"] = "s";
option["long"] = "stream"; option["long"] = "stream";
option["help"] = "The name of the stream that this connector will transmit."; option["help"] = "The name of the stream that this connector will provide in player mode";
config->addOption("streamname", option); config->addOption("streamname", option);
option.null();
option["short"] = "p"; capa["optional"]["debug"]["name"] = "debug";
option["long"] = "player"; capa["optional"]["debug"]["help"] = "The debug level at which messages need to be printed.";
option["help"] = "Makes this connector into a player"; capa["optional"]["debug"]["option"] = "--debug";
config->addOption("player", option); capa["optional"]["debug"]["type"] = "uint";
packTime = 0; packTime = 0;
lastActive = Util::epoch(); lastActive = Util::epoch();
@ -99,7 +99,7 @@ namespace Mist {
int Input::run() { int Input::run() {
if (config->getBool("json")) { if (config->getBool("json")) {
std::cerr << capa.toString() << std::endl; std::cout << capa.toString() << std::endl;
return 0; return 0;
} }
if (!setup()) { if (!setup()) {
@ -113,7 +113,7 @@ namespace Mist {
} }
parseHeader(); parseHeader();
if (!config->getBool("player")){ if (!config->getString("streamname").size()){
//check filename for no - //check filename for no -
if (config->getString("output") != "-"){ if (config->getString("output") != "-"){
std::string filename = config->getString("output"); std::string filename = config->getString("output");

82
src/input/input_buffer.cpp
View file

@ -14,26 +14,25 @@
namespace Mist { namespace Mist {
inputBuffer::inputBuffer(Util::Config * cfg) : Input(cfg) { inputBuffer::inputBuffer(Util::Config * cfg) : Input(cfg) {
capa["name"] = "Buffer";
JSON::Value option; JSON::Value option;
option["arg"] = "integer"; option["arg"] = "integer";
option["long"] = "buffer"; option["long"] = "buffer";
option["short"] = "b"; option["short"] = "b";
option["help"] = "Buffertime for this stream."; option["help"] = "DVR buffer time in ms";
option["value"].append(30000LL); option["value"].append(30000LL);
config->addOption("bufferTime", option); config->addOption("bufferTime", option);
capa["optional"]["DVR"]["name"] = "Buffer time (ms)";
capa["desc"] = "Enables buffered live input"; 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.";
capa["optional"]["DVR"]["option"] = "--buffer";
capa["optional"]["DVR"]["type"] = "uint";
capa["optional"]["DVR"]["default"] = 30000LL;
capa["source_match"] = "push://*";
capa["priority"] = 9ll;
capa["desc"] = "Provides buffered live input";
capa["codecs"][0u][0u].append("*"); capa["codecs"][0u][0u].append("*");
capa["codecs"][0u][1u].append("*"); capa["codecs"][0u][1u].append("*");
capa["codecs"][0u][2u].append("*"); capa["codecs"][0u][2u].append("*");
capa["codecs"][0u][3u].append("*");
capa["codecs"][0u][4u].append("*");
capa["codecs"][0u][5u].append("*");
capa["codecs"][0u][6u].append("*");
capa["codecs"][0u][7u].append("*");
capa["codecs"][0u][8u].append("*");
capa["codecs"][0u][9u].append("*");
DEBUG_MSG(DLVL_DEVEL, "Started MistInBuffer");
isBuffer = true; isBuffer = true;
singleton = this; singleton = this;
bufferTime = 0; bufferTime = 0;
@ -41,6 +40,15 @@ namespace Mist {
} }
inputBuffer::~inputBuffer(){
if (myMeta.tracks.size()){
DEBUG_MSG(DLVL_DEVEL, "Cleaning up, removing last keyframes");
for(std::map<int,DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
while (removeKey(it->first)){}
}
}
}
void inputBuffer::updateMeta(){ void inputBuffer::updateMeta(){
long long unsigned int firstms = 0xFFFFFFFFFFFFFFFFull; long long unsigned int firstms = 0xFFFFFFFFFFFFFFFFull;
long long unsigned int lastms = 0; long long unsigned int lastms = 0;
@ -55,15 +63,18 @@ namespace Mist {
myMeta.bufferWindow = lastms - firstms; myMeta.bufferWindow = lastms - firstms;
myMeta.vod = false; myMeta.vod = false;
myMeta.live = true; myMeta.live = true;
myMeta.writeTo(metaPage.mapped);
IPC::semaphore liveMeta(std::string("liveMeta@" + config->getString("streamname")).c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1); IPC::semaphore liveMeta(std::string("liveMeta@" + config->getString("streamname")).c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1);
liveMeta.wait(); liveMeta.wait();
myMeta.writeTo(metaPage.mapped);
memset(metaPage.mapped+myMeta.getSendLen(), 0, metaPage.len > myMeta.getSendLen() ? std::min(metaPage.len-myMeta.getSendLen(), 4ll) : 0); memset(metaPage.mapped+myMeta.getSendLen(), 0, metaPage.len > myMeta.getSendLen() ? std::min(metaPage.len-myMeta.getSendLen(), 4ll) : 0);
liveMeta.post(); liveMeta.post();
} }
bool inputBuffer::removeKey(unsigned int tid){ bool inputBuffer::removeKey(unsigned int tid){
if (myMeta.tracks[tid].keys.size() < 2 || myMeta.tracks[tid].fragments.size() < 2){ if ((myMeta.tracks[tid].keys.size() < 2 || myMeta.tracks[tid].fragments.size() < 2) && config->is_active){
return false;
}
if (!myMeta.tracks[tid].keys.size()){
return false; return false;
} }
DEBUG_MSG(DLVL_HIGH, "Erasing key %d:%d", tid, myMeta.tracks[tid].keys[0].getNumber()); DEBUG_MSG(DLVL_HIGH, "Erasing key %d:%d", tid, myMeta.tracks[tid].keys[0].getNumber());
@ -144,16 +155,19 @@ namespace Mist {
//Skip value 0 as this indicates an empty track //Skip value 0 as this indicates an empty track
continue; continue;
} }
if (counter == 126 || counter == 127 || counter == 254 || counter == 255){ if (pushedLoc[value] == thisData){
if (negotiateTracks.count(value)){ if (counter == 126 || counter == 127 || counter == 254 || counter == 255){
negotiateTracks.erase(value); pushedLoc.erase(value);
metaPages.erase(value); if (negotiateTracks.count(value)){
negotiateTracks.erase(value);
metaPages.erase(value);
}
if (data[4] == 0xFF && data[5] == 0xFF && givenTracks.count(value)){
givenTracks.erase(value);
inputLoc.erase(value);
}
continue;
} }
if (data[4] == 0xFF && data[5] == 0xFF && givenTracks.count(value)){
givenTracks.erase(value);
inputLoc.erase(value);
}
continue;
} }
if (value & 0x80000000){ if (value & 0x80000000){
//Track is set to "New track request", assign new track id and create shared memory page //Track is set to "New track request", assign new track id and create shared memory page
@ -226,6 +240,7 @@ namespace Mist {
} }
} }
givenTracks.insert(finalMap); givenTracks.insert(finalMap);
pushedLoc[finalMap] = thisData;
if (!myMeta.tracks.count(finalMap)){ if (!myMeta.tracks.count(finalMap)){
DEBUG_MSG(DLVL_HIGH, "Inserting metadata for track number %d", finalMap); DEBUG_MSG(DLVL_HIGH, "Inserting metadata for track number %d", finalMap);
myMeta.tracks[finalMap] = tmpMeta.tracks.begin()->second; myMeta.tracks[finalMap] = tmpMeta.tracks.begin()->second;
@ -249,7 +264,7 @@ namespace Mist {
} }
} }
} }
if (givenTracks.count(value)){ if (givenTracks.count(value) && pushedLoc[value] == thisData){
//First check if the previous page has been finished: //First check if the previous page has been finished:
if (!inputLoc[value].count(dataPages[value].rbegin()->first) || !inputLoc[value][dataPages[value].rbegin()->first].curOffset){ if (!inputLoc[value].count(dataPages[value].rbegin()->first) || !inputLoc[value][dataPages[value].rbegin()->first].curOffset){
if (dataPages[value].size() > 1){ if (dataPages[value].size() > 1){
@ -307,16 +322,21 @@ namespace Mist {
if (!bufferTime){ if (!bufferTime){
bufferTime = config->getInteger("bufferTime"); bufferTime = config->getInteger("bufferTime");
} }
JSON::Value servConf = JSON::fromFile(Util::getTmpFolder() + "streamlist");
if (servConf.isMember("streams") && servConf["streams"].isMember(config->getString("streamname"))){ IPC::sharedPage serverCfg("!mistConfig", 4*1024*1024); ///< Contains server configuration and capabilities
JSON::Value & streamConfig = servConf["streams"][config->getString("streamname")]; IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
if (streamConfig.isMember("DVR") && streamConfig["DVR"].asInt()){ configLock.wait();
if (bufferTime != streamConfig["DVR"].asInt()){ DTSC::Scan streamCfg = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("streams").getMember(config->getString("streamname"));
DEBUG_MSG(DLVL_DEVEL, "Setting bufferTime from %u to new value of %lli", bufferTime, streamConfig["DVR"].asInt()); if (streamCfg && streamCfg.getMember("DVR")){
bufferTime = streamConfig["DVR"].asInt(); long long bufTime = streamCfg.getMember("DVR").asInt();
} if (bufferTime != bufTime){
DEBUG_MSG(DLVL_DEVEL, "Setting bufferTime from %u to new value of %lli", bufferTime, bufTime);
bufferTime = bufTime;
} }
} }
configLock.post();
configLock.close();
return true; return true;
} }

2
src/input/input_buffer.h
View file

@ -6,6 +6,7 @@ namespace Mist {
class inputBuffer : public Input { class inputBuffer : public Input {
public: public:
inputBuffer(Util::Config * cfg); inputBuffer(Util::Config * cfg);
~inputBuffer();
private: private:
unsigned int bufferTime; unsigned int bufferTime;
unsigned int cutTime; unsigned int cutTime;
@ -26,6 +27,7 @@ namespace Mist {
std::map<unsigned long, IPC::sharedPage> metaPages; std::map<unsigned long, IPC::sharedPage> metaPages;
///Maps trackid to a pagenum->pageData map ///Maps trackid to a pagenum->pageData map
std::map<unsigned long, std::map<unsigned long, DTSCPageData> > inputLoc; std::map<unsigned long, std::map<unsigned long, DTSCPageData> > inputLoc;
std::map<unsigned long, char *> pushedLoc;
inputBuffer * singleton; inputBuffer * singleton;
}; };
} }

5
src/input/input_dtsc.cpp
View file

@ -11,7 +11,10 @@
namespace Mist { namespace Mist {
inputDTSC::inputDTSC(Util::Config * cfg) : Input(cfg) { inputDTSC::inputDTSC(Util::Config * cfg) : Input(cfg) {
capa["name"] = "DTSC";
capa["decs"] = "Enables DTSC Input"; capa["decs"] = "Enables DTSC Input";
capa["priority"] = 9ll;
capa["source_match"] = "/*.dtsc";
capa["codecs"][0u][0u].append("H264"); capa["codecs"][0u][0u].append("H264");
capa["codecs"][0u][0u].append("H263"); capa["codecs"][0u][0u].append("H263");
capa["codecs"][0u][0u].append("VP6"); capa["codecs"][0u][0u].append("VP6");
@ -26,7 +29,7 @@ namespace Mist {
std::cerr << "Input from stdin not yet supported" << std::endl; std::cerr << "Input from stdin not yet supported" << std::endl;
return false; return false;
} }
if (!config->getBool("player")){ if (!config->getString("streamname").size()){
if (config->getString("output") == "-") { if (config->getString("output") == "-") {
std::cerr << "Output to stdout not yet supported" << std::endl; std::cerr << "Output to stdout not yet supported" << std::endl;
return false; return false;

5
src/input/input_flv.cpp
View file

@ -13,7 +13,10 @@
namespace Mist { namespace Mist {
inputFLV::inputFLV(Util::Config * cfg) : Input(cfg) { inputFLV::inputFLV(Util::Config * cfg) : Input(cfg) {
capa["name"] = "FLV";
capa["decs"] = "Enables FLV Input"; capa["decs"] = "Enables FLV Input";
capa["source_match"] = "/*.flv";
capa["priority"] = 9ll;
capa["codecs"][0u][0u].append("H264"); capa["codecs"][0u][0u].append("H264");
capa["codecs"][0u][0u].append("H263"); capa["codecs"][0u][0u].append("H263");
capa["codecs"][0u][0u].append("VP6"); capa["codecs"][0u][0u].append("VP6");
@ -26,7 +29,7 @@ namespace Mist {
std::cerr << "Input from stdin not yet supported" << std::endl; std::cerr << "Input from stdin not yet supported" << std::endl;
return false; return false;
} }
if (!config->getBool("player")){ if (!config->getString("streamname").size()){
if (config->getString("output") == "-") { if (config->getString("output") == "-") {
std::cerr << "Output to stdout not yet supported" << std::endl; std::cerr << "Output to stdout not yet supported" << std::endl;
return false; return false;

3
src/input/input_ogg.cpp
View file

@ -14,6 +14,7 @@
namespace Mist { namespace Mist {
inputOGG::inputOGG(Util::Config * cfg) : Input(cfg) { inputOGG::inputOGG(Util::Config * cfg) : Input(cfg) {
capa["name"] = "OGG";
capa["decs"] = "Enables OGG Input"; capa["decs"] = "Enables OGG Input";
capa["codecs"][0u][0u].append("theora"); capa["codecs"][0u][0u].append("theora");
capa["codecs"][0u][1u].append("vorbis"); capa["codecs"][0u][1u].append("vorbis");
@ -24,7 +25,7 @@ namespace Mist {
std::cerr << "Input from stdin not yet supported" << std::endl; std::cerr << "Input from stdin not yet supported" << std::endl;
return false; return false;
} }
if (!config->getBool("player")){ if (!config->getString("streamname").size()){
if (config->getString("output") == "-") { if (config->getString("output") == "-") {
std::cerr << "Output to stdout not yet supported" << std::endl; std::cerr << "Output to stdout not yet supported" << std::endl;
return false; return false;

4
src/output/output.cpp
View file

@ -25,7 +25,7 @@ namespace Mist {
return ((long long int)timePoint[0] << 56) | ((long long int)timePoint[1] << 48) | ((long long int)timePoint[2] << 40) | ((long long int)timePoint[3] << 32) | ((long long int)timePoint[4] << 24) | ((long long int)timePoint[5] << 16) | ((long long int)timePoint[6] << 8) | timePoint[7]; return ((long long int)timePoint[0] << 56) | ((long long int)timePoint[1] << 48) | ((long long int)timePoint[2] << 40) | ((long long int)timePoint[3] << 32) | ((long long int)timePoint[4] << 24) | ((long long int)timePoint[5] << 16) | ((long long int)timePoint[6] << 8) | timePoint[7];
} }
void Output::init(Util::Config * cfg){ void Output::init(Util::Config * cfg){
capa["optional"]["debug"]["name"] = "debug"; capa["optional"]["debug"]["name"] = "debug";
capa["optional"]["debug"]["help"] = "The debug level at which messages need to be printed."; capa["optional"]["debug"]["help"] = "The debug level at which messages need to be printed.";
capa["optional"]["debug"]["option"] = "--debug"; capa["optional"]["debug"]["option"] = "--debug";
@ -226,7 +226,7 @@ namespace Mist {
if (streamIndex.mapped){ if (streamIndex.mapped){
return; return;
} }
if (!Util::Stream::getStream(streamName)){ if (!Util::startInput(streamName)){
DEBUG_MSG(DLVL_FAIL, "Opening stream disallowed - aborting initalization"); DEBUG_MSG(DLVL_FAIL, "Opening stream disallowed - aborting initalization");
onFail(); onFail();
return; return;

39
src/output/output_rtmp.cpp
View file

@ -2,6 +2,7 @@
#include <mist/http_parser.h> #include <mist/http_parser.h>
#include <mist/defines.h> #include <mist/defines.h>
#include <mist/stream.h> #include <mist/stream.h>
#include <sys/stat.h>
#include <cstring> #include <cstring>
#include <cstdlib> #include <cstdlib>
@ -446,30 +447,34 @@ namespace Mist {
if ((amfData.getContentP(0)->StrValue() == "publish")) { if ((amfData.getContentP(0)->StrValue() == "publish")) {
if (amfData.getContentP(3)) { if (amfData.getContentP(3)) {
streamName = amfData.getContentP(3)->StrValue(); streamName = amfData.getContentP(3)->StrValue();
Util::Stream::sanitizeName(streamName); Util::sanitizeName(streamName);
//pull the server configuration //pull the server configuration
JSON::Value servConf = JSON::fromFile(Util::getTmpFolder() + "streamlist"); IPC::sharedPage serverCfg("!mistConfig", 4*1024*1024); ///< Contains server configuration and capabilities
if (servConf.isMember("streams") && servConf["streams"].isMember(streamName)){ IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
JSON::Value & streamConfig = servConf["streams"][streamName]; configLock.wait();
if (!streamConfig.isMember("source") || streamConfig["source"].asStringRef().substr(0, 7) != "push://"){
DEBUG_MSG(DLVL_FAIL, "Push rejected - stream not a push-able stream. (%s != push://*)", streamConfig["source"].asStringRef().c_str()); DTSC::Scan streamCfg = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("streams").getMember(streamName);
if (streamCfg){
if (streamCfg.getMember("source").asString().substr(0, 7) != "push://"){
DEBUG_MSG(DLVL_FAIL, "Push rejected - stream not a push-able stream. (%s != push://*)", streamCfg.getMember("source").asString().c_str());
myConn.close(); myConn.close();
return; }else{
} std::string source = streamCfg.getMember("source").asString().substr(7);
std::string source = streamConfig["source"].asStringRef().substr(7); std::string IP = source.substr(0, source.find('@'));
std::string IP = source.substr(0, source.find('@')); if (IP != ""){
if (IP != ""){ if (!myConn.isAddress(IP)){
if (!myConn.isAddress(IP)){ DEBUG_MSG(DLVL_FAIL, "Push rejected - source host not whitelisted");
DEBUG_MSG(DLVL_FAIL, "Push rejected - source host not whitelisted"); myConn.close();
myConn.close(); }
return;
} }
} }
}else{ }else{
DEBUG_MSG(DLVL_FAIL, "Push rejected - stream not configured."); DEBUG_MSG(DLVL_FAIL, "Push rejected - stream not configured.");
myConn.close(); myConn.close();
return;
} }
configLock.post();
configLock.close();
if (!myConn){return;}//do not initialize if rejected
initialize(); initialize();
} }
//send a _result reply //send a _result reply
@ -513,8 +518,8 @@ namespace Mist {
//handle variables //handle variables
if (streamName.find('?') != std::string::npos){ if (streamName.find('?') != std::string::npos){
std::string tmpVars = streamName.substr(streamName.find('?') + 1); std::string tmpVars = streamName.substr(streamName.find('?') + 1);
Util::Stream::sanitizeName(streamName);
parseVars(tmpVars); parseVars(tmpVars);
Util::sanitizeName(streamName);
} }
initialize(); initialize();