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

Fully implemented DTSC pull support

Browse source
This commit is contained in:
Erik Zandvliet 2016-03-15 11:29:01 +01:00 • committed by Thulinma
parent 668560ff05
commit dda0ea669c
27 changed files with 930 additions and 272 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
CMakeLists.txt
View file

@ -160,6 +160,7 @@ set(libHeaders
${SOURCE_DIR}/lib/tinythread.h
${SOURCE_DIR}/lib/ts_packet.h
${SOURCE_DIR}/lib/ts_stream.h
${SOURCE_DIR}/lib/util.h
${SOURCE_DIR}/lib/vorbis.h
${SOURCE_DIR}/lib/triggers.h
)
@ -203,6 +204,7 @@ set(libSources
${SOURCE_DIR}/lib/tinythread.cpp
${SOURCE_DIR}/lib/ts_packet.cpp
${SOURCE_DIR}/lib/ts_stream.cpp
${SOURCE_DIR}/lib/util.cpp
${SOURCE_DIR}/lib/vorbis.cpp
${SOURCE_DIR}/lib/triggers.cpp
)
@ -290,9 +292,6 @@ macro(makeInput inputName format)
#Set compile definitions
unset(my_definitions)
if (";${ARGN};" MATCHES ";nolock;")#Currently only used in TSStream
list(APPEND my_definitions "INPUT_NOLOCK")
endif()
if (";${ARGN};" MATCHES ";tslive;")
list(APPEND my_definitions "TSLIVE_INPUT")
endif()
@ -320,7 +319,7 @@ makeInput(Buffer buffer)
makeInput(ISMV ismv)#LTS
makeInput(MP4 mp4)#LTS
makeInput(TS ts)#LTS
makeInput(TSStream ts nolock tslive)#LTS
makeInput(TSStream ts tslive)#LTS
makeInput(Folder folder folder)#LTS
########################################

2
flow_input
View file

@ -1,7 +1,7 @@
- Construct input
- Parse arguments
- Stream wordt gelocked IFF !nolock
- Stream wordt gelocked IFF conv.needsLock()
- Start .run()
- setup(): opent files/sockets/etc waar nodig
- set "isStream" naar true

4
lib/config.cpp
View file

