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

Implemented triggers.

Browse source 
Merged from code by Wouter Spruit, with additions by yours truly.
This commit is contained in:
Thulinma 2015-10-07 13:23:27 +02:00
parent eb6b98b219
commit 279add438a
18 changed files with 597 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
CMakeLists.txt
View file

@ -150,6 +150,7 @@ set(libHeaders
${SOURCE_DIR}/lib/ts_packet.h
${SOURCE_DIR}/lib/ts_stream.h
${SOURCE_DIR}/lib/vorbis.h
${SOURCE_DIR}/lib/triggers.h
)
########################################
@ -191,6 +192,7 @@ set(libSources
${SOURCE_DIR}/lib/ts_packet.cpp
${SOURCE_DIR}/lib/ts_stream.cpp
${SOURCE_DIR}/lib/vorbis.cpp
${SOURCE_DIR}/lib/triggers.cpp
)
########################################
# MistLib - Build #

2
Doxyfile.in
View file

@ -228,7 +228,7 @@ TAB_SIZE = 2
# "Side Effects:". You can put \n's in the value part of an alias to insert
# newlines.
ALIASES = "api=\xrefitem api \"API call\" \"API calls\""
ALIASES = "api=\xrefitem api \"API call\" \"API calls\"" "triggers=\xrefitem triggers \"Trigger\" \"Triggers\""
# This tag can be used to specify a number of word-keyword mappings (TCL only).
# A mapping has the form "name=value". For example adding "class=itcl::class"

1
lib/defines.h
View file

