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

Library code style update + some minor backports from Pro edition

Browse source
This commit is contained in:
Erik Zandvliet 2018-12-03 23:51:48 +01:00 • committed by Thulinma
parent 593b291e85
commit 2607113727
68 changed files with 4538 additions and 4665 deletions
Download patch file Download diff file Expand all files Collapse all files

212
lib/stream.cpp
View file

@ -1,44 +1,39 @@
/// \file stream.cpp
/// Utilities for handling streams.
#include "stream.h"
#include "config.h"
#include "defines.h"
#include "dtsc.h"
#include "json.h"
#include "procs.h"
#include "shared_memory.h"
#include "socket.h"
#include <semaphore.h>
#include <stdlib.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <stdlib.h>
#include <semaphore.h>
#include "json.h"
#include "stream.h"
#include "procs.h"
#include "config.h"
#include "socket.h"
#include "defines.h"
#include "shared_memory.h"
#include "dtsc.h"
std::string Util::getTmpFolder() {
std::string Util::getTmpFolder(){
std::string dir;
char * tmp_char = 0;
if (!tmp_char) {
tmp_char = getenv("TMP");
}
if (!tmp_char) {
tmp_char = getenv("TEMP");
}
if (!tmp_char) {
tmp_char = getenv("TMPDIR");
}
if (tmp_char) {
char *tmp_char = 0;
if (!tmp_char){tmp_char = getenv("TMP");}
if (!tmp_char){tmp_char = getenv("TEMP");}
if (!tmp_char){tmp_char = getenv("TMPDIR");}
if (tmp_char){
dir = tmp_char;
dir += "/mist";
} else {
}else{
#if defined(_WIN32) || defined(_CYGWIN_)
dir = "C:/tmp/mist";
#else
dir = "/tmp/mist";
#endif
}
if (access(dir.c_str(), 0) != 0) {
mkdir(dir.c_str(), S_IRWXU | S_IRWXG | S_IRWXO); //attempt to create mist folder - ignore failures
if (access(dir.c_str(), 0) != 0){
mkdir(dir.c_str(),
S_IRWXU | S_IRWXG | S_IRWXO); // attempt to create mist folder - ignore failures
}
return dir + "/";
}
@ -47,36 +42,37 @@ std::string Util::getTmpFolder() {
/// letters to lowercase. If a '?' character is found, everything following
/// that character is deleted. The original string is modified. If a '+' or space
/// exists, then only the part before that is sanitized.
void Util::sanitizeName(std::string & streamname) {
//strip anything that isn't numbers, digits or underscores
void Util::sanitizeName(std::string &streamname){
// strip anything that isn't numbers, digits or underscores
size_t index = streamname.find_first_of("+ ");
if(index != std::string::npos){
std::string preplus = streamname.substr(0,index);
if (index != std::string::npos){
std::string preplus = streamname.substr(0, index);
sanitizeName(preplus);
std::string postplus = streamname.substr(index+1);
std::string postplus = streamname.substr(index + 1);
if (postplus.find('?') != std::string::npos){
postplus = postplus.substr(0, (postplus.find('?')));
}
streamname = preplus+"+"+postplus;
streamname = preplus + "+" + postplus;
return;
}
for (std::string::iterator i = streamname.end() - 1; i >= streamname.begin(); --i) {
if (*i == '?') {
for (std::string::iterator i = streamname.end() - 1; i >= streamname.begin(); --i){
if (*i == '?'){
streamname.erase(i, streamname.end());
break;
}
if ( !isalpha( *i) && !isdigit( *i) && *i != '_' && *i != '.'){
if (!isalpha(*i) && !isdigit(*i) && *i != '_' && *i != '.'){
streamname.erase(i);
} else {
}else{
*i = tolower(*i);
}
}
}
JSON::Value Util::getStreamConfig(const std::string & streamname){
JSON::Value Util::getStreamConfig(const std::string &streamname){
JSON::Value result;
if (streamname.size() > 100){
FAIL_MSG("Stream opening denied: %s is longer than 100 characters (%lu).", streamname.c_str(), streamname.size());
FAIL_MSG("Stream opening denied: %s is longer than 100 characters (%lu).", streamname.c_str(),
streamname.size());
return result;
}
std::string smp = streamname.substr(0, streamname.find_first_of("+ "));
@ -92,7 +88,7 @@ JSON::Value Util::getStreamConfig(const std::string & streamname){
return stream_cfg.asJSON();
}
DTSC::Meta Util::getStreamMeta(const std::string & streamname){
DTSC::Meta Util::getStreamMeta(const std::string &streamname){
DTSC::Meta ret;
char pageId[NAME_BUFFER_SIZE];
snprintf(pageId, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, streamname.c_str());
@ -108,12 +104,12 @@ DTSC::Meta Util::getStreamMeta(const std::string & streamname){
/// Checks if the given streamname has an active input serving it. Returns true if this is the case.
/// Assumes the streamname has already been through sanitizeName()!
bool Util::streamAlive(std::string & streamname){
bool Util::streamAlive(std::string &streamname){
char semName[NAME_BUFFER_SIZE];
snprintf(semName, NAME_BUFFER_SIZE, SEM_INPUT, streamname.c_str());
IPC::semaphore playerLock(semName, O_RDWR, ACCESSPERMS, 1, true);
if (!playerLock){return false;}
if (!playerLock.tryWait()) {
if (!playerLock.tryWait()){
playerLock.close();
return true;
}else{
@ -126,120 +122,118 @@ bool Util::streamAlive(std::string & streamname){
/// Assures the input for the given stream name is active.
/// 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, return true.
/// 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 isProvider, const std::map<std::string, std::string> & overrides, pid_t * spawn_pid ) {
/// 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 isProvider,
const std::map<std::string, std::string> &overrides, pid_t *spawn_pid){
sanitizeName(streamname);
if (streamname.size() > 100){
FAIL_MSG("Stream opening denied: %s is longer than 100 characters (%lu).", streamname.c_str(), streamname.size());
FAIL_MSG("Stream opening denied: %s is longer than 100 characters (%lu).", streamname.c_str(),
streamname.size());
return false;
}
//Check if the stream is already active.
//If yes, don't activate again to prevent duplicate inputs.
//It's still possible a duplicate starts anyway, this is caught in the inputs initializer.
//Note: this uses the _whole_ stream name, including + (if any).
//This means "test+a" and "test+b" have separate locks and do not interact with each other.
// Check if the stream is already active.
// If yes, don't activate again to prevent duplicate inputs.
// It's still possible a duplicate starts anyway, this is caught in the inputs initializer.
// Note: this uses the _whole_ stream name, including + (if any).
// This means "test+a" and "test+b" have separate locks and do not interact with each other.
uint8_t streamStat = getStreamStatus(streamname);
//Wait for a maximum of 240 x 250ms sleeps = 60 seconds
// Wait for a maximum of 240 x 250ms sleeps = 60 seconds
size_t sleeps = 0;
while (++sleeps < 240 && streamStat != STRMSTAT_OFF && streamStat != STRMSTAT_READY && (!isProvider || streamStat != STRMSTAT_WAIT)){
if (streamStat == STRMSTAT_BOOT && overrides.count("throughboot")){
break;
}
while (++sleeps < 240 && streamStat != STRMSTAT_OFF && streamStat != STRMSTAT_READY &&
(!isProvider || streamStat != STRMSTAT_WAIT)){
if (streamStat == STRMSTAT_BOOT && overrides.count("throughboot")){break;}
Util::sleep(250);
streamStat = getStreamStatus(streamname);
}
if (streamAlive(streamname) && !overrides.count("alwaysStart")){
DEBUG_MSG(DLVL_MEDIUM, "Stream %s already active; continuing", streamname.c_str());
MEDIUM_MSG("Stream %s already active; continuing", streamname.c_str());
return true;
}
//Find stream base name
// Find stream base name
std::string smp = streamname.substr(0, streamname.find_first_of("+ "));
//check if base name (everything before + or space) exists
// check if base name (everything before + or space) exists
const JSON::Value stream_cfg = getStreamConfig(streamname);
if (!stream_cfg){
DEBUG_MSG(DLVL_HIGH, "Stream %s not configured - attempting to ignore", streamname.c_str());
HIGH_MSG("Stream %s not configured - attempting to ignore", streamname.c_str());
}
//Only use configured source if not manually overridden. Abort if no config is available.
// Only use configured source if not manually overridden. Abort if no config is available.
if (!filename.size()){
if (!stream_cfg){
DEBUG_MSG(DLVL_MEDIUM, "Stream %s not configured, no source manually given, cannot start", streamname.c_str());
MEDIUM_MSG("Stream %s not configured, no source manually given, cannot start", streamname.c_str());
return false;
}
filename = stream_cfg["source"].asStringRef();
}
const JSON::Value input = getInputBySource(filename, isProvider);
if (!input){return false;}
//copy the necessary arguments to separate storage so we can unlock the config semaphore safely
// copy the necessary arguments to separate storage so we can unlock the config semaphore safely
std::map<std::string, std::string> str_args;
//check required parameters
// check required parameters
if (input.isMember("required")){
jsonForEachConst(input["required"], prm){
const std::string opt = (*prm)["option"].asStringRef();
//check for overrides
// check for overrides
if (overrides.count(opt)){
str_args[opt] = overrides.at(opt);
}else{
if (!stream_cfg.isMember(prm.key())){
FAIL_MSG("Required parameter %s for stream %s missing", prm.key().c_str(), streamname.c_str());
FAIL_MSG("Required parameter %s for stream %s missing", prm.key().c_str(),
streamname.c_str());
return false;
}
str_args[opt] = stream_cfg[opt].asStringRef();
}
}
}
//check optional parameters
// check optional parameters
if (input.isMember("optional")){
jsonForEachConst(input["optional"], prm){
const std::string opt = (*prm)["option"].asStringRef();
//check for overrides
// check for overrides
if (overrides.count(opt)){
str_args[opt] = overrides.at(opt);
}else{
if (stream_cfg.isMember(prm.key())){
str_args[opt] = stream_cfg[prm.key()].asStringRef();
}
}
if (!prm->isMember("type") && str_args.count(opt)){
str_args[opt] = "";
if (stream_cfg.isMember(prm.key())){str_args[opt] = stream_cfg[prm.key()].asStringRef();}
}
if (!prm->isMember("type") && str_args.count(opt)){str_args[opt] = "";}
}
}
if (isProvider){
//Set environment variable so we can know if we have a provider when re-exec'ing.
// Set environment variable so we can know if we have a provider when re-exec'ing.
setenv("MISTPROVIDER", "1", 1);
}
std::string player_bin = Util::getMyPath() + "MistIn" + input["name"].asStringRef();
INFO_MSG("Starting %s -s %s %s", player_bin.c_str(), streamname.c_str(), filename.c_str());
char * argv[30] = {(char *)player_bin.c_str(), (char *)"-s", (char *)streamname.c_str(), (char *)filename.c_str()};
char *argv[30] ={(char *)player_bin.c_str(), (char *)"-s", (char *)streamname.c_str(),
(char *)filename.c_str()};
int argNum = 3;
std::string debugLvl;
if (Util::Config::printDebugLevel != DEBUG && !str_args.count("--debug")){
debugLvl = JSON::Value((long long)Util::Config::printDebugLevel).asString();
debugLvl = JSON::Value(Util::Config::printDebugLevel).asString();
argv[++argNum] = (char *)"--debug";
argv[++argNum] = (char *)debugLvl.c_str();
}
for (std::map<std::string, std::string>::iterator it = str_args.begin(); it != str_args.end(); ++it){
for (std::map<std::string, std::string>::iterator it = str_args.begin(); it != str_args.end();
++it){
argv[++argNum] = (char *)it->first.c_str();
if (it->second.size()){
argv[++argNum] = (char *)it->second.c_str();
}
if (it->second.size()){argv[++argNum] = (char *)it->second.c_str();}
}
argv[++argNum] = (char *)0;
Util::Procs::setHandler();
int pid = 0;
if (forkFirst){
DEBUG_MSG(DLVL_DONTEVEN, "Forking");
DONTEVEN_MSG("Forking");
pid = fork();
if (pid == -1) {
if (pid == -1){
FAIL_MSG("Forking process for stream %s failed: %s", streamname.c_str(), strerror(errno));
return false;
}
@ -248,15 +242,16 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
Util::Procs::remember(pid);
}
}else{
DEBUG_MSG(DLVL_DONTEVEN, "Not forking");
DONTEVEN_MSG("Not forking");
}
if (pid == 0){
Socket::Connection io(0, 1);
io.drop();
DEBUG_MSG(DLVL_DONTEVEN, "execvp");
DONTEVEN_MSG("execvp");
execvp(argv[0], argv);
FAIL_MSG("Starting process %s for stream %s failed: %s", argv[0], streamname.c_str(), strerror(errno));
FAIL_MSG("Starting process %s for stream %s failed: %s", argv[0], streamname.c_str(),
strerror(errno));
_exit(42);
}else if (spawn_pid != NULL){
*spawn_pid = pid;
@ -277,16 +272,16 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
JSON::Value Util::getInputBySource(const std::string &filename, bool isProvider){
JSON::Value ret;
//Attempt to load up configuration and find this stream
// Attempt to load up configuration and find this stream
Util::DTSCShmReader rCapa(SHM_CAPA);
DTSC::Scan inputs = rCapa.getMember("inputs");
//Abort if not available
// Abort if not available
if (!inputs){
FAIL_MSG("Capabilities not available, aborting! Is MistController running?");
return false;
}
//check in curConf for <naam>-priority/source_match
// check in curConf for <naam>-priority/source_match
bool selected = false;
long long int curPrio = -1;
DTSC::Scan input;
@ -294,17 +289,19 @@ JSON::Value Util::getInputBySource(const std::string &filename, bool isProvider)
bool noProviderNoPick = false;
for (unsigned int i = 0; i < input_size; ++i){
DTSC::Scan tmp_input = inputs.getIndice(i);
//if match voor current stream && priority is hoger dan wat we al hebben
// if match voor current stream && priority is hoger dan wat we al hebben
if (tmp_input.getMember("source_match") && curPrio < tmp_input.getMember("priority").asInt()){
if (tmp_input.getMember("source_match").getSize()){
for(unsigned int j = 0; j < tmp_input.getMember("source_match").getSize(); ++j){
for (unsigned int j = 0; j < tmp_input.getMember("source_match").getSize(); ++j){
std::string source = tmp_input.getMember("source_match").getIndice(j).asString();
std::string front = source.substr(0,source.find('*'));
std::string back = source.substr(source.find('*')+1);
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){
std::string front = source.substr(0, source.find('*'));
std::string back = source.substr(source.find('*') + 1);
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;
@ -316,11 +313,13 @@ JSON::Value Util::getInputBySource(const std::string &filename, bool isProvider)
}
}else{
std::string source = tmp_input.getMember("source_match").asString();
std::string front = source.substr(0,source.find('*'));
std::string back = source.substr(source.find('*')+1);
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){
std::string front = source.substr(0, source.find('*'));
std::string back = source.substr(source.find('*') + 1);
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;
@ -330,7 +329,6 @@ JSON::Value Util::getInputBySource(const std::string &filename, bool isProvider)
input = tmp_input;
}
}
}
}
if (!selected){
@ -345,7 +343,7 @@ JSON::Value Util::getInputBySource(const std::string &filename, bool isProvider)
return ret;
}
uint8_t Util::getStreamStatus(const std::string & streamname){
uint8_t Util::getStreamStatus(const std::string &streamname){
char pageName[NAME_BUFFER_SIZE];
snprintf(pageName, NAME_BUFFER_SIZE, SHM_STREAM_STATE, streamname.c_str());
IPC::sharedPage streamStatus(pageName, 1, false, false);