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

Various fixes, among which:

Browse source 
- Fixed segfault when attempting to initialseek on disconnected streams
- Fix 100% CPU bug in controller's stats code
- WebRTC UDP bind socket improvements
- Several segfault fixes
- Increased packet reordering buffer size from 30 to 150 packets
- Tweaks to default output/buffer behaviour for incoming pushes
- Added message for load balancer checks
- Fixed HLS content type
- Stats fixes
- Exit reason fixes
- Fixed socket IP address detection
- Fixed non-string arguments for stream settings
- Added caching for getConnectedBinHost()
- Added WebRTC playback rate control
- Added/completed VP8/VP9 support to WebRTC/RTSP
- Added live seek option to WebRTC
- Fixed seek to exactly newest timestamp
- Fixed HLS input

# Conflicts:
#	lib/defines.h
#	src/input/input.cpp
This commit is contained in:
Thulinma 2021-04-21 18:11:46 +02:00
parent 2b99f2f5ea
commit 0af992d405
75 changed files with 1512 additions and 790 deletions
Download patch file Download diff file Expand all files Collapse all files

16
src/analysers/analyser_dtsc.cpp
View file

@ -9,6 +9,10 @@ void AnalyserDTSC::init(Util::Config &conf){
opt["short"] = "H";
opt["help"] = "Parse entire file or streams as a single headless DTSC packet";
conf.addOption("headless", opt);
opt["long"] = "sizeprepended";
opt["short"] = "s";
opt["help"] = "If set, data of packets is considered to be size-prepended";
conf.addOption("sizeprepended", opt);
opt.null();
}
@ -21,6 +25,7 @@ bool AnalyserDTSC::open(const std::string &filename){
AnalyserDTSC::AnalyserDTSC(Util::Config &conf) : Analyser(conf){
headLess = conf.getBool("headless");
sizePrepended = conf.getBool("sizeprepended");
}
bool AnalyserDTSC::parsePacket(){
@ -60,10 +65,19 @@ bool AnalyserDTSC::parsePacket(){
char *payDat;
size_t payLen;
P.getString("data", payDat, payLen);
uint32_t currLen = 0;
uint64_t byteCounter = 0;
for (uint64_t i = 0; i < payLen; ++i){
if ((i % 32) == 0){std::cout << std::endl;}
if (sizePrepended && !currLen){
currLen = 4+Bit::btohl(payDat+i);
byteCounter = 0;
}
if ((byteCounter % 32) == 0){std::cout << std::endl;}
std::cout << std::hex << std::setw(2) << std::setfill('0') << (unsigned int)payDat[i];
++byteCounter;
--currLen;
}
std::cout << std::dec << std::endl;
}
break;
}

1
src/analysers/analyser_dtsc.h
View file

@ -10,6 +10,7 @@ public:
private:
bool headLess;
bool sizePrepended;
DTSC::Packet P;
Socket::Connection conn;
uint64_t totalBytes;

10
src/analysers/analyser_rtmp.cpp
View file

@ -16,6 +16,11 @@ void AnalyserRTMP::init(Util::Config &conf){
opt.null();
}
/// Checks if standard input is still valid.
bool AnalyserRTMP::isOpen(){
return (*isActive) && (std::cin.good() || strbuf.size());
}
AnalyserRTMP::AnalyserRTMP(Util::Config &conf) : Analyser(conf){
if (conf.getString("reconstruct") != ""){
reconstruct.open(conf.getString("reconstruct").c_str());
@ -43,7 +48,10 @@ bool AnalyserRTMP::parsePacket(){
// While we can't parse a packet,
while (!next.Parse(strbuf)){
// fill our internal buffer "strbuf" in (up to) 1024 byte chunks
if (!std::cin.good()){return false;}
if (!std::cin.good()){
strbuf.clear();
return false;
}
size_t charCount = 0;
std::string tmpbuffer;
tmpbuffer.reserve(1024);

1
src/analysers/analyser_rtmp.h
View file

@ -18,4 +18,5 @@ public:
static void init(Util::Config &conf);
bool parsePacket();
virtual bool open(const std::string &filename);
virtual bool isOpen();
};

2
src/controller/controller.cpp
View file

@ -343,7 +343,7 @@ int main_loop(int argc, char **argv){
WARN_MSG("You have very little free RAM available (%" PRIu64
" MiB). While Mist will run just fine with this amount, do note that random crashes "
"may occur should you ever run out of free RAM. Please be pro-active and keep an "
"eye on the RAM usage!");
"eye on the RAM usage!", (mem_free + mem_bufcache)/1024);
}
if (shm_free < 1024 * 1024 && mem_total > 1024 * 1024 * 1.12){
WARN_MSG("You have very little shared memory available (%" PRIu64

6
src/controller/controller_api.cpp
View file

@ -502,6 +502,12 @@ void Controller::handleAPICommands(JSON::Value &Request, JSON::Value &Response){
Controller::triggerStats[Request["trigger_fail"].asStringRef()].failCount++;
return;
}
if (Request.isMember("push_status_update")){
JSON::Value &statUp = Request["push_status_update"];
if (statUp.isMember("id") && statUp.isMember("status")){
setPushStatus(statUp["id"].asInt(), statUp["status"]);
}
}
/*LTS-END*/
// Parse config and streams from the request.
if (Request.isMember("config") && Request["config"].isObject()){

6
src/controller/controller_capabilities.cpp
View file

@ -200,6 +200,12 @@ namespace Controller{
trgs["DEFAULT_STREAM"]["response_action"] =
"Overrides the default stream setting (for this view) to the response value. If empty, "
"fails loading the stream and returns an error to the viewer/user.";
trgs["PUSH_END"]["when"] = "Every time a push stops, for any reason";
trgs["PUSH_END"]["stream_specific"] = true;
trgs["PUSH_END"]["payload"] = "push ID (integer)\nstream name (string)\ntarget URI, before variables/triggers affected it (string)\ntarget URI, afterwards, as actually used (string)\nlast 10 log messages (JSON array string)\nmost recent push status (JSON object string)";
trgs["PUSH_END"]["response"] = "ignored";
trgs["PUSH_END"]["response_action"] = "None.";
}
/// Aquire list of available protocols, storing in global 'capabilities' JSON::Value.

4
src/controller/controller_connectors.cpp
View file

@ -164,8 +164,6 @@ namespace Controller{
std::set<std::string> runningConns;
// used for building args
int zero = 0;
int out = fileno(stdout);
int err = fileno(stderr);
char *argarr[15]; // approx max # of args (with a wide margin)
int i;
@ -259,7 +257,7 @@ namespace Controller{
JSON::Value p = JSON::fromString(*runningConns.begin());
buildPipedArguments(p, (char **)&argarr, capabilities);
// start piped w/ generated args
currentConnectors[*runningConns.begin()] = Util::Procs::StartPiped(argarr, &zero, &out, &err);
currentConnectors[*runningConns.begin()] = Util::Procs::StartPiped(argarr, 0, 0, &err);
Triggers::doTrigger("OUTPUT_START", *runningConns.begin()); // LTS
}
runningConns.erase(runningConns.begin());

53
src/controller/controller_push.cpp
View file

@ -7,6 +7,7 @@
#include <mist/procs.h>
#include <mist/stream.h>
#include <mist/tinythread.h>
#include <mist/triggers.h>
#include <string>
namespace Controller{
@ -38,6 +39,38 @@ namespace Controller{
}
}
void setPushStatus(uint64_t id, const JSON::Value & status){
if (!activePushes.count(id)){return;}
activePushes[id][5].extend(status);
}
void pushLogMessage(uint64_t id, const JSON::Value & msg){
JSON::Value &log = activePushes[id][4];
log.append(msg);
log.shrink(10);
}
bool isPushActive(uint64_t id){
while (Controller::conf.is_active && !pushListRead){Util::sleep(100);}
return activePushes.count(id);
}
/// Only used internally, to remove pushes
static void removeActivePush(pid_t id){
//ignore if the push does not exist
if (!activePushes.count(id)){return;}
JSON::Value p = activePushes[id];
if (Triggers::shouldTrigger("PUSH_END", p[1].asStringRef())){
std::string payload = p[0u].asString() + "\n" + p[1u].asString() + "\n" + p[2u].asString() + "\n" + p[3u].asString() + "\n" + p[4u].toString() + "\n" + p[5u].toString();
Triggers::doTrigger("PUSH_END", payload, p[1].asStringRef());
}
//actually remove, make sure next pass the new list is written out too
activePushes.erase(id);
mustWritePushList = true;
}
/// Returns true if the push is currently active, false otherwise.
bool isPushActive(const std::string &streamname, const std::string &target){
while (Controller::conf.is_active && !pushListRead){Util::sleep(100);}
@ -52,8 +85,7 @@ namespace Controller{
}
}
while (toWipe.size()){
activePushes.erase(*toWipe.begin());
mustWritePushList = true;
removeActivePush(*toWipe.begin());
toWipe.erase(toWipe.begin());
}
return false;
@ -75,8 +107,7 @@ namespace Controller{
}
}
while (toWipe.size()){
activePushes.erase(*toWipe.begin());
mustWritePushList = true;
removeActivePush(*toWipe.begin());
toWipe.erase(toWipe.begin());
}
}
@ -198,6 +229,17 @@ namespace Controller{
break;
}
}
//Check if any pushes have ended, clean them up
std::set<pid_t> toWipe;
for (std::map<pid_t, JSON::Value>::iterator it = activePushes.begin(); it != activePushes.end(); ++it){
if (!Util::Procs::isActive(it->first)){toWipe.insert(it->first);}
}
while (toWipe.size()){
removeActivePush(*toWipe.begin());
toWipe.erase(toWipe.begin());
mustWritePushList = true;
}
//write push list to shared memory, for restarting/crash recovery/etc
if (mustWritePushList && pushPage.mapped){
writePushList(pushPage.mapped);
mustWritePushList = false;
@ -227,8 +269,7 @@ namespace Controller{
}
}
while (toWipe.size()){
activePushes.erase(*toWipe.begin());
mustWritePushList = true;
removeActivePush(*toWipe.begin());
toWipe.erase(toWipe.begin());
}
}

3
src/controller/controller_push.h
View file

@ -8,6 +8,9 @@ namespace Controller{
void startPush(const std::string &streamname, std::string &target);
void stopPush(unsigned int ID);
void listPush(JSON::Value &output);
void pushLogMessage(uint64_t id, const JSON::Value & msg);
void setPushStatus(uint64_t id, const JSON::Value & status);
bool isPushActive(uint64_t id);
// Functions for automated pushes, add/remove
void addPush(JSON::Value &request);

93
src/controller/controller_statistics.cpp
View file

@ -50,6 +50,7 @@ bool Controller::killOnExit = KILL_ON_EXIT;
tthread::mutex Controller::statsMutex;
unsigned int Controller::maxConnsPerIP = 0;
uint64_t Controller::statDropoff = 0;
static uint64_t cpu_use = 0;
char noBWCountMatches[1717];
uint64_t bwLimit = 128 * 1024 * 1024; // gigabit default limit
@ -76,6 +77,21 @@ void Controller::updateBandwidthConfig(){
}
}
}
//Localhost is always excepted from counts
{
std::string newbins = Socket::getBinForms("::1");
if (offset + newbins.size() < 1700){
memcpy(noBWCountMatches + offset, newbins.data(), newbins.size());
offset += newbins.size();
}
}
{
std::string newbins = Socket::getBinForms("127.0.0.1/8");
if (offset + newbins.size() < 1700){
memcpy(noBWCountMatches + offset, newbins.data(), newbins.size());
offset += newbins.size();
}
}
}
// For server-wide totals. Local to this file only.
@ -107,7 +123,7 @@ Controller::sessIndex::sessIndex(){
/// into strings. This extracts the host, stream name, connector and crc field, ignoring everything
/// else.
Controller::sessIndex::sessIndex(const Comms::Statistics &statComm, size_t id){
host = statComm.getHost(id);
Socket::hostBytesToStr(statComm.getHost(id).data(), 16, host);
streamName = statComm.getStream(id);
connector = statComm.getConnector(id);
crc = statComm.getCRC(id);
@ -336,6 +352,28 @@ void Controller::SharedMemStats(void *config){
bool firstRun = true;
while (((Util::Config *)config)->is_active){
{
std::ifstream cpustat("/proc/stat");
if (cpustat){
char line[300];
while (cpustat.getline(line, 300)){
static unsigned long long cl_total = 0, cl_idle = 0;
unsigned long long c_user, c_nice, c_syst, c_idle, c_total;
if (sscanf(line, "cpu %Lu %Lu %Lu %Lu", &c_user, &c_nice, &c_syst, &c_idle) == 4){
c_total = c_user + c_nice + c_syst + c_idle;
if (c_total > cl_total){
cpu_use = (long long)(1000 - ((c_idle - cl_idle) * 1000) / (c_total - cl_total));
}else{
cpu_use = 0;
}
cl_total = c_total;
cl_idle = c_idle;
break;
}
}
}
}
{
tthread::lock_guard<tthread::mutex> guard(Controller::configMutex);
tthread::lock_guard<tthread::mutex> guard2(statsMutex);
cacheLock->wait(); /*LTS*/
@ -522,7 +560,8 @@ uint32_t Controller::statSession::kill(){
/// Updates the given active connection with new stats data.
void Controller::statSession::update(uint64_t index, Comms::Statistics &statComm){
std::string myHost = statComm.getHost(index);
std::string myHost;
Socket::hostBytesToStr(statComm.getHost(index).data(), 16, myHost);
std::string myStream = statComm.getStream(index);
std::string myConnector = statComm.getConnector(index);
// update the sync byte: 0 = requesting fill, 2 = requesting refill, 1 = needs checking, > 2 =
@ -613,12 +652,12 @@ void Controller::statSession::update(uint64_t index, Comms::Statistics &statComm
}
if (currDown + currUp >= COUNTABLE_BYTES){
if (sessionType == SESS_UNSET){
if (myConnector == "INPUT"){
if (myConnector.size() >= 5 && myConnector.substr(0, 5) == "INPUT"){
++servInputs;
streamStats[myStream].inputs++;
streamStats[myStream].currIns++;
sessionType = SESS_INPUT;
}else if (myConnector == "OUTPUT"){
}else if (myConnector.size() >= 6 && myConnector.substr(0, 6) == "OUTPUT"){
++servOutputs;
streamStats[myStream].outputs++;
streamStats[myStream].currOuts++;
@ -632,19 +671,21 @@ void Controller::statSession::update(uint64_t index, Comms::Statistics &statComm
}
// If previous < COUNTABLE_BYTES, we haven't counted any data so far.
// We need to count all the data in that case, otherwise we only count the difference.
if (prevUp + prevDown < COUNTABLE_BYTES){
if (!myStream.size() || myStream[0] == 0){
if (streamStats.count(myStream)){streamStats.erase(myStream);}
if (noBWCount != 2){ //only count connections that are countable
if (prevUp + prevDown < COUNTABLE_BYTES){
if (!myStream.size() || myStream[0] == 0){
if (streamStats.count(myStream)){streamStats.erase(myStream);}
}else{
streamStats[myStream].upBytes += currUp;
streamStats[myStream].downBytes += currDown;
}
}else{
streamStats[myStream].upBytes += currUp;
streamStats[myStream].downBytes += currDown;
}
}else{
if (!myStream.size() || myStream[0] == 0){
if (streamStats.count(myStream)){streamStats.erase(myStream);}
}else{
streamStats[myStream].upBytes += currUp - prevUp;
streamStats[myStream].downBytes += currDown - prevDown;
if (!myStream.size() || myStream[0] == 0){
if (streamStats.count(myStream)){streamStats.erase(myStream);}
}else{
streamStats[myStream].upBytes += currUp - prevUp;
streamStats[myStream].downBytes += currDown - prevDown;
}
}
}
}
@ -1437,29 +1478,9 @@ void Controller::handlePrometheus(HTTP::Parser &H, Socket::Connection &conn, int
H.StartResponse("200", "OK", H, conn, true);
// Collect core server stats
uint64_t cpu_use = 0;
uint64_t mem_total = 0, mem_free = 0, mem_bufcache = 0;
uint64_t bw_up_total = 0, bw_down_total = 0;
{
std::ifstream cpustat("/proc/stat");
if (cpustat){
char line[300];
while (cpustat.getline(line, 300)){
static unsigned long long cl_total = 0, cl_idle = 0;
unsigned long long c_user, c_nice, c_syst, c_idle, c_total;
if (sscanf(line, "cpu %Lu %Lu %Lu %Lu", &c_user, &c_nice, &c_syst, &c_idle) == 4){
c_total = c_user + c_nice + c_syst + c_idle;
if (c_total > cl_total){
cpu_use = (long long int)(1000 - ((c_idle - cl_idle) * 1000) / (c_total - cl_total));
}else{
cpu_use = 0;
}
cl_total = c_total;
cl_idle = c_idle;
break;
}
}
}
std::ifstream meminfo("/proc/meminfo");
if (meminfo){
char line[300];

4
src/controller/controller_storage.cpp
View file

@ -1,5 +1,6 @@
#include "controller_capabilities.h"
#include "controller_storage.h"
#include "controller_push.h" //LTS
#include <algorithm>
#include <fstream>
#include <iostream>
@ -41,7 +42,7 @@ namespace Controller{
///\brief Store and print a log message.
///\param kind The type of message.
///\param message The message to be logged.
void Log(const std::string &kind, const std::string &message, const std::string &stream, bool noWriteToLog){
void Log(const std::string &kind, const std::string &message, const std::string &stream, uint64_t progPid, bool noWriteToLog){
if (noWriteToLog){
tthread::lock_guard<tthread::mutex> guard(logMutex);
JSON::Value m;
@ -52,6 +53,7 @@ namespace Controller{
m.append(stream);
Storage["log"].append(m);
Storage["log"].shrink(100); // limit to 100 log messages
if (isPushActive(progPid)){pushLogMessage(progPid, m);} //LTS
logCounter++;
if (rlxLogs && rlxLogs->isReady()){
if (!firstLog){firstLog = logCounter;}

2
src/controller/controller_storage.h
View file

@ -22,7 +22,7 @@ namespace Controller{
Util::RelAccX *streamsAccessor();
/// Store and print a log message.
void Log(const std::string &kind, const std::string &message, const std::string &stream = "",
void Log(const std::string &kind, const std::string &message, const std::string &stream = "", uint64_t progPid = 0,
bool noWriteToLog = false);
void logAccess(const std::string &sessId, const std::string &strm, const std::string &conn,
const std::string &host, uint64_t duration, uint64_t up, uint64_t down,

253
src/input/input.cpp
View file

@ -77,7 +77,8 @@ namespace Mist{
option["help"] = "Generate .dtsh, then exit";
config->addOption("headeronly", option);
/*LTS-START
/*LTS-START*/
/*
//Encryption
option.null();
option["arg"] = "string";
@ -86,31 +87,6 @@ namespace Mist{
option["help"] = "a KID:KEY combo for auto-encrypting tracks";
config->addOption("encryption", option);
option.null();
option["long"] = "realtime";
option["short"] = "r";
option["help"] = "Feed the results of this input in realtime to the buffer";
config->addOption("realtime", option);
capa["optional"]["realtime"]["name"] = "Simulated Live";
capa["optional"]["realtime"]["help"] = "Make this input run as a simulated live stream";
capa["optional"]["realtime"]["option"] = "--realtime";
option.null();
option["long"] = "simulated-starttime";
option["arg"] = "integer";
option["short"] = "S";
option["help"] = "Unix timestamp on which the simulated start of the stream is based.";
option["value"].append(0);
config->addOption("simulated-starttime", option);
capa["optional"]["simulated-starttime"]["name"] = "Simulated start time";
capa["optional"]["simulated-starttime"]["help"] =
"The unix timestamp on which this stream is assumed to have started playback, or 0 for "
"automatic";
capa["optional"]["simulated-starttime"]["option"] = "--simulated-starttime";
capa["optional"]["simulated-starttime"]["type"] = "uint";
capa["optional"]["simulated-starttime"]["default"] = 0;
option.null();
option["arg"] = "string";
option["short"] = "B";
@ -175,8 +151,33 @@ namespace Mist{
capa["optional"]["playready"]["help"] = "The header to use for PlayReady encryption.";
capa["optional"]["playready"]["option"] = "--playready";
capa["optional"]["playready"]["type"] = "string";
LTS-END*/
*/
option.null();
option["long"] = "realtime";
option["short"] = "r";
option["help"] = "Feed the results of this input in realtime to the buffer";
config->addOption("realtime", option);
capa["optional"]["realtime"]["name"] = "Simulated Live";
capa["optional"]["realtime"]["help"] = "Make this input run as a simulated live stream";
capa["optional"]["realtime"]["option"] = "--realtime";
option.null();
option["long"] = "simulated-starttime";
option["arg"] = "integer";
option["short"] = "S";
option["help"] = "Unix timestamp on which the simulated start of the stream is based.";
option["value"].append(0);
config->addOption("simulated-starttime", option);
capa["optional"]["simulated-starttime"]["name"] = "Simulated start time";
capa["optional"]["simulated-starttime"]["help"] =
"The unix timestamp on which this stream is assumed to have started playback, or 0 for "
"automatic";
capa["optional"]["simulated-starttime"]["option"] = "--simulated-starttime";
capa["optional"]["simulated-starttime"]["type"] = "uint";
capa["optional"]["simulated-starttime"]["default"] = 0;
/*LTS-END*/
capa["optional"]["debug"]["name"] = "debug";
capa["optional"]["debug"]["help"] = "The debug level at which messages need to be printed.";
capa["optional"]["debug"]["option"] = "--debug";
@ -637,7 +638,7 @@ namespace Mist{
if (streamStatus){streamStatus.mapped[0] = STRMSTAT_SHUTDOWN;}
config->is_active = false;
finish();
INFO_MSG("Input for stream %s closing clean", streamName.c_str());
INFO_MSG("Input closing clean, reason: %s", Util::exitReason);
userSelect.clear();
if (streamStatus){streamStatus.mapped[0] = STRMSTAT_OFF;}
}
@ -669,6 +670,9 @@ namespace Mist{
}
}
/*LTS-END*/
if (!ret && ((Util::bootSecs() - activityCounter) >= INPUT_TIMEOUT)){
Util::logExitReason("no activity for %u seconds", Util::bootSecs() - activityCounter);
}
return ret;
}
@ -682,28 +686,29 @@ namespace Mist{
/// - call getNext() in a loop, buffering packets
void Input::stream(){
if (!config->getBool("realtime") && Util::streamAlive(streamName)){
WARN_MSG("Stream already online, cancelling");
return;
}
std::map<std::string, std::string> overrides;
overrides["throughboot"] = "";
if (isSingular()){
if (Util::streamAlive(streamName)){
WARN_MSG("Stream already online, cancelling");
return;
}
overrides["singular"] = "";
}
if (config->getBool("realtime") ||
(capa.isMember("hardcoded") && capa["hardcoded"].isMember("resume") && capa["hardcoded"]["resume"])){
overrides["resume"] = "1";
}
if (!Util::startInput(streamName, "push://INTERNAL_ONLY:" + config->getString("input"), true,
true, overrides)){// manually override stream url to start the buffer
WARN_MSG("Could not start buffer, cancelling");
return;
if (isSingular()){
if (!config->getBool("realtime") && Util::streamAlive(streamName)){
WARN_MSG("Stream already online, cancelling");
return;
}
overrides["singular"] = "";
if (!Util::startInput(streamName, "push://INTERNAL_ONLY:" + config->getString("input"), true,
true, overrides)){// manually override stream url to start the buffer
WARN_MSG("Could not start buffer, cancelling");
return;
}
}else{
if (!Util::startInput(streamName, "push://INTERNAL_PUSH:" + capa["name"].asStringRef(), true,
true, overrides)){// manually override stream url to start the buffer
WARN_MSG("Could not start buffer, cancelling");
return;
}
}
INFO_MSG("Input started");
@ -715,19 +720,23 @@ namespace Mist{
}
parseStreamHeader();
std::set<size_t> validTracks = M.getMySourceTracks(getpid());
if (!validTracks.size()){
userSelect.clear();
finish();
INFO_MSG("No tracks found, cancelling");
return;
std::set<size_t> validTracks;
if (publishesTracks()){
validTracks = M.getMySourceTracks(getpid());
if (!validTracks.size()){
userSelect.clear();
finish();
INFO_MSG("No tracks found, cancelling");
return;
}
}
timeOffset = 0;
uint64_t minFirstMs = 0;
// If resume mode is on, find matching tracks and set timeOffset values to make sure we append to the tracks.
if (config->getBool("realtime")){
if (publishesTracks() && config->getBool("realtime")){
seek(0);
minFirstMs = 0xFFFFFFFFFFFFFFFFull;
@ -736,12 +745,11 @@ namespace Mist{
uint64_t maxLastMs = 0;
// track lowest firstms value
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin();
it != myMeta.tracks.end(); ++it){
if (it->second.firstms < minFirstMs){minFirstMs = it->second.firstms;}
if (it->second.firstms > maxFirstMs){maxFirstMs = it->second.firstms;}
if (it->second.lastms < minLastMs){minLastMs = it->second.lastms;}
if (it->second.lastms > maxLastMs){maxLastMs = it->second.lastms;}
for (std::set<size_t>::iterator it = validTracks.begin(); it != validTracks.end(); ++it){
if (meta.getFirstms(*it) < minFirstMs){minFirstMs = meta.getFirstms(*it);}
if (meta.getFirstms(*it) > maxFirstMs){maxFirstMs = meta.getFirstms(*it);}
if (meta.getLastms(*it) < minLastMs){minLastMs = meta.getLastms(*it);}
if (meta.getLastms(*it) > maxLastMs){maxLastMs = meta.getLastms(*it);}
}
if (maxFirstMs - minFirstMs > 500){
WARN_MSG("Begin timings of tracks for this file are %" PRIu64
@ -756,10 +764,8 @@ namespace Mist{
maxLastMs - minLastMs, minLastMs, maxLastMs);
}
// find highest current time
for (std::map<unsigned int, DTSC::Track>::iterator secondIt = tmpM.tracks.begin();
secondIt != tmpM.tracks.end(); ++secondIt){
VERYHIGH_MSG("Track %u starts at %" PRIu64, secondIt->first, secondIt->second.lastms);
timeOffset = std::max(timeOffset, (int64_t)secondIt->second.lastms);
for (std::set<size_t>::iterator it = validTracks.begin(); it != validTracks.end(); ++it){
timeOffset = std::max(timeOffset, (int64_t)meta.getLastms(*it));
}
if (timeOffset){
@ -771,13 +777,11 @@ namespace Mist{
timeOffset -= minFirstMs; // we don't need to add the lowest firstms value to the offset, as it's already there
}
}
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin();
it != myMeta.tracks.end(); it++){
it->second.firstms += timeOffset;
it->second.lastms = 0;
selectedTracks.insert(it->first);
it->second.minKeepAway = SIMULATED_LIVE_BUFFER;
if (publishesTracks()){
for (std::set<size_t>::iterator it = validTracks.begin(); it != validTracks.end(); ++it){
meta.setFirstms(*it, meta.getFirstms(*it)+timeOffset);
meta.setLastms(*it, 0);
}
}
simStartTime = config->getInteger("simulated-starttime");
@ -785,9 +789,9 @@ namespace Mist{
std::string reason;
if (config->getBool("realtime")){
reason = realtimeMainLoop();
realtimeMainLoop();
}else{
reason = streamMainLoop();
streamMainLoop();
}
closeStreamSource();
@ -795,11 +799,68 @@ namespace Mist{
userSelect.clear();
finish();
INFO_MSG("Input closing clean; reason: %s", reason.c_str());
INFO_MSG("Input closing clean; reason: %s", Util::exitReason);
return;
}
std::string Input::streamMainLoop(){
void Input::streamMainLoop(){
uint64_t statTimer = 0;
uint64_t startTime = Util::bootSecs();
size_t tid;
size_t idx;
Comms::Statistics statComm;
getNext();
tid = thisPacket.getTrackId();
idx = M.trackIDToIndex(tid, getpid());
if (thisPacket && !userSelect.count(idx)){
userSelect[idx].reload(streamName, idx, COMM_STATUS_SOURCE | COMM_STATUS_DONOTTRACK);
}
while (thisPacket && config->is_active && userSelect[idx].isAlive()){
if (userSelect[idx].getStatus() == COMM_STATUS_REQDISCONNECT){
Util::logExitReason("buffer requested shutdown");
break;
}
bufferLivePacket(thisPacket);
userSelect[idx].keepAlive();
getNext();
if (!thisPacket){
Util::logExitReason("invalid packet from getNext");
break;
}
tid = thisPacket.getTrackId();
idx = M.trackIDToIndex(tid, getpid());
if (thisPacket && !userSelect.count(idx)){
userSelect[idx].reload(streamName, idx, COMM_STATUS_SOURCE | COMM_STATUS_DONOTTRACK);
}
if (Util::bootSecs() - statTimer > 1){
// Connect to stats for INPUT detection
if (!statComm){statComm.reload();}
if (statComm){
if (!statComm.isAlive()){
config->is_active = false;
Util::logExitReason("received shutdown request from controller");
return;
}
uint64_t now = Util::bootSecs();
statComm.setNow(now);
statComm.setCRC(getpid());
statComm.setStream(streamName);
statComm.setConnector("INPUT:" + capa["name"].asStringRef());
statComm.setUp(0);
statComm.setDown(streamByteCount());
statComm.setTime(now - startTime);
statComm.setLastSecond(0);
statComm.setHost(getConnectedBinHost());
statComm.keepAlive();
}
statTimer = Util::bootSecs();
}
}
}
void Input::realtimeMainLoop(){
uint64_t statTimer = 0;
uint64_t startTime = Util::bootSecs();
Comms::Statistics statComm;
@ -809,7 +870,18 @@ namespace Mist{
userSelect[tid].reload(streamName, tid, COMM_STATUS_SOURCE | COMM_STATUS_DONOTTRACK);
}
while (thisPacket && config->is_active && userSelect[thisPacket.getTrackId()].isAlive()){
thisPacket.nullMember("bpos");
while (config->is_active && userSelect[thisPacket.getTrackId()].isAlive() &&
Util::bootMS() + SIMULATED_LIVE_BUFFER < (thisPacket.getTime() + timeOffset) + simStartTime){
Util::sleep(std::min(((thisPacket.getTime() + timeOffset) + simStartTime) - (Util::getMS() + SIMULATED_LIVE_BUFFER),
(uint64_t)1000));
userSelect[thisPacket.getTrackId()].keepAlive();
}
uint64_t originalTime = thisPacket.getTime();
thisPacket.setTime(originalTime + timeOffset);
bufferLivePacket(thisPacket);
thisPacket.setTime(originalTime);
userSelect[thisPacket.getTrackId()].keepAlive();
getNext();
if (thisPacket && !userSelect.count(thisPacket.getTrackId())){
@ -823,50 +895,31 @@ namespace Mist{
if (statComm){
if (!statComm.isAlive()){
config->is_active = false;
return "received shutdown request from controller";
Util::logExitReason("received shutdown request from controller");
return;
}
uint64_t now = Util::bootSecs();
statComm.setNow(now);
statComm.setCRC(getpid());
statComm.setStream(streamName);
statComm.setConnector("INPUT");
statComm.setConnector("INPUT:" + capa["name"].asStringRef());
statComm.setUp(0);
statComm.setDown(streamByteCount());
statComm.setTime(now - startTime);
statComm.setLastSecond(0);
statComm.setHost(getConnectedBinHost());
statComm.keepAlive();
}
statTimer = Util::bootSecs();
}
}
if (!nProxy.userClient.isAlive()){return "buffer shutdown";}
if (!config->is_active){return "received deactivate signal";}
if (!thisPacket){return "Invalid packet";}
return "Unknown";
}
std::string Input::realtimeMainLoop(){
getNext();
while (thisPacket && config->is_active && nProxy.userClient.isAlive()){
thisPacket.nullMember("bpos");
while (config->is_active && nProxy.userClient.isAlive() &&
Util::bootMS() + SIMULATED_LIVE_BUFFER < (thisPacket.getTime() + timeOffset) + simStartTime){
Util::sleep(std::min(((thisPacket.getTime() + timeOffset) + simStartTime) - (Util::getMS() + SIMULATED_LIVE_BUFFER),
(uint64_t)1000));
nProxy.userClient.keepAlive();
}
uint64_t originalTime = thisPacket.getTime();
thisPacket.setTime(originalTime + timeOffset);
nProxy.bufferLivePacket(thisPacket, myMeta);
thisPacket.setTime(originalTime);
getNext();
nProxy.userClient.keepAlive();
if (!thisPacket){
Util::logExitReason("invalid packet from getNext");
}
if (thisPacket && !userSelect[thisPacket.getTrackId()].isAlive()){
Util::logExitReason("buffer shutdown");
}
if (!thisPacket){return "end of file";}
if (!config->is_active){return "received deactivate signal";}
if (!userSelect[thisPacket.getTrackId()].isAlive()){return "buffer shutdown";}
return "Unknown";
}
void Input::finish(){

6
src/input/input.h
View file

@ -33,6 +33,7 @@ namespace Mist{
bool hasMeta() const;
static Util::Config *config;
virtual bool needsLock(){return !config->getBool("realtime");}
virtual bool publishesTracks(){return true;}
protected:
virtual bool checkArguments() = 0;
@ -54,11 +55,12 @@ namespace Mist{
virtual void convert();
virtual void serve();
virtual void stream();
virtual std::string getConnectedBinHost(){return std::string("\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\001", 16);}
virtual size_t streamByteCount(){
return 0;
}; // For live streams: to update the stats with correct values.
virtual std::string streamMainLoop();
virtual std::string realtimeMainLoop();
virtual void streamMainLoop();
virtual void realtimeMainLoop();
bool isAlwaysOn();
virtual void userLeadIn();

67
src/input/input_balancer.cpp
View file

@ -16,6 +16,73 @@ namespace Mist{
capa["source_match"] = "balance:*";
capa["priority"] = 9;
capa["morphic"] = 1;
JSON::Value option;
option["arg"] = "integer";
option["long"] = "buffer";
option["short"] = "b";
option["help"] = "DVR buffer time in ms";
option["value"].append(50000);
config->addOption("bufferTime", option);
capa["optional"]["DVR"]["name"] = "Buffer time (ms)";
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 as "
"well as the minimum duration needed for stable playback.";
capa["optional"]["DVR"]["option"] = "--buffer";
capa["optional"]["DVR"]["type"] = "uint";
capa["optional"]["DVR"]["default"] = 50000;
option.null();
option["arg"] = "integer";
option["long"] = "cut";
option["short"] = "c";
option["help"] = "Any timestamps before this will be cut from the live buffer";
option["value"].append(0);
config->addOption("cut", option);
capa["optional"]["cut"]["name"] = "Cut time (ms)";
capa["optional"]["cut"]["help"] =
"Any timestamps before this will be cut from the live buffer.";
capa["optional"]["cut"]["option"] = "--cut";
capa["optional"]["cut"]["type"] = "uint";
capa["optional"]["cut"]["default"] = 0;
option.null();
option["arg"] = "integer";
option["long"] = "resume";
option["short"] = "R";
option["help"] = "Enable resuming support (1) or disable resuming support (0, default)";
option["value"].append(0);
config->addOption("resume", option);
capa["optional"]["resume"]["name"] = "Resume support";
capa["optional"]["resume"]["help"] =
"If enabled, the buffer will linger after source disconnect to allow resuming the stream "
"later. If disabled, the buffer will instantly close on source disconnect.";
capa["optional"]["resume"]["option"] = "--resume";
capa["optional"]["resume"]["type"] = "select";
capa["optional"]["resume"]["select"][0u][0u] = "0";
capa["optional"]["resume"]["select"][0u][1u] = "Disabled";
capa["optional"]["resume"]["select"][1u][0u] = "1";
capa["optional"]["resume"]["select"][1u][1u] = "Enabled";
capa["optional"]["resume"]["default"] = 0;
option.null();
option["arg"] = "integer";
option["long"] = "segment-size";
option["short"] = "S";
option["help"] = "Target time duration in milliseconds for segments";
option["value"].append(5000);
config->addOption("segmentsize", option);
capa["optional"]["segmentsize"]["name"] = "Segment size (ms)";
capa["optional"]["segmentsize"]["help"] = "Target time duration in milliseconds for segments.";
capa["optional"]["segmentsize"]["option"] = "--segment-size";
capa["optional"]["segmentsize"]["type"] = "uint";
capa["optional"]["segmentsize"]["default"] = 5000;
capa["codecs"][0u][0u].append("*");
capa["codecs"][0u][1u].append("*");
capa["codecs"][0u][2u].append("*");
}
int inputBalancer::boot(int argc, char *argv[]){

89
src/input/input_buffer.cpp
View file

@ -30,6 +30,7 @@ namespace Mist{
capa["optional"].removeMember("realtime");
lastReTime = 0; /*LTS*/
finalMillis = 0;
liveMeta = 0;
capa["name"] = "Buffer";
@ -118,6 +119,7 @@ namespace Mist{
cutTime = 0;
segmentSize = 1900;
hasPush = false;
everHadPush = false;
resumeMode = false;
}
@ -360,7 +362,7 @@ namespace Mist{
if (!(users.getStatus(i) & COMM_STATUS_SOURCE)){continue;}
if (users.getTrack(i) != tid){continue;}
// We have found the right track here (pid matches, and COMM_STATUS_SOURCE set)
users.setStatus(COMM_STATUS_DISCONNECT, i);
users.setStatus(COMM_STATUS_REQDISCONNECT, i);
break;
}
@ -450,7 +452,7 @@ namespace Mist{
// firstVideo = 1 happens when there are no tracks, in which case we don't care any more
uint32_t firstKey = keys.getFirstValid();
uint32_t endKey = keys.getEndValid();
if (type != "video"){
if (type != "video" && videoFirstms != 0xFFFFFFFFFFFFFFFFull){
if ((endKey - firstKey) < 2 || keys.getTime(firstKey + 1) > videoFirstms){continue;}
}
// Buffer cutting
@ -464,19 +466,6 @@ namespace Mist{
}
}
updateMeta();
if (config->is_active){
if (streamStatus){streamStatus.mapped[0] = hasPush ? STRMSTAT_READY : STRMSTAT_WAIT;}
}
static bool everHadPush = false;
if (hasPush){
hasPush = false;
everHadPush = true;
}else if (everHadPush && !resumeMode && config->is_active){
INFO_MSG("Shutting down buffer because resume mode is disabled and the source disconnected");
if (streamStatus){streamStatus.mapped[0] = STRMSTAT_SHUTDOWN;}
config->is_active = false;
userSelect.clear();
}
}
void inputBuffer::userLeadIn(){
@ -487,22 +476,21 @@ namespace Mist{
/*LTS-END*/
connectedUsers = 0;
//Store child process PIDs in generatePids.
//These are controlled by the buffer (usually processes) and should not count towards incoming pushes
generatePids.clear();
for (std::map<std::string, pid_t>::iterator it = runningProcs.begin(); it != runningProcs.end(); it++){
generatePids.insert(it->second);
}
hasPush = false;
}
void inputBuffer::userOnActive(size_t id){
///\todo Add tracing of earliest watched keys, to prevent data going out of memory for
/// still-watching viewers
if (users.getStatus(id) != COMM_STATUS_DISCONNECT && users.getStatus(id) & COMM_STATUS_SOURCE){
sourcePids[users.getPid(id)].insert(users.getTrack(id));
if (!M.trackValid(users.getTrack(id))){
users.setStatus(COMM_STATUS_DISCONNECT, id);
return;
}
// GeneratePids holds the pids of the process that generate data, so ignore those for determining if a push is ingested.
if (!generatePids.count(users.getPid(id))){hasPush = true;}
if (M.trackValid(users.getTrack(id)) && !generatePids.count(users.getPid(id))){hasPush = true;}
}
if (!(users.getStatus(id) & COMM_STATUS_DONOTTRACK)){++connectedUsers;}
@ -516,11 +504,20 @@ namespace Mist{
}
}
void inputBuffer::userLeadOut(){
if (config->is_active && streamStatus){streamStatus.mapped[0] = hasPush ? STRMSTAT_READY : STRMSTAT_WAIT;}
if (hasPush){everHadPush = true;}
if (!hasPush && everHadPush && !resumeMode && config->is_active){
Util::logExitReason("source disconnected for non-resumable stream");
if (streamStatus){streamStatus.mapped[0] = STRMSTAT_SHUTDOWN;}
config->is_active = false;
userSelect.clear();
}
/*LTS-START*/
static std::set<size_t> prevValidTracks;
std::set<size_t> validTracks = M.getValidTracks();
if (validTracks != prevValidTracks){
MEDIUM_MSG("Valid tracks count changed from %lu to %lu", prevValidTracks.size(), validTracks.size());
prevValidTracks = validTracks;
if (Triggers::shouldTrigger("LIVE_TRACK_LIST")){
JSON::Value triggerPayload;
@ -545,7 +542,6 @@ namespace Mist{
bool inputBuffer::preRun(){
// This function gets run periodically to make sure runtime updates of the config get parsed.
lastReTime = Util::epoch(); /*LTS*/
std::string strName = config->getString("streamname");
Util::sanitizeName(strName);
strName = strName.substr(0, (strName.find_first_of("+ ")));
@ -553,16 +549,21 @@ namespace Mist{
snprintf(tmpBuf, NAME_BUFFER_SIZE, SHM_STREAM_CONF, strName.c_str());
Util::DTSCShmReader rStrmConf(tmpBuf);
DTSC::Scan streamCfg = rStrmConf.getScan();
if (streamCfg){
JSON::Value configuredProcesses = streamCfg.getMember("processes").asJSON();
checkProcesses(configuredProcesses);
}
//Check if bufferTime setting is correct
uint64_t tmpNum = retrieveSetting(streamCfg, "DVR", "bufferTime");
if (tmpNum < 1000){tmpNum = 1000;}
// if the new value is different, print a message and apply it
if (bufferTime != tmpNum){
DEVEL_MSG("Setting bufferTime from %" PRIu64 " to new value of %" PRIu64, bufferTime, tmpNum);
bufferTime = tmpNum;
}
/*LTS-START*/
//Check if cutTime setting is correct
tmpNum = retrieveSetting(streamCfg, "cut");
// if the new value is different, print a message and apply it
if (cutTime != tmpNum){
@ -570,28 +571,27 @@ namespace Mist{
cutTime = tmpNum;
}
//Check if resume setting is correct
tmpNum = retrieveSetting(streamCfg, "resume");
// if the new value is different, print a message and apply it
if (resumeMode != (bool)tmpNum){
INFO_MSG("Setting resume mode from %s to new value of %s",
resumeMode ? "enabled" : "disabled", tmpNum ? "enabled" : "disabled");
resumeMode = tmpNum;
}
if (!meta){return true;}//abort the rest if we can't write metadata
lastReTime = Util::epoch(); /*LTS*/
//Check if segmentsize setting is correct
tmpNum = retrieveSetting(streamCfg, "segmentsize");
if (M && tmpNum < M.biggestFragment() / 2){tmpNum = M.biggestFragment() / 2;}
// if the new value is different, print a message and apply it
if (tmpNum < meta.biggestFragment() / 2){tmpNum = meta.biggestFragment() / 2;}
segmentSize = meta.getMinimumFragmentDuration();
if (segmentSize != tmpNum){
INFO_MSG("Setting segmentSize from %" PRIu64 " to new value of %" PRIu64, segmentSize, tmpNum);
segmentSize = tmpNum;
if (M && M.getMinimumFragmentDuration() == 0){
meta.setMinimumFragmentDuration(segmentSize);
}
}
if (streamCfg){
JSON::Value configuredProcesses = streamCfg.getMember("processes").asJSON();
checkProcesses(configuredProcesses);
meta.setMinimumFragmentDuration(segmentSize);
}
/*LTS-END*/
return true;
}
@ -643,11 +643,8 @@ namespace Mist{
void inputBuffer::checkProcesses(const JSON::Value &procs){
if (!M.getValidTracks().size()){return;}
std::set<std::string> newProcs;
std::map<std::string, std::string> wouldSelect;
// used for building args
int zero = 0;
int out = fileno(stdout);
int err = fileno(stderr);
char *argarr[3];
@ -660,22 +657,14 @@ namespace Mist{
continue;
}
if (tmp.isMember("source_track")){
std::string sourceTrack = tmp["source_track"].asString();
if (sourceTrack != "null" && findTrack(sourceTrack) == INVALID_TRACK_ID){
// No match - skip this process
continue;
}
std::set<size_t> wouldSelect = Util::findTracks(M, JSON::Value(), "", tmp["source_track"].asStringRef());
// No match - skip this process
if (!wouldSelect.size()){continue;}
}
std::stringstream s;
if (tmp.isMember("track_select")){
std::set<size_t> wouldSelect = Util::wouldSelect(M, tmp["track_select"].asStringRef());
if (!wouldSelect.size()){
// No match - skip this process
continue;
}
for (std::set<size_t>::iterator it = wouldSelect.begin(); it != wouldSelect.end(); it++){
s << *it << " ";
}
// No match - skip this process
if (!wouldSelect.size()){continue;}
}
if (tmp.isMember("track_inhibit")){
std::set<size_t> wouldSelect = Util::wouldSelect(
@ -693,7 +682,6 @@ namespace Mist{
}
}
newProcs.insert(tmp.toString());
wouldSelect[tmp.toString()] = s.str();
}
// shut down deleted/changed processes
@ -722,8 +710,7 @@ namespace Mist{
argarr[1] = (char *)config.c_str();
argarr[2] = 0;
INFO_MSG("Starting process: %s %s", argarr[0], argarr[1]);
INFO_MSG(" WouldSelect is %s", wouldSelect.at(*newProcs.begin()).c_str());
runningProcs[*newProcs.begin()] = Util::Procs::StartPiped(argarr, &zero, &out, &err);
runningProcs[*newProcs.begin()] = Util::Procs::StartPiped(argarr, 0, 0, &err);
}
newProcs.erase(newProcs.begin());
}

3
src/input/input_buffer.h
View file

@ -17,7 +17,8 @@ namespace Mist{
size_t segmentSize; /*LTS*/
uint64_t lastReTime; /*LTS*/
uint64_t finalMillis;
bool hasPush;
bool hasPush;//Is a push currently being received?
bool everHadPush;//Was there ever a push received?
bool resumeMode;
IPC::semaphore *liveMeta;

1
src/input/input_dtsc.cpp
View file

@ -361,7 +361,6 @@ namespace Mist{
thisPacket.reInit(srcConn); // read the next packet before continuing
continue; // parse the next packet before returning
}
thisPacket = DTSC::Packet(thisPacket, M.trackIDToIndex(thisPacket.getTrackId(), getpid()));
return; // we have a packet
}
}

5
src/input/input_dtsc.h
View file

@ -27,6 +27,11 @@ namespace Mist{
inputDTSC(Util::Config *cfg);
bool needsLock();
virtual std::string getConnectedBinHost(){
if (srcConn){return srcConn.getBinHost();}
return Input::getConnectedBinHost();
}
protected:
// Private Functions
bool openStreamSource();

17
src/input/input_ebml.cpp
View file

@ -114,12 +114,17 @@ namespace Mist{
while (ptr.size() < needed){
if (!ptr.allocate(needed)){return false;}
int64_t toRead = needed - ptr.size();
if (!fread(ptr + ptr.size(), toRead, 1, inFile)){
// We assume if there is no current data buffered, that we are at EOF and don't print a warning
if (ptr.size()){
FAIL_MSG("Could not read more data! (have %zu, need %" PRIu32 ")", ptr.size(), needed);
int readResult = 0;
while (!readResult){
readResult = fread(ptr + ptr.size(), toRead, 1, inFile);
if (!readResult){
if (errno == EINTR){continue;}
// At EOF we don't print a warning
if (!feof(inFile)){
FAIL_MSG("Could not read more data! (have %zu, need %" PRIu32 ")", ptr.size(), needed);
}
return false;
}
return false;
}
totalBytes += toRead;
ptr.size() = needed;
@ -463,7 +468,7 @@ namespace Mist{
}break;
}
}
thisPacket.genericFill(C.time, C.offset, M.trackIDToIndex(C.track, getpid()), C.ptr, C.dsize,
thisPacket.genericFill(C.time, C.offset, C.track, C.ptr, C.dsize,
C.bpos, C.key);
}

84
src/input/input_hls.cpp
View file

@ -525,8 +525,7 @@ namespace Mist{
std::string test = root.link(entry.filename).getFilePath();
fileSource.open(test.c_str(), std::ios::ate | std::ios::binary);
if (!fileSource.good()){WARN_MSG("file: %s, error: %s", test.c_str(), strerror(errno));}
entry.byteEnd = fileSource.tellg();
totalBytes += entry.byteEnd;
totalBytes += fileSource.tellg();
}
entry.timestamp = lastTimestamp + startTime;
@ -592,12 +591,9 @@ namespace Mist{
void inputHLS::parseStreamHeader(){
if (!initPlaylist(config->getString("input"))){
FAIL_MSG("Failed to load HLS playlist, aborting");
myMeta = DTSC::Meta();
return;
}
myMeta = DTSC::Meta();
myMeta.live = true;
myMeta.vod = false;
meta.reInit(config->getString("streamname"), false);
INFO_MSG("Parsing live stream to create header...");
TS::Packet packet; // to analyse and extract data
int counter = 1;
@ -612,7 +608,7 @@ namespace Mist{
for (std::deque<playListEntries>::iterator entryIt = pListIt->second.begin();
entryIt != pListIt->second.end(); ++entryIt){
nProxy.userClient.keepAlive();
keepAlive();
if (!segDowner.loadSegment(*entryIt)){
WARN_MSG("Skipping segment that could not be loaded in an attempt to recover");
tsStream.clear();
@ -633,21 +629,24 @@ namespace Mist{
tsStream.getEarliestPacket(headerPack);
int tmpTrackId = headerPack.getTrackId();
uint64_t packetId = pidMapping[(((uint64_t)pListIt->first) << 32) + tmpTrackId];
if (packetId == 0){
pidMapping[(((uint64_t)pListIt->first) << 32) + headerPack.getTrackId()] = counter;
pidMappingR[counter] = (((uint64_t)pListIt->first) << 32) + headerPack.getTrackId();
packetId = counter;
VERYHIGH_MSG("Added file %s, trackid: %zu, mapped to: %d",
entryIt->filename.c_str(), headerPack.getTrackId(), counter);
VERYHIGH_MSG("Added file %s, trackid: %zu, mapped to: %d", entryIt->filename.c_str(),
headerPack.getTrackId(), counter);
counter++;
}
if ((!myMeta.tracks.count(packetId) || !myMeta.tracks[packetId].codec.size())){
tsStream.initializeMetadata(myMeta, tmpTrackId, packetId);
myMeta.tracks[packetId].minKeepAway = globalWaitTime * 2000;
VERYHIGH_MSG("setting minKeepAway = %d for track: %" PRIu64,
myMeta.tracks[packetId].minKeepAway, packetId);
size_t idx = M.trackIDToIndex(packetId, getpid());
if ((idx == INVALID_TRACK_ID || !M.getCodec(idx).size())){
tsStream.initializeMetadata(meta, tmpTrackId, packetId);
idx = M.trackIDToIndex(packetId, getpid());
if (idx != INVALID_TRACK_ID){
meta.setMinKeepAway(idx, globalWaitTime * 2000);
VERYHIGH_MSG("setting minKeepAway = %" PRIu32 " for track: %zu", globalWaitTime * 2000, idx);
}
}
}
break; // we have all tracks discovered, next playlist!
@ -655,8 +654,6 @@ namespace Mist{
}while (!segDowner.atEnd());
if (preCounter < counter){break;}// We're done reading this playlist!
}
in.close();
}
tsStream.clear();
currentPlaylist = 0;
@ -673,8 +670,6 @@ namespace Mist{
meta.reInit(config->getString("streamname"), config->getString("input") + ".dtsh");
hasHeader = (bool)M;
if (M){return true;}
if (!hasHeader){meta.reInit(config->getString("streamname"), true);}
TS::Packet packet; // to analyse and extract data
@ -704,7 +699,7 @@ namespace Mist{
DTSC::Packet headerPack;
tsStream.getEarliestPacket(headerPack);
int tmpTrackId = headerPack.getTrackId();
size_t tmpTrackId = headerPack.getTrackId();
uint64_t packetId = pidMapping[(((uint64_t)pListIt->first) << 32) + tmpTrackId];
if (packetId == 0){
@ -717,10 +712,8 @@ namespace Mist{
}
size_t idx = M.trackIDToIndex(packetId, getpid());
INFO_MSG("PacketID: %" PRIu64 ", pid: %d, mapped to %zu", packetId, getpid(), idx);
if (!hasHeader && (idx == INVALID_TRACK_ID || !M.getCodec(idx).size())){
tsStream.initializeMetadata(meta, tmpTrackId, packetId);
INFO_MSG("InitializingMeta for track %zu -> %zu", tmpTrackId, packetId);
idx = M.trackIDToIndex(packetId, getpid());
}
@ -757,6 +750,7 @@ namespace Mist{
counter++;
}
size_t idx = M.trackIDToIndex(packetId, getpid());
if (!hasHeader && (idx == INVALID_TRACK_ID || !M.getCodec(idx).size())){
tsStream.initializeMetadata(meta, tmpTrackId, packetId);
idx = M.trackIDToIndex(packetId, getpid());
@ -781,7 +775,6 @@ namespace Mist{
INFO_MSG("write header file...");
M.toFile((config->getString("input") + ".dtsh").c_str());
in.close();
return true;
}
@ -794,26 +787,29 @@ namespace Mist{
INSANE_MSG("Getting next");
uint32_t tid = 0;
bool finished = false;
if (userSelect.size()){tid = userSelect.begin()->first;}
thisPacket.null();
while (config->is_active && (needsLock() || keepAlive())){
// Check if we have a packet
bool hasPacket = false;
if (streamIsLive){
if (idx == INVALID_TRACK_ID){
hasPacket = tsStream.hasPacketOnEachTrack() || (segDowner.atEnd() && tsStream.hasPacket());
}else{
hasPacket = tsStream.hasPacket(M.getID(idx) & 0xFFFF);
hasPacket = tsStream.hasPacket(getMappedTrackId(M.getID(idx)));
}
// Yes? Excellent! Read and return it.
if (hasPacket){
// Read
if (M.getLive()){
if (idx == INVALID_TRACK_ID){
tsStream.getEarliestPacket(thisPacket);
tid = M.trackIDToIndex((((uint64_t)currentPlaylist) << 16) + thisPacket.getTrackId(), getpid());
tid = getOriginalTrackId(currentPlaylist, thisPacket.getTrackId());
if (!tid){
INFO_MSG("Track %" PRIu64 " on PLS %" PRIu64 " -> %" PRIu32, thisPacket.getTrackId(), currentPlaylist, tid);
continue;
}
}else{
tsStream.getPacket(M.getID(idx) & 0xFFFF, thisPacket);
tsStream.getPacket(getMappedTrackId(M.getID(idx)), thisPacket);
}
if (!thisPacket){
FAIL_MSG("Could not getNext TS packet!");
@ -850,8 +846,8 @@ namespace Mist{
plsTimeOffset[currentPlaylist] +=
(int64_t)(plsLastTime[currentPlaylist] + plsInterval[currentPlaylist]) - (int64_t)newTime;
newTime = thisPacket.getTime() + plsTimeOffset[currentPlaylist];
INFO_MSG("[Guess; New offset: %" PRId64 " -> %" PRId64 "] Packet %" PRIu32
"@%" PRIu64 "ms -> %" PRIu64 "ms",
INFO_MSG("[Guess; New offset: %" PRId64 " -> %" PRId64 "] Packet %" PRIu32 "@%" PRIu64
"ms -> %" PRIu64 "ms",
prevOffset, plsTimeOffset[currentPlaylist], tid, thisPacket.getTime(), newTime);
}
}
@ -891,24 +887,28 @@ namespace Mist{
// No? Then we want to try reading the next file.
// No segments? Wait until next playlist reloading time.
currentPlaylist = firstSegment();
if (idx != INVALID_TRACK_ID){
currentPlaylist = getMappedTrackPlaylist(M.getID(idx));
}else{
currentPlaylist = firstSegment();
}
if (currentPlaylist < 0){
VERYHIGH_MSG("Waiting for segments...");
if (nProxy.userClient.isAlive()){nProxy.userClient.keepAlive();}
keepAlive();
Util::wait(500);
continue;
}
// Now that we know our playlist is up-to-date, actually try to read the file.
VERYHIGH_MSG("Moving on to next TS segment (variant %" PRIu32 ")", currentPlaylist);
VERYHIGH_MSG("Moving on to next TS segment (variant %" PRIu64 ")", currentPlaylist);
if (readNextFile()){
MEDIUM_MSG("Next segment read successfully");
finished = false;
continue; // Success! Continue regular parsing.
}else{
if (selectedTracks.size() > 1){
if (userSelect.size() > 1){
// failed to read segment for playlist, dropping it
WARN_MSG("Dropping variant %" PRIu32 " because we couldn't read anything from it", currentPlaylist);
WARN_MSG("Dropping variant %" PRIu64 " because we couldn't read anything from it", currentPlaylist);
tthread::lock_guard<tthread::mutex> guard(entryMutex);
listEntries.erase(currentPlaylist);
if (listEntries.size()){continue;}
@ -946,17 +946,17 @@ namespace Mist{
currentIndex = plistEntry - 1;
currentPlaylist = getMappedTrackPlaylist(trackId);
INFO_MSG("Seeking to index %d on playlist %d", currentIndex, currentPlaylist);
INFO_MSG("Seeking to index %zu on playlist %" PRIu64, currentIndex, currentPlaylist);
{// Lock mutex for listEntries
tthread::lock_guard<tthread::mutex> guard(entryMutex);
if (!listEntries.count(currentPlaylist)){
WARN_MSG("Playlist %d not loaded, aborting seek", currentPlaylist);
WARN_MSG("Playlist %" PRIu64 " not loaded, aborting seek", currentPlaylist);
return;
}
std::deque<playListEntries> &curPlaylist = listEntries[currentPlaylist];
if (curPlaylist.size() <= currentIndex){
WARN_MSG("Playlist %d has %zu <= %d entries, aborting seek", currentPlaylist,
WARN_MSG("Playlist %" PRIu64 " has %zu <= %zu entries, aborting seek", currentPlaylist,
curPlaylist.size(), currentIndex);
return;
}
@ -1179,7 +1179,7 @@ namespace Mist{
tthread::lock_guard<tthread::mutex> guard(entryMutex);
std::deque<playListEntries> &curList = listEntries[currentPlaylist];
if (!curList.size()){
WARN_MSG("no entries found in playlist: %d!", currentPlaylist);
WARN_MSG("no entries found in playlist: %" PRIu64 "!", currentPlaylist);
return false;
}
if (!streamIsLive){
@ -1204,7 +1204,7 @@ namespace Mist{
if (Util::bootSecs() < ntry.timestamp){
VERYHIGH_MSG("Slowing down to realtime...");
while (Util::bootSecs() < ntry.timestamp){
if (nProxy.userClient.isAlive()){nProxy.userClient.keepAlive();}
keepAlive();
Util::wait(250);
}
}
@ -1228,7 +1228,7 @@ namespace Mist{
/// this will keep the playlists in sync while reading segments.
size_t inputHLS::firstSegment(){
// Only one selected? Immediately return the right playlist.
if (userSelect.size() == 1){return ((M.getID(userSelect.begin()->first) >> 16) & 0xFFFF);}
if (userSelect.size() == 1){return getMappedTrackPlaylist(M.getID(userSelect.begin()->first));}
uint64_t firstTimeStamp = 0;
int tmpId = -1;
int segCount = 0;

6
src/input/input_hls.h
View file

@ -121,7 +121,7 @@ namespace Mist{
Socket::Connection conn;
TS::Packet tsBuf;
int firstSegment();
size_t firstSegment();
void waitForNextSegment();
void readPMT();
bool checkArguments();
@ -130,7 +130,6 @@ namespace Mist{
bool needHeader(){return true;}
void getNext(size_t idx = INVALID_TRACK_ID);
void seek(uint64_t seekTime, size_t idx = INVALID_TRACK_ID);
FILE *inFile;
FILE *tsFile;
@ -141,6 +140,9 @@ namespace Mist{
void parseStreamHeader();
uint32_t getMappedTrackId(uint64_t id);
uint32_t getMappedTrackPlaylist(uint64_t id);
uint64_t getOriginalTrackId(uint32_t playlistId, uint32_t id);
size_t getEntryId(uint32_t playlistId, uint64_t bytePos);
};
}// namespace Mist

16
src/input/input_playlist.cpp
View file

@ -39,17 +39,19 @@ namespace Mist{
return true;
}
std::string inputPlaylist::streamMainLoop(){
void inputPlaylist::streamMainLoop(){
bool seenValidEntry = true;
uint64_t startTime = Util::bootMS();
while (config->is_active && nProxy.userClient.isAlive()){
while (config->is_active){
struct tm *wTime;
time_t nowTime = time(0);
wTime = localtime(&nowTime);
wallTime = wTime->tm_hour * 60 + wTime->tm_min;
nProxy.userClient.keepAlive();
reloadPlaylist();
if (!playlist.size()){return "No entries in playlist";}
if (!playlist.size()){
Util::logExitReason("No entries in playlist");
return;
}
++playlistIndex;
if (playlistIndex >= playlist.size()){
if (!seenValidEntry){
@ -103,7 +105,7 @@ namespace Mist{
continue;
}
seenValidEntry = true;
while (Util::Procs::isRunning(spawn_pid) && nProxy.userClient.isAlive() && config->is_active){
while (Util::Procs::isRunning(spawn_pid) && config->is_active){
Util::sleep(1000);
if (reloadOn != 0xFFFF){
time_t nowTime = time(0);
@ -117,13 +119,9 @@ namespace Mist{
Util::Procs::Stop(spawn_pid);
}
}
nProxy.userClient.keepAlive();
}
if (!config->is_active && Util::Procs::isRunning(spawn_pid)){Util::Procs::Stop(spawn_pid);}
}
if (!config->is_active){return "received deactivate signal";}
if (!nProxy.userClient.isAlive()){return "buffer shutdown";}
return "Unknown";
}
void inputPlaylist::reloadPlaylist(){

5
src/input/input_playlist.h
View file

@ -11,9 +11,10 @@ namespace Mist{
protected:
bool checkArguments();
bool readHeader(){return true;}
virtual void parseStreamHeader(){myMeta.tracks[1].codec = "PLACEHOLDER";}
std::string streamMainLoop();
virtual void parseStreamHeader(){}
void streamMainLoop();
virtual bool needHeader(){return false;}
virtual bool publishesTracks(){return false;}
private:
void reloadPlaylist();

52
src/input/input_rtsp.cpp
View file

@ -45,6 +45,8 @@ namespace Mist{
capa["codecs"][0u][0u].append("H264");
capa["codecs"][0u][0u].append("HEVC");
capa["codecs"][0u][0u].append("MPEG2");
capa["codecs"][0u][0u].append("VP8");
capa["codecs"][0u][0u].append("VP9");
capa["codecs"][0u][1u].append("AAC");
capa["codecs"][0u][1u].append("MP3");
capa["codecs"][0u][1u].append("AC3");
@ -194,45 +196,45 @@ namespace Mist{
tcpCon.close();
}
std::string InputRTSP::streamMainLoop(){
IPC::sharedClient statsPage = IPC::sharedClient(SHM_STATISTICS, STAT_EX_SIZE, true);
void InputRTSP::streamMainLoop(){
Comms::Statistics statComm;
uint64_t startTime = Util::epoch();
uint64_t lastPing = Util::bootSecs();
uint64_t lastSecs = 0;
while (keepAlive() && parsePacket()){
uint64_t currSecs = Util::bootSecs();
handleUDP();
if (Util::bootSecs() - lastPing > 30){
sendCommand("GET_PARAMETER", url.getUrl(), "");
lastPing = Util::bootSecs();
}
if (lastSecs != currSecs){
if (!statsPage.getData()){
statsPage = IPC::sharedClient(SHM_STATISTICS, STAT_EX_SIZE, true);
}
if (statsPage.getData()){
if (!statsPage.isAlive()){
lastSecs = currSecs;
// Connect to stats for INPUT detection
statComm.reload();
if (statComm){
if (!statComm.isAlive()){
config->is_active = false;
statsPage.finish();
return "received shutdown request from controller";
Util::logExitReason("received shutdown request from controller");
return;
}
uint64_t now = Util::epoch();
IPC::statExchange tmpEx(statsPage.getData());
tmpEx.now(now);
tmpEx.crc(getpid());
tmpEx.streamName(streamName);
tmpEx.connector("INPUT");
tmpEx.up(tcpCon.dataUp());
tmpEx.down(tcpCon.dataDown());
tmpEx.time(now - startTime);
tmpEx.lastSecond(0);
statsPage.keepAlive();
uint64_t now = Util::bootSecs();
statComm.setNow(now);
statComm.setCRC(getpid());
statComm.setStream(streamName);
statComm.setConnector("INPUT:" + capa["name"].asStringRef());
statComm.setUp(tcpCon.dataUp());
statComm.setDown(tcpCon.dataDown());
statComm.setTime(now - startTime);
statComm.setLastSecond(0);
statComm.setHost(getConnectedBinHost());
statComm.keepAlive();
}
}
}
statsPage.finish();
if (!tcpCon){return "TCP connection closed";}
if (!config->is_active){return "received deactivate signal";}
if (!keepAlive()){return "buffer shutdown";}
return "Unknown";
if (!tcpCon){
Util::logExitReason("TCP connection closed");
}
}
bool InputRTSP::parsePacket(bool mustHave){

7
src/input/input_rtsp.h
View file

@ -17,6 +17,11 @@ namespace Mist{
void incoming(const DTSC::Packet &pkt);
void incomingRTP(const uint64_t track, const RTP::Packet &p);
virtual std::string getConnectedBinHost(){
if (tcpCon){return tcpCon.getBinHost();}
return Input::getConnectedBinHost();
}
protected:
// Private Functions
bool checkArguments();
@ -29,7 +34,7 @@ namespace Mist{
const std::map<std::string, std::string> *extraHeaders = 0, bool reAuth = true);
bool parsePacket(bool mustHave = false);
bool handleUDP();
std::string streamMainLoop();
void streamMainLoop();
Socket::Connection tcpCon;
HTTP::Parser sndH, recH;
HTTP::URL url;

18
src/input/input_ts.cpp
View file

@ -463,7 +463,7 @@ namespace Mist{
tmpIdx = meta.addTrack(0, 0, 0, 0);
}
std::string inputTS::streamMainLoop(){
void inputTS::streamMainLoop(){
meta.removeTrack(tmpIdx);
INFO_MSG("Removed temptrack %zu", tmpIdx);
Comms::Statistics statComm;
@ -495,7 +495,8 @@ namespace Mist{
}
if (!tcpCon){
config->is_active = false;
return "end of streamed input";
Util::logExitReason("end of streamed input");
return;
}
}else{
std::string leftData;
@ -557,17 +558,19 @@ namespace Mist{
if (statComm){
if (!statComm.isAlive()){
config->is_active = false;
return "received shutdown request from controller";
Util::logExitReason("received shutdown request from controller");
return;
}
uint64_t now = Util::bootSecs();
statComm.setNow(now);
statComm.setCRC(getpid());
statComm.setStream(streamName);
statComm.setConnector("INPUT");
statComm.setConnector("INPUT:" + capa["name"].asStringRef());
statComm.setUp(0);
statComm.setDown(downCounter + tcpCon.dataDown());
statComm.setTime(now - startTime);
statComm.setLastSecond(0);
statComm.setHost(getConnectedBinHost());
statComm.keepAlive();
}
@ -577,7 +580,8 @@ namespace Mist{
if (hasStarted && !threadTimer.size()){
if (!isAlwaysOn()){
config->is_active = false;
return "no active threads and we had input in the past";
Util::logExitReason("no active threads and we had input in the past");
return;
}else{
hasStarted = false;
}
@ -607,13 +611,13 @@ namespace Mist{
if (Util::bootSecs() - noDataSince > 20){
if (!isAlwaysOn()){
config->is_active = false;
return "No packets received for 20 seconds - terminating";
Util::logExitReason("no packets received for 20 seconds");
return;
}else{
noDataSince = Util::bootSecs();
}
}
}
return "received shutdown request";
}
void inputTS::finish(){

7
src/input/input_ts.h
View file

@ -14,6 +14,11 @@ namespace Mist{
~inputTS();
bool needsLock();
virtual std::string getConnectedBinHost(){
if (tcpCon){return tcpCon.getBinHost();}
/// \TODO Handle UDP
return Input::getConnectedBinHost();
}
protected:
// Private Functions
bool checkArguments();
@ -25,7 +30,7 @@ namespace Mist{
void readPMT();
bool openStreamSource();
void parseStreamHeader();
std::string streamMainLoop();
void streamMainLoop();
void finish();
FILE *inFile; ///< The input file with ts data
TS::Stream tsStream; ///< Used for parsing the incoming ts stream

16
src/io.cpp
View file

@ -303,10 +303,15 @@ namespace Mist{
/// Initiates/continues negotiation with the buffer as well
///\param packet The packet to buffer
void InOutBase::bufferLivePacket(const DTSC::Packet &packet){
size_t idx = M.trackIDToIndex(packet.getTrackId(), getpid());
if (idx == INVALID_TRACK_ID){
INFO_MSG("Packet for track %zu has no valid index!", packet.getTrackId());
return;
}
char *data;
size_t dataLen;
packet.getString("data", data, dataLen);
bufferLivePacket(packet.getTime(), packet.getInt("offset"), packet.getTrackId(), data, dataLen,
bufferLivePacket(packet.getTime(), packet.getInt("offset"), idx, data, dataLen,
packet.getInt("bpos"), packet.getFlag("keyframe"));
/// \TODO META Build something that should actually be able to deal with "extra" values
}
@ -329,7 +334,7 @@ namespace Mist{
// Assume this is the first packet on the track
isKeyframe = true;
}else{
if (packTime - tPages.getInt("lastkeytime", tPages.getEndPos() - 1) >= 5000){
if (packTime - tPages.getInt("lastkeytime", tPages.getEndPos() - 1) >= AUDIO_KEY_INTERVAL){
isKeyframe = true;
}
}
@ -343,9 +348,8 @@ namespace Mist{
packTime, M.getLastms(packTrack));
return;
}
if (packet.getTime() > myMeta.tracks[tid].lastms + 30000 && myMeta.tracks[tid].lastms){
WARN_MSG("Sudden jump in timestamp from %" PRIu64 " to %" PRIu64, myMeta.tracks[tid].lastms,
packet.getTime());
if (packTime > M.getLastms(packTrack) + 30000 && M.getLastms(packTrack)){
WARN_MSG("Sudden jump in timestamp from %" PRIu64 " to %" PRIu64, M.getLastms(packTrack), packTime);
}
}
@ -361,7 +365,7 @@ namespace Mist{
tPages.setInt("firstkey", 0, 0);
tPages.setInt("firsttime", packTime, 0);
tPages.setInt("size", DEFAULT_DATA_PAGE_SIZE, 0);
tPages.setInt("keycount", 0, endPage);
tPages.setInt("keycount", 0, 0);
tPages.setInt("avail", 0, 0);
++endPage;
}

4
src/output/mist_out.cpp
View file

@ -12,7 +12,7 @@ int spawnForked(Socket::Connection &S){
void handleUSR1(int signum, siginfo_t *sigInfo, void *ignore){
HIGH_MSG("USR1 received - triggering rolling restart");
Util::Config::is_restarting = true;
Util::Config::logExitReason("setting is_active to false because of received USR1");
Util::logExitReason("signal USR1");
Util::Config::is_active = false;
}
@ -47,5 +47,7 @@ int main(int argc, char *argv[]){
return tmp.run();
}
}
INFO_MSG("Exit reason: %s", Util::exitReason);
return 0;
}

325
src/output/output.cpp
View file

@ -48,7 +48,6 @@ namespace Mist{
Output::Output(Socket::Connection &conn) : myConn(conn){
pushing = false;
pushIsOngoing = false;
firstTime = 0;
firstPacketTime = 0xFFFFFFFFFFFFFFFFull;
lastPacketTime = 0;
@ -67,6 +66,10 @@ namespace Mist{
lastRecv = Util::bootSecs();
if (myConn){
setBlocking(true);
//Make sure that if the socket is a non-stdio socket, we close it when forking
if (myConn.getSocket() > 2){
Util::Procs::socketList.insert(myConn.getSocket());
}
}else{
WARN_MSG("Warning: MistOut created with closed socket!");
}
@ -129,6 +132,7 @@ namespace Mist{
}else{
MEDIUM_MSG("onFail '%s': %s", streamName.c_str(), msg.c_str());
}
Util::logExitReason(msg.c_str());
isInitialized = false;
wantRequest = true;
parseData = false;
@ -143,7 +147,7 @@ namespace Mist{
}
reconnect();
// if the connection failed, fail
if (streamName.size() < 1){
if (!meta || streamName.size() < 1){
onFail("Could not connect to stream", true);
return;
}
@ -181,19 +185,8 @@ namespace Mist{
if (shmSessions.mapped){
char shmEmpty[SHM_SESSIONS_ITEM];
memset(shmEmpty, 0, SHM_SESSIONS_ITEM);
std::string host = statComm.getHost();
if (host.substr(0, 12) ==
std::string("\000\000\000\000\000\000\000\000\000\000\377\377", 12)){
char tmpstr[16];
snprintf(tmpstr, 16, "%hhu.%hhu.%hhu.%hhu", host[12], host[13], host[14], host[15]);
host = tmpstr;
}else{
char tmpstr[40];
snprintf(tmpstr, 40, "%.2x%.2x:%.2x%.2x:%.2x%.2x:%.2x%.2x:%.2x%.2x:%.2x%.2x:%.2x%.2x:%.2x%.2x",
host[0], host[1], host[2], host[3], host[4], host[5], host[6], host[7],
host[8], host[9], host[10], host[11], host[12], host[13], host[14], host[15]);
host = tmpstr;
}
std::string host;
Socket::hostBytesToStr(statComm.getHost().data(), 16, host);
uint32_t shmOffset = 0;
const std::string &cName = capa["name"].asStringRef();
while (shmOffset + SHM_SESSIONS_ITEM < SHM_SESSIONS_SIZE){
@ -259,12 +252,19 @@ namespace Mist{
std::string Output::getConnectedHost(){return myConn.getHost();}
std::string Output::getConnectedBinHost(){return myConn.getBinHost();}
std::string Output::getConnectedBinHost(){
if (!prevHost.size()){
if (myConn && myConn.getPureSocket() != -1){
prevHost = myConn.getBinHost();
}
if (!prevHost.size()){prevHost.assign("\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\001", 16);}
}
return prevHost;
}
bool Output::isReadyForPlay(){
// If a protocol does not support any codecs, we assume you know what you're doing
if (!capa.isMember("codecs")){return true;}
if (isPushing()){return true;}
if (!isInitialized){initialize();}
meta.refresh();
if (getSupportedTracks().size()){
@ -363,27 +363,25 @@ namespace Mist{
while (!meta && ++attempts < 20 && Util::streamAlive(streamName)){
meta.reInit(streamName, false);
}
if (!meta){
onFail("Could not connect to stream data", true);
return;
}
if (!meta){return;}
meta.refresh();
isInitialized = true;
statComm.reload();
stats(true);
if (!pushing){selectDefaultTracks();}
if (!M.getVod() && !isReadyForPlay()){
uint64_t waitUntil = Util::epoch() + 30;
if (isPushing()){return;}
if (!isRecording() && !M.getVod() && !isReadyForPlay()){
uint64_t waitUntil = Util::bootSecs() + 45;
while (!M.getVod() && !isReadyForPlay()){
if (Util::epoch() > waitUntil + 45 || (!userSelect.size() && Util::epoch() > waitUntil)){
INFO_MSG("Giving up waiting for playable tracks. Stream: %s, IP: %s", streamName.c_str(),
getConnectedHost().c_str());
if (Util::bootSecs() > waitUntil || (!userSelect.size() && Util::bootSecs() > waitUntil)){
INFO_MSG("Giving up waiting for playable tracks. IP: %s", getConnectedHost().c_str());
break;
}
Util::wait(500);
meta.refresh();
stats();
}
}
selectDefaultTracks();
}
std::set<size_t> Output::getSupportedTracks(const std::string &type) const{
@ -403,22 +401,11 @@ namespace Mist{
bool autoSeek = buffer.size();
uint64_t seekTarget = currentTime();
std::set<size_t> newSelects =
Util::wouldSelect(myMeta, targetParams, capa, UA, autoSeek ? seekTarget : 0);
std::set<size_t> oldSel;
for (std::set<unsigned long>::iterator selIt = selectedTracks.begin();
selIt != selectedTracks.end(); ++selIt){
oldSel.insert(*selIt);
}
if (oldSel == newSelects){
// No new selections? Do nothing, return no change.
return false;
}
Util::wouldSelect(M, targetParams, capa, UA, autoSeek ? seekTarget : 0);
if (autoSeek){
std::set<size_t> toRemove;
for (std::set<size_t>::iterator it = wouldSelect.begin(); it != wouldSelect.end(); it++){
for (std::set<size_t>::iterator it = newSelects.begin(); it != newSelects.end(); it++){
// autoSeeking and target not in bounds? Drop it too.
if (M.getLastms(*it) < std::max(seekTarget, (uint64_t)6000lu) - 6000){
toRemove.insert(*it);
@ -426,7 +413,7 @@ namespace Mist{
}
// remove those from selectedtracks
for (std::set<size_t>::iterator it = toRemove.begin(); it != toRemove.end(); it++){
wouldSelect.erase(*it);
newSelects.erase(*it);
}
}
@ -435,7 +422,7 @@ namespace Mist{
userSelect.clear();
// Select tracks here!
for (std::set<size_t>::iterator it = wouldSelect.begin(); it != wouldSelect.end(); it++){
for (std::set<size_t>::iterator it = newSelects.begin(); it != newSelects.end(); it++){
userSelect[*it].reload(streamName, *it);
}
@ -453,6 +440,14 @@ namespace Mist{
parseData = false;
}
///Returns the timestamp of the next upcoming keyframe after thisPacket, or 0 if that cannot be determined (yet).
uint64_t Output::nextKeyTime(){
DTSC::Keys keys(M.keys(getMainSelectedTrack()));
if (!keys.getValidCount()){return 0;}
size_t keyNum = keys.getNumForTime(lastPacketTime);
return keys.getTime(keyNum+1);
}
uint64_t Output::pageNumForKey(size_t trackId, size_t keyNum){
const Util::RelAccX &tPages = M.pages(trackId);
for (uint64_t i = tPages.getDeleted(); i < tPages.getEndPos(); i++){
@ -671,7 +666,7 @@ namespace Mist{
for (std::set<size_t>::iterator it = seekTracks.begin(); it != seekTracks.end(); it++){
seek(*it, pos, false);
}
firstTime = Util::bootMS() - currentTime();
firstTime = Util::bootMS() - (currentTime() * realTime / 1000);
}
bool Output::seek(size_t tid, uint64_t pos, bool getNextKey){
@ -690,7 +685,7 @@ namespace Mist{
stats();
}
}
if (meta.getLastms(tid) <= pos){
if (meta.getLastms(tid) < pos){
WARN_MSG("Aborting seek to %" PRIu64 "ms in track %zu: past end of track (= %" PRIu64 "ms).",
pos, tid, meta.getLastms(tid));
userSelect.erase(tid);
@ -751,7 +746,7 @@ namespace Mist{
if (curPage[tid].mapped[tmp.offset]){return seek(tid, pos, getNextKey);}
FAIL_MSG("Track %zu no data (key %zu@%" PRIu64 ") - timeout", tid, keyNum + (getNextKey ? 1 : 0), tmp.offset);
userSelect.erase(tid);
firstTime = Util::bootMS() - buffer.begin()->time;
firstTime = Util::bootMS() - (buffer.begin()->time * realTime / 1000);
return false;
}
@ -761,6 +756,7 @@ namespace Mist{
/// needsLookAhead+minKeepAway ms from the end. Unless lastms < 5000, then it seeks to the first
/// keyframe of the main track. Aborts if there is no main track or it has no keyframes.
void Output::initialSeek(){
if (!meta){return;}
uint64_t seekPos = 0;
if (meta.getLive()){
size_t mainTrack = getMainSelectedTrack();
@ -980,6 +976,7 @@ namespace Mist{
/// It seeks to the last sync'ed keyframe of the main track, no closer than needsLookAhead+minKeepAway ms from the end.
/// Aborts if not live, there is no main track or it has no keyframes.
bool Output::liveSeek(){
if (!realTime){return false;}//Makes no sense when playing in turbo mode
static uint32_t seekCount = 2;
uint64_t seekPos = 0;
if (!meta.getLive()){return false;}
@ -989,11 +986,15 @@ namespace Mist{
uint64_t cTime = thisPacket.getTime();
uint64_t mKa = getMinKeepAway();
if (!maxSkipAhead){
bool noReturn = false;
uint64_t newSpeed = 1000;
if (lMs - mKa - needsLookAhead - extraKeepAway > cTime + 50){
// We need to speed up!
uint64_t diff = (lMs - mKa - needsLookAhead - extraKeepAway) - cTime;
if (diff > 1000){
if (diff > 3000){
noReturn = true;
newSpeed = 1000;
}else if (diff > 1000){
newSpeed = 750;
}else if (diff > 500){
newSpeed = 900;
@ -1002,13 +1003,11 @@ namespace Mist{
}
}
if (realTime != newSpeed){
HIGH_MSG("Changing playback speed from %" PRIu64 " to %" PRIu64 "(%" PRIu64
" ms LA, %" PRIu64 " ms mKA, %lu eKA)",
realTime, newSpeed, needsLookAhead, mKa, extraKeepAway);
firstTime = Util::bootMS() - cTime;
HIGH_MSG("Changing playback speed from %" PRIu64 " to %" PRIu64 "(%" PRIu64 " ms LA, %" PRIu64 " ms mKA, %lu eKA)", realTime, newSpeed, needsLookAhead, mKa, extraKeepAway);
firstTime = Util::bootMS() - (cTime * newSpeed / 1000);
realTime = newSpeed;
}
return false;
if (!noReturn){return false;}
}
// cancel if there are no keys in the main track
if (mainTrack == INVALID_TRACK_ID){return false;}
@ -1167,8 +1166,14 @@ namespace Mist{
while (keepGoing() && (wantRequest || parseData)){
if (wantRequest){requestHandler();}
if (parseData){
if (!isInitialized){initialize();}
if (!sentHeader){
if (!isInitialized){
initialize();
if (!isInitialized){
onFail("Stream initialization failed");
break;
}
}
if (!sentHeader && keepGoing()){
DONTEVEN_MSG("sendHeader");
sendHeader();
}
@ -1186,6 +1191,8 @@ namespace Mist{
Util::sleep(std::min(thisPacket.getTime() -
((((Util::bootMS() - firstTime) * 1000) + maxSkipAhead) / realTime),
1000ul));
//Make sure we stay responsive to requests and stats while waiting
if (wantRequest){requestHandler();}
stats();
}
}
@ -1217,6 +1224,8 @@ namespace Mist{
}else{
playbackSleep(sleepTime);
}
//Make sure we stay responsive to requests and stats while waiting
if (wantRequest){requestHandler();}
stats();
}
if (!timeoutTries){
@ -1237,6 +1246,7 @@ namespace Mist{
INFO_MSG("Switching to next push target filename: %s", newTarget.c_str());
if (!connectToFile(newTarget)){
FAIL_MSG("Failed to open file, aborting: %s", newTarget.c_str());
Util::logExitReason("failed to open file, aborting: %s", newTarget.c_str());
onFinish();
break;
}
@ -1247,13 +1257,14 @@ namespace Mist{
}else{
if (!onFinish()){
INFO_MSG("Shutting down because planned stopping point reached");
Util::logExitReason("planned stopping point reached");
break;
}
}
}
sendNext();
}else{
INFO_MSG("Shutting down because of stream end");
Util::logExitReason("end of stream");
/*LTS-START*/
if (Triggers::shouldTrigger("CONN_STOP", streamName)){
std::string payload =
@ -1265,18 +1276,14 @@ namespace Mist{
}
}
if (!meta){
Util::Config::logExitReason("No connection to the metadata");
Util::logExitReason("lost internal connection to stream data");
break;
}
}
stats();
}
MEDIUM_MSG("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");
if (Util::Config::exitReason.size()){
INFO_MSG("Logged exit reason: %s", Util::Config::exitReason.c_str());
}
if (!myConn){Util::logExitReason("remote connection closed");}
INFO_MSG("Client handler shutting down, exit reason: %s", Util::exitReason);
onFinish();
/*LTS-START*/
@ -1460,42 +1467,66 @@ namespace Mist{
DTSC::Keys keys(M.keys(nxt.tid));
size_t thisKey = keys.getNumForTime(nxt.time);
while (!nextTime && keepGoing()){
// The next packet is either not available yet, or on another page
// Check if we have a next valid packet
if (memcmp(curPage[nxt.tid].mapped + nxt.offset + preLoad.getDataLen(), "\000\000\000\000", 4)){
nextTime = getDTSCTime(curPage[nxt.tid].mapped, nxt.offset + preLoad.getDataLen());
// After 500ms, we assume the time will not update anymore
if (++emptyCount >= 20){break;}
// Check if we have a next valid packet
if (memcmp(curPage[nxt.tid].mapped + nxt.offset + preLoad.getDataLen(), "\000\000\000\000", 4)){
nextTime = getDTSCTime(curPage[nxt.tid].mapped, nxt.offset + preLoad.getDataLen());
if (!nextTime){
WARN_MSG("Next packet is available, but has no time. Please warn the developers if you see this message!");
dropTrack(nxt.tid, "EOP: invalid next packet");
return false;
}
}else{
//no next packet yet!
//Check if this is the last packet of a VoD stream. Return success and drop the track.
if (M.getVod() && nxt.time >= M.getLastms(nxt.tid)){
thisPacket.reInit(curPage[nxt.tid].mapped + nxt.offset, 0, true);
thisIdx = nxt.tid;
dropTrack(nxt.tid, "end of VoD track reached", false);
return true;
}
//Check if there exists a different page for the next key
size_t nextKeyPage = M.getPageNumberForKey(nxt.tid, thisKey + 1);
if (nextKeyPage != INVALID_KEY_NUM && nextKeyPage != currentPage[nxt.tid]){
// If so, the next key is our next packet
nextTime = keys.getTime(thisKey + 1);
}else{
if (M.getVod() && nxt.time >= M.getLastms(nxt.tid)){break;}
if (M.getPageNumberForKey(nxt.tid, thisKey + 1) != currentPage[nxt.tid]){
// Check if its on a different page
nextTime = keys.getTime(thisKey + 1);
}
//Okay, there's no next page yet, and no next packet on this page either.
//That means we're waiting for data to show up, somewhere.
// after ~25 seconds, give up and drop the track.
if (++emptyCount >= 1000){
// after ~25 seconds, give up and drop the track.
dropTrack(nxt.tid, "EOP: data wait timeout");
return false;
}
Util::sleep(25);
// we're waiting for new data to show up
if (emptyCount % 640 == 0){
reconnect(); // reconnect every 16 seconds
// if we don't have a connection to the metadata here, this means the stream has gone offline in the meanwhile.
if (!meta){return false;}
//every ~1 second, check if the stream is not offline
if (emptyCount % 40 == 0 && M.getLive() && Util::getStreamStatus(streamName) == STRMSTAT_OFF){
Util::logExitReason("Stream source shut down");
thisPacket.null();
return true;
}
//every ~16 seconds, reconnect to metadata
if (emptyCount % 640 == 0){
reconnect();
if (!meta){
onFail("Could not connect to stream data", true);
thisPacket.null();
return true;
}
// if we don't have a connection to the metadata here, this means the stream has gone offline in the meanwhile.
if (!meta){
Util::logExitReason("Attempted reconnect to source failed");
thisPacket.null();
return true;
}
return false;//no sleep after reconnect
}
//Fine! We didn't want a packet, anyway. Let's try again later.
Util::sleep(25);
return false;
}
}
if (M.getVod() && nxt.time >= M.getLastms(nxt.tid)){
thisPacket.reInit(curPage[nxt.tid].mapped + nxt.offset, 0, true);
thisIdx = nxt.tid;
dropTrack(nxt.tid, "end of VoD track reached", false);
return true;
}
// If we don't have a timestamp at all, this is due to a different cause.
if (!nextTime && (emptyCount < 20)){return false;}
//If the next packet should've been before the current packet, something is wrong. Abort, abort!
if (nextTime < nxt.time){
dropTrack(nxt.tid, "time going backwards");
return false;
@ -1513,7 +1544,7 @@ namespace Mist{
emptyCount = 0; // valid packet - reset empty counter
if (!userSelect[nxt.tid].isAlive()){
INFO_MSG("Track %zu is not alive!", nxt.tid);
INFO_MSG("Track %zu is not alive!", nxt.tid);
return false;
}
@ -1526,12 +1557,6 @@ namespace Mist{
// exchange the current packet in the buffer for the next one
buffer.erase(buffer.begin());
if (M.getVod() && nxt.time > M.getLastms(nxt.tid)){
dropTrack(nxt.tid, "detected last VoD packet");
return true;
}
buffer.insert(nxt);
return true;
@ -1541,8 +1566,10 @@ namespace Mist{
/// Outputs used as an input should return INPUT, outputs used for automation should return OUTPUT, others should return their proper name.
/// The default implementation is usually good enough for all the non-INPUT types.
std::string Output::getStatsName(){
if (isPushing()){return "INPUT";}
if (config->hasOption("target") && config->getString("target").size()){return "OUTPUT";}
if (isPushing()){return "INPUT:" + capa["name"].asStringRef();}
if (config->hasOption("target") && config->getString("target").size()){
return "OUTPUT:" + capa["name"].asStringRef();
}
return capa["name"].asStringRef();
}
@ -1554,6 +1581,25 @@ namespace Mist{
uint64_t now = Util::bootSecs();
if (now == lastStats && !force){return;}
if (isRecording()){
static uint64_t lastPushUpdate = now;
if (lastPushUpdate + 5 <= now){
JSON::Value pStat;
pStat["push_status_update"]["id"] = getpid();
JSON::Value & pData = pStat["push_status_update"]["status"];
pData["mediatime"] = currentTime();
for (std::map<size_t, Comms::Users>::iterator it = userSelect.begin(); it != userSelect.end(); it++){
pData["tracks"].append(it->first);
}
pData["bytes"] = myConn.dataUp();
pData["active_seconds"] = (now - myConn.connTime());
Socket::UDPConnection uSock;
uSock.SetDestination("localhost", 4242);
uSock.SendNow(pStat.toString());
lastPushUpdate = now;
}
}
if (!statComm){statComm.reload();}
if (!statComm){return;}
@ -1562,16 +1608,13 @@ namespace Mist{
HIGH_MSG("Writing stats: %s, %s, %u, %lu, %lu", getConnectedHost().c_str(), streamName.c_str(),
crc & 0xFFFFFFFFu, myConn.dataUp(), myConn.dataDown());
/*LTS-START*/
if (statComm.getStatus() == COMM_STATUS_DISCONNECT){
if (statComm.getStatus() == COMM_STATUS_REQDISCONNECT){
onFail("Shutting down on controller request");
return;
}
/*LTS-END*/
statComm.setNow(now);
if (statComm.getHost() ==
std::string("\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000", 16)){
statComm.setHost(getConnectedBinHost());
}
statComm.setHost(getConnectedBinHost());
statComm.setCRC(crc);
statComm.setStream(streamName);
statComm.setConnector(getStatsName());
@ -1597,16 +1640,38 @@ namespace Mist{
doSync();
for (std::map<size_t, Comms::Users>::iterator it = userSelect.begin(); it != userSelect.end(); it++){
it->second.keepAlive();
if (it->second.isAlive() && M.getLive() && it->second.getStatus() & COMM_STATUS_SOURCE){
if (Util::bootSecs() - M.getLastUpdated(it->first) > 3){
INFO_MSG("Not updating data for track %zu?", it->first);
if (isPushing()){
for (std::map<size_t, Comms::Users>::iterator it = userSelect.begin(); it != userSelect.end(); it++){
it->second.keepAlive();
if (it->second.getStatus() == COMM_STATUS_REQDISCONNECT){
if (dropPushTrack(it->second.getTrack(), "disconnect request from buffer")){break;}
}
if (!it->second.isAlive()){
if (dropPushTrack(it->second.getTrack(), "track mapping no longer valid")){break;}
}
//if (Util::bootSecs() - M.getLastUpdated(it->first) > 5){
// if (dropPushTrack(it->second.getTrack(), "track updates being ignored by buffer")){break;}
//}
}
}else{
for (std::map<size_t, Comms::Users>::iterator it = userSelect.begin(); it != userSelect.end(); it++){
it->second.keepAlive();
}
}
}
bool Output::dropPushTrack(uint32_t trackId, const std::string & dropReason){
for (std::map<size_t, Comms::Users>::iterator it = userSelect.begin(); it != userSelect.end(); it++){
if (it->second.getTrack() == trackId){
WARN_MSG("Dropping input track %" PRIu32 ": %s", trackId, dropReason.c_str());
userSelect.erase(it);
return true;
break;
}
}
return false;
}
void Output::onRequest(){
// simply clear the buffer, we don't support any kind of input by default
myConn.Received().clear();
@ -1651,7 +1716,7 @@ namespace Mist{
// Initialize the stream source if needed, connect to it
waitForStreamPushReady();
// pull the source setting from metadata
strmSource = meta.getSource();
if (meta){strmSource = meta.getSource();}
if (!strmSource.size()){
FAIL_MSG("Push rejected - stream %s not configured or unavailable", streamName.c_str());
@ -1683,13 +1748,10 @@ namespace Mist{
}
}
std::string smp = streamName.substr(0, streamName.find_first_of("+ "));
if (Triggers::shouldTrigger("STREAM_PUSH", smp)){
std::string payload =
streamName + "\n" + getConnectedHost() + "\n" + capa["name"].asStringRef() + "\n" + reqUrl;
if (!Triggers::doTrigger("STREAM_PUSH", payload, smp)){
FAIL_MSG("Push from %s to %s rejected - STREAM_PUSH trigger denied the push",
getConnectedHost().c_str(), streamName.c_str());
if (Triggers::shouldTrigger("STREAM_PUSH", streamName)){
std::string payload = streamName + "\n" + getConnectedHost() + "\n" + capa["name"].asStringRef() + "\n" + reqUrl;
if (!Triggers::doTrigger("STREAM_PUSH", payload, streamName)){
WARN_MSG("Push from %s rejected by STREAM_PUSH trigger", getConnectedHost().c_str());
pushing = false;
return false;
}
@ -1698,8 +1760,7 @@ namespace Mist{
if (IP != ""){
if (!myConn.isAddress(IP)){
FAIL_MSG("Push from %s to %s rejected - source host not whitelisted",
getConnectedHost().c_str(), streamName.c_str());
WARN_MSG("Push from %s rejected; not whitelisted", getConnectedHost().c_str());
pushing = false;
return false;
}
@ -1712,7 +1773,31 @@ namespace Mist{
void Output::waitForStreamPushReady(){
uint8_t streamStatus = Util::getStreamStatus(streamName);
MEDIUM_MSG("Current status for %s buffer is %u", streamName.c_str(), streamStatus);
while (streamStatus != STRMSTAT_WAIT && streamStatus != STRMSTAT_READY && keepGoing()){
if (streamStatus == STRMSTAT_READY){
reconnect();
std::set<size_t> vTracks = M.getValidTracks(true);
INFO_MSG("Stream already active (%zu valid tracks) - check if it's not shutting down...", vTracks.size());
uint64_t oneTime = 0;
uint64_t twoTime = 0;
for (std::set<size_t>::iterator it = vTracks.begin(); it != vTracks.end(); ++it){
if (M.getLastms(*it) > oneTime){oneTime = M.getLastms(*it);}
}
Util::wait(2000);
for (std::set<size_t>::iterator it = vTracks.begin(); it != vTracks.end(); ++it){
if (M.getLastms(*it) > twoTime){twoTime = M.getLastms(*it);}
}
if (twoTime <= oneTime+500){
disconnect();
INFO_MSG("Waiting for stream reset before attempting push input accept (%" PRIu64 " <= %" PRIu64 "+500)", twoTime, oneTime);
while (streamStatus != STRMSTAT_OFF && keepGoing()){
userSelect.clear();
Util::wait(1000);
streamStatus = Util::getStreamStatus(streamName);
}
reconnect();
}
}
while (((streamStatus != STRMSTAT_WAIT && streamStatus != STRMSTAT_READY) || !meta) && keepGoing()){
INFO_MSG("Waiting for %s buffer to be ready... (%u)", streamName.c_str(), streamStatus);
disconnect();
userSelect.clear();
@ -1720,11 +1805,15 @@ namespace Mist{
streamStatus = Util::getStreamStatus(streamName);
if (streamStatus == STRMSTAT_OFF || streamStatus == STRMSTAT_WAIT || streamStatus == STRMSTAT_READY){
INFO_MSG("Reconnecting to %s buffer... (%u)", streamName.c_str(), streamStatus);
Util::wait(500);
reconnect();
streamStatus = Util::getStreamStatus(streamName);
}
}
if (streamStatus == STRMSTAT_READY || streamStatus == STRMSTAT_WAIT){reconnect();}
if (!meta){
onFail("Could not connect to stream data", true);
}
}
void Output::selectAllTracks(){

4
src/output/output.h
View file

@ -65,6 +65,7 @@ namespace Mist{
/// This function is called whenever a packet is ready for sending.
/// Inside it, thisPacket is guaranteed to contain a valid packet.
virtual void sendNext(){}// REQUIRED! Others are optional.
virtual bool dropPushTrack(uint32_t trackId, const std::string & dropReason);
bool getKeyFrame();
bool prepareNext();
virtual void dropTrack(size_t trackId, const std::string &reason, bool probablyBad = true);
@ -105,6 +106,7 @@ namespace Mist{
std::set<sortedPageInfo> buffer; ///< A sorted list of next-to-be-loaded packets.
bool sought; ///< If a seek has been done, this is set to true. Used for seeking on
///< prepareNext().
std::string prevHost; ///< Old value for getConnectedBinHost, for caching
protected: // these are to be messed with by child classes
virtual bool inlineRestartCapable() const{
return false;
@ -128,6 +130,7 @@ namespace Mist{
Comms::Statistics statComm;
bool isBlocking; ///< If true, indicates that myConn is blocking.
uint32_t crc; ///< Checksum, if any, for usage in the stats.
uint64_t nextKeyTime();
// stream delaying variables
uint64_t maxSkipAhead; ///< Maximum ms that we will go ahead of the intended timestamps.
@ -148,7 +151,6 @@ namespace Mist{
virtual bool isPushing(){return pushing;};
bool allowPush(const std::string &passwd);
void waitForStreamPushReady();
bool pushIsOngoing;
uint64_t firstPacketTime;
uint64_t lastPacketTime;

11
src/output/output_cmaf.cpp
View file

@ -13,6 +13,8 @@
#include <mist/stream.h>
#include <mist/timing.h>
uint64_t bootMsOffset;
namespace Mist{
OutCMAF::OutCMAF(Socket::Connection &conn) : HTTPOutput(conn){
@ -68,6 +70,8 @@ namespace Mist{
void OutCMAF::onHTTP(){
initialize();
bootMsOffset = 0;
if (M.getLive()){bootMsOffset = M.getBootMsOffset();}
if (H.url.size() < streamName.length() + 7){
H.Clean();
@ -440,6 +444,12 @@ namespace Mist{
}
void hlsSegment(uint64_t start, uint64_t duration, std::stringstream &s, bool first){
if (bootMsOffset){
uint64_t unixMs = start + bootMsOffset + (Util::unixMS() - Util::bootMS());
time_t uSecs = unixMs/1000;
struct tm * tVal = gmtime(&uSecs);
s << "#EXT-X-PROGRAM-DATE-TIME: " << (tVal->tm_year+1900) << "-" << std::setw(2) << std::setfill('0') << (tVal->tm_mon+1) << "-" << std::setw(2) << std::setfill('0') << tVal->tm_mday << "T" << std::setw(2) << std::setfill('0') << tVal->tm_hour << ":" << std::setw(2) << std::setfill('0') << tVal->tm_min << ":" << std::setw(2) << std::setfill('0') << tVal->tm_sec << "." << std::setw(3) << std::setfill('0') << (unixMs%1000) << "Z" << std::endl;
}
s << "#EXTINF:" << (((double)duration) / 1000) << ",\r\nchunk_" << start << ".m4s" << std::endl;
}
@ -537,7 +547,6 @@ namespace Mist{
result << "#EXTM3U\r\n"
"#EXT-X-VERSION:7\r\n"
"#EXT-X-DISCONTINUITY\r\n"
"#EXT-X-TARGETDURATION:"
<< targetDuration << "\r\n";
if (M.getLive()){result << "#EXT-X-MEDIA-SEQUENCE:" << firstFragment << "\r\n";}

2
src/output/output_dtsc.cpp
View file

@ -63,7 +63,7 @@ namespace Mist{
config = cfg;
}
std::string OutDTSC::getStatsName(){return (pushing ? "INPUT" : "OUTPUT");}
std::string OutDTSC::getStatsName(){return (pushing ? "INPUT:DTSC" : "OUTPUT:DTSC");}
/// Seeks to the first sync'ed keyframe of the main track.
/// Aborts if there is no main track or it has no keyframes.

23
src/output/output_hls.cpp
View file

@ -6,6 +6,8 @@
namespace Mist{
bool OutHLS::isReadyForPlay(){
if (!isInitialized){initialize();}
meta.refresh();
if (!M.getValidTracks().size()){return false;}
uint32_t mainTrack = M.mainTrack();
if (mainTrack == INVALID_TRACK_ID){return false;}
@ -244,7 +246,11 @@ namespace Mist{
bool isTS = (HTTP::URL(H.url).getExt().substr(0, 3) != "m3u");
H.Clean();
H.setCORSHeaders();
H.SetHeader("Content-Type", "application/octet-stream");
if (isTS){
H.SetHeader("Content-Type", "video/mp2t");
}else{
H.SetHeader("Content-Type", "application/vnd.apple.mpegurl");
}
if (isTS && !hasSessionIDs()){
H.SetHeader("Cache-Control", "public, max-age=600, immutable");
H.SetHeader("Pragma", "");
@ -343,6 +349,7 @@ namespace Mist{
std::string request = H.url.substr(H.url.find("/", 5) + 1);
H.Clean();
H.setCORSHeaders();
H.SetHeader("Content-Type", "application/vnd.apple.mpegurl");
if (!M.getValidTracks().size()){
H.SendResponse("404", "Not online or found", myConn);
H.Clean();
@ -378,16 +385,14 @@ namespace Mist{
wantRequest = true;
parseData = false;
// Ensure alignment of contCounters for selected tracks, to prevent discontinuities.
for (std::map<size_t, Comms::Users>::iterator it = userSelect.begin(); it != userSelect.end(); it++){
uint32_t pkgPid = 255 + it->first;
uint16_t &contPkg = contCounters[pkgPid];
if (contPkg % 16 != 0){
// Ensure alignment of contCounters, to prevent discontinuities.
for (std::map<size_t, uint16_t>::iterator it = contCounters.begin(); it != contCounters.end(); it++){
if (it->second % 16 != 0){
packData.clear();
packData.setPID(pkgPid);
packData.setPID(it->first);
packData.addStuffing();
while (contPkg % 16 != 0){
packData.setContinuityCounter(++contPkg);
while (it->second % 16 != 0){
packData.setContinuityCounter(++it->second);
sendTS(packData.checkAndGetBuffer());
}
packData.clear();

7
src/output/output_http_internal.cpp
View file

@ -111,6 +111,7 @@ namespace Mist{
capa["provides"] = "HTTP";
capa["protocol"] = "http://";
capa["url_rel"] = "/$.html";
capa["codecs"][0u][0u].append("+*");
capa["url_match"].append("/crossdomain.xml");
capa["url_match"].append("/clientaccesspolicy.xml");
capa["url_match"].append("/$.html");
@ -1028,7 +1029,7 @@ namespace Mist{
snprintf(pageName, NAME_BUFFER_SIZE, SHM_STREAM_STATE, streamName.c_str());
IPC::sharedPage streamStatus(pageName, 1, false, false);
uint8_t prevState, newState, pingCounter = 0;
uint64_t prevTracks;
std::set<size_t> prevTracks;
prevState = newState = STRMSTAT_INVALID;
while (keepGoing()){
if (!streamStatus || !streamStatus.exists()){streamStatus.init(pageName, 1, false, false);}
@ -1038,10 +1039,10 @@ namespace Mist{
newState = streamStatus.mapped[0];
}
if (newState != prevState || (newState == STRMSTAT_READY && M.getValidTracks().size() != prevTracks)){
if (newState != prevState || (newState == STRMSTAT_READY && M.getValidTracks() != prevTracks)){
if (newState == STRMSTAT_READY){
reconnect();
prevTracks = M.getValidTracks().size();
prevTracks = M.getValidTracks();
}else{
disconnect();
}

5
src/output/output_https.cpp
View file

@ -92,6 +92,7 @@ namespace Mist{
char error_buf[200];
mbedtls_strerror(ret, error_buf, 200);
MEDIUM_MSG("Could not handshake, SSL error: %s (%d)", error_buf, ret);
Util::logExitReason("Could not handshake, SSL error: %s (%d)", error_buf, ret);
C.close();
return;
}else{
@ -110,6 +111,7 @@ namespace Mist{
int fd[2];
if (socketpair(PF_LOCAL, SOCK_STREAM, 0, fd) != 0){
FAIL_MSG("Could not open anonymous socket for SSL<->HTTP connection!");
Util::logExitReason("Could not open anonymous socket for SSL<->HTTP connection!");
return 1;
}
std::deque<std::string> args;
@ -135,6 +137,7 @@ namespace Mist{
close(fd[1]);
if (http_proc < 2){
FAIL_MSG("Could not spawn MistOutHTTP process for SSL connection!");
Util::logExitReason("Could not spawn MistOutHTTP process for SSL connection!");
return 1;
}
Socket::Connection http(fd[0]);
@ -150,6 +153,7 @@ namespace Mist{
if (ret != MBEDTLS_ERR_SSL_WANT_READ && ret != MBEDTLS_ERR_SSL_WANT_WRITE){
if (ret <= 0){
HIGH_MSG("SSL disconnect!");
Util::logExitReason("SSL client disconnected");
break;
}
// we received ret bytes of data to pass on. Do so.
@ -168,6 +172,7 @@ namespace Mist{
ret = mbedtls_ssl_write(&ssl, (const unsigned char *)http_buf.get().data() + done, toSend - done);
if (ret == MBEDTLS_ERR_NET_CONN_RESET || ret == MBEDTLS_ERR_SSL_CLIENT_RECONNECT){
HIGH_MSG("SSL disconnect!");
Util::logExitReason("SSL client disconnected");
http.close();
break;
}

2
src/output/output_rtmp.cpp
View file

@ -1229,7 +1229,7 @@ namespace Mist{
static std::map<size_t, AMF::Object> pushMeta;
static std::map<size_t, uint64_t> lastTagTime;
static std::map<size_t, size_t> reTrackToID;
if (!isInitialized){
if (!isInitialized || !meta){
MEDIUM_MSG("Received useless media data");
onFinish();
break;

2
src/output/output_rtsp.cpp
View file

@ -91,6 +91,8 @@ namespace Mist{
capa["codecs"][0u][0u].append("H264");
capa["codecs"][0u][0u].append("HEVC");
capa["codecs"][0u][0u].append("MPEG2");
capa["codecs"][0u][0u].append("VP8");
capa["codecs"][0u][0u].append("VP9");
capa["codecs"][0u][1u].append("AAC");
capa["codecs"][0u][1u].append("MP3");
capa["codecs"][0u][1u].append("AC3");

2
src/output/output_ts.cpp
View file

@ -189,7 +189,7 @@ namespace Mist{
std::string OutTS::getStatsName(){
if (!parseData){
return "INPUT";
return "INPUT:" + capa["name"].asStringRef();
}else{
return Output::getStatsName();
}

7
src/output/output_ts_base.cpp
View file

@ -65,17 +65,12 @@ namespace Mist{
std::string type = M.getType(thisIdx);
std::string codec = M.getCodec(thisIdx);
bool video = (type == "video");
size_t pkgPid = M.getID(thisIdx);
if (pkgPid < 255){pkgPid += 255;}
size_t pkgPid = TS::getUniqTrackID(M, thisIdx);
bool &firstPack = first[thisIdx];
uint16_t &contPkg = contCounters[pkgPid];
uint64_t packTime = thisPacket.getTime();
bool keyframe = thisPacket.getInt("keyframe");
firstPack = true;
char *dataPointer = 0;
size_t dataLen = 0;
thisPacket.getString("data", dataPointer, dataLen); // data

200
src/output/output_webrtc.cpp
View file

@ -3,6 +3,7 @@
#include <mist/procs.h>
#include <mist/sdp.h>
#include <mist/timing.h>
#include <mist/url.h>
#include <netdb.h> // ifaddr, listing ip addresses.
namespace Mist{
@ -31,10 +32,12 @@ namespace Mist{
/* ------------------------------------------------ */
OutWebRTC::OutWebRTC(Socket::Connection &myConn) : HTTPOutput(myConn){
lastPackMs = 0;
vidTrack = INVALID_TRACK_ID;
prevVidTrack = INVALID_TRACK_ID;
audTrack = INVALID_TRACK_ID;
stayLive = true;
target_rate = 0.0;
firstKey = true;
repeatInit = true;
@ -95,6 +98,7 @@ namespace Mist{
capa["url_match"] = "/webrtc/$";
capa["codecs"][0u][0u].append("H264");
capa["codecs"][0u][0u].append("VP8");
capa["codecs"][0u][0u].append("VP9");
capa["codecs"][0u][1u].append("opus");
capa["codecs"][0u][1u].append("ALAW");
capa["codecs"][0u][1u].append("ULAW");
@ -107,7 +111,7 @@ namespace Mist{
capa["optional"]["preferredvideocodec"]["help"] =
"Comma separated list of video codecs you want to support in preferred order. e.g. "
"H264,VP8";
capa["optional"]["preferredvideocodec"]["default"] = "H264,VP8";
capa["optional"]["preferredvideocodec"]["default"] = "H264,VP9,VP8";
capa["optional"]["preferredvideocodec"]["type"] = "string";
capa["optional"]["preferredvideocodec"]["option"] = "--webrtc-video-codecs";
capa["optional"]["preferredvideocodec"]["short"] = "V";
@ -129,14 +133,26 @@ namespace Mist{
capa["optional"]["bindhost"]["option"] = "--bindhost";
capa["optional"]["bindhost"]["short"] = "B";
capa["optional"]["mergesessions"]["name"] = "Merge sessions";
capa["optional"]["mergesessions"]["name"] = "merge sessions";
capa["optional"]["mergesessions"]["help"] =
"If enabled, merges together all views from a single user into a single combined session. "
"If disabled, each view (reconnection of the signalling websocket) is a separate session.";
"if enabled, merges together all views from a single user into a single combined session. "
"if disabled, each view (reconnection of the signalling websocket) is a separate session.";
capa["optional"]["mergesessions"]["option"] = "--mergesessions";
capa["optional"]["mergesessions"]["short"] = "m";
capa["optional"]["mergesessions"]["default"] = 0;
capa["optional"]["nackdisable"]["name"] = "Disallow NACKs for viewers";
capa["optional"]["nackdisable"]["help"] = "Disallows viewers to send NACKs for lost packets";
capa["optional"]["nackdisable"]["option"] = "--nackdisable";
capa["optional"]["nackdisable"]["short"] = "n";
capa["optional"]["nackdisable"]["default"] = 0;
capa["optional"]["jitterlog"]["name"] = "Write jitter log";
capa["optional"]["jitterlog"]["help"] = "Writes log of frame transmit jitter to /tmp/ for each outgoing connection";
capa["optional"]["jitterlog"]["option"] = "--jitterlog";
capa["optional"]["jitterlog"]["short"] = "J";
capa["optional"]["jitterlog"]["default"] = 0;
config->addOptionsFromCapabilities(capa);
}
@ -280,13 +296,61 @@ namespace Mist{
parseData = true;
selectDefaultTracks();
}
stayLive = (endTime() < seek_time + 5000);
stayLive = (target_rate == 0.0) && (endTime() < seek_time + 5000);
if (command["seek_time"].asStringRef() == "live"){stayLive = true;}
if (stayLive){seek_time = endTime();}
seek(seek_time, true);
JSON::Value commandResult;
commandResult["type"] = "on_seek";
commandResult["result"] = true;
if (M.getLive()){commandResult["live_point"] = stayLive;}
if (target_rate == 0.0){
commandResult["play_rate_curr"] = "auto";
}else{
commandResult["play_rate_curr"] = target_rate;
}
webSock->sendFrame(commandResult.toString());
onIdle();
return;
}
if (command["type"] == "set_speed"){
if (!command.isMember("play_rate")){
sendSignalingError("on_speed", "Received a playback speed setting request but no `play_rate` property.");
return;
}
double set_rate = command["play_rate"].asDouble();
if (!parseData){
parseData = true;
selectDefaultTracks();
}
JSON::Value commandResult;
commandResult["type"] = "on_speed";
if (target_rate == 0.0){
commandResult["play_rate_prev"] = "auto";
}else{
commandResult["play_rate_prev"] = target_rate;
}
if (set_rate == 0.0){
commandResult["play_rate_curr"] = "auto";
}else{
commandResult["play_rate_curr"] = set_rate;
}
if (target_rate != set_rate){
target_rate = set_rate;
if (target_rate == 0.0){
realTime = 1000;//set playback speed to default
firstTime = Util::bootMS() - currentTime();
maxSkipAhead = 0;//enabled automatic rate control
}else{
stayLive = false;
//Set new realTime speed
realTime = 1000 / target_rate;
firstTime = Util::bootMS() - (currentTime() / target_rate);
maxSkipAhead = 1;//disable automatic rate control
}
}
if (M.getLive()){commandResult["live_point"] = stayLive;}
webSock->sendFrame(commandResult.toString());
onIdle();
return;
@ -337,6 +401,15 @@ namespace Mist{
sendSignalingError(command["type"].asString(), "Unhandled command type: " + command["type"].asString());
}
bool OutWebRTC::dropPushTrack(uint32_t trackId, const std::string & dropReason){
JSON::Value commandResult;
commandResult["type"] = "on_track_drop";
commandResult["track"] = trackId;
commandResult["mediatype"] = M.getType(trackId);
webSock->sendFrame(commandResult.toString());
return Output::dropPushTrack(trackId, dropReason);
}
void OutWebRTC::sendSignalingError(const std::string &commandType, const std::string &errorMessage){
JSON::Value commandResult;
commandResult["type"] = "on_error";
@ -357,6 +430,12 @@ namespace Mist{
initialize();
selectDefaultTracks();
if (config && config->hasOption("jitterlog") && config->getBool("jitterlog")){
std::string fileName = "/tmp/jitter_"+JSON::Value(getpid()).asString();
jitterLog.open(fileName.c_str());
lastPackMs = Util::bootMS();
}
if (0 == udpPort){bindUDPSocketOnLocalCandidateAddress(0);}
std::string videoCodec;
@ -389,8 +468,10 @@ namespace Mist{
return false;
}
videoTrack.rtpPacketizer = RTP::Packet(videoTrack.payloadType, rand(), 0, videoTrack.SSRC, 0);
// Enabled NACKs
sdpAnswer.videoLossPrevention = SDP_LOSS_PREVENTION_NACK;
if (!config || !config->hasOption("nackdisable") || !config->getBool("nackdisable")){
// Enable NACKs
sdpAnswer.videoLossPrevention = SDP_LOSS_PREVENTION_NACK;
}
videoTrack.sorter.tmpVideoLossPrevention = sdpAnswer.videoLossPrevention;
}
}
@ -500,7 +581,7 @@ namespace Mist{
capa["codecs"].null();
const char *videoCodecPreference[] ={"H264", "VP8", NULL};
const char *videoCodecPreference[] ={"H264", "VP9", "VP8", NULL};
const char **videoCodec = videoCodecPreference;
SDP::Media *videoMediaOffer = sdpSession.getMediaForType("video");
if (videoMediaOffer){
@ -535,12 +616,12 @@ namespace Mist{
if (0 == udpPort){bindUDPSocketOnLocalCandidateAddress(0);}
std::string prefVideoCodec = "VP8,H264";
std::string prefVideoCodec = "VP9,VP8,H264";
if (config && config->hasOption("preferredvideocodec")){
prefVideoCodec = config->getString("preferredvideocodec");
if (prefVideoCodec.empty()){
WARN_MSG("No preferred video codec value set; resetting to default.");
prefVideoCodec = "VP8,H264";
prefVideoCodec = "VP9,VP8,H264";
}
}
@ -579,10 +660,11 @@ namespace Mist{
SDP::MediaFormat *fmtRED = sdpSession.getMediaFormatByEncodingName("video", "RED");
SDP::MediaFormat *fmtULPFEC = sdpSession.getMediaFormatByEncodingName("video", "ULPFEC");
if (fmtRED && fmtULPFEC){
if (fmtRED || fmtULPFEC){
videoTrack.ULPFECPayloadType = fmtULPFEC->payloadType;
videoTrack.REDPayloadType = fmtRED->payloadType;
payloadTypeToWebRTCTrack[fmtRED->payloadType] = videoTrack.payloadType;
payloadTypeToWebRTCTrack[fmtULPFEC->payloadType] = videoTrack.payloadType;
}
sdpAnswer.videoLossPrevention = SDP_LOSS_PREVENTION_NACK;
videoTrack.sorter.tmpVideoLossPrevention = sdpAnswer.videoLossPrevention;
@ -594,7 +676,7 @@ namespace Mist{
videoTrack.rtpToDTSC.setProperties(meta, vIdx);
videoTrack.rtpToDTSC.setCallbacks(onDTSCConverterHasPacketCallback, onDTSCConverterHasInitDataCallback);
videoTrack.sorter.setCallback(vIdx, onRTPSorterHasPacketCallback);
videoTrack.sorter.setCallback(M.getID(vIdx), onRTPSorterHasPacketCallback);
userSelect[vIdx].reload(streamName, vIdx, COMM_STATUS_SOURCE);
INFO_MSG("Video push received on track %zu", vIdx);
@ -616,7 +698,7 @@ namespace Mist{
audioTrack.rtpToDTSC.setProperties(meta, aIdx);
audioTrack.rtpToDTSC.setCallbacks(onDTSCConverterHasPacketCallback, onDTSCConverterHasInitDataCallback);
audioTrack.sorter.setCallback(aIdx, onRTPSorterHasPacketCallback);
audioTrack.sorter.setCallback(M.getID(aIdx), onRTPSorterHasPacketCallback);
userSelect[aIdx].reload(streamName, aIdx, COMM_STATUS_SOURCE);
INFO_MSG("Audio push received on track %zu", aIdx);
@ -639,13 +721,47 @@ namespace Mist{
return false;
}
udpPort =
udp.bind(port, (config && config->hasOption("bindhost") && config->getString("bindhost").size())
? config->getString("bindhost")
: myConn.getBoundAddress());
Util::Procs::socketList.insert(udp.getSock());
sdpAnswer.setCandidate(externalAddr, udpPort);
std::string bindAddr;
//If a bind host has been put in as override, use it
if (config && config->hasOption("bindhost") && config->getString("bindhost").size()){
bindAddr = config->getString("bindhost");
udpPort = udp.bind(port, bindAddr);
if (!udpPort){
WARN_MSG("UDP bind address not valid - ignoring setting and using best guess instead");
bindAddr.clear();
}else{
INFO_MSG("Bound to pre-configured UDP bind address");
}
}
//use the best IPv4 guess we have
if (!bindAddr.size()){
bindAddr = Socket::resolveHostToBestExternalAddrGuess(externalAddr, AF_INET, myConn.getBoundAddress());
if (!bindAddr.size()){
WARN_MSG("UDP bind to best guess failed - using same address as incoming connection as a last resort");
bindAddr.clear();
}else{
udpPort = udp.bind(port, bindAddr);
if (!udpPort){
WARN_MSG("UDP bind to best guess failed - using same address as incoming connection as a last resort");
bindAddr.clear();
}else{
INFO_MSG("Bound to public UDP bind address derived from hostname");
}
}
}
if (!bindAddr.size()){
bindAddr = myConn.getBoundAddress();
udpPort = udp.bind(port, bindAddr);
if (!udpPort){
FAIL_MSG("UDP bind to connected address failed - we're out of options here, I'm afraid...");
bindAddr.clear();
}else{
INFO_MSG("Bound to same UDP address as TCP address - this is potentially wrong, but used as a last resort");
}
}
Util::Procs::socketList.insert(udp.getSock());
sdpAnswer.setCandidate(bindAddr, udpPort);
return true;
}
@ -813,8 +929,8 @@ namespace Mist{
if (idx == INVALID_TRACK_ID || !webrtcTracks.count(idx)){
FAIL_MSG("Received an RTP packet for a track that we didn't prepare for. PayloadType is "
"%" PRIu32,
rtp_pkt.getPayloadType());
"%" PRIu32 ", idx %zu",
rtp_pkt.getPayloadType(), idx);
return;
}
@ -830,6 +946,8 @@ namespace Mist{
FAIL_MSG("Failed to unprotect a RTP packet.");
return;
}
RTP::Packet unprotPack(udp.data, len);
DONTEVEN_MSG("%s", unprotPack.toString().c_str());
// Here follows a very rudimentary algo for requesting lost
// packets; I guess after some experimentation a better
@ -843,11 +961,11 @@ namespace Mist{
rtcTrack.prevReceivedSequenceNumber = currSeqNum;
if (rtp_pkt.getPayloadType() == rtcTrack.REDPayloadType){
if (rtp_pkt.getPayloadType() == rtcTrack.REDPayloadType || rtp_pkt.getPayloadType() == rtcTrack.ULPFECPayloadType){
rtcTrack.sorter.addREDPacket(udp.data, len, rtcTrack.payloadType, rtcTrack.REDPayloadType,
rtcTrack.ULPFECPayloadType);
}else{
rtcTrack.sorter.addPacket(RTP::Packet(udp.data, len));
rtcTrack.sorter.addPacket(unprotPack);
}
}else if ((pt >= 64) && (pt < 96)){
@ -909,7 +1027,7 @@ namespace Mist{
void OutWebRTC::onDTSCConverterHasPacket(const DTSC::Packet &pkt){
// extract meta data (init data, width/height, etc);
size_t idx = pkt.getTrackId();
size_t idx = M.trackIDToIndex(pkt.getTrackId(), getpid());
std::string codec = M.getCodec(idx);
if (codec == "H264"){
if (M.getInit(idx).empty()){
@ -919,9 +1037,10 @@ namespace Mist{
}
if (codec == "VP8" && pkt.getFlag("keyframe")){extractFrameSizeFromVP8KeyFrame(pkt);}
if (codec == "VP9" && pkt.getFlag("keyframe")){extractFrameSizeFromVP8KeyFrame(pkt);}
// create rtcp packet (set bitrate and request keyframe).
if (codec == "H264" || codec == "VP8"){
if (codec == "H264" || codec == "VP8" || codec == "VP9"){
uint64_t now = Util::bootMS();
if (now >= rtcpTimeoutInMillis){
@ -942,13 +1061,15 @@ namespace Mist{
INFO_MSG("Validated track %zu in meta", idx);
meta.validateTrack(idx);
}
DONTEVEN_MSG("DTSC: %s", pkt.toSummary().c_str());
bufferLivePacket(pkt);
}
void OutWebRTC::onDTSCConverterHasInitData(size_t idx, const std::string &initData){
void OutWebRTC::onDTSCConverterHasInitData(size_t trackId, const std::string &initData){
size_t idx = M.trackIDToIndex(trackId, getpid());
if (idx == INVALID_TRACK_ID || !webrtcTracks.count(idx)){
ERROR_MSG(
"Recieved init data for a track that we don't manager. TrackID %zu /PayloadType: %zu",
"Recieved init data for a track that we don't manage. TrackID %zu /PayloadType: %zu",
idx, M.getID(idx));
return;
}
@ -972,9 +1093,10 @@ namespace Mist{
meta.setInit(idx, avccbox.payload(), avccbox.payloadSize());
}
void OutWebRTC::onRTPSorterHasPacket(size_t idx, const RTP::Packet &pkt){
void OutWebRTC::onRTPSorterHasPacket(size_t trackId, const RTP::Packet &pkt){
size_t idx = M.trackIDToIndex(trackId, getpid());
if (idx == INVALID_TRACK_ID || !webrtcTracks.count(idx)){
ERROR_MSG("Received a sorted RTP packet for track %zu but we don't manage this track.", idx);
ERROR_MSG("Received a sorted RTP packet for payload %zu (idx %zu) but we don't manage this track.", trackId, idx);
return;
}
webrtcTracks[idx].rtpToDTSC.addRTP(pkt);
@ -989,7 +1111,7 @@ namespace Mist{
int protectedSize = nbytes;
if (srtpWriter.protectRtp((uint8_t *)(void *)rtpOutBuffer, &protectedSize) != 0){
ERROR_MSG("Failed to protect the RTCP message.");
ERROR_MSG("Failed to protect the RTP message.");
return;
}
@ -1076,6 +1198,14 @@ namespace Mist{
// If we see this is audio or video, use the webrtc track we negotiated
if (M.getType(tid) == "video" && webrtcTracks.count(vidTrack)){
trackPointer = &webrtcTracks[vidTrack];
if (lastPackMs){
uint64_t newMs = Util::bootMS();
jitterLog << (newMs - lastPackMs) << std::endl;
lastPackMs = newMs;
}
}
if (M.getType(tid) == "audio" && webrtcTracks.count(audTrack)){
trackPointer = &webrtcTracks[audTrack];
@ -1093,19 +1223,17 @@ namespace Mist{
WebRTCTrack &rtcTrack = *trackPointer;
uint64_t timestamp = thisPacket.getTime();
rtcTrack.rtpPacketizer.setTimestamp(timestamp * SDP::getMultiplier(&M, thisIdx));
uint64_t newTime = timestamp * SDP::getMultiplier(&M, thisIdx);
rtcTrack.rtpPacketizer.setTimestamp(newTime);
bool isKeyFrame = thisPacket.getFlag("keyframe");
didReceiveKeyFrame = isKeyFrame;
if (M.getCodec(thisIdx) == "H264"){
if (isKeyFrame && firstKey){
char *data;
size_t dataLen;
thisPacket.getString("data", data, dataLen);
size_t offset = 0;
while (offset + 4 < dataLen){
size_t nalLen = Bit::btohl(data + offset);
uint8_t nalType = data[offset + 4] & 0x1F;
size_t nalLen = Bit::btohl(dataPointer + offset);
uint8_t nalType = dataPointer[offset + 4] & 0x1F;
if (nalType == 7 || nalType == 8){// Init data already provided in-band, skip repeating
// it.
repeatInit = false;

5
src/output/output_webrtc.h
View file

@ -67,6 +67,7 @@
#include <mist/stun.h>
#include <mist/tinythread.h>
#include <mist/websocket.h>
#include <fstream>
#define NACK_BUFFER_SIZE 1024
@ -130,6 +131,7 @@ namespace Mist{
virtual void sendNext();
virtual void onWebsocketFrame();
virtual void preWebsocketConnect();
virtual bool dropPushTrack(uint32_t trackId, const std::string & dropReason);
void onIdle();
bool onFinish();
bool doesWebsockets(){return true;}
@ -142,6 +144,8 @@ namespace Mist{
void onRTPPacketizerHasRTCPPacket(const char *data, uint32_t nbytes);
private:
uint64_t lastPackMs;
std::ofstream jitterLog;
std::string externalAddr;
void ackNACK(uint32_t SSRC, uint16_t seq);
bool handleWebRTCInputOutput(); ///< Reads data from the UDP socket. Returns true when we read
@ -198,6 +202,7 @@ namespace Mist{
///< the signaling channel. Defaults to 6mbit.
size_t audTrack, vidTrack, prevVidTrack;
double target_rate; ///< Target playback speed rate (1.0 = normal, 0 = auto)
bool didReceiveKeyFrame; /* TODO burst delay */
int64_t packetOffset; ///< For timestamp rewrite with BMO

23
src/process/process_exec.cpp
View file

@ -9,7 +9,7 @@
#include <sys/types.h> //for stat
#include <unistd.h> //for stat
int pipein[2], pipeout[2], pipeerr[2];
int pipein[2], pipeout[2];
Util::Config co;
Util::Config conf;
@ -134,8 +134,14 @@ int main(int argc, char *argv[]){
}
// create pipe pair before thread
pipe(pipein);
pipe(pipeout);
if (pipe(pipein) || pipe(pipeout)){
FAIL_MSG("Could not create pipes for process!");
return 1;
}
Util::Procs::socketList.insert(pipeout[0]);
Util::Procs::socketList.insert(pipeout[1]);
Util::Procs::socketList.insert(pipein[0]);
Util::Procs::socketList.insert(pipein[1]);
// stream which connects to input
tthread::thread source(sourceThread, 0);
@ -188,9 +194,18 @@ namespace Mist{
int ffer = 2;
pid_t execd_proc = -1;
std::string streamName = opt["sink"].asString();
if (!streamName.size()){streamName = opt["source"].asStringRef();}
Util::streamVariables(streamName, opt["source"].asStringRef());
//Do variable substitution on command
std::string tmpCmd = opt["exec"].asStringRef();
Util::streamVariables(tmpCmd, streamName, opt["source"].asStringRef());
// exec command
char exec_cmd[10240];
strncpy(exec_cmd, opt["exec"].asString().c_str(), 10240);
strncpy(exec_cmd, tmpCmd.c_str(), 10240);
INFO_MSG("Executing command: %s", exec_cmd);
uint8_t argCnt = 0;
char *startCh = 0;

14
src/process/process_exec.h
View file

@ -21,7 +21,9 @@ namespace Mist{
class ProcessSink : public InputEBML{
public:
ProcessSink(Util::Config *cfg) : InputEBML(cfg){};
ProcessSink(Util::Config *cfg) : InputEBML(cfg){
capa["name"] = "MKVExec";
};
void getNext(size_t idx = INVALID_TRACK_ID){
static bool recurse = false;
if (recurse){return InputEBML::getNext(idx);}
@ -52,7 +54,15 @@ namespace Mist{
class ProcessSource : public OutEBML{
public:
ProcessSource(Socket::Connection &c) : OutEBML(c){realTime = 1000;};
bool isRecording(){return false;}
ProcessSource(Socket::Connection &c) : OutEBML(c){
capa["name"] = "MKVExec";
realTime = 0;
};
void sendHeader(){
realTime = 0;
OutEBML::sendHeader();
};
void sendNext(){
extraKeepAway = 0;
needsLookAhead = 0;

37
src/process/process_ffmpeg.cpp
View file

@ -13,7 +13,7 @@
int ofin = -1, ofout = 1, oferr = 2;
int ifin = -1, ifout = -1, iferr = 2;
int pipein[2], pipeout[2], pipeerr[2];
int pipein[2], pipeout[2];
Util::Config co;
Util::Config conf;
@ -307,8 +307,12 @@ int main(int argc, char *argv[]){
}
// create pipe pair before thread
pipe(pipein);
pipe(pipeout);
if (pipe(pipein) || pipe(pipeout)){
FAIL_MSG("Could not create pipes for process!");
return 1;
}
Util::Procs::socketList.insert(pipeout[0]);
Util::Procs::socketList.insert(pipein[1]);
// stream which connects to input
tthread::thread source(sourceThread, 0);
@ -384,14 +388,14 @@ namespace Mist{
std::string EncodeOutputEBML::getTrackType(int tid){return M.getType(tid);}
void EncodeOutputEBML::setVideoTrack(std::string tid){
std::set<size_t> tracks = Util::findTracks(M, "video", tid);
std::set<size_t> tracks = Util::findTracks(M, capa, "video", tid);
for (std::set<size_t>::iterator it = tracks.begin(); it != tracks.end(); it++){
userSelect[*it].reload(streamName, *it);
}
}
void EncodeOutputEBML::setAudioTrack(std::string tid){
std::set<size_t> tracks = Util::findTracks(M, "audio", tid);
std::set<size_t> tracks = Util::findTracks(M, capa, "audio", tid);
for (std::set<size_t>::iterator it = tracks.begin(); it != tracks.end(); it++){
userSelect[*it].reload(streamName, *it);
}
@ -571,7 +575,7 @@ namespace Mist{
if (!preset.empty()){options.append(" -preset " + preset);}
snprintf(ffcmd, 10240, "ffmpeg -hide_banner -loglevel warning -f lavfi -i color=c=black:s=%dx%d %s %s -c:v %s %s %s %s -an -f matroska - ",
snprintf(ffcmd, 10240, "ffmpeg -hide_banner -loglevel warning -f lavfi -i color=c=black:s=%dx%d %s %s -c:v %s %s %s %s -an -force_key_frames source -f matroska - ",
res_x, res_y, s_input.c_str(), s_overlay.c_str(), codec.c_str(), options.c_str(),
getBitrateSetting().c_str(), flags.c_str());
@ -681,16 +685,26 @@ namespace Mist{
}else{
// sources array missing, create empty object in array
opt["sources"][0u]["src"] = "-";
WARN_MSG("No stdin input set in config, adding input stream with default settings");
if (opt.isMember("resolution")){
opt["sources"][0u]["width"] = -1;
opt["sources"][0u]["height"] = res_y;
opt["sources"][0u]["anchor"] = "center";
}
INFO_MSG("Default source: input stream at preserved-aspect same height");
stdinSource = true;
}
if (!stdinSource){
// no stdin source item found in sources configuration, add source object at the beginning
opt["sources"].prepend(JSON::fromString("{\"src\':\"-\"}"));
WARN_MSG("No stdin input stream found in 'inputs' config, adding stdin input stream at the "
"beginning of the array");
JSON::Value nOpt;
nOpt["src"] = "-";
if (opt.isMember("resolution")){
nOpt["width"] = -1;
nOpt["height"] = res_y;
nOpt["anchor"] = "center";
}
opt["sources"].prepend(nOpt);
WARN_MSG("Source is not used: adding source stream at preserved-aspect same height");
}
return true;
@ -799,7 +813,6 @@ namespace Mist{
}
prepareCommand();
MEDIUM_MSG("Starting ffmpeg process...");
ffout = p.StartPiped(args, &pipein[0], &pipeout[1], &ffer);
while (conf.is_active && p.isRunning(ffout)){Util::sleep(200);}

1
src/process/process_ffmpeg.h
View file

@ -52,6 +52,7 @@ namespace Mist{
class EncodeOutputEBML : public OutEBML{
public:
EncodeOutputEBML(Socket::Connection &c) : OutEBML(c){}; // realTime = 0;};
bool isRecording(){return false;}
void setVideoTrack(std::string tid);
void setAudioTrack(std::string tid);
void sendNext();

1
src/sourcery.cpp
View file

@ -13,6 +13,7 @@
#include <stdint.h>
#include <string.h>
#include <string>
#include <sstream>
#include <unistd.h>
std::string getContents(const char *fileName){

4
src/utils/util_load.cpp
View file

@ -664,7 +664,9 @@ void handleServer(void *hostEntryPointer){
entry->state = STATE_ERROR;
}else{
if (down){
WARN_MSG("Connection established with %s", url.host.c_str());
std::string ipStr;
Socket::hostBytesToStr(DL.getSocket().getBinHost().data(), 16, ipStr);
WARN_MSG("Connection established with %s (%s)", url.host.c_str(), ipStr.c_str());
memcpy(entry->details->binHost, DL.getSocket().getBinHost().data(), 16);
entry->state = STATE_ONLINE;
down = false;