@ -75,6 +75,7 @@ static const char * DBG_LVL_LIST[] = {"NONE", "FAIL", "ERROR", "WARN", "INFO", "
#define SHM_TRACK_DATA "MstDATA%s@%lu_%lu" //%s stream name, %lu track ID, %lu page #
#define SHM_STATISTICS "MstSTAT"
#define SHM_USERS "MstUSER%s" //%s stream name
#define SHM_TRIGGER "MstTRIG%s" //%s trigger name
#define SEM_LIVE "MstLIVE%s" //%s stream name
#define NAME_BUFFER_SIZE 200 //char buffer size for snprintf'ing shm filenames

20
lib/stream.cpp
View file

@ -14,6 +14,7 @@
#include "defines.h"
#include "shared_memory.h"
#include "dtsc.h"
#include "triggers.h"//LTS
std::string Util::getTmpFolder() {
std::string dir;
@ -116,6 +117,17 @@ bool Util::streamAlive(std::string & streamname){
/// Then, checks if an input is already active by running streamAlive(). If yes, aborts.
/// If no, loads up the server configuration and attempts to start the given stream according to current config.
/// At this point, fails and aborts if MistController isn't running.
/// \triggers
/// The `"STREAM_LOAD"` trigger is stream-specific, and is ran right before launching an input for an inactive stream. If cancelled, the input is not launched. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// ~~~~~~~~~~~~~~~
/// The `"STREAM_SOURCE"` trigger is stream-specific, and is ran right before launching an input for an inactive stream. It cannot be cancelled, but an invalid source can be returned; which is effectively equivalent to cancelling.
/// This trigger is special: the response is used as source override for this stream, and not handled as normal. If used, the handler for this trigger MUST return a valid source to allow the stream input to load up at all. If used multiple times, the last defined handler overrides any and all previous handlers.
/// Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// ~~~~~~~~~~~~~~~
bool Util::startInput(std::string streamname, std::string filename, bool forkFirst) {
sanitizeName(streamname);
if (streamname.size() > 100){
@ -160,6 +172,14 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
if (stream_cfg && stream_cfg.getMember("hardlimit_active")) {
return false;
}
if(Triggers::shouldTrigger("STREAM_LOAD", smp)){
if (!Triggers::doTrigger("STREAM_LOAD", streamname, smp)){
return false;
}
}
if(Triggers::shouldTrigger("STREAM_SOURCE", smp)){
Triggers::doTrigger("STREAM_SOURCE", streamname, smp, false, filename);
}
/*LTS-END*/

242
lib/triggers.cpp Normal file
View file

@ -0,0 +1,242 @@
/// \page triggers Triggers
/// \brief Listing of all available triggers and their payloads.
/// MistServer reports certain occurances as configurable triggers to a URL or executable. This page describes the triggers system in full.
///
/// Triggers are the preferred way of responding to server events. Each trigger has a name and a payload, and may be stream-specific or global.
///
/// Triggers may be handled by a URL or an executable. If the handler contains ://, a HTTP URL is assumed. Otherwise, an executable is assumed.
/// If handled by an URL, a POST request is sent to the URL with an extra X-Trigger header containing the trigger name and the payload as the POST body.
/// If handled by an executable, it's started with the trigger name as its only argument, and the payload is piped into the executable over standard input.
///
/// Currently, all triggers are handled asynchronously and responses (if any) are completely ignored. In the future this may change.
///
#include <string.h>//for strncmp
#include "triggers.h"
#include "http_parser.h"//for sending http request
#include "defines.h" //for FAIL_MSG and INFO_MSG
#include "procs.h" //for StartPiped
#include "shared_memory.h"
#include "bitfields.h" //for strToBool
namespace Triggers{
///\brief Handles a trigger by sending a payload to a destination.
///\param trigger Trigger event type.
///\param value Destination. This can be an (HTTP)URL, or an absolute path to a binary/script
///\param payload This data will be sent to the destionation URL/program
///\param sync If true, handler is executed blocking and uses the response data.
///\returns String, false if further processing should be aborted.
std::string handleTrigger(const std::string &trigger, const std::string &value, const std::string &payload, int sync){
if(!value.size()){
WARN_MSG("Trigger requested with empty destination");
return "true";
}
INFO_MSG("Executing %s trigger: %s (%s)", trigger.c_str(), value.c_str(), sync ? "blocking" : "asynchronous");
if (value.substr(0, 7) == "http://"){ //interpret as url
std::string url = value.substr(value.find("://") + 3); //contains server+url
std::string server = url.substr(0, url.find('/'));
int port=80;
if (server.find(':') != std::string::npos){
port = atoi(server.data() + server.find(':') + 1);
server.erase(server.find(':'));
}
url = url.substr(url.find('/'));
Socket::Connection conn(server,port,false);
HTTP::Parser H;
H.url = url;
H.method = "POST";
H.SetHeader("Host", server + ":" + JSON::Value((long long)port).toString());
H.SetHeader("Content-Type", "application/x-www-form-urlencoded");
H.SetHeader("X-Trigger", trigger);
H.SetBody(payload);
H.SendRequest(conn);
H.Clean();
if(sync){ //if sync!=0 wait for response
while (conn && (!conn.spool() || !H.Read(conn))) {}
conn.close();
/// \todo Handle errors!
return H.body;
}else{
conn.close();
return "true";
}
} else { //send payload to stdin of newly forked process
int fdIn=-1;
int fdOut=-1;
int fdErr=-1;
char * argv[3];
argv[0]=(char*)value.c_str();
argv[1]=(char*)trigger.c_str();
argv[2]=NULL;
pid_t myProc = Util::Procs::StartPiped(argv, &fdIn,&fdOut,&fdErr); //start new process and return stdin file desc.
if ( fdIn == -1 || fdOut == -1 || fdErr == -1 ){ //verify fdIn
FAIL_MSG("StartPiped returned invalid fd");
return "true";/// \todo Return true/false based on config here.
}
write(fdIn, payload.data(), payload.size());
shutdown(fdIn, SHUT_RDWR);
close(fdIn);
if(sync){ //if sync!=0 wait for response
while (Util::Procs::isActive(myProc)) {
Util::sleep(100);
}
std::string ret;
FILE * outFile = fdopen(fdOut, "r");
char * fileBuf = 0;
size_t fileBufLen = 0;
while (!(feof(outFile) || ferror(outFile)) && (getline(&fileBuf, &fileBufLen, outFile) != -1)) {
ret += fileBuf;
}
fclose(outFile);
free(fileBuf);
close(fdOut);
close(fdErr);
return ret;
}
close(fdOut);
close(fdErr);
return "true";
}
}
static std::string empty;
///\brief Checks if one or more triggers are defined that should be handled for all streams (for a trigger event type)
///\param type Trigger event type.
///\return returns true, if so
///calls doTrigger with dryRun set to true
bool shouldTrigger(const std::string type){ //returns true if a trigger of the specified type should be handled for all streams
static std::string empty;
empty.clear();
return doTrigger(type, empty, empty, true, empty);
}
///\brief returns true if a trigger of the specified type should be handled for a specified stream (, or entire server)
///\param type Trigger event type.
///\param streamName the stream to be handled
///\return returns true if so
///calls doTrigger with dryRun set to true
bool shouldTrigger(const std::string type, const std::string &streamName){ //returns true if a trigger of the specified type should be handled for a specified stream (, or entire server)
empty.clear();
return doTrigger(type, empty, streamName, true, empty);
}
///\brief handles triggers for a specific trigger event type, without a payload, server-wide
///\param type Trigger event type.
///\returns Boolean, false if further processing should be aborted.
///calls doTrigger with dryRun set to false
bool doTrigger(const std::string type){
empty.clear();
return doTrigger(type, empty, empty, false, empty);
}
///\brief handles triggers for a specific trigger event type, with a payload, server-wide
///\param type Trigger event type.
///\param payload Trigger type-specific data
///\returns Boolean, false if further processing should be aborted.
///calls doTrigger with dryRun set to false
bool doTrigger(const std::string type, const std::string &payload){
empty.clear();
return doTrigger(type, payload, empty, false, empty);
}
///\brief handles triggers for a specific trigger event type, with a payload, for a specified stream, and/or server-wide
///\param type Trigger event type.
///\param payload Trigger type-specific data
///\param streamName The name of a stream.
///\returns Boolean, false if further processing should be aborted.
///calls doTrigger with dryRun set to false
bool doTrigger(const std::string type, const std::string &payload, const std::string &streamName){
empty.clear();
return doTrigger(type, payload, streamName, false, empty);
}
///\brief
///\param type Trigger event type
///\param payload Trigger type-specific data
///\param streamName Name of a stream to check for stream-specific triggers
///\param dryRun determines the mode of operation for this function
///\param response Returns the last received response by reference
///\returns Boolean, false if further processing should be aborted
///This function attempts to open and parse a shared memory page with the config for a trigger event type, in order to parse the triggers defined for that trigger event type.
///The function can be used for two separate purposes, determined by the value of dryRun
///-if this function is called with dryRun==true (for example, from a handleTrigger function), the return value will be true, if at least one trigger should be handled for the requested type/stream.
///this can be used to make sure a payload is only generated if at least one trigger should be handled.
///-if this function is called with dryRun==false (for example, from one of the overloaded doTrigger functions), handleTrigger is called for all configured triggers. In that case, the return value does not matter, it will probably be false in all cases.
bool doTrigger(const std::string type, const std::string &payload, const std::string &streamName, bool dryRun, std::string & response){
//open SHM page for this type:
char thisPageName[NAME_BUFFER_SIZE];
snprintf(thisPageName, NAME_BUFFER_SIZE, SHM_TRIGGER, type.c_str());
IPC::sharedPage typePage(thisPageName, 8*1024, false, false);
if(!typePage.mapped){ //page doesn't exist?
HIGH_MSG("No triggers for %s defined: list does not exist", type.c_str());
return false;
}
char* bytepos = typePage.mapped; //not checking page size. will probably be fine.
char* startBytepos=bytepos;
unsigned int totalLen = ((unsigned int *)bytepos)[0];
bool retVal = true;
VERYHIGH_MSG("Parsing %lu bytes of triggers for %s, stream: %s", totalLen, type.c_str(), streamName.c_str());
std::string uri;
unsigned int sync=0;
while( totalLen != 0 && bytepos < typePage.mapped + typePage.len ){
unsigned int uriLen = ((unsigned int *)bytepos)[1];
bytepos+=4+4;
uri=std::string(bytepos,uriLen);
bytepos+=uriLen;
sync=bytepos[0];
bytepos++;
bool isHandled = false;
if(totalLen>((unsigned int)(bytepos-startBytepos))){
while( totalLen>((unsigned int)(bytepos-startBytepos)) ){
unsigned int stringLen=((unsigned int *)bytepos)[0];
bytepos+=4;
if (strncmp(bytepos, streamName.c_str(), stringLen) == 0){
isHandled = true;
}
bytepos+=stringLen;
}
if (!streamName.size()){
isHandled = true;
}
} else if(totalLen==((unsigned int)(bytepos-startBytepos))){
//no streams explicitly defined for this trigger, return true for all streams.
isHandled = true;
}
if (isHandled){
VERYHIGH_MSG("%s trigger handled by %s", type.c_str(), uri.c_str());
if(dryRun){
return true;
}
response = handleTrigger(type,uri,payload,sync); //do it.
retVal &= Util::stringToBool(response);
}
if(totalLen!=((unsigned int)(bytepos-startBytepos))){ //if this is not the case, something bad might have happened.
ERROR_MSG("Error in %s trigger, totalLen: %d current position from startBytepos: %d", type.c_str(),totalLen, (unsigned int)(bytepos-startBytepos));
break; //stop reading hypothetical garbage
}
startBytepos=startBytepos+totalLen; //init next iteration
bytepos=startBytepos;
totalLen = ((unsigned int *)bytepos)[0]; //read next size
}
if (dryRun){
return false;
}else{
return retVal;
}
}
} //end namespace Controller

14
lib/triggers.h Normal file
View file

@ -0,0 +1,14 @@
#pragma once
#include <string>
namespace Triggers{
bool doTrigger(const std::string triggerType, const std::string &payload, const std::string &streamName, bool dryRun, std::string & response);
std::string handleTrigger(const std::string &triggerType, const std::string &value, const std::string &payload, int sync);
//All of the below are just shorthands for specific usage of the doTrigger function above:
bool shouldTrigger(const std::string triggerType);
bool shouldTrigger(const std::string triggerType, const std::string &streamName);
bool doTrigger(const std::string triggerType);
bool doTrigger(const std::string triggerType, const std::string &payload);
bool doTrigger(const std::string triggerType, const std::string &payload, const std::string &streamName);
}

1
lsp/mist.js
View file

@ -3545,6 +3545,7 @@ var UI = {
['STREAM_LOAD', 'STREAM_LOAD: right before stream input is loaded in memory'],
['STREAM_READY', 'STREAM_READY: when the stream input is loaded and ready for playback'],
['STREAM_UNLOAD', 'STREAM_UNLOAD: right before the stream input is removed from memory'],
['STREAM_PUSH', 'STREAM_PUSH: right before an incoming push is accepted'],
['STREAM_TRACK_ADD', 'STREAM_TRACK_ADD: right before a track will be added to a stream; e.g.: additional push received'],
['STREAM_TRACK_REMOVE', 'STREAM_TRACK_REMOVE: right before a track will be removed track from a stream; e.g.: push timeout'],
['CONN_OPEN', 'CONN_OPEN: right after a new incoming connection has been received'],

30
src/controller/controller.cpp
View file

@ -1,4 +1,5 @@
/// \page api API calls
/// \brief Listing of all controller API calls.
/// The controller listens for commands through a JSON-based API. This page describes the API in full.
///
/// A default interface implementing this API as a single HTML page is included in the controller itself. This default interface will be send for invalid API requests, and is thus triggered by default when a browser attempts to access the API port directly.
@ -20,7 +21,9 @@
///
/// You may also include a `"callback"` or `"jsonp"` HTTP variable, to trigger JSONP compatibility mode. JSONP is useful for getting around the cross-domain scripting protection in most modern browsers. Developers creating non-JavaScript applications will most likely not want to use JSONP mode, though nothing is stopping you if you really want to.
///
/// \brief Listing of all controller API calls.
/// \file controller.cpp
/// Contains all code for the controller executable.
@ -47,6 +50,7 @@
#include "controller_connectors.h"
#include "controller_statistics.h"
/*LTS-START*/
#include <mist/triggers.h>
#include "controller_updater.h"
#include "controller_limits.h"
#include "controller_uplink.h"
@ -132,6 +136,12 @@ void statusMonitor(void * np){
}
///\brief The main entry point for the controller.
///
/// \triggers
/// The `"SYSTEM_STOP"` trigger is global, and is ran when the controller shuts down. If cancelled, the controller does not shut down and will attempt to re-open the API socket. Its payload is:
/// ~~~~~~~~~~~~~~~
/// shutdown reason
/// ~~~~~~~~~~~~~~~
int main(int argc, char ** argv){
Controller::Storage = JSON::fromFile("config.json");
@ -214,6 +224,7 @@ int main(int argc, char ** argv){
if (Controller::Storage["config"]["controller"]["username"]){
Controller::conf.getOption("username", true)[0u] = Controller::Storage["config"]["controller"]["username"];
}
Controller::writeConfig();
Controller::checkAvailProtocols();
createAccount(Controller::conf.getString("account"));
@ -299,6 +310,7 @@ int main(int argc, char ** argv){
tthread::thread uplinkThread(Controller::uplinkConnection, 0);/*LTS*/
//start main loop
while (Controller::conf.is_active){/*LTS*/
Controller::conf.serveThreadedSocket(Controller::handleAPIConnection);
//print shutdown reason
std::string shutdown_reason;
@ -311,9 +323,21 @@ int main(int argc, char ** argv){
if (Controller::restarting){
shutdown_reason = "update (on request)";
}
if(Triggers::shouldTrigger("SYSTEM_STOP")){
if (!Triggers::doTrigger("SYSTEM_STOP", shutdown_reason)){
Controller::conf.is_active = true;
Controller::restarting = false;
Util::sleep(1000);
}else{
Controller::conf.is_active = false;
Controller::Log("CONF", "Controller shutting down because of "+shutdown_reason);
}
}else{
Controller::conf.is_active = false;
Controller::Log("CONF", "Controller shutting down because of "+shutdown_reason);
}
}//indentation intentionally wrong, to minimize Pro/nonPro diffs
/*LTS-END*/
Controller::Log("CONF", "Controller shutting down because of "+shutdown_reason);
Controller::conf.is_active = false;
//join all joinable threads
statsThread.join();
monitorThread.join();

13
src/controller/controller_connectors.cpp
View file

@ -10,6 +10,7 @@
#include <mist/defines.h>
#include "controller_storage.h"
#include "controller_connectors.h"
#include <mist/triggers.h>
#include <iostream>
#include <unistd.h>
@ -74,6 +75,16 @@ namespace Controller {
///\param p An object containing all protocols.
///\param capabilities An object containing the detected capabilities.
///\returns True if any action was taken
///
/// \triggers
/// The `"OUTPUT_START"` trigger is global, and is ran whenever a new protocol listener is started. It cannot be cancelled. Its payload is:
/// ~~~~~~~~~~~~~~~
/// output listener commandline
/// ~~~~~~~~~~~~~~~
/// The `"OUTPUT_STOP"` trigger is global, and is ran whenever a protocol listener is terminated. It cannot be cancelled. Its payload is:
/// ~~~~~~~~~~~~~~~
/// output listener commandline
/// ~~~~~~~~~~~~~~~
bool CheckProtocols(JSON::Value & p, const JSON::Value & capabilities){
std::set<std::string> runningConns;
@ -146,6 +157,7 @@ namespace Controller {
Log("CONF", "Stopping connector " + it->first);
action = true;
Util::Procs::Stop(it->second);
Triggers::doTrigger("OUTPUT_STOP",it->first); //LTS
}
currentConnectors.erase(it);
if (!currentConnectors.size()){
@ -168,6 +180,7 @@ namespace Controller {
buildPipedArguments(p, (char **)&argarr, capabilities);
// start piped w/ generated args
currentConnectors[*runningConns.begin()] = Util::Procs::StartPiped(argarr, &zero, &out, &err);
Triggers::doTrigger("OUTPUT_START", *runningConns.begin());//LTS
}
runningConns.erase(runningConns.begin());
}

84
src/controller/controller_storage.cpp
View file

@ -7,6 +7,7 @@
#include <mist/defines.h>
#include "controller_storage.h"
#include "controller_capabilities.h"
#include <mist/triggers.h>//LTS
///\brief Holds everything unique to the controller.
namespace Controller {
@ -75,6 +76,10 @@ namespace Controller {
}
/// Writes the current config to shared memory to be used in other processes
/// \triggers
/// The `"SYSTEM_START"` trigger is global, and is ran as soon as the server configuration is first stable. It has no payload. If cancelled, the system immediately shuts down again.
/// \n
/// The `"SYSTEM_CONFIG"` trigger is global, and is ran every time the server configuration is updated. Its payload is the new configuration in JSON format. This trigger cannot be cancelled.
void writeConfig(){
static JSON::Value writeConf;
bool changed = false;
@ -104,6 +109,83 @@ namespace Controller {
memcpy(mistConfOut.mapped, temp.data(), std::min(temp.size(), (size_t)mistConfOut.len));
//unlock semaphore
configLock.post();
}
/*LTS-START*/
static std::map<std::string,IPC::sharedPage> pageForType; //should contain one page for every trigger type
char tmpBuf[NAME_BUFFER_SIZE];
//for all shm pages that hold triggers
pageForType.clear();
if( writeConf["config"]["triggers"].size() ){//if triggers are defined...
jsonForEach(writeConf["config"]["triggers"], it){//for all types defined in config
snprintf(tmpBuf,NAME_BUFFER_SIZE,SHM_TRIGGER,(it.key()).c_str()); //create page
pageForType[it.key()].init(tmpBuf, 8*1024, true, false);// todo: should this be false/why??
char * bytePos=pageForType[it.key()].mapped;
//write data to page
jsonForEach(*it, triggIt){ //for all defined
unsigned int tmpUrlSize=(*triggIt)[(unsigned int) 0].asStringRef().size();
unsigned int tmpStreamNames=0;// (*triggIt)[2ul].packedSize();
std::string namesArray="";
if( (triggIt->size() >= 3) && (*triggIt)[2ul].size()){
jsonForEach((*triggIt)[2ul], shIt){
unsigned int tmpLen=shIt->asString().size();
tmpStreamNames+= 4+tmpLen;
//INFO_MSG("adding string: %s len: %d", shIt->asString().c_str() , tmpLen );
((unsigned int*)tmpBuf)[0] = tmpLen; //NOTE: namesArray may be replaced by writing directly to tmpBuf.
namesArray.append(tmpBuf,4);
namesArray.append(shIt->asString());
}
}
unsigned int totalLen=9+tmpUrlSize+tmpStreamNames; //4Btotal len, 4Burl len ,XB tmpurl, 1B sync , XB tmpstreamnames
if(totalLen > (pageForType[it.key()].len-(bytePos-pageForType[it.key()].mapped)) ){ //check if totalLen fits on the page
ERROR_MSG("trigger does not fit on page. size: %d bytes left on page: %d skipping.",totalLen,(pageForType[it.key()].len-(bytePos-pageForType[it.key()].mapped))); //doesnt fit
continue;
}
((unsigned int*)bytePos)[0] = totalLen;
bytePos+=4;
((unsigned int*)bytePos)[0] = tmpUrlSize;
bytePos+=4;
memcpy(bytePos, (*triggIt)[(unsigned int) 0].asStringRef().data(), (*triggIt)[(unsigned int) 0].asStringRef().size());
bytePos+=(*triggIt)[(unsigned int) 0].asStringRef().size();
(bytePos++)[0] = (*triggIt)[1ul].asBool() ? '\001' : '\000';
if(tmpStreamNames){
memcpy(bytePos,namesArray.data(),tmpStreamNames); //contains a string of 4Blen,XBstring pairs
bytePos+=tmpStreamNames;
}
}
}
}
static bool serverStartTriggered;
if(!serverStartTriggered){
if (!Triggers::doTrigger("SYSTEM_START")){
conf.is_active = false;
}
serverStartTriggered++;
}
if (Triggers::shouldTrigger("SYSTEM_CONFIG")){
std::string payload = writeConf.toString();
Triggers::doTrigger("SYSTEM_CONFIG", payload);
}
/*LTS-END*/
}
}
/*NOTES:
4B size (total size of entry 9B+XB(URL)+ 0..XB(nameArrayLen)) (if 0x00, stop reading)
4B url_len
XB url
1B async
for(number of strings)
4B stringLen
XB string
)
*/

42
src/controller/controller_streams.cpp
View file

@ -10,6 +10,7 @@
#include "controller_storage.h"
#include "controller_statistics.h"
#include "controller_limits.h" /*LTS*/
#include <mist/triggers.h> //LTS
#include <sys/stat.h>
#include <map>
@ -155,16 +156,45 @@ namespace Controller {
return false;
}
///
/// \triggers
/// The `"STREAM_ADD"` trigger is stream-specific, and is ran whenever a new stream is added to the server configuration. If cancelled, the stream is not added. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// configuration in JSON format
/// ~~~~~~~~~~~~~~~
/// The `"STREAM_CONFIG"` trigger is stream-specific, and is ran whenever a stream's configuration is changed. If cancelled, the configuration is not changed. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// configuration in JSON format
/// ~~~~~~~~~~~~~~~
///
void AddStreams(JSON::Value & in, JSON::Value & out){
//check for new streams and updates
jsonForEach(in, jit) {
if (out.isMember(jit.key())){
if ( !streamsEqual((*jit), out[jit.key()])){
/*LTS-START*/
if(Triggers::shouldTrigger("STREAM_CONFIG")){
std::string payload = jit.key()+"\n"+jit->toString();
if (!Triggers::doTrigger("STREAM_CONFIG", payload, jit.key())){
continue;
}
}
/*LTS-END*/
out[jit.key()] = (*jit);
out[jit.key()]["name"] = jit.key();
Log("STRM", std::string("Updated stream ") + jit.key());
}
}else{
/*LTS-START*/
if(Triggers::shouldTrigger("STREAM_ADD")){
std::string payload = jit.key()+"\n"+jit->toString();
if (!Triggers::doTrigger("STREAM_ADD", payload, jit.key())){
continue;
}
}
/*LTS-END*/
out[jit.key()] = (*jit);
out[jit.key()]["name"] = jit.key();
Log("STRM", std::string("New stream ") + jit.key());
@ -255,10 +285,22 @@ namespace Controller {
}
/// \triggers
/// The `"STREAM_REMOVE"` trigger is stream-specific, and is ran whenever a stream is removed from the server configuration. If cancelled, the stream is not removed. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// ~~~~~~~~~~~~~~~
void deleteStream(const std::string & name, JSON::Value & out) {
if (!out.isMember(name)){
return;
}
/*LTS-START*/
if(Triggers::shouldTrigger("STREAM_REMOVE")){
if (!Triggers::doTrigger("STREAM_REMOVE", name, name)){
return;
}
}
/*LTS-END*/
Log("STRM", std::string("Deleted stream ") + name);
out.removeMember(name);
if (inputProcesses.count(name)){

36
src/input/input.cpp
View file

@ -3,6 +3,7 @@
#include <sys/stat.h>
#include <mist/stream.h>
#include <mist/triggers.h>
#include <mist/defines.h>
#include "input.h"
#include <sstream>
@ -185,7 +186,21 @@ namespace Mist {
}
}
void Input::serve() {
/// The main loop for inputs in stream serving mode.
///
/// \triggers
/// The `"STREAM_READY"` trigger is stream-specific, and is ran whenever an input finished loading and started serving a stream. If cancelled, the input is immediately shut down again. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// input name
/// ~~~~~~~~~~~~~~~
/// The `"STREAM_UNLOAD"` trigger is stream-specific, and is ran right before an input shuts down and stops serving a stream. If cancelled, the shut down is delayed. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// input name
/// ~~~~~~~~~~~~~~~
//
void Input::serve(){
char userPageName[NAME_BUFFER_SIZE];
snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, streamName.c_str());
#ifdef INPUT_LIVE
@ -219,6 +234,14 @@ namespace Mist {
}
userClient.finish();
#else
/*LTS-START*/
if(Triggers::shouldTrigger("STREAM_READY", config->getString("streamname"))){
std::string payload = config->getString("streamname")+"\n" +capa["name"].asStringRef()+"\n";
if (!Triggers::doTrigger("STREAM_READY", payload, config->getString("streamname"))){
config->is_active = false;
}
}
/*LTS-END*/
userPage.init(userPageName, PLAY_EX_SIZE, true);
if (!isBuffer) {
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++) {
@ -239,6 +262,17 @@ namespace Mist {
} else {
DEBUG_MSG(DLVL_INSANE, "Timer running");
}
/*LTS-START*/
if ((Util::bootSecs() - activityCounter) >= 10 || !config->is_active){//10 second timeout
if(Triggers::shouldTrigger("STREAM_UNLOAD", config->getString("streamname"))){
std::string payload = config->getString("streamname")+"\n" +capa["name"].asStringRef()+"\n";
if (!Triggers::doTrigger("STREAM_UNLOAD", payload, config->getString("streamname"))){
activityCounter = Util::bootSecs();
config->is_active = true;
}
}
}
/*LTS-END*/
}
#endif
finish();

23
src/input/input_buffer.cpp
View file

@ -9,6 +9,7 @@
#include <string>
#include <mist/stream.h>
#include <mist/defines.h>
#include <mist/triggers.h>
#include "input_buffer.h"
@ -299,6 +300,12 @@ namespace Mist {
}
}
/// \triggers
/// The `"STREAM_TRACK_REMOVE"` trigger is stream-specific, and is ran whenever a track is fully removed from a live strean buffer. It cannot be cancelled. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// trackID
/// ~~~~~~~~~~~~~~~
void inputBuffer::removeUnused(){
//first remove all tracks that have not been updated for too long
bool changed = true;
@ -335,6 +342,12 @@ namespace Mist {
}else{
INFO_MSG("Erasing inactive track %u because it was inactive for 5+ seconds and contains data (%us - %us), while active tracks are (%us - %us), which is more than %us seconds apart.", it->first, it->second.firstms / 1000, it->second.lastms / 1000, compareFirst/1000, compareLast/1000, bufferTime / 1000);
}
/*LTS-START*/
if(Triggers::shouldTrigger("STREAM_TRACK_REMOVE")){
std::string payload = config->getString("streamname")+"\n"+JSON::Value((long long)it->first).asString()+"\n";
Triggers::doTrigger("STREAM_TRACK_REMOVE", payload, config->getString("streamname"));
}
/*LTS-END*/
lastUpdated.erase(tid);
/// \todo Consider replacing with eraseTrackDataPages(it->first)?
while (bufferLocations[tid].size()){
@ -397,6 +410,12 @@ namespace Mist {
updateMeta();
}
/// \triggers
/// The `"STREAM_TRACK_ADD"` trigger is stream-specific, and is ran whenever a new track is added to a live strean buffer. It cannot be cancelled. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// trackID
/// ~~~~~~~~~~~~~~~
void inputBuffer::userCallback(char * data, size_t len, unsigned int id){
/*LTS-START*/
//Reload the configuration to make sure we stay up to date with changes through the api
@ -532,6 +551,10 @@ namespace Mist {
//No collision has been detected, assign a new final number
finalMap = (myMeta.tracks.size() ? myMeta.tracks.rbegin()->first : 0) + 1;
DEBUG_MSG(DLVL_DEVEL, "No colision detected for temporary track %lu from user %u, assigning final track number %lu", value, id, finalMap);
if(Triggers::shouldTrigger("STREAM_TRACK_ADD")){
std::string payload = config->getString("streamname")+"\n"+JSON::Value((long long)finalMap).asString()+"\n";
Triggers::doTrigger("STREAM_TRACK_ADD", payload, config->getString("streamname"));
}
}
/*LTS-END*/
//Resume either if we have more than 1 keyframe on the replacement track (assume it was already pushing before the track "dissapeared")

15
src/input/input_ts.cpp
View file

@ -13,6 +13,21 @@
#include <mist/mp4_generic.h>
#include "input_ts.h"
/// \todo Implement this trigger equivalent...
/*
if(Triggers::shouldTrigger("STREAM_PUSH", smp)){
std::string payload = streamName+"\n" + myConn.getHost() +"\n"+capa["name"].asStringRef()+"\n"+reqUrl;
if (!Triggers::doTrigger("STREAM_PUSH", payload, smp)){
DEBUG_MSG(DLVL_FAIL, "Push from %s to %s rejected - STREAM_PUSH trigger denied the push", myConn.getHost().c_str(), streamName.c_str());
myConn.close();
configLock.post();
configLock.close();
return;
}
}
*/
namespace Mist {

54
src/output/output.cpp
View file

@ -14,6 +14,7 @@
#include "output.h"
/*LTS-START*/
#include <mist/triggers.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netdb.h>
@ -102,6 +103,14 @@ namespace Mist {
myConn.close();
}
/// \triggers
/// The `"CONN_PLAY"` trigger is stream-specific, and is ran when an active connection first opens a stream. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// connected client host
/// output handler name
/// request URL (if any)
/// ~~~~~~~~~~~~~~~
void Output::initialize(){
if (isInitialized){
return;
@ -121,6 +130,14 @@ namespace Mist {
}
selectDefaultTracks();
sought = false;
/*LTS-START*/
if(Triggers::shouldTrigger("CONN_PLAY", streamName)){
std::string payload = streamName+"\n" + myConn.getHost() +"\n"+capa["name"].asStringRef()+"\n"+reqUrl;
if (!Triggers::doTrigger("CONN_PLAY", payload, streamName)){
myConn.close();
}
}
/*LTS-END*/
}
/// Connects or reconnects to the stream.
@ -701,7 +718,30 @@ namespace Mist {
}
}
/// \triggers
/// The `"CONN_OPEN"` trigger is stream-specific, and is ran when a connection is made or passed to a new handler. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// connected client host
/// output handler name
/// request URL (if any)
/// ~~~~~~~~~~~~~~~
/// The `"CONN_CLOSE"` trigger is stream-specific, and is ran when a connection closes. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// connected client host
/// output handler name
/// request URL (if any)
/// ~~~~~~~~~~~~~~~
int Output::run() {
/*LTS-START*/
if(Triggers::shouldTrigger("CONN_OPEN", streamName)){
std::string payload = streamName+"\n" + myConn.getHost() +"\n"+capa["name"].asStringRef()+"\n"+reqUrl;
if (!Triggers::doTrigger("CONN_OPEN", payload, streamName)){
return 1;
}
}
/*LTS-END*/
DEBUG_MSG(DLVL_MEDIUM, "MistOut client handler started");
while (config->is_active && myConn.connected() && (wantRequest || parseData)){
stats();
@ -727,6 +767,14 @@ namespace Mist {
}
}
DEBUG_MSG(DLVL_MEDIUM, "MistOut client handler shutting down: %s, %s, %s", myConn.connected() ? "conn_active" : "conn_closed", wantRequest ? "want_request" : "no_want_request", parseData ? "parsing_data" : "not_parsing_data");
/*LTS-START*/
if(Triggers::shouldTrigger("CONN_CLOSE", streamName)){
std::string payload = streamName+"\n"+myConn.getHost()+"\n"+capa["name"].asStringRef()+"\n"+reqUrl; ///\todo generate payload
Triggers::doTrigger("CONN_CLOSE", payload, streamName); //no stream specified
}
/*LTS-END*/
stats();
userClient.finish();
statsPage.finish();
@ -777,6 +825,12 @@ namespace Mist {
thisPacket.null();
DEBUG_MSG(DLVL_DEVEL, "Buffer completely played out");
onFinish();
/*LTS-START*/
if(Triggers::shouldTrigger("CONN_STOP", streamName)){
std::string payload = streamName+"\n" + myConn.getHost() +"\n"+capa["name"].asStringRef()+"\n";
Triggers::doTrigger("CONN_STOP", payload, streamName);
}
/*LTS-END*/
return;
}
sortedPageInfo nxt = *(buffer.begin());

1
src/output/output.h
View file

@ -51,6 +51,7 @@ namespace Mist {
static GeoIP * geoIP4;
static GeoIP * geoIP6;
#endif
std::string reqUrl;
/*LTS-END*/
//non-virtual generic functions
int run();

1
src/output/output_http.cpp
View file

@ -159,6 +159,7 @@ namespace Mist {
if (!myConn.Received().size()){
if (myConn.peek() && H.Read(myConn)){
std::string handler = getHandler();
reqUrl = H.getUrl();//LTS
DEBUG_MSG(DLVL_MEDIUM, "Received request: %s => %s (%s)", H.getUrl().c_str(), handler.c_str(), H.GetVar("stream").c_str());
if (!handler.size()){
H.Clean();

22
src/output/output_rtmp.cpp
View file

@ -2,6 +2,7 @@
#include <mist/http_parser.h>
#include <mist/defines.h>
#include <mist/stream.h>
#include <mist/triggers.h>
#include <sys/stat.h>
#include <cstring>
#include <cstdlib>
@ -325,6 +326,14 @@ namespace Mist {
///\param amfData The received request.
///\param messageType The type of message.
///\param streamId The ID of the AMF stream.
/// \triggers
/// The `"STREAM_PUSH"` trigger is stream-specific, and is ran right before an incoming push is accepted. If cancelled, the push is denied. Its payload is:
/// ~~~~~~~~~~~~~~~
/// streamname
/// connected client host
/// output handler name
/// request URL (if any)
/// ~~~~~~~~~~~~~~~
void OutRTMP::parseAMFCommand(AMF::Object & amfData, int messageType, int streamId) {
#if DEBUG >= 5
fprintf(stderr, "Received command: %s\n", amfData.Print().c_str());
@ -359,6 +368,7 @@ namespace Mist {
}
#endif
app_name = amfData.getContentP(2)->getContentP("tcUrl")->StrValue();
reqUrl = app_name;//LTS
app_name = app_name.substr(app_name.find('/', 7) + 1);
RTMPStream::chunk_snd_max = 4096;
myConn.SendNow(RTMPStream::SendCTL(1, RTMPStream::chunk_snd_max)); //send chunk size max (msg 1)
@ -467,6 +477,7 @@ namespace Mist {
if ((amfData.getContentP(0)->StrValue() == "publish")) {
if (amfData.getContentP(3)) {
streamName = amfData.getContentP(3)->StrValue();
reqUrl += "/"+streamName;//LTS
if (streamName.find('/')){
streamName = streamName.substr(0, streamName.find('/'));
@ -513,6 +524,16 @@ namespace Mist {
}
}
}
if(Triggers::shouldTrigger("STREAM_PUSH", smp)){
std::string payload = streamName+"\n" + myConn.getHost() +"\n"+capa["name"].asStringRef()+"\n"+reqUrl;
if (!Triggers::doTrigger("STREAM_PUSH", payload, smp)){
DEBUG_MSG(DLVL_FAIL, "Push from %s to %s rejected - STREAM_PUSH trigger denied the push", myConn.getHost().c_str(), streamName.c_str());
myConn.close();
configLock.post();
configLock.close();
return;
}
}
/*LTS-END*/
if (IP != ""){
if (!myConn.isAddress(IP)){
@ -567,6 +588,7 @@ namespace Mist {
int playMessageType = messageType;
int playStreamId = streamId;
streamName = amfData.getContentP(3)->StrValue();
reqUrl += "/"+streamName;//LTS
//handle variables
if (streamName.find('?') != std::string::npos){