@ -264,6 +264,10 @@ bool Util::Config::parseArgs(int & argc, char ** & argv) {
return true;
}
bool Util::Config::hasOption(const std::string & optname) {
return vals.isMember(optname);
}
/// Returns a reference to the current value of an option or default if none was set.
/// If the option does not exist, this exits the application with a return code of 37.
JSON::Value & Util::Config::getOption(std::string optname, bool asArray) {

1
lib/config.h
View file

@ -30,6 +30,7 @@ namespace Util {
void addOption(std::string optname, JSON::Value option);
void printHelp(std::ostream & output);
bool parseArgs(int & argc, char ** & argv);
bool hasOption(const std::string & optname);
JSON::Value & getOption(std::string optname, bool asArray = false);
std::string getString(std::string optname);
long long int getInteger(std::string optname);

5
lib/dtsc.h
View file

@ -109,6 +109,7 @@ namespace DTSC {
void operator = (const Packet & rhs);
operator bool() const;
packType getVersion() const;
void reInit(Socket::Connection & src);
void reInit(const char * data_, unsigned int len, bool noCopy = false);
void genericFill(long long packTime, long long packOffset, long long packTrack, const char * packData, long long packDataSize, long long packBytePos, bool isKeyframe);
void getString(const char * identifier, char *& result, unsigned int & len) const;
@ -354,8 +355,8 @@ namespace DTSC {
void update(long long packTime, long long packOffset, long long packTrack, long long packDataSize, long long packBytePos, bool isKeyframe, long long packSendSize = 0, unsigned long segment_size = 5000);
LTS*/
void update(long long packTime, long long packOffset, long long packTrack, long long packDataSize, long long packBytePos, bool isKeyframe, long long packSendSize = 0, unsigned long segment_size = 5000, const char * iVec = 0);
unsigned int getSendLen(bool skipDynamic = false);
void send(Socket::Connection & conn, bool skipDynamic = false);
unsigned int getSendLen(bool skipDynamic = false, std::set<unsigned long> selectedTracks = std::set<unsigned long>());
void send(Socket::Connection & conn, bool skipDynamic = false, std::set<unsigned long> selectedTracks = std::set<unsigned long>());
void writeTo(char * p);
JSON::Value toJSON();
void reset();

38
lib/dtscmeta.cpp
View file

@ -109,6 +109,32 @@ namespace DTSC {
}
}
void Packet::reInit(Socket::Connection & src) {
int sleepCount = 0;
null();
int toReceive = 0;
while (src.connected()){
if (!toReceive && src.Received().available(8)){
if (src.Received().copy(2) != "DT"){
INFO_MSG("Invalid DTSC Packet header encountered (%s)", src.Received().copy(4).c_str());
break;
}
toReceive = Bit::btohl(src.Received().copy(8).data() + 4);
}
if (toReceive && src.Received().available(toReceive + 8)){
std::string dataBuf = src.Received().remove(toReceive + 8);
reInit(dataBuf.data(), dataBuf.size());
return;
}
if(!src.spool()){
if (sleepCount++ > 5){
return;
}
Util::sleep(500);
}
}
}
///\brief Initializes a packet with new data
///\param data_ The new data for the packet
///\param len The length of the data pointed to by data_
@ -1530,7 +1556,7 @@ namespace DTSC {
} else if (type == "video") {
result += 48;
}
if (missedFrags) {
if (!skipDynamic && missedFrags) {
result += 23;
}
return result;
@ -1709,11 +1735,13 @@ namespace DTSC {
}
///\brief Determines the "packed" size of a meta object
unsigned int Meta::getSendLen(bool skipDynamic) {
unsigned int Meta::getSendLen(bool skipDynamic, std::set<unsigned long> selectedTracks) {
unsigned int dataLen = 16 + (vod ? 14 : 0) + (live ? 15 : 0) + (merged ? 17 : 0) + (bufferWindow ? 24 : 0) + 21;
for (std::map<unsigned int, Track>::iterator it = tracks.begin(); it != tracks.end(); it++) {
if (!selectedTracks.size() || selectedTracks.count(it->first)){
dataLen += it->second.getSendLen(skipDynamic);
}
}
return dataLen + 8; //add 8 bytes header
}
@ -1749,14 +1777,16 @@ namespace DTSC {
}
///\brief Writes a meta object to a socket
void Meta::send(Socket::Connection & conn, bool skipDynamic) {
int dataLen = getSendLen(skipDynamic) - 8; //strip 8 bytes header
void Meta::send(Socket::Connection & conn, bool skipDynamic, std::set<unsigned long> selectedTracks) {
int dataLen = getSendLen(skipDynamic, selectedTracks) - 8; //strip 8 bytes header
conn.SendNow(DTSC::Magic_Header, 4);
conn.SendNow(convertInt(dataLen), 4);
conn.SendNow("\340\000\006tracks\340", 10);
for (std::map<unsigned int, Track>::iterator it = tracks.begin(); it != tracks.end(); it++) {
if (!selectedTracks.size() || selectedTracks.count(it->first)){
it->second.send(conn, skipDynamic);
}
}
conn.SendNow("\000\000\356", 3);//End tracks object
if (vod) {
conn.SendNow("\000\003vod\001", 6);

58
lib/shared_memory.cpp
View file

@ -1068,6 +1068,54 @@ namespace IPC {
///\brief The deconstructor
sharedClient::~sharedClient() {
mySemaphore.close();
}
bool sharedClient::isSingleEntry() {
semaphore tmpSem(baseName.c_str(), O_RDWR);
if (!tmpSem) {
HIGH_MSG("Creating semaphore %s failed: %s, assuming we're alone", baseName.c_str(), strerror(errno));
return true;
}
//Empty is used to compare for emptyness. This is not needed when the page uses a counter
char * empty = 0;
if (!hasCounter) {
empty = (char *)malloc(payLen * sizeof(char));
if (!empty) {
HIGH_MSG("Failed to allocate %u bytes for empty payload, assuming we're not alone", payLen);
return false;
}
memset(empty, 0, payLen);
}
bool result = true;
{
semGuard tmpGuard(&tmpSem);
for (char i = 'A'; i <= 'Z'; i++) {
sharedPage tmpPage(baseName.substr(1) + i, (4096 << (i - 'A')), false, false);
if (!tmpPage.mapped) {
break;
}
int offset = 0;
while (offset + payLen + (hasCounter ? 1 : 0) <= tmpPage.len) {
//Skip our own entry
if (tmpPage.name == myPage.name && offset == offsetOnPage){
offset += payLen + (hasCounter ? 1 : 0);
continue;
}
if (!((hasCounter && tmpPage.mapped[offset] == 0) || (!hasCounter && !memcmp(tmpPage.mapped + offset, empty, payLen)))) {
result = false;
break;
}
offset += payLen + (hasCounter ? 1 : 0);
}
}
}
if (empty) {
free(empty);
}
return result;
}
///\brief Writes data to the shared data
@ -1114,6 +1162,16 @@ namespace IPC {
return (myPage.mapped + offsetOnPage + (hasCounter ? 1 : 0));
}
int sharedClient::getCounter() {
if (!hasCounter){
return -1;
}
if (!myPage.mapped) {
return 0;
}
return *(myPage.mapped + offsetOnPage);
}
userConnection::userConnection(char * _data) {
data = _data;
if (!data){

4
lib/shared_memory.h
View file

@ -60,7 +60,7 @@ namespace IPC {
class semaphore {
public:
semaphore();
semaphore(const char * name, int oflag, mode_t mode, unsigned int value);
semaphore(const char * name, int oflag, mode_t mode = 0, unsigned int value = 0);
~semaphore();
operator bool() const;
void open(const char * name, int oflag, mode_t mode = 0, unsigned int value = 0);
@ -220,6 +220,8 @@ namespace IPC {
void finish();
void keepAlive();
char * getData();
int getCounter();
bool isSingleEntry();
private:
///\brief The basename of the shared pages.
std::string baseName;

22
lib/stream.cpp
View file

@ -204,6 +204,7 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
//check in curConf for capabilities-inputs-<naam>-priority/source_match
std::string player_bin;
bool pullMode = false;
bool selected = false;
long long int curPrio = -1;
DTSC::Scan inputs = config.getMember("capabilities").getMember("inputs");
@ -224,6 +225,20 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
curPrio = input.getMember("priority").asInt();
selected = true;
}
if (input.hasMember("stream_match")){
source = input.getMember("stream_match").asString();
front = source.substr(0,source.find('*'));
back = source.substr(source.find('*')+1);
DEBUG_MSG(DLVL_MEDIUM, "Checking input %s: %s (%s)", inputs.getIndiceName(i).c_str(), input.getMember("name").asString().c_str(), source.c_str());
if (filename.substr(0,front.size()) == front && filename.substr(filename.size()-back.size()) == back){
player_bin = Util::getMyPath() + "MistIn" + input.getMember("name").asString();
curPrio = input.getMember("priority").asInt();
pullMode = true;
selected = true;
}
}
}
}
@ -261,9 +276,16 @@ bool Util::startInput(std::string streamname, std::string filename, bool forkFir
//finally, unlock the config semaphore
configLock.post();
if (pullMode){
DEBUG_MSG(DLVL_MEDIUM, "Starting %s -p -s %s %s", player_bin.c_str(), streamname.c_str(), filename.c_str());
}else{
DEBUG_MSG(DLVL_MEDIUM, "Starting %s -s %s %s", player_bin.c_str(), streamname.c_str(), filename.c_str());
}
char * argv[30] = {(char *)player_bin.c_str(), (char *)"-s", (char *)streamname.c_str(), (char *)filename.c_str()};
int argNum = 3;
if (pullMode){
argv[++argNum] = (char*)"--pull";
}
std::string debugLvl;
if (Util::Config::printDebugLevel != DEBUG && !str_args.count("--debug")){
debugLvl = JSON::Value((long long)Util::Config::printDebugLevel).asString();

40
lib/util.cpp Normal file
View file

@ -0,0 +1,40 @@
#include "util.h"
#include <iostream>
namespace Util {
bool stringScan(const std::string & src, const std::string & pattern, std::deque<std::string> & result){
result.clear();
std::deque<size_t> positions;
size_t pos = pattern.find("%", 0);
while (pos != std::string::npos){
positions.push_back(pos);
pos = pattern.find("%", pos + 1);
}
if (positions.size() == 0){
return false;
}
size_t sourcePos = 0;
size_t patternPos = 0;
std::deque<size_t>::iterator posIter = positions.begin();
while (sourcePos != std::string::npos){
//Match first part of the string
if (pattern.substr(patternPos, *posIter - patternPos) != src.substr(sourcePos, *posIter - patternPos)){
break;
}
sourcePos += *posIter - patternPos;
std::deque<size_t>::iterator nxtIter = posIter + 1;
if (nxtIter != positions.end()){
patternPos = *posIter+2;
size_t tmpPos = src.find(pattern.substr(*posIter+2, *nxtIter - patternPos), sourcePos);
result.push_back(src.substr(sourcePos, tmpPos - sourcePos));
sourcePos = tmpPos;
}else{
result.push_back(src.substr(sourcePos));
sourcePos = std::string::npos;
}
posIter++;
}
return result.size() == positions.size();
}
}

6
lib/util.h Normal file
View file

@ -0,0 +1,6 @@
#include <string>
#include <deque>
namespace Util {
bool stringScan(const std::string & src, const std::string & pattern, std::deque<std::string> & result);
}

141
src/input/input.cpp
View file

@ -83,6 +83,7 @@ namespace Mist {
singleton = this;
isBuffer = false;
streamMode = false;
}
void Input::checkHeaderTimes(std::string streamFile) {
@ -115,16 +116,27 @@ namespace Mist {
}
}
bool Input::needsLock() {
return !(config->hasOption("pull") && config->getBool("pull"));
}
int Input::run() {
if (config->getBool("json")) {
std::cout << capa.toString() << std::endl;
return 0;
}
if (streamName != "") {
config->getOption("streamname") = streamName;
}
streamName = config->getString("streamname");
nProxy.streamName = streamName;
if (config->getBool("json")) {
std::cout << capa.toString() << std::endl;
return 0;
}
streamMode = config->hasOption("pull") && config->getBool("pull");
INFO_MSG("Stream %s in %s mode", streamName.c_str(), streamMode ? "stream" : "non-stream");
if (!setup()) {
std::cerr << config->getString("cmd") << " setup failed." << std::endl;
return 0;
@ -139,6 +151,8 @@ namespace Mist {
if (!streamName.size()) {
convert();
} else if (streamMode) {
stream();
}else{
serve();
}
@ -243,53 +257,92 @@ namespace Mist {
/// Main loop for stream-style inputs.
/// This loop will start the buffer without resume support, and then repeatedly call ..... followed by ....
void Input::stream(){
IPC::semaphore pullLock;
pullLock.open(std::string("/MstPull_" + streamName).c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1);
if (!pullLock.tryWait()){
DEBUG_MSG(DLVL_DEVEL, "A pull process for stream %s is already running", streamName.c_str());
return;
}
if (Util::streamAlive(streamName)){
pullLock.post();
pullLock.close();
return;
}
if (!Util::startInput(streamName, "push://")) {//manually override stream url to start the buffer
pullLock.post();
pullLock.close();
return;
}
char userPageName[NAME_BUFFER_SIZE];
snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, streamName.c_str());
/*LTS-START*/
if(Triggers::shouldTrigger("STREAM_READY", config->getString("streamname"))){
std::string payload = config->getString("streamname")+"\n" +capa["name"].asStringRef()+"\n";
if (!Triggers::doTrigger("STREAM_READY", payload, config->getString("streamname"))){
config->is_active = false;
}
}
/*LTS-END*/
userPage.init(userPageName, PLAY_EX_SIZE, true);
if (!isBuffer) {
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++) {
bufferFrame(it->first, 1);
}
}
nProxy.userClient = IPC::sharedClient(userPageName, PLAY_EX_SIZE, true);
DEBUG_MSG(DLVL_DEVEL, "Input for stream %s started", streamName.c_str());
long long int activityCounter = Util::bootSecs();
while ((Util::bootSecs() - activityCounter) < 10 && config->is_active) { //10 second timeout
userPage.parseEach(callbackWrapper);
removeUnused();
if (userPage.amount) {
activityCounter = Util::bootSecs();
DEBUG_MSG(DLVL_INSANE, "Connected users: %d", userPage.amount);
} else {
DEBUG_MSG(DLVL_INSANE, "Timer running");
}
/*LTS-START*/
if ((Util::bootSecs() - activityCounter) >= 10 || !config->is_active){//10 second timeout
if(Triggers::shouldTrigger("STREAM_UNLOAD", config->getString("streamname"))){
std::string payload = config->getString("streamname")+"\n" +capa["name"].asStringRef()+"\n";
if (!Triggers::doTrigger("STREAM_UNLOAD", payload, config->getString("streamname"))){
activityCounter = Util::bootSecs();
config->is_active = true;
}
}
}
/*LTS-END*/
if (config->is_active){
Util::sleep(1000);
}
if (!openStreamSource()){
FAIL_MSG("Unable to connect to source");
pullLock.post();
pullLock.close();
return;
}
parseStreamHeader();
if (myMeta.tracks.size() == 0){
nProxy.userClient.finish();
finish();
DEBUG_MSG(DLVL_DEVEL, "Input for stream %s closing clean", streamName.c_str());
//end player functionality
pullLock.post();
pullLock.close();
return;
}
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
it->second.firstms = 0;
it->second.lastms = 0;
}
getNext();
unsigned long long lastTime = Util::getMS();
unsigned long long lastActive = Util::getMS();
while (thisPacket && config->is_active){
nProxy.bufferLivePacket(thisPacket, myMeta);
getNext();
nProxy.userClient.keepAlive();
if (Util::getMS() - lastTime >= 1000){
lastTime = Util::getMS();
if (nProxy.userClient.isSingleEntry()){
if (lastTime - lastActive >= 10000){//10sec timeout
config->is_active = false;
}
}else{
lastActive = lastTime;
}
}
}
closeStreamSource();
while (config->is_active){
Util::sleep(500);
nProxy.userClient.keepAlive();
if (Util::getMS() - lastTime >= 1000){
lastTime = Util::getMS();
if (nProxy.userClient.isSingleEntry()){
if (lastTime - lastActive >= 10000){//10sec timeout
config->is_active = false;
}
}else{
lastActive = lastTime;
}
}
}
nProxy.userClient.finish();
finish();
pullLock.post();
pullLock.close();
return;
}
void Input::finish() {

8
src/input/input.h
View file

@ -23,6 +23,8 @@ namespace Mist {
virtual void onCrash(){}
virtual void argumentsParsed(){}
virtual ~Input() {};
virtual bool needsLock();
protected:
static void callbackWrapper(char * data, size_t len, unsigned int id);
virtual bool setup() = 0;
@ -31,6 +33,9 @@ namespace Mist {
virtual void getNext(bool smart = true) {};
virtual void seek(int seekTime){};
virtual void finish();
virtual bool openStreamSource() { return false; };
virtual void closeStreamSource() {};
virtual void parseStreamHeader() {};
void play(int until = 0);
void playOnce();
void quitPlay();
@ -40,6 +45,9 @@ namespace Mist {
virtual void userCallback(char * data, size_t len, unsigned int id);
virtual void convert();
virtual void serve();
virtual void stream();
bool streamMode;
virtual void parseHeader();
bool bufferFrame(unsigned int track, unsigned int keyNum);

86
src/input/input_buffer.cpp
View file

@ -256,7 +256,9 @@ namespace Mist {
long long unsigned int lastms = 0;
long long unsigned int fragCount = 0xFFFFull;
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++) {
if (it->second.type == "meta" || !it->second.type.size()){continue;}
if (it->second.type == "meta" || !it->second.type.size()) {
continue;
}
if (it->second.init.size()) {
if (!initData.count(it->first) || initData[it->first] != it->second.init) {
initData[it->first] = it->second.init;
@ -569,6 +571,66 @@ namespace Mist {
//Track is set to "New track request", assign new track id and create shared memory page
//This indicates that the 'current key' part of the element is set to contain the original track id from the pushing process
if (value & 0x80000000) {
if (value & 0x40000000) {
unsigned long finalMap = value & ~0xC0000000;
//Register the new track as an active track.
activeTracks.insert(finalMap);
//Register the time of registration as initial value for the lastUpdated field, plus an extra 5 seconds just to be sure.
lastUpdated[finalMap] = Util::bootSecs() + 5;
//Register the user thats is pushing this element
pushLocation[finalMap] = data;
//Initialize the metadata for this track
if (!myMeta.tracks.count(finalMap)) {
DEBUG_MSG(DLVL_MEDIUM, "Inserting metadata for track number %d", finalMap);
IPC::sharedPage tMeta;
char tempMetaName[NAME_BUFFER_SIZE];
snprintf(tempMetaName, NAME_BUFFER_SIZE, SHM_TRACK_META, config->getString("streamname").c_str(), finalMap);
tMeta.init(tempMetaName, 8388608, false);
//The page exist, now we try to read in the metadata of the track
//Store the size of the dtsc packet to read.
unsigned int len = ntohl(((int *)tMeta.mapped)[1]);
//Temporary variable, won't be used again
unsigned int tempForReadingMeta = 0;
//Read in the metadata through a temporary JSON object
///\todo Optimize this part. Find a way to not have to store the metadata in JSON first, but read it from the page immediately
JSON::Value tempJSONForMeta;
JSON::fromDTMI((const unsigned char *)tMeta.mapped + 8, len, tempForReadingMeta, tempJSONForMeta);
tMeta.master = true;
//Construct a metadata object for the current track
DTSC::Meta trackMeta(tempJSONForMeta);
myMeta.tracks[finalMap] = trackMeta.tracks.begin()->second;
myMeta.tracks[finalMap].firstms = 0;
myMeta.tracks[finalMap].lastms = 0;
userConn.setTrackId(index, finalMap);
userConn.setKeynum(index, 0x0000);
char firstPage[NAME_BUFFER_SIZE];
snprintf(firstPage, NAME_BUFFER_SIZE, SHM_TRACK_INDEX, config->getString("streamname").c_str(), finalMap);
nProxy.metaPages[finalMap].init(firstPage, 8192, false);
INFO_MSG("Meh %d", finalMap);
//Update the metadata for this track
updateTrackMeta(finalMap);
INFO_MSG("Setting hasPush to true, quickNegotiate");
hasPush = true;
}
//Write the final mapped track number and keyframe number to the user page element
//This is used to resume pushing as well as pushing new tracks
userConn.setTrackId(index, finalMap);
userConn.setKeynum(index, myMeta.tracks[finalMap].keys.size());
//Update the metadata to reflect all changes
updateMeta();
continue;
}
//Set the temporary track id for this item, and increase the temporary value for use with the next track
unsigned long long tempMapping = nextTempId++;
//Add the temporary track id to the list of tracks that are currently being negotiated
@ -827,7 +889,9 @@ namespace Mist {
tmpNum = config->getOption("bufferTime").asInt();
}
}
if (tmpNum < 1000){tmpNum = 1000;}
if (tmpNum < 1000) {
tmpNum = 1000;
}
//if the new value is different, print a message and apply it
if (bufferTime != tmpNum) {
DEBUG_MSG(DLVL_DEVEL, "Setting bufferTime from %u to new value of %lli", bufferTime, tmpNum);
@ -943,28 +1007,18 @@ namespace Mist {
++it;
}
if (streamCfg
&& streamCfg.getMember("record")
&& streamCfg.getMember("record").asString().size() > 0
&& has_keyframes
)
{
if (streamCfg && streamCfg.getMember("record") && streamCfg.getMember("record").asString().size() > 0 && has_keyframes) {
// @todo check if output is already running ?
if (recordingPid == -1 && config != NULL) {
INFO_MSG("The stream %s has a value specified for the recording. We're goint to start an output and record into %s", config->getString("streamname").c_str(), streamCfg.getMember("record").asString().c_str());
configLock.post();
configLock.close();
INFO_MSG("The stream %s has a value specified for the recording. "
"We're going to start an output and record into %s",
config->getString("streamname").c_str(),
streamCfg.getMember("record").asString().c_str());
recordingPid = Util::startRecording(config->getString("streamname"));
if (recordingPid < 0) {
FAIL_MSG("Failed to start the recording for %s", config->getString("streamname").c_str());
// @todo shouldn't we do configList.post(), configLock.close() and return false?
// @todo discuss with Jaron. 2015.09.26, remove this comment when discussed.
}
INFO_MSG("We started an output for recording with PID: %d", recordingPid);
return true;

198
src/input/input_dtsc.cpp
View file

@ -7,6 +7,9 @@
#include <mist/stream.h>
#include <mist/defines.h>
#include <mist/util.h>
#include <mist/bitfields.h>
#include "input_dtsc.h"
namespace Mist {
@ -15,6 +18,7 @@ namespace Mist {
capa["desc"] = "Enables DTSC Input";
capa["priority"] = 9ll;
capa["source_match"] = "/*.dtsc";
capa["stream_match"] = "dtsc://*";
capa["codecs"][0u][0u].append("H264");
capa["codecs"][0u][0u].append("H263");
capa["codecs"][0u][0u].append("VP6");
@ -22,9 +26,149 @@ namespace Mist {
capa["codecs"][0u][1u].append("AAC");
capa["codecs"][0u][1u].append("MP3");
capa["codecs"][0u][1u].append("vorbis");
JSON::Value option;
option["long"] = "pull";
option["short"] = "p";
option["help"] = "Start this input in pull mode.";
option["value"].append(0ll);
config->addOption("pull", option);
}
void parseDTSCURI(const std::string & src, std::string & host, uint16_t & port, std::string & password, std::string & streamName) {
host = "";
port = 4200;
password = "";
streamName = "";
std::deque<std::string> matches;
if (Util::stringScan(src, "%s:%s@%s/%s", matches)) {
host = matches[0];
port = atoi(matches[1].c_str());
password = matches[2];
streamName = matches[3];
return;
}
//Using default streamname
if (Util::stringScan(src, "%s:%s@%s", matches)) {
host = matches[0];
port = atoi(matches[1].c_str());
password = matches[2];
return;
}
//Without password
if (Util::stringScan(src, "%s:%s/%s", matches)) {
host = matches[0];
port = atoi(matches[1].c_str());
streamName = matches[2];
return;
}
//Using default port
if (Util::stringScan(src, "%s@%s/%s", matches)) {
host = matches[0];
password = matches[1];
streamName = matches[2];
return;
}
//Default port, no password
if (Util::stringScan(src, "%s/%s", matches)) {
host = matches[0];
streamName = matches[1];
return;
}
//No password, default streamname
if (Util::stringScan(src, "%s:%s", matches)) {
host = matches[0];
port = atoi(matches[1].c_str());
return;
}
//Default port and streamname
if (Util::stringScan(src, "%s@%s", matches)) {
host = matches[0];
password = matches[1];
return;
}
//Default port and streamname, no password
if (Util::stringScan(src, "%s", matches)) {
host = matches[0];
return;
}
}
void inputDTSC::parseStreamHeader() {
while (srcConn.connected()){
srcConn.spool();
if (srcConn.Received().available(8)){
if (srcConn.Received().copy(4) == "DTCM" || srcConn.Received().copy(4) == "DTSC") {
// Command message
std::string toRec = srcConn.Received().copy(8);
unsigned long rSize = Bit::btohl(toRec.c_str() + 4);
if (!srcConn.Received().available(8 + rSize)) {
continue; //abort - not enough data yet
}
//Ignore initial DTCM message, as this is a "hi" message from the server
if (srcConn.Received().copy(4) == "DTCM"){
srcConn.Received().remove(8 + rSize);
}else{
std::string dataPacket = srcConn.Received().remove(8+rSize);
DTSC::Packet metaPack(dataPacket.data(), dataPacket.size());
myMeta.reinit(metaPack);
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
continueNegotiate(it->first, true);
}
break;
}
}else{
INFO_MSG("Received a wrong type of packet - '%s'", srcConn.Received().copy(4).c_str());
break;
}
}
}
}
bool inputDTSC::openStreamSource() {
std::string source = config->getString("input");
if (source.find("dtsc://") == 0) {
source.erase(0, 7);
}
std::string host;
uint16_t port;
std::string password;
std::string streamName;
parseDTSCURI(source, host, port, password, streamName);
std::string givenStream = config->getString("streamname");
if (streamName == "") {
streamName = givenStream;
}else{
if (givenStream.find("+") != std::string::npos){
streamName += givenStream.substr(givenStream.find("+"));
}
}
srcConn = Socket::Connection(host, port, true);
if (!srcConn.connected()){
return false;
}
JSON::Value prep;
prep["cmd"] = "play";
prep["version"] = "MistServer " PACKAGE_VERSION;
prep["stream"] = streamName;
srcConn.SendNow("DTCM");
char sSize[4] = {0, 0, 0, 0};
Bit::htobl(sSize, prep.packedSize());
srcConn.SendNow(sSize, 4);
prep.sendTo(srcConn);
return true;
}
void inputDTSC::closeStreamSource(){
srcConn.close();
}
bool inputDTSC::setup() {
if (streamMode) {
return true;
} else {
if (config->getString("input") == "-") {
std::cerr << "Input from stdin not yet supported" << std::endl;
return false;
@ -46,10 +190,14 @@ namespace Mist {
if (!inFile) {
return false;
}
}
return true;
}
bool inputDTSC::readHeader() {
if (streamMode) {
return true;
}
if (!inFile) {
return false;
}
@ -69,6 +217,55 @@ namespace Mist {
}
void inputDTSC::getNext(bool smart) {
if (streamMode){
thisPacket.reInit(srcConn);
if (thisPacket.getVersion() == DTSC::DTCM){
std::string cmd;
thisPacket.getString("cmd", cmd);
if (cmd == "reset"){
//Read next packet
thisPacket.reInit(srcConn);
if (thisPacket.getVersion() == DTSC::DTSC_HEAD){
DTSC::Meta newMeta;
newMeta.reinit(thisPacket);
//Detect new tracks
std::set<unsigned int> newTracks;
for (std::map<unsigned int, DTSC::Track>::iterator it = newMeta.tracks.begin(); it != newMeta.tracks.end(); it++){
if (!myMeta.tracks.count(it->first)){
newTracks.insert(it->first);
}
}
for (std::set<unsigned int>::iterator it = newTracks.begin(); it != newTracks.end(); it++){
INFO_MSG("Adding track %d to internal metadata", *it);
myMeta.tracks[*it] = newMeta.tracks[*it];
continueNegotiate(*it, true);
}
//Detect removed tracks
std::set<unsigned int> deletedTracks;
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (!newMeta.tracks.count(it->first)){
deletedTracks.insert(it->first);
}
}
for(std::set<unsigned int>::iterator it = deletedTracks.begin(); it != deletedTracks.end(); it++){
INFO_MSG("Deleting track %d from internal metadata", *it);
myMeta.tracks.erase(*it);
}
//Read next packet before returning
thisPacket.reInit(srcConn);
}else{
myMeta = DTSC::Meta();
}
}else{
//Read next packet before returning
thisPacket.reInit(srcConn);
}
}
}else{
if (smart) {
inFile.seekNext();
} else {
@ -76,6 +273,7 @@ namespace Mist {
}
thisPacket = inFile.getPacket();
}
}
void inputDTSC::seek(int seekTime) {
inFile.seek_time(seekTime);

5
src/input/input_dtsc.h
View file

@ -7,6 +7,9 @@ namespace Mist {
inputDTSC(Util::Config * cfg);
protected:
//Private Functions
bool openStreamSource();
void closeStreamSource();
void parseStreamHeader();
bool setup();
bool readHeader();
void getNext(bool smart = true);
@ -14,6 +17,8 @@ namespace Mist {
void trackSelect(std::string trackSpec);
DTSC::File inFile;
Socket::Connection srcConn;
};
}

4
src/input/input_ts.cpp
View file

@ -441,6 +441,10 @@ namespace Mist {
}
} while (threadCount);
}
bool inputTS::needsLock() {
return false;
}
#endif
}

3
src/input/input_ts.h
View file

@ -13,6 +13,9 @@ namespace Mist {
public:
inputTS(Util::Config * cfg);
~inputTS();
#ifdef TSLIVE_INPUT
bool needsLock();
#endif
protected:
//Private Functions
bool setup();

17
src/input/mist_in.cpp
View file

@ -17,8 +17,9 @@ int main(int argc, char * argv[]) {
if (conf.parseArgs(argc, argv)) {
std::string streamName = conf.getString("streamname");
conv.argumentsParsed();
#ifndef INPUT_NOLOCK
IPC::semaphore playerLock;
if (conv.needsLock()){
if (streamName.size()){
char semName[NAME_BUFFER_SIZE];
snprintf(semName, NAME_BUFFER_SIZE, SEM_INPUT, streamName.c_str());
@ -28,21 +29,21 @@ int main(int argc, char * argv[]) {
return 1;
}
}
#endif
}
conf.activate();
while (conf.is_active){
pid_t pid = fork();
if (pid == 0){
#ifndef INPUT_NOLOCK
if (conv.needsLock()){
playerLock.close();
#endif
}
return conv.run();
}
if (pid == -1){
DEBUG_MSG(DLVL_FAIL, "Unable to spawn player process");
#ifndef INPUT_NOLOCK
if (conv.needsLock()){
playerLock.post();
#endif
}
return 2;
}
//wait for the process to exit
@ -71,11 +72,11 @@ int main(int argc, char * argv[]) {
DEBUG_MSG(DLVL_DEVEL, "Input for stream %s uncleanly shut down! Restarting...", streamName.c_str());
}
}
#ifndef INPUT_NOLOCK
if (conv.needsLock()){
playerLock.post();
playerLock.unlink();
playerLock.close();
#endif
}
}
return 0;
}

44
src/io.cpp
View file

@ -559,10 +559,11 @@ namespace Mist {
bufferNext(packet, myMeta);
}
void InOutBase::continueNegotiate(unsigned long tid) {
nProxy.continueNegotiate(tid, myMeta);
void InOutBase::continueNegotiate(unsigned long tid, bool quickNegotiate) {
nProxy.continueNegotiate(tid, myMeta, quickNegotiate);
}
void negotiationProxy::continueNegotiate(unsigned long tid, DTSC::Meta & myMeta) {
void negotiationProxy::continueNegotiate(unsigned long tid, DTSC::Meta & myMeta, bool quickNegotiate) {
if (!tid) {
return;
}
@ -618,12 +619,49 @@ namespace Mist {
unsigned long offset = 6 * trackOffset[tid];
//If we have a new track to negotiate
if (!trackState.count(tid)) {
memset(tmp + offset, 0, 4);
if (quickNegotiate){
unsigned long finalTid = getpid() + tid;
unsigned short firstPage = 1;
INFO_MSG("HANDLING quick negotiation for track %d ~> %d", tid, finalTid)
MEDIUM_MSG("Buffer has indicated that incoming track %lu should start writing on track %lu, page %lu", tid, finalTid, firstPage);
trackMap[tid] = finalTid;
if (myMeta.tracks.count(finalTid) && myMeta.tracks[finalTid].lastms){
myMeta.tracks[finalTid].lastms = 0;
}
trackState[tid] = FILL_ACC;
char pageName[NAME_BUFFER_SIZE];
snprintf(pageName, NAME_BUFFER_SIZE, SHM_TRACK_META, streamName.c_str(), finalTid);
metaPages[tid].init(pageName, 8 * 1024 * 1024, true);
metaPages[tid].master = false;
DTSC::Meta tmpMeta;
tmpMeta.tracks[finalTid] = myMeta.tracks[tid];
tmpMeta.tracks[finalTid].trackID = finalTid;
JSON::Value tmpVal = tmpMeta.toJSON();
std::string tmpStr = tmpVal.toNetPacked();
memcpy(metaPages[tid].mapped, tmpStr.data(), tmpStr.size());
snprintf(pageName, NAME_BUFFER_SIZE, SHM_TRACK_INDEX, streamName.c_str(), finalTid);
metaPages[tid].init(pageName, 8 * 1024 * 1024, true);
metaPages[tid].master = false;
Bit::htobl(tmp + offset, finalTid | 0xC0000000);
Bit::htobs(tmp + offset + 4, firstPage);
}else{
INFO_MSG("Starting negotiation for incoming track %lu, at offset %lu", tid, trackOffset[tid]);
memset(tmp + offset, 0, 4);
tmp[offset] = 0x80;
tmp[offset + 4] = ((tid >> 8) & 0xFF);
tmp[offset + 5] = (tid & 0xFF);
trackState[tid] = FILL_NEW;
}
return;
}
#if defined(__CYGWIN__) || defined(_WIN32)

4
src/io.h
View file

@ -59,7 +59,7 @@ namespace Mist {
std::map<int,unsigned long long int> iVecs;
IPC::sharedPage encryptionPage;
void continueNegotiate(unsigned long tid, DTSC::Meta & myMeta);
void continueNegotiate(unsigned long tid, DTSC::Meta & myMeta, bool quickNegotiate = false);
};
///\brief Class containing all basic input and output functions.
@ -74,7 +74,7 @@ namespace Mist {
void bufferLivePacket(JSON::Value & packet);
void bufferLivePacket(DTSC::Packet & packet);
protected:
void continueNegotiate(unsigned long tid);
void continueNegotiate(unsigned long tid, bool quickNegotiate = false);

44
src/output/output.cpp
View file

@ -278,6 +278,16 @@ namespace Mist {
onFail();
return;
}
if (!source.size()){
std::string strName = streamName;
Util::sanitizeName(strName);
IPC::sharedPage serverCfg("!mistConfig", DEFAULT_CONF_PAGE_SIZE, false, false); ///< Contains server configuration and capabilities
IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
configLock.wait();
DTSC::Scan streamCfg = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("streams").getMember(strName);
source = streamCfg.getMember("source").asString();
configLock.post();
}
char pageId[NAME_BUFFER_SIZE];
snprintf(pageId, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, streamName.c_str());
nProxy.metaPages.clear();
@ -416,9 +426,20 @@ namespace Mist {
// when we don't see this explicitly it makes debugging the recording feature
// a bit painfull :)
if (selectedTracks.size() == 0) {
WARN_MSG("We didn't find any tracks which that we can use. selectedTrack.size() is 0.");
INSANE_MSG("We didn't find any tracks which that we can use. selectedTrack.size() is 0.");
for (std::map<unsigned int,DTSC::Track>::iterator trit = myMeta.tracks.begin(); trit != myMeta.tracks.end(); trit++){
WARN_MSG("Found track/codec: %s", trit->second.codec.c_str());
INSANE_MSG("Found track/codec: %s", trit->second.codec.c_str());
}
if (!myMeta.tracks.size() && (source.find("dtsc://") == 0)){
//Wait 5 seconds and try again. Keep a counter, try at most 3 times
static int counter = 0;
if (counter++ < 10){
Util::wait(1000);
nProxy.userClient.keepAlive();
stats();
updateMeta();
selectDefaultTracks();
}
}
}
/*end-roxlu*/
@ -898,6 +919,25 @@ namespace Mist {
}
if ( !sentHeader){
DEBUG_MSG(DLVL_DONTEVEN, "sendHeader");
bool waitLonger = false;
if (!myMeta.tracks.size()){
waitLonger = true;
}else{
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (!it->second.keys.size()){
waitLonger = true;
break;
}
}
}
if (waitLonger){
updateMeta();
Util::sleep(1000);
static unsigned int metaTries = 0;
if(++metaTries < 7){
continue;
}
}
sendHeader();
}
prepareNext();

1
src/output/output.h
View file

@ -108,6 +108,7 @@ namespace Mist {
bool sought;///<If a seek has been done, this is set to true. Used for seeking on prepareNext().
bool completeKeyReadyTimeOut;//a bool to see if there has been a keyframe TimeOut for complete keys in Live
protected://these are to be messed with by child classes
std::string source;
virtual std::string getConnectedHost();
virtual std::string getConnectedBinHost();

45
src/output/output_dtsc.cpp
View file

@ -29,6 +29,7 @@ namespace Mist {
myConn.SendNow(sSize, 4);
prep.sendTo(myConn);
pushing = false;
fastAsPossibleTime = 0;
}
OutDTSC::~OutDTSC() {}
@ -44,12 +45,52 @@ namespace Mist {
}
void OutDTSC::sendNext(){
if (!realTime && thisPacket.getTime() >= fastAsPossibleTime){
realTime = 1000;
}
if (thisPacket.getFlag("keyframe")){
std::set<unsigned long> availableTracks;
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (it->second.type == "video" || it->second.type == "audio"){
availableTracks.insert(it->first);
}
}
if (availableTracks != selectedTracks){
//reset, resendheader
JSON::Value prep;
prep["cmd"] = "reset";
/// \todo Make this securererer.
unsigned long sendSize = prep.packedSize();
myConn.SendNow("DTCM");
char sSize[4] = {0, 0, 0, 0};
Bit::htobl(sSize, prep.packedSize());
myConn.SendNow(sSize, 4);
prep.sendTo(myConn);
}
}
myConn.SendNow(thisPacket.getData(), thisPacket.getDataLen());
}
void OutDTSC::sendHeader(){
sentHeader = true;
myMeta.send(myConn, true);
selectedTracks.clear();
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (it->second.type == "video" || it->second.type == "audio"){
selectedTracks.insert(it->first);
}
}
myMeta.send(myConn, true, selectedTracks);
if (myMeta.live){
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (!fastAsPossibleTime || it->second.lastms < fastAsPossibleTime){
fastAsPossibleTime = it->second.lastms;
realTime = 0;
}
}
}else{
fastAsPossibleTime = 50000;//50 seconds
realTime = 0;
}
}
void OutDTSC::onRequest(){
@ -76,6 +117,8 @@ namespace Mist {
streamName = dScan.getMember("stream").asString();
Util::sanitizeName(streamName);
parseData = true;
INFO_MSG("Handled play for stream %s", streamName.c_str());
setBlocking(false);
}
void OutDTSC::handlePush(DTSC::Scan & dScan){

1
src/output/output_dtsc.h
View file

@ -15,6 +15,7 @@ namespace Mist {
bool pushing;
void handlePush(DTSC::Scan & dScan);
void handlePlay(DTSC::Scan & dScan);
unsigned long long fastAsPossibleTime;
};
}

21
src/output/output_hls.cpp
View file

@ -6,6 +6,27 @@ namespace Mist {
///\brief Builds an index file for HTTP Live streaming.
///\return The index file for HTTP Live Streaming.
std::string OutHLS::liveIndex() {
static int timer = 0;
bool checkWait = true;
while (checkWait && ++timer < 10){
checkWait = false;
if (!myMeta.tracks.size()){
checkWait = true;
}
for (std::map<unsigned int,DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (it->second.keys.size() <= 3){
checkWait = true;
break;
}
}
if (checkWait){
Util::sleep(500);
INFO_MSG("SLeeping timer %d", timer);
updateMeta();
}
}
std::stringstream result;
result << "#EXTM3U\r\n";
int audioId = -1;

25
src/output/output_progressive_mp4.cpp
View file

@ -43,6 +43,9 @@ namespace Mist {
///\todo This function does not indicate errors anywhere... maybe fix this...
std::string OutProgressiveMP4::DTSCMeta2MP4Header(long long & size, int fragmented) {
if (myMeta.live){
completeKeysOnly = true;
}
//Make sure we have a proper being value for the size...
size = 0;
//Stores the result of the function
@ -745,6 +748,28 @@ namespace Mist {
void OutProgressiveMP4::setvidTrack() {
vidTrack = 0;
static int timer = 0;
bool checkWait = true;
while (checkWait && ++timer < 10){
checkWait = false;
if (!myMeta.tracks.size()){
checkWait = true;
}
for (std::map<unsigned int,DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (!it->second.keys.size()){
checkWait = true;
break;
}
}
if (checkWait){
Util::sleep(500);
updateMeta();
}
}
if (!selectedTracks.size()){
selectDefaultTracks();
}
for (std::set<long unsigned int>::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++) {
//Find video track
if (myMeta.tracks[*it].type == "video") {