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

Backported various little edits from Pro edition.

Browse source
This commit is contained in:
Thulinma 2016-05-30 15:17:54 +02:00
parent ef9938da0c
commit 4c9c6fa7ba
78 changed files with 2334 additions and 1266 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/output/mist_out.cpp
View file

@ -1,6 +1,7 @@
#include OUTPUTTYPE
#include <mist/config.h>
#include <mist/socket.h>
#include <mist/defines.h>
int spawnForked(Socket::Connection & S){
mistOut tmp(S);
@ -15,6 +16,7 @@ int main(int argc, char * argv[]) {
std::cout << mistOut::capa.toString() << std::endl;
return -1;
}
conf.activate();
if (mistOut::listenMode()){
conf.serveForkedSocket(spawnForked);
}else{

685
src/output/output.cpp
View file

@ -44,7 +44,6 @@ namespace Mist {
maxSkipAhead = 7500;
minSkipAhead = 5000;
realTime = 1000;
completeKeyReadyTimeOut = false;
if (myConn){
setBlocking(true);
}else{
@ -57,26 +56,31 @@ namespace Mist {
isBlocking = blocking;
myConn.setBlocking(isBlocking);
}
Output::~Output(){}
void Output::updateMeta(){
//read metadata from page to myMeta variable
if (nProxy.metaPages[0].mapped){
IPC::semaphore * liveSem = 0;
if (!myMeta.vod){
static char liveSemName[NAME_BUFFER_SIZE];
snprintf(liveSemName, NAME_BUFFER_SIZE, SEM_LIVE, streamName.c_str());
IPC::semaphore liveMeta(liveSemName, O_CREAT | O_RDWR, ACCESSPERMS, 1);
bool lock = myMeta.live;
if (lock){
liveMeta.wait();
}
if (metaPages[0].mapped){
DTSC::Packet tmpMeta(metaPages[0].mapped, metaPages[0].len, true);
liveSem = new IPC::semaphore(liveSemName, O_RDWR, ACCESSPERMS, 1);
if (*liveSem){
liveSem->wait();
}else{
delete liveSem;
liveSem = 0;
}
}
DTSC::Packet tmpMeta(nProxy.metaPages[0].mapped, nProxy.metaPages[0].len, true);
if (tmpMeta.getVersion()){
myMeta.reinit(tmpMeta);
}
}
if (lock){
liveMeta.post();
if (liveSem){
liveSem->post();
delete liveSem;
liveSem = 0;
}
}
}
@ -95,7 +99,7 @@ namespace Mist {
if (isInitialized){
return;
}
if (metaPages[0].mapped){
if (nProxy.metaPages[0].mapped){
return;
}
if (streamName.size() < 1){
@ -120,23 +124,24 @@ namespace Mist {
return myConn.getBinHost();
}
bool Output::isReadyForPlay() {
if (myMeta.tracks.size()){
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (it->second.keys.size() >= 2){
return true;
}
}
}
return false;
}
/// Connects or reconnects to the stream.
/// Assumes streamName class member has been set already.
/// Will start input if not currently active, calls onFail() if this does not succeed.
/// After assuring stream is online, clears metaPages, then sets metaPages[0], statsPage and userClient to (hopefully) valid handles.
/// Finally, calls updateMeta() and stats()
/// After assuring stream is online, clears nProxy.metaPages, then sets nProxy.metaPages[0], statsPage and nProxy.userClient to (hopefully) valid handles.
/// Finally, calls updateMeta()
void Output::reconnect(){
if (!Util::startInput(streamName)){
DEBUG_MSG(DLVL_FAIL, "Opening stream failed - aborting initalization");
onFail();
return;
}
char pageId[NAME_BUFFER_SIZE];
snprintf(pageId, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, streamName.c_str());
metaPages.clear();
metaPages[0].init(pageId, DEFAULT_META_PAGE_SIZE);
if (!metaPages[0].mapped){
FAIL_MSG("Could not connect to server for %s", streamName.c_str());
FAIL_MSG("Opening stream %s failed - aborting initalization", streamName.c_str());
onFail();
return;
}
@ -144,14 +149,35 @@ namespace Mist {
statsPage.finish();
}
statsPage = IPC::sharedClient(SHM_STATISTICS, STAT_EX_SIZE, true);
if (userClient.getData()){
userClient.finish();
if (nProxy.userClient.getData()){
nProxy.userClient.finish();
}
char userPageName[NAME_BUFFER_SIZE];
snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, streamName.c_str());
userClient = IPC::sharedClient(userPageName, PLAY_EX_SIZE, true);
stats();
nProxy.userClient = IPC::sharedClient(userPageName, PLAY_EX_SIZE, true);
char pageId[NAME_BUFFER_SIZE];
snprintf(pageId, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, streamName.c_str());
nProxy.metaPages.clear();
nProxy.metaPages[0].init(pageId, DEFAULT_STRM_PAGE_SIZE);
if (!nProxy.metaPages[0].mapped){
FAIL_MSG("Could not connect to server for %s", streamName.c_str());
onFail();
return;
}
stats(true);
updateMeta();
if (myMeta.live && !isReadyForPlay()){
unsigned long long waitUntil = Util::epoch() + 15;
while (!isReadyForPlay()){
if (Util::epoch() > waitUntil){
INFO_MSG("Giving up waiting for playable tracks. Stream: %s, IP: %s", streamName.c_str(), getConnectedHost().c_str());
break;
}
Util::wait(750);
stats();
updateMeta();
}
}
}
void Output::selectDefaultTracks(){
@ -192,10 +218,12 @@ namespace Mist {
}
if (!found){
for (std::map<unsigned int,DTSC::Track>::iterator trit = myMeta.tracks.begin(); trit != myMeta.tracks.end(); trit++){
if (trit->second.codec == (*itc).asStringRef()){
if (trit->second.codec == (*itc).asStringRef() || (*itc).asStringRef() == "*"){
genCounter++;
found = true;
break;
if ((*itc).asStringRef() != "*"){
break;
}
}
}
}
@ -206,16 +234,16 @@ namespace Mist {
if (selCounter + genCounter > bestSoFarCount){
bestSoFarCount = selCounter + genCounter;
bestSoFar = index;
DEBUG_MSG(DLVL_HIGH, "Match (%u/%u): %s", selCounter, selCounter+genCounter, (*it).toString().c_str());
HIGH_MSG("Match (%u/%u): %s", selCounter, selCounter+genCounter, (*it).toString().c_str());
}
}else{
DEBUG_MSG(DLVL_VERYHIGH, "Not a match for currently selected tracks: %s", (*it).toString().c_str());
VERYHIGH_MSG("Not a match for currently selected tracks: %s", (*it).toString().c_str());
}
}
index++;
}
DEBUG_MSG(DLVL_MEDIUM, "Trying to fill: %s", capa["codecs"][bestSoFar].toString().c_str());
MEDIUM_MSG("Trying to fill: %s", capa["codecs"][bestSoFar].toString().c_str());
//try to fill as many codecs simultaneously as possible
if (capa["codecs"][bestSoFar].size() > 0){
jsonForEach(capa["codecs"][bestSoFar], itb) {
@ -233,10 +261,12 @@ namespace Mist {
}
if (!found){
for (std::map<unsigned int,DTSC::Track>::iterator trit = myMeta.tracks.begin(); trit != myMeta.tracks.end(); trit++){
if (trit->second.codec == (*itc).asStringRef()){
if (trit->second.codec == (*itc).asStringRef() || (*itc).asStringRef() == "*"){
selectedTracks.insert(trit->first);
found = true;
break;
if ((*itc).asStringRef() != "*"){
break;
}
}
}
}
@ -259,6 +289,37 @@ namespace Mist {
DEBUG_MSG(DLVL_MEDIUM, "Selected tracks: %s (%lu)", selected.str().c_str(), selectedTracks.size());
}
if (selectedTracks.size() == 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++){
INSANE_MSG("Found track/codec: %s", trit->second.codec.c_str());
}
static std::string source;
if (!source.size()){
IPC::sharedPage serverCfg(SHM_CONF, DEFAULT_CONF_PAGE_SIZE, false, false); ///< Contains server configuration and capabilities
IPC::semaphore configLock(SEM_CONF, O_CREAT | O_RDWR, ACCESSPERMS, 1);
configLock.wait();
std::string smp = streamName.substr(0, streamName.find_first_of("+ "));
//check if smp (everything before + or space) exists
DTSC::Scan streamCfg = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("streams").getMember(smp);
if (streamCfg){
source = streamCfg.getMember("source").asString();
}
configLock.post();
configLock.close();
}
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();
}
}
}
}
/// Clears the buffer, sets parseData to false, and generally makes not very much happen at all.
@ -287,14 +348,15 @@ namespace Mist {
}
int Output::pageNumForKey(long unsigned int trackId, long long int keyNum){
if (!metaPages.count(trackId)){
if (!nProxy.metaPages.count(trackId) || !nProxy.metaPages[trackId].mapped){
char id[NAME_BUFFER_SIZE];
snprintf(id, NAME_BUFFER_SIZE, SHM_TRACK_INDEX, streamName.c_str(), trackId);
metaPages[trackId].init(id, 8 * 1024);
nProxy.metaPages[trackId].init(id, SHM_TRACK_INDEX_SIZE);
}
int len = metaPages[trackId].len / 8;
if (!nProxy.metaPages[trackId].mapped){return -1;}
int len = nProxy.metaPages[trackId].len / 8;
for (int i = 0; i < len; i++){
int * tmpOffset = (int *)(metaPages[trackId].mapped + (i * 8));
int * tmpOffset = (int *)(nProxy.metaPages[trackId].mapped + (i * 8));
long amountKey = ntohl(tmpOffset[1]);
if (amountKey == 0){continue;}
long tmpKey = ntohl(tmpOffset[0]);
@ -304,19 +366,40 @@ namespace Mist {
}
return -1;
}
/// Gets the highest page number available for the given trackId.
int Output::pageNumMax(long unsigned int trackId){
if (!nProxy.metaPages.count(trackId) || !nProxy.metaPages[trackId].mapped){
char id[NAME_BUFFER_SIZE];
snprintf(id, NAME_BUFFER_SIZE, SHM_TRACK_INDEX, streamName.c_str(), trackId);
nProxy.metaPages[trackId].init(id, SHM_TRACK_INDEX_SIZE);
}
if (!nProxy.metaPages[trackId].mapped){return -1;}
int len = nProxy.metaPages[trackId].len / 8;
int highest = -1;
for (int i = 0; i < len; i++){
int * tmpOffset = (int *)(nProxy.metaPages[trackId].mapped + (i * 8));
long amountKey = ntohl(tmpOffset[1]);
if (amountKey == 0){continue;}
long tmpKey = ntohl(tmpOffset[0]);
if (tmpKey > highest){highest = tmpKey;}
}
return highest;
}
void Output::loadPageForKey(long unsigned int trackId, long long int keyNum){
if (myMeta.vod && keyNum > myMeta.tracks[trackId].keys.rbegin()->getNumber()){
curPage.erase(trackId);
INFO_MSG("Seek in track %lu to key %lld aborted, is > %lld", trackId, keyNum, myMeta.tracks[trackId].keys.rbegin()->getNumber());
nProxy.curPage.erase(trackId);
currKeyOpen.erase(trackId);
return;
}
DEBUG_MSG(DLVL_VERYHIGH, "Loading track %lu, containing key %lld", trackId, keyNum);
VERYHIGH_MSG("Loading track %lu, containing key %lld", trackId, keyNum);
unsigned int timeout = 0;
unsigned long pageNum = pageNumForKey(trackId, keyNum);
while (pageNum == -1){
if (!timeout){
DEBUG_MSG(DLVL_HIGH, "Requesting page with key %lu:%lld", trackId, keyNum);
HIGH_MSG("Requesting page with key %lu:%lld", trackId, keyNum);
}
++timeout;
//if we've been waiting for this page for 3 seconds, reconnect to the stream - something might be going wrong...
@ -325,8 +408,8 @@ namespace Mist {
reconnect();
}
if (timeout > 100){
DEBUG_MSG(DLVL_FAIL, "Timeout while waiting for requested page %lld for track %lu. Aborting.", keyNum, trackId);
curPage.erase(trackId);
FAIL_MSG("Timeout while waiting for requested page %lld for track %lu. Aborting.", keyNum, trackId);
nProxy.curPage.erase(trackId);
currKeyOpen.erase(trackId);
return;
}
@ -335,7 +418,7 @@ namespace Mist {
}else{
nxtKeyNum[trackId] = 0;
}
stats();
stats(true);
Util::wait(100);
pageNum = pageNumForKey(trackId, keyNum);
}
@ -345,20 +428,20 @@ namespace Mist {
}else{
nxtKeyNum[trackId] = 0;
}
stats();
nxtKeyNum[trackId] = pageNum;
stats(true);
if (currKeyOpen.count(trackId) && currKeyOpen[trackId] == (unsigned int)pageNum){
return;
}
char id[NAME_BUFFER_SIZE];
snprintf(id, NAME_BUFFER_SIZE, SHM_TRACK_DATA, streamName.c_str(), trackId, pageNum);
curPage[trackId].init(id, DEFAULT_DATA_PAGE_SIZE);
if (!(curPage[trackId].mapped)){
DEBUG_MSG(DLVL_FAIL, "Initializing page %s failed", curPage[trackId].name.c_str());
nProxy.curPage[trackId].init(id, DEFAULT_DATA_PAGE_SIZE);
if (!(nProxy.curPage[trackId].mapped)){
FAIL_MSG("Initializing page %s failed", nProxy.curPage[trackId].name.c_str());
return;
}
currKeyOpen[trackId] = pageNum;
VERYHIGH_MSG("Page %s loaded for %s", id, streamName.c_str());
}
/// Prepares all tracks from selectedTracks for seeking to the specified ms position.
@ -373,44 +456,58 @@ namespace Mist {
if (myMeta.live){
updateMeta();
}
DEBUG_MSG(DLVL_MEDIUM, "Seeking to %llums", pos);
MEDIUM_MSG("Seeking to %llums", pos);
for (std::set<long unsigned int>::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++){
seek(*it, pos);
if (myMeta.tracks.count(*it)){
seek(*it, pos);
}
}
}
bool Output::seek(unsigned int tid, unsigned long long pos, bool getNextKey){
loadPageForKey(tid, getKeyForTime(tid, pos) + (getNextKey?1:0));
if (!curPage.count(tid) || !curPage[tid].mapped){
INFO_MSG("Aborting seek to %llums in track %u, not available.", pos, tid);
if (myMeta.tracks[tid].lastms < pos){
INFO_MSG("Aborting seek to %llums in track %u: past end of track (= %llums).", pos, tid, myMeta.tracks[tid].lastms);
return false;
}
unsigned int keyNum = getKeyForTime(tid, pos);
if (myMeta.tracks[tid].getKey(keyNum).getTime() > pos){
if (myMeta.live){
INFO_MSG("Actually seeking to %d, for %d is not available any more", myMeta.tracks[tid].getKey(keyNum).getTime(), pos);
pos = myMeta.tracks[tid].getKey(keyNum).getTime();
}
}
loadPageForKey(tid, keyNum + (getNextKey?1:0));
if (!nProxy.curPage.count(tid) || !nProxy.curPage[tid].mapped){
INFO_MSG("Aborting seek to %llums in track %u: not available.", pos, tid);
return false;
}
sortedPageInfo tmp;
tmp.tid = tid;
tmp.offset = 0;
DTSC::Packet tmpPack;
tmpPack.reInit(curPage[tid].mapped + tmp.offset, 0, true);
tmpPack.reInit(nProxy.curPage[tid].mapped + tmp.offset, 0, true);
tmp.time = tmpPack.getTime();
char * mpd = curPage[tid].mapped;
char * mpd = nProxy.curPage[tid].mapped;
while ((long long)tmp.time < pos && tmpPack){
tmp.offset += tmpPack.getDataLen();
tmpPack.reInit(mpd + tmp.offset, 0, true);
tmp.time = tmpPack.getTime();
}
if (tmpPack){
HIGH_MSG("Sought to time %d in %s@%u", tmp.time, streamName.c_str(), tid);
buffer.insert(tmp);
return true;
}else{
//don't print anything for empty packets - not sign of corruption, just unfinished stream.
if (curPage[tid].mapped[tmp.offset] != 0){
DEBUG_MSG(DLVL_FAIL, "Noes! Couldn't find packet on track %d because of some kind of corruption error or somesuch.", tid);
if (nProxy.curPage[tid].mapped[tmp.offset] != 0){
FAIL_MSG("Noes! Couldn't find packet on track %d because of some kind of corruption error or somesuch.", tid);
}else{
VERYHIGH_MSG("Track %d no data (key %u @ %u) - waiting...", tid, getKeyForTime(tid, pos) + (getNextKey?1:0), tmp.offset);
unsigned int i = 0;
while (curPage[tid].mapped[tmp.offset] == 0 && ++i < 42){
while (nProxy.curPage[tid].mapped[tmp.offset] == 0 && ++i < 42){
Util::wait(100);
}
if (curPage[tid].mapped[tmp.offset] == 0){
if (nProxy.curPage[tid].mapped[tmp.offset] == 0){
FAIL_MSG("Track %d no data (key %u) - timeout", tid, getKeyForTime(tid, pos) + (getNextKey?1:0));
}else{
return seek(tid, pos, getNextKey);
@ -434,9 +531,8 @@ namespace Mist {
}
int Output::run() {
DEBUG_MSG(DLVL_MEDIUM, "MistOut client handler started");
while (config->is_active && myConn.connected() && (wantRequest || parseData)){
stats();
DONTEVEN_MSG("MistOut client handler started");
while (config->is_active && myConn && (wantRequest || parseData)){
if (wantRequest){
requestHandler();
}
@ -445,11 +541,67 @@ namespace Mist {
initialize();
}
if ( !sentHeader){
DEBUG_MSG(DLVL_DONTEVEN, "sendHeader");
DONTEVEN_MSG("sendHeader");
sendHeader();
}
prepareNext();
if (!sought){
if (myMeta.live){
long unsigned int mainTrack = getMainSelectedTrack();
//cancel if there are no keys in the main track
if (!myMeta.tracks.count(mainTrack) || !myMeta.tracks[mainTrack].keys.size()){break;}
//seek to the newest keyframe, unless that is <5s, then seek to the oldest keyframe
unsigned long long seekPos = myMeta.tracks[mainTrack].keys.rbegin()->getTime();
if (seekPos < 5000){
seekPos = myMeta.tracks[mainTrack].keys.begin()->getTime();
}
seek(seekPos);
}else{
seek(0);
}
}
if (prepareNext()){
if (thisPacket){
//slow down processing, if real time speed is wanted
if (realTime){
while (thisPacket.getTime() > (((Util::getMS() - firstTime)*1000)+maxSkipAhead)/realTime) {
Util::sleep(std::min(thisPacket.getTime() - (((Util::getMS() - firstTime)*1000)+minSkipAhead)/realTime, 1000llu));
stats();
}
}
//delay the stream until its current keyframe is complete, if only complete keys wanted
if (completeKeysOnly){
bool completeKeyReady = false;
int timeoutTries = 40;//wait default 250ms*40=10 seconds
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (it->second.keys.size() >1){
int thisTimeoutTries = ((it->second.lastms - it->second.firstms) / (it->second.keys.size()-1)) / 125;
if (thisTimeoutTries > timeoutTries) timeoutTries = thisTimeoutTries;
}
}
while(!completeKeyReady && timeoutTries>0){
completeKeyReady = true;
for (std::set<unsigned long>::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++){
if (!myMeta.tracks[*it].keys.size() || myMeta.tracks[*it].keys.rbegin()->getTime() + myMeta.tracks[*it].keys.rbegin()->getLength() <= thisPacket.getTime() ){
completeKeyReady = false;
break;
}
}
if (!completeKeyReady){
timeoutTries--;//we count down
stats();
Util::wait(250);
updateMeta();
}
}
if (timeoutTries<=0){
WARN_MSG("Waiting for key frame timed out");
completeKeysOnly = false;
}
}
sendNext();
}else{
if (!onFinish()){
@ -458,9 +610,12 @@ namespace Mist {
}
}
}
DEBUG_MSG(DLVL_MEDIUM, "MistOut client handler shutting down: %s, %s, %s", myConn.connected() ? "conn_active" : "conn_closed", wantRequest ? "want_request" : "no_want_request", parseData ? "parsing_data" : "not_parsing_data");
stats();
userClient.finish();
}
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");
stats(true);
nProxy.userClient.finish();
statsPage.finish();
myConn.close();
return 0;
@ -473,250 +628,258 @@ namespace Mist {
return 0;
}
for (std::set<long unsigned int>::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++){
if (myMeta.tracks[*it].type == "video"){
if (myMeta.tracks.count(*it) && myMeta.tracks[*it].type == "video"){
return *it;
}
}
return *(selectedTracks.begin());
}
void Output::prepareNext(){
static int nonVideoCount = 0;
if (!sought){
if (myMeta.live){
long unsigned int mainTrack = getMainSelectedTrack();
if (myMeta.tracks[mainTrack].keys.size() < 2){
if (!myMeta.tracks[mainTrack].keys.size()){
myConn.close();
return;
}else{
seek(myMeta.tracks[mainTrack].keys.begin()->getTime());
prepareNext();
return;
}
}
unsigned long long seekPos = myMeta.tracks[mainTrack].keys.rbegin()->getTime();
if (seekPos < 5000){
seekPos = 0;
}
seek(seekPos);
}else{
seek(0);
void Output::dropTrack(uint32_t trackId, std::string reason, bool probablyBad){
//depending on whether this is probably bad and the current debug level, print a message
unsigned int printLevel = DLVL_INFO;
if (probablyBad){
printLevel = DLVL_WARN;
}
DEBUG_MSG(printLevel, "Dropping %s (%s) track %lu@k%lu (nextP=%d, lastP=%d): %s", streamName.c_str(), myMeta.tracks[trackId].codec.c_str(), (long unsigned)trackId, nxtKeyNum[trackId]+1, pageNumForKey(trackId, nxtKeyNum[trackId]+1), pageNumMax(trackId), reason.c_str());
//now actually drop the track from the buffer
for (std::set<sortedPageInfo>::iterator it = buffer.begin(); it != buffer.end(); ++it){
if (it->tid == trackId){
buffer.erase(it);
break;
}
}
selectedTracks.erase(trackId);
}
///Attempts to prepare a new packet for output.
///If thisPacket evaluates to false, playback has completed.
///Could be called repeatedly in a loop if you really really want a new packet.
/// \returns true if thisPacket was filled with the next packet.
/// \returns false if we could not reliably determine the next packet yet.
bool Output::prepareNext(){
static bool atLivePoint = false;
static int nonVideoCount = 0;
static unsigned int emptyCount = 0;
if (!buffer.size()){
thisPacket.null();
DEBUG_MSG(DLVL_DEVEL, "Buffer completely played out");
onFinish();
return;
INFO_MSG("Buffer completely played out");
return true;
}
sortedPageInfo nxt = *(buffer.begin());
buffer.erase(buffer.begin());
DEBUG_MSG(DLVL_DONTEVEN, "Loading track %u (next=%lu), %llu ms", nxt.tid, nxtKeyNum[nxt.tid], nxt.time);
if (nxt.offset >= curPage[nxt.tid].len){
if (!myMeta.tracks.count(nxt.tid)){
dropTrack(nxt.tid, "disappeared from metadata", true);
return false;
}
DONTEVEN_MSG("Loading track %u (next=%lu), %llu ms", nxt.tid, nxtKeyNum[nxt.tid], nxt.time);
//if we're going to read past the end of the data page, load the next page
//this only happens for VoD
if (nxt.offset >= nProxy.curPage[nxt.tid].len){
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
loadPageForKey(nxt.tid, ++nxtKeyNum[nxt.tid]);
nxt.offset = 0;
if (curPage.count(nxt.tid) && curPage[nxt.tid].mapped){
if (getDTSCTime(curPage[nxt.tid].mapped, nxt.offset) < nxt.time){
ERROR_MSG("Time going backwards in track %u - dropping track.", nxt.tid);
if (nProxy.curPage.count(nxt.tid) && nProxy.curPage[nxt.tid].mapped){
if (getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset) < nxt.time){
dropTrack(nxt.tid, "time going backwards");
}else{
nxt.time = getDTSCTime(curPage[nxt.tid].mapped, nxt.offset);
nxt.time = getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset);
//swap out the next object in the buffer with a new one
buffer.erase(buffer.begin());
buffer.insert(nxt);
}
prepareNext();
return;
}else{
dropTrack(nxt.tid, "page load failure", true);
}
}
if (!curPage.count(nxt.tid) || !curPage[nxt.tid].mapped){
//mapping failure? Drop this track and go to next.
//not an error - usually means end of stream.
DEBUG_MSG(DLVL_DEVEL, "Track %u no page - dropping track.", nxt.tid);
prepareNext();
return;
return false;
}
//have we arrived at the end of the memory page? (4 zeroes mark the end)
if (!memcmp(curPage[nxt.tid].mapped + nxt.offset, "\000\000\000\000", 4)){
if (!memcmp(nProxy.curPage[nxt.tid].mapped + nxt.offset, "\000\000\000\000", 4)){
//if we don't currently know where we are, we're lost. We should drop the track.
if (!nxt.time){
DEBUG_MSG(DLVL_DEVEL, "Timeless empty packet on track %u - dropping track.", nxt.tid);
prepareNext();
return;
dropTrack(nxt.tid, "timeless empty packet");
return false;
}
//if this is a live stream, we might have just reached the live point.
//check where the next key is
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
int nextPage = pageNumForKey(nxt.tid, nxtKeyNum[nxt.tid]+1);
//are we live, and the next key hasn't shown up on another page? then we're waiting.
//are we live, and the next key hasn't shown up on another page, then we're waiting.
if (myMeta.live && currKeyOpen.count(nxt.tid) && (currKeyOpen[nxt.tid] == (unsigned int)nextPage || nextPage == -1)){
if (myMeta && ++emptyCount < 42){
//we're waiting for new data. Simply retry.
buffer.insert(nxt);
}else{
//after ~10 seconds, give up and drop the track.
DEBUG_MSG(DLVL_DEVEL, "Empty packet on track %u @ key %lu (next=%d) - could not reload, dropping track.", nxt.tid, nxtKeyNum[nxt.tid]+1, nextPage);
}
//keep updating the metadata at 250ms intervals while waiting for more data
Util::sleep(250);
updateMeta();
}else{
//if we're not live, we've simply reached the end of the page. Load the next key.
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
loadPageForKey(nxt.tid, ++nxtKeyNum[nxt.tid]);
nxt.offset = 0;
if (curPage.count(nxt.tid) && curPage[nxt.tid].mapped){
unsigned long long nextTime = getDTSCTime(curPage[nxt.tid].mapped, nxt.offset);
if (nextTime && nextTime < nxt.time){
DEBUG_MSG(DLVL_DEVEL, "Time going backwards in track %u - dropping track.", nxt.tid);
if (++emptyCount < 100){
Util::wait(250);
//we're waiting for new data to show up
if (emptyCount % 8 == 0){
reconnect();//reconnect every 2 seconds
}else{
if (nextTime){
nxt.time = nextTime;
if (emptyCount % 4 == 0){
updateMeta();
}
buffer.insert(nxt);
DEBUG_MSG(DLVL_MEDIUM, "Next page for track %u starts at %llu.", nxt.tid, nxt.time);
}
}else{
DEBUG_MSG(DLVL_DEVEL, "Could not load next memory page for track %u - dropping track.", nxt.tid);
//after ~25 seconds, give up and drop the track.
dropTrack(nxt.tid, "could not reload empty packet");
}
return false;
}
prepareNext();
return;
}
thisPacket.reInit(curPage[nxt.tid].mapped + nxt.offset, 0, true);
if (thisPacket){
if (thisPacket.getTime() != nxt.time && nxt.time){
WARN_MSG("Loaded track %ld@%llu instead of %ld@%llu", thisPacket.getTrackId(), thisPacket.getTime(), nxt.tid, nxt.time);
}
if ((myMeta.tracks[nxt.tid].type == "video" && thisPacket.getFlag("keyframe")) || (++nonVideoCount % 30 == 0)){
if (myMeta.live){
updateMeta();
}
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
DEBUG_MSG(DLVL_VERYHIGH, "Track %u @ %llums = key %lu", nxt.tid, thisPacket.getTime(), nxtKeyNum[nxt.tid]);
}
emptyCount = 0;
}
nxt.offset += thisPacket.getDataLen();
if (realTime){
while (nxt.time > (((Util::getMS() - firstTime)*1000)+maxSkipAhead)/realTime) {
Util::sleep(nxt.time - (((Util::getMS() - firstTime)*1000)+minSkipAhead)/realTime);
}
}
//delay the stream until its current keyframe is complete
if (completeKeysOnly){
bool completeKeyReady = false;
int timeoutTries = 40;//attempts to updateMeta before timeOut and moving on; default is approximately 10 seconds
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (it->second.keys.size() >1){
int thisTimeoutTries = ((it->second.lastms - it->second.firstms) / (it->second.keys.size()-1)) / 125;
if (thisTimeoutTries > timeoutTries) timeoutTries = thisTimeoutTries;
}
}
while(!completeKeyReady && timeoutTries>0){
completeKeyReady = true;
for (std::set<unsigned long>::iterator it = selectedTracks.begin(); it != selectedTracks.end(); it++){
if (!myMeta.tracks[*it].keys.size() || myMeta.tracks[*it].keys.rbegin()->getTime() + myMeta.tracks[*it].keys.rbegin()->getLength() <= nxt.time ){
completeKeyReady = false;
break;
//We've simply reached the end of the page. Load the next key = next page.
loadPageForKey(nxt.tid, ++nxtKeyNum[nxt.tid]);
nxt.offset = 0;
if (nProxy.curPage.count(nxt.tid) && nProxy.curPage[nxt.tid].mapped){
unsigned long long nextTime = getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset);
if (nextTime && nextTime < nxt.time){
dropTrack(nxt.tid, "time going backwards");
}else{
if (nextTime){
nxt.time = nextTime;
}
//swap out the next object in the buffer with a new one
buffer.erase(buffer.begin());
buffer.insert(nxt);
MEDIUM_MSG("Next page for track %u starts at %llu.", nxt.tid, nxt.time);
}
if (!completeKeyReady){
if (completeKeyReadyTimeOut){
INSANE_MSG("Complete Key not ready and time-out is being skipped");
timeoutTries = 0;
}else{
INSANE_MSG("Timeout: %d",timeoutTries);
timeoutTries--;//we count down
stats();
Util::wait(250);
updateMeta();
}
}
}
if (timeoutTries<=0){
if (!completeKeyReadyTimeOut){
INFO_MSG("Wait for keyframe Timeout triggered! Ended to avoid endless loops");
}
completeKeyReadyTimeOut = true;
}else{
//untimeout handling
completeKeyReadyTimeOut = false;
dropTrack(nxt.tid, "page load failure");
}
return false;
}
if (curPage[nxt.tid]){
if (nxt.offset < curPage[nxt.tid].len){
unsigned long long nextTime = getDTSCTime(curPage[nxt.tid].mapped, nxt.offset);
if (nextTime){
nxt.time = nextTime;
}else{
++nxt.time;
//we've handled all special cases - at this point the packet should exist
//let's load it
thisPacket.reInit(nProxy.curPage[nxt.tid].mapped + nxt.offset, 0, true);
//if it failed, drop the track and continue
if (!thisPacket){
dropTrack(nxt.tid, "packet load failure");
return false;
}
emptyCount = 0;//valid packet - reset empty counter
//if there's a timestamp mismatch, print this.
//except for live, where we never know the time in advance
if (thisPacket.getTime() != nxt.time && nxt.time && !atLivePoint){
static int warned = 0;
if (warned < 5){
WARN_MSG("Loaded %s track %ld@%llu in stead of %u@%llu (%dms, %s)", streamName.c_str(), thisPacket.getTrackId(), thisPacket.getTime(), nxt.tid, nxt.time, (int)((long long)thisPacket.getTime() - (long long)nxt.time), myMeta.tracks[nxt.tid].codec.c_str());
if (++warned == 5){
WARN_MSG("Further warnings about time mismatches printed on HIGH level.");
}
}else{
HIGH_MSG("Loaded %s track %ld@%llu in stead of %u@%llu (%dms, %s)", streamName.c_str(), thisPacket.getTrackId(), thisPacket.getTime(), nxt.tid, nxt.time, (int)((long long)thisPacket.getTime() - (long long)nxt.time), myMeta.tracks[nxt.tid].codec.c_str());
}
buffer.insert(nxt);
}
stats();
//when live, every keyframe, check correctness of the keyframe number
if (myMeta.live && thisPacket.getFlag("keyframe")){
//Check whether returned keyframe is correct. If not, wait for approximately 10 seconds while checking.
//Failure here will cause tracks to drop due to inconsistent internal state.
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
int counter = 0;
while(counter < 40 && myMeta.tracks[nxt.tid].getKey(nxtKeyNum[nxt.tid]).getTime() != thisPacket.getTime()){
if (counter++){
//Only sleep 250ms if this is not the first updatemeta try
Util::wait(250);
}
updateMeta();
nxtKeyNum[nxt.tid] = getKeyForTime(nxt.tid, thisPacket.getTime());
}
if (myMeta.tracks[nxt.tid].getKey(nxtKeyNum[nxt.tid]).getTime() != thisPacket.getTime()){
WARN_MSG("Keyframe value is not correct - state will now be inconsistent.");
}
EXTREME_MSG("Track %u @ %llums = key %lu", nxt.tid, thisPacket.getTime(), nxtKeyNum[nxt.tid]);
}
//always assume we're not at the live point
atLivePoint = false;
//we assume the next packet is the next on this same page
nxt.offset += thisPacket.getDataLen();
if (nxt.offset < nProxy.curPage[nxt.tid].len){
unsigned long long nextTime = getDTSCTime(nProxy.curPage[nxt.tid].mapped, nxt.offset);
if (nextTime){
nxt.time = nextTime;
}else{
++nxt.time;
//no packet -> we are at the live point
atLivePoint = true;
}
}
//exchange the current packet in the buffer for the next one
buffer.erase(buffer.begin());
buffer.insert(nxt);
return true;
}
void Output::stats(){
static bool setHost = true;
if (!isInitialized){
return;
/// Returns the name as it should be used in statistics.
/// 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 (config->hasOption("target") && config->getString("target").size()){
return "OUTPUT";
}else{
return capa["name"].asStringRef();
}
}
void Output::stats(bool force){
//cancel stats update if not initialized
if (!isInitialized){return;}
//also cancel if it has been less than a second since the last update
//unless force is set to true
unsigned long long int now = Util::epoch();
if (now == lastStats && !force){return;}
lastStats = now;
EXTREME_MSG("Writing stats: %s, %s, %lu", getConnectedHost().c_str(), streamName.c_str(), crc & 0xFFFFFFFFu);
if (statsPage.getData()){
unsigned long long int now = Util::epoch();
if (now != lastStats){
lastStats = now;
IPC::statExchange tmpEx(statsPage.getData());
tmpEx.now(now);
if (setHost){
tmpEx.host(getConnectedBinHost());
setHost = false;
}
tmpEx.crc(crc);
tmpEx.streamName(streamName);
tmpEx.connector(capa["name"].asString());
tmpEx.up(myConn.dataUp());
tmpEx.down(myConn.dataDown());
tmpEx.time(now - myConn.connTime());
if (thisPacket){
tmpEx.lastSecond(thisPacket.getTime());
}else{
tmpEx.lastSecond(0);
}
statsPage.keepAlive();
IPC::statExchange tmpEx(statsPage.getData());
tmpEx.now(now);
if (tmpEx.host() == std::string("\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000", 16)){
tmpEx.host(getConnectedBinHost());
}
tmpEx.crc(crc);
tmpEx.streamName(streamName);
tmpEx.connector(getStatsName());
tmpEx.up(myConn.dataUp());
tmpEx.down(myConn.dataDown());
tmpEx.time(now - myConn.connTime());
if (thisPacket){
tmpEx.lastSecond(thisPacket.getTime());
}else{
tmpEx.lastSecond(0);
}
statsPage.keepAlive();
}
int tNum = 0;
if (!userClient.getData()){
if (!nProxy.userClient.getData()){
char userPageName[NAME_BUFFER_SIZE];
snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, streamName.c_str());
userClient = IPC::sharedClient(userPageName, PLAY_EX_SIZE, true);
if (!userClient.getData()){
DEBUG_MSG(DLVL_WARN, "Player connection failure - aborting output");
nProxy.userClient = IPC::sharedClient(userPageName, PLAY_EX_SIZE, true);
if (!nProxy.userClient.getData()){
WARN_MSG("Player connection failure - aborting output");
myConn.close();
return;
}
}
if (!trackMap.size()){
if (!nProxy.userClient.isAlive()){
myConn.close();
return;
}
if (!nProxy.trackMap.size()){
IPC::userConnection userConn(nProxy.userClient.getData());
for (std::set<unsigned long>::iterator it = selectedTracks.begin(); it != selectedTracks.end() && tNum < SIMUL_TRACKS; it++){
unsigned int tId = *it;
char * thisData = userClient.getData() + (6 * tNum);
thisData[0] = ((tId >> 24) & 0xFF);
thisData[1] = ((tId >> 16) & 0xFF);
thisData[2] = ((tId >> 8) & 0xFF);
thisData[3] = ((tId) & 0xFF);
thisData[4] = ((nxtKeyNum[tId] >> 8) & 0xFF);
thisData[5] = ((nxtKeyNum[tId]) & 0xFF);
userConn.setTrackId(tNum, *it);
userConn.setKeynum(tNum, nxtKeyNum[*it]);
tNum ++;
}
}
userClient.keepAlive();
nProxy.userClient.keepAlive();
if (tNum > SIMUL_TRACKS){
DEBUG_MSG(DLVL_WARN, "Too many tracks selected, using only first %d", SIMUL_TRACKS);
WARN_MSG("Too many tracks selected, using only first %d", SIMUL_TRACKS);
}
}
@ -730,4 +893,24 @@ namespace Mist {
//just set the sentHeader bool to true, by default
sentHeader = true;
}
bool Output::connectToFile(std::string file) {
int flags = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH;
int mode = O_RDWR | O_CREAT | O_TRUNC;
int outFile = open(file.c_str(), mode, flags);
if (outFile < 0) {
ERROR_MSG("Failed to open file %s, error: %s", file.c_str(), strerror(errno));
return false;
}
int r = dup2(outFile, myConn.getSocket());
if (r == -1) {
ERROR_MSG("Failed to create an alias for the socket using dup2: %s.", strerror(errno));
return false;
}
close(outFile);
return true;
}
}

17
src/output/output.h
View file

@ -37,13 +37,12 @@ namespace Mist {
public:
//constructor and destructor
Output(Socket::Connection & conn);
virtual ~Output();
//static members for initialization and capabilities
static void init(Util::Config * cfg);
static JSON::Value capa;
//non-virtual generic functions
int run();
void stats();
void stats(bool force = false);
void seek(unsigned long long pos);
bool seek(unsigned int tid, unsigned long long pos, bool getNextKey = false);
void stop();
@ -51,10 +50,13 @@ namespace Mist {
long unsigned int getMainSelectedTrack();
void updateMeta();
void selectDefaultTracks();
bool connectToFile(std::string file);
static bool listenMode(){return true;}
virtual bool isReadyForPlay();
//virtuals. The optional virtuals have default implementations that do as little as possible.
virtual void sendNext() {}//REQUIRED! Others are optional.
virtual void prepareNext();
bool prepareNext();
virtual void dropTrack(uint32_t trackId, std::string reason, bool probablyBad = true);
virtual void onRequest();
virtual bool onFinish() {
return false;
@ -68,21 +70,21 @@ namespace Mist {
std::map<unsigned long, unsigned int> currKeyOpen;
void loadPageForKey(long unsigned int trackId, long long int keyNum);
int pageNumForKey(long unsigned int trackId, long long int keyNum);
int pageNumMax(long unsigned int trackId);
unsigned int lastStats;///<Time of last sending of stats.
long long unsigned int firstTime;///< Time of first packet after last seek. Used for real-time sending.
std::map<unsigned long, unsigned long> nxtKeyNum;///< Contains the number of the next key, for page seeking purposes.
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().
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
virtual std::string getConnectedHost();
virtual std::string getConnectedBinHost();
virtual std::string getStatsName();
virtual bool hasSessionIDs(){return false;}
IPC::sharedClient statsPage;///< Shared memory used for statistics reporting.
bool isBlocking;///< If true, indicates that myConn is blocking.
unsigned int crc;///< Checksum, if any, for usage in the stats.
uint32_t crc;///< Checksum, if any, for usage in the stats.
unsigned int getKeyForTime(long unsigned int trackId, long long timeStamp);
//stream delaying variables
@ -103,3 +105,4 @@ namespace Mist {
};
}

4
src/output/output_hds.cpp
View file

@ -159,7 +159,7 @@ namespace Mist {
capa["codecs"][0u][1u].append("G711mu");
capa["methods"][0u]["handler"] = "http";
capa["methods"][0u]["type"] = "flash/11";
capa["methods"][0u]["priority"] = 7ll;
capa["methods"][0u]["priority"] = 6ll;
capa["methods"][0u]["player_url"] = "/flashplayer.swf";
}
@ -229,7 +229,7 @@ namespace Mist {
myConn.close();
break;
}
Util::sleep(500);
Util::wait(500);
updateMeta();
}
mstime = myMeta.tracks[tid].getKey(myMeta.tracks[tid].fragments[fragNum - myMeta.tracks[tid].missedFrags].getNumber()).getTime();

54
src/output/output_hls.cpp
View file

@ -3,17 +3,26 @@
#include <unistd.h>
namespace Mist {
bool OutHLS::isReadyForPlay() {
if (myMeta.tracks.size()){
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (it->second.fragments.size() >= 3){
return true;
}
}
}
return false;
}
///\brief Builds an index file for HTTP Live streaming.
///\return The index file for HTTP Live Streaming.
std::string OutHLS::liveIndex(){
std::stringstream result;
result << "#EXTM3U\r\n";
int audioId = -1;
std::string audioName;
for (std::map<unsigned int,DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (it->second.codec == "AAC"){
audioId = it->first;
audioName = it->second.getIdentifier();
break;
}
}
@ -33,7 +42,7 @@ namespace Mist {
if (audioId != -1){
result << "_" << audioId;
}
result << "/index.m3u8\r\n";
result << "/index.m3u8?sessId=" << getpid() << "\r\n";
}
}
if (!vidTracks && audioId){
@ -44,13 +53,13 @@ namespace Mist {
return result.str();
}
std::string OutHLS::liveIndex(int tid){
std::string OutHLS::liveIndex(int tid, std::string & sessId) {
updateMeta();
std::stringstream result;
//parse single track
int longestFragment = 0;
if (!myMeta.tracks[tid].fragments.size()){
DEBUG_MSG(DLVL_FAIL, "liveIndex called with track %d, which has no fragments!", tid);
INFO_MSG("liveIndex called with track %d, which has no fragments!", tid);
return "";
}
for (std::deque<DTSC::Fragment>::iterator it = myMeta.tracks[tid].fragments.begin(); (it + 1) != myMeta.tracks[tid].fragments.end(); it++){
@ -66,15 +75,18 @@ namespace Mist {
std::deque<std::string> lines;
for (std::deque<DTSC::Fragment>::iterator it = myMeta.tracks[tid].fragments.begin(); it != myMeta.tracks[tid].fragments.end(); it++){
long long int starttime = myMeta.tracks[tid].getKey(it->getNumber()).getTime();
std::stringstream line;
long long duration = it->getDuration();
if (duration <= 0){
duration = myMeta.tracks[tid].lastms - starttime;
}
line << "#EXTINF:" << ((duration + 500) / 1000) << ", no desc\r\n" << starttime << "_" << duration + starttime << ".ts\r\n";
lines.push_back(line.str());
char lineBuf[400];
if (sessId.size()){
snprintf(lineBuf, 400, "#EXTINF:%lld, no desc\r\n%lld_%lld.ts?sessId=%s\r\n", ((duration + 500) / 1000), starttime, starttime + duration, sessId.c_str());
}else{
snprintf(lineBuf, 400, "#EXTINF:%lld, no desc\r\n%lld_%lld.ts\r\n", ((duration + 500) / 1000), starttime, starttime + duration);
}
lines.push_back(lineBuf);
}
unsigned int skippedLines = 0;
if (myMeta.live){
//only print the last segment when VoD
@ -135,7 +147,7 @@ namespace Mist {
void OutHLS::onHTTP(){
std::string method = H.method;
std::string sessId = H.GetVar("sessId");
if (H.url == "/crossdomain.xml"){
H.Clean();
@ -152,12 +164,24 @@ namespace Mist {
H.Clean(); //clean for any possible next requests
return;
} //crossdomain.xml
if (H.method == "OPTIONS") {
H.Clean();
H.SetHeader("Content-Type", "application/octet-stream");
H.SetHeader("Cache-Control", "no-cache");
H.setCORSHeaders();
H.SetBody("");
H.SendResponse("200", "OK", myConn);
H.Clean();
return;
}
if (H.url.find("hls") == std::string::npos){
myConn.close();
return;
}
appleCompat = (H.GetHeader("User-Agent").find("iPad") != std::string::npos) || (H.GetHeader("User-Agent").find("iPod") != std::string::npos)|| (H.GetHeader("User-Agent").find("iPhone") != std::string::npos);
bool VLCworkaround = false;
if (H.GetHeader("User-Agent").substr(0, 3) == "VLC"){
@ -167,6 +191,7 @@ namespace Mist {
VLCworkaround = true;
}
}
initialize();
if (H.url.find(".m3u") == std::string::npos){
std::string tmpStr = H.getUrl().substr(5+streamName.size());
@ -189,6 +214,11 @@ namespace Mist {
selectedTracks.insert(vidTrack);
selectedTracks.insert(audTrack);
}
for (std::map<unsigned int,DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (it->second.codec == "ID3"){
selectedTracks.insert(it->first);
}
}
if (myMeta.live){
unsigned int timeout = 0;
@ -202,7 +232,7 @@ namespace Mist {
myConn.close();
break;
}
Util::sleep(500);
Util::wait(500);
updateMeta();
}
}while (myConn && seekable > 0);
@ -266,7 +296,7 @@ namespace Mist {
manifest = liveIndex();
}else{
int selectId = atoi(request.substr(0,request.find("/")).c_str());
manifest = liveIndex(selectId);
manifest = liveIndex(selectId, sessId);
}
H.SetBody(manifest);
H.SendResponse("200", "OK", myConn);

4
src/output/output_hls.h
View file

@ -9,9 +9,11 @@ namespace Mist {
static void init(Util::Config * cfg);
void sendTS(const char * tsData, unsigned int len=188);
void onHTTP();
bool isReadyForPlay();
protected:
bool hasSessionIDs(){return true;}
std::string liveIndex();
std::string liveIndex(int tid);
std::string liveIndex(int tid, std::string & sessId);
int canSeekms(unsigned int ms);
int keysToSend;
unsigned int vidTrack;

23
src/output/output_hss.cpp
View file

@ -5,6 +5,7 @@
#include <mist/mp4_generic.h>
#include <mist/http_parser.h>
#include <mist/stream.h>
#include <mist/bitfields.h>
#include <mist/checksum.h>
#include <unistd.h>
@ -56,11 +57,9 @@ namespace Mist {
capa["methods"][0u]["handler"] = "http";
capa["methods"][0u]["type"] = "html5/application/vnd.ms-ss";
capa["methods"][0u]["priority"] = 9ll;
capa["methods"][0u]["nolive"] = 1;
capa["methods"][1u]["handler"] = "http";
capa["methods"][1u]["type"] = "silverlight";
capa["methods"][1u]["priority"] = 1ll;
capa["methods"][1u]["nolive"] = 1;
}
void OutHSS::sendNext() {
@ -132,7 +131,7 @@ namespace Mist {
myConn.close();
break;
}
Util::sleep(500);
Util::wait(500);
updateMeta();
}
}while (myConn && seekable > 0);
@ -201,11 +200,11 @@ namespace Mist {
//Wrap everything in mp4 boxes
MP4::MFHD mfhd_box;
mfhd_box.setSequenceNumber(((keyObj.getNumber() - 1) * 2) + tid);///\todo Urgent: Check this for multitrack... :P wtf... :P
mfhd_box.setSequenceNumber(((keyObj.getNumber() - 1) * 2) + (myMeta.tracks[tid].type == "video" ? 1 : 2));
MP4::TFHD tfhd_box;
tfhd_box.setFlags(MP4::tfhdSampleFlag);
tfhd_box.setTrackID(tid);
tfhd_box.setTrackID((myMeta.tracks[tid].type == "video" ? 1 : 2));
if (myMeta.tracks[tid].type == "video") {
tfhd_box.setDefaultSampleFlags(0x00004001);
} else {
@ -254,19 +253,24 @@ namespace Mist {
//If the stream is live, we want to have a fragref box if possible
//////HEREHEREHERE
if (myMeta.live) {
MP4::UUID_TFXD tfxd_box;
tfxd_box.setTime(keyObj.getTime());
tfxd_box.setDuration(keyObj.getLength());
traf_box.setContent(tfxd_box, 3);
MP4::UUID_TrackFragmentReference fragref_box;
fragref_box.setVersion(1);
fragref_box.setFragmentCount(0);
int fragCount = 0;
for (unsigned int i = 0; fragCount < 2 && i < myMeta.tracks[tid].keys.size() - 1; i++) {
if (myMeta.tracks[tid].keys[i].getTime() > seekTime) {
DEBUG_MSG(DLVL_HIGH, "Key %d added to fragRef box, time %ld > %lld", i, myMeta.tracks[tid].keys[i].getTime(), seekTime);
DEBUG_MSG(DLVL_HIGH, "Key %d added to fragRef box, time %llu > %lld", i, myMeta.tracks[tid].keys[i].getTime(), seekTime);
fragref_box.setTime(fragCount, myMeta.tracks[tid].keys[i].getTime() * 10000);
fragref_box.setDuration(fragCount, myMeta.tracks[tid].keys[i].getLength() * 10000);
fragref_box.setFragmentCount(++fragCount);
}
}
traf_box.setContent(fragref_box, 3);
traf_box.setContent(fragref_box, 4);
}
MP4::MOOF moof_box;
@ -467,9 +471,4 @@ namespace Mist {
sendHeader();
}
}
void OutHSS::initialize() {
Output::initialize();
}
}

2
src/output/output_hss.h
View file

@ -9,10 +9,8 @@ namespace Mist {
static void init(Util::Config * cfg);
void onHTTP();
void sendNext();
void initialize();/*LTS*/
void sendHeader();
protected:
JSON::Value encryption;
std::string smoothIndex();
int canSeekms(unsigned int ms);
int keysToSend;

27
src/output/output_http.cpp
View file

@ -2,6 +2,7 @@
#include "output_http.h"
#include <mist/stream.h>
#include <mist/checksum.h>
#include <set>
namespace Mist {
HTTPOutput::HTTPOutput(Socket::Connection & conn) : Output(conn) {
@ -104,9 +105,9 @@ namespace Mist {
}
//loop over the connectors
IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
IPC::semaphore configLock(SEM_CONF, O_CREAT | O_RDWR, ACCESSPERMS, 1);
configLock.wait();
IPC::sharedPage serverCfg("!mistConfig", DEFAULT_CONF_PAGE_SIZE);
IPC::sharedPage serverCfg(SHM_CONF, DEFAULT_CONF_PAGE_SIZE);
DTSC::Scan capa = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("capabilities").getMember("connectors");
unsigned int capa_ctr = capa.getSize();
for (unsigned int i = 0; i < capa_ctr; ++i){
@ -171,7 +172,7 @@ namespace Mist {
if (handler != capa["name"].asStringRef() || H.GetVar("stream") != streamName){
DEBUG_MSG(DLVL_MEDIUM, "Switching from %s (%s) to %s (%s)", capa["name"].asStringRef().c_str(), streamName.c_str(), handler.c_str(), H.GetVar("stream").c_str());
streamName = H.GetVar("stream");
userClient.finish();
nProxy.userClient.finish();
statsPage.finish();
reConnector(handler);
H.Clean();
@ -209,8 +210,19 @@ namespace Mist {
void HTTPOutput::onRequest(){
while (H.Read(myConn)){
std::string ua = H.GetHeader("User-Agent");
crc = checksum::crc32(0, ua.data(), ua.size());
if (hasSessionIDs()){
if (H.GetVar("sessId").size()){
std::string ua = H.GetVar("sessId");
crc = checksum::crc32(0, ua.data(), ua.size());
}else{
std::string ua = JSON::Value((long long)getpid()).asString();
crc = checksum::crc32(0, ua.data(), ua.size());
}
}else{
std::string ua = H.GetHeader("User-Agent") + H.GetHeader("X-Playback-Session-Id");
crc = checksum::crc32(0, ua.data(), ua.size());
}
INFO_MSG("Received request %s", H.getUrl().c_str());
selectedTracks.clear();
if (H.GetVar("audio") != ""){
@ -239,6 +251,7 @@ namespace Mist {
for (std::set<unsigned long>::iterator it = toRemove.begin(); it != toRemove.end(); it++){
selectedTracks.erase(*it);
}
onHTTP();
if (!H.bufferChunks){
H.Clean();
@ -275,9 +288,9 @@ namespace Mist {
for (int i=0; i<20; i++){argarr[i] = 0;}
int id = -1;
IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
IPC::semaphore configLock(SEM_CONF, O_CREAT | O_RDWR, ACCESSPERMS, 1);
configLock.wait();
IPC::sharedPage serverCfg("!mistConfig", DEFAULT_CONF_PAGE_SIZE);
IPC::sharedPage serverCfg(SHM_CONF, DEFAULT_CONF_PAGE_SIZE);
DTSC::Scan prots = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("config").getMember("protocols");
unsigned int prots_ctr = prots.getSize();

14
src/output/output_http_internal.cpp
View file

@ -248,9 +248,9 @@ namespace Mist {
std::string port, url_rel;
IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
IPC::semaphore configLock(SEM_CONF, O_CREAT | O_RDWR, ACCESSPERMS, 1);
configLock.wait();
IPC::sharedPage serverCfg("!mistConfig", DEFAULT_CONF_PAGE_SIZE);
IPC::sharedPage serverCfg(SHM_CONF, DEFAULT_CONF_PAGE_SIZE);
DTSC::Scan prtcls = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("config").getMember("protocols");
DTSC::Scan capa = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("capabilities").getMember("connectors").getMember("RTMP");
unsigned int pro_cnt = prtcls.getSize();
@ -320,11 +320,13 @@ namespace Mist {
}
response = "// Generating info code for stream " + streamName + "\n\nif (!mistvideo){var mistvideo = {};}\n";
JSON::Value json_resp;
IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
IPC::semaphore metaLocker(std::string("liveMeta@" + streamName).c_str(), O_CREAT | O_RDWR, ACCESSPERMS, 1);
IPC::semaphore configLock(SEM_CONF, O_CREAT | O_RDWR, ACCESSPERMS, 1);
static char liveSemName[NAME_BUFFER_SIZE];
snprintf(liveSemName, NAME_BUFFER_SIZE, SEM_LIVE, streamName.c_str());
IPC::semaphore metaLocker(liveSemName, O_CREAT | O_RDWR, ACCESSPERMS, 1);
bool metaLock = false;
configLock.wait();
IPC::sharedPage serverCfg("!mistConfig", DEFAULT_CONF_PAGE_SIZE);
IPC::sharedPage serverCfg(SHM_CONF, DEFAULT_CONF_PAGE_SIZE);
DTSC::Scan strm = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("streams").getMember(streamName).getMember("meta");
IPC::sharedPage streamIndex;
if (!strm){
@ -333,7 +335,7 @@ namespace Mist {
if (Util::startInput(streamName)){
char pageId[NAME_BUFFER_SIZE];
snprintf(pageId, NAME_BUFFER_SIZE, SHM_STREAM_INDEX, streamName.c_str());
streamIndex.init(pageId, DEFAULT_META_PAGE_SIZE);
streamIndex.init(pageId, DEFAULT_STRM_PAGE_SIZE);
if (streamIndex.mapped){
metaLock = true;
metaLocker.wait();

10
src/output/output_httpts.cpp
View file

@ -25,13 +25,19 @@ namespace Mist {
}
void OutHTTPTS::onHTTP(){
std::string method = H.method;
initialize();
H.Clean();
H.SetHeader("Content-Type", "video/mp2t");
H.setCORSHeaders();
if(method == "OPTIONS" || method == "HEAD"){
H.SendResponse("200", "OK", myConn);
H.Clean();
return;
}
H.StartResponse(H, myConn);
parseData = true;
wantRequest = false;
H.Clean(); //clean for any possible next requests
}
void OutHTTPTS::sendTS(const char * tsData, unsigned int len){

7
src/output/output_httpts.h
View file

@ -9,13 +9,6 @@ namespace Mist {
static void init(Util::Config * cfg);
void onHTTP();
void sendTS(const char * tsData, unsigned int len=188);
protected:
int keysToSend;
long long int playUntil;
long long unsigned int lastVid;
long long unsigned int until;
unsigned int vidTrack;
unsigned int audTrack;
};
}

1
src/output/output_progressive_flv.cpp
View file

@ -2,7 +2,6 @@
namespace Mist {
OutProgressiveFLV::OutProgressiveFLV(Socket::Connection & conn) : HTTPOutput(conn){}
OutProgressiveFLV::~OutProgressiveFLV() {}
void OutProgressiveFLV::init(Util::Config * cfg){
HTTPOutput::init(cfg);

1
src/output/output_progressive_flv.h
View file

@ -5,7 +5,6 @@ namespace Mist {
class OutProgressiveFLV : public HTTPOutput {
public:
OutProgressiveFLV(Socket::Connection & conn);
~OutProgressiveFLV();
static void init(Util::Config * cfg);
void onHTTP();
void sendNext();

1
src/output/output_progressive_mp3.cpp
View file

@ -2,7 +2,6 @@
namespace Mist {
OutProgressiveMP3::OutProgressiveMP3(Socket::Connection & conn) : HTTPOutput(conn){}
OutProgressiveMP3::~OutProgressiveMP3(){}
void OutProgressiveMP3::init(Util::Config * cfg){
HTTPOutput::init(cfg);

1
src/output/output_progressive_mp3.h
View file

@ -5,7 +5,6 @@ namespace Mist {
class OutProgressiveMP3 : public HTTPOutput {
public:
OutProgressiveMP3(Socket::Connection & conn);
~OutProgressiveMP3();
static void init(Util::Config * cfg);
void onHTTP();
void sendNext();

8
src/output/output_raw.cpp
View file

@ -4,11 +4,11 @@ namespace Mist {
OutRaw::OutRaw(Socket::Connection & conn) : Output(conn) {
streamName = config->getString("streamname");
initialize();
selectedTracks.clear();
std::string tracks = config->getString("tracks");
if (tracks.size()){
selectedTracks.clear();
unsigned int currTrack = 0;
//loop over tracks, add any found track IDs to selectedTracks
if (tracks != ""){
for (unsigned int i = 0; i < tracks.size(); ++i){
if (tracks[i] >= '0' && tracks[i] <= '9'){
currTrack = currTrack*10 + (tracks[i] - '0');
@ -46,8 +46,7 @@ namespace Mist {
capa["optional"]["seek"]["help"] = "The time in milliseconds to seek to, 0 by default.";
capa["optional"]["seek"]["type"] = "int";
capa["optional"]["seek"]["option"] = "--seek";
capa["codecs"][0u][0u].append("H264");
capa["codecs"][0u][1u].append("AAC");
capa["codecs"][0u][0u].append("*");
cfg->addOption("streamname",
JSON::fromString("{\"arg\":\"string\",\"short\":\"s\",\"long\":\"stream\",\"help\":\"The name of the stream that this connector will transmit.\"}"));
cfg->addOption("tracks",
@ -68,3 +67,4 @@ namespace Mist {
}
}

178
src/output/output_rtmp.cpp
View file

@ -10,10 +10,10 @@
namespace Mist {
OutRTMP::OutRTMP(Socket::Connection & conn) : Output(conn) {
setBlocking(true);
while (!conn.Received().available(1537) && conn.connected()) {
while (!conn.Received().available(1537) && conn.connected() && config->is_active) {
conn.spool();
}
if (!conn){
if (!conn || !config->is_active){
return;
}
RTMPStream::handshake_in.append(conn.Received().remove(1537));
@ -21,21 +21,41 @@ namespace Mist {
if (RTMPStream::doHandshake()) {
conn.SendNow(RTMPStream::handshake_out);
while (!conn.Received().available(1536) && conn.connected()) {
while (!conn.Received().available(1536) && conn.connected() && config->is_active) {
conn.spool();
}
conn.Received().remove(1536);
RTMPStream::rec_cnt += 1536;
DEBUG_MSG(DLVL_HIGH, "Handshake success!");
HIGH_MSG("Handshake success");
} else {
DEBUG_MSG(DLVL_DEVEL, "Handshake fail!");
MEDIUM_MSG("Handshake fail (this is not a problem, usually)");
}
setBlocking(false);
maxSkipAhead = 1500;
minSkipAhead = 500;
}
OutRTMP::~OutRTMP() {}
bool OutRTMP::isReadyForPlay(){
if (isPushing){
return true;
}
if (myMeta.tracks.size()){
for (std::map<unsigned int, DTSC::Track>::iterator it = myMeta.tracks.begin(); it != myMeta.tracks.end(); it++){
if (it->second.keys.size() >= 2){
return true;
}
}
}
return false;
}
std::string OutRTMP::getStatsName(){
if (isPushing){
return "INPUT";
}else{
return Output::getStatsName();
}
}
void OutRTMP::parseVars(std::string data){
std::string varname;
@ -94,7 +114,7 @@ namespace Mist {
void OutRTMP::init(Util::Config * cfg) {
Output::init(cfg);
capa["name"] = "RTMP";
capa["desc"] = "Enables the RTMP protocol which is used by Adobe Flash Player.";
capa["desc"] = "Enables ingest and output over Adobe's RTMP protocol.";
capa["deps"] = "";
capa["url_rel"] = "/play/$";
capa["codecs"][0u][0u].append("H264");
@ -114,13 +134,28 @@ namespace Mist {
capa["codecs"][0u][1u].append("G711mu");
capa["methods"][0u]["handler"] = "rtmp";
capa["methods"][0u]["type"] = "flash/10";
capa["methods"][0u]["priority"] = 6ll;
capa["methods"][0u]["priority"] = 7ll;
capa["methods"][0u]["player_url"] = "/flashplayer.swf";
cfg->addConnectorOptions(1935, capa);
config = cfg;
}
void OutRTMP::sendNext() {
//If there are now more selectable tracks, select the new track and do a seek to the current timestamp
//Set sentHeader to false to force it to send init data
if (selectedTracks.size() < 2 && myMeta.tracks.size() > 1){
size_t prevTrackCount = selectedTracks.size();
selectDefaultTracks();
if (selectedTracks.size() > prevTrackCount){
INFO_MSG("Picked up new track - selecting it and resetting state.");
sentHeader = false;
seek(thisPacket.getTime());
}
return;
}
char rtmpheader[] = {0, //byte 0 = cs_id | ch_type
0, 0, 0, //bytes 1-3 = timestamp
0, 0, 0, //bytes 4-6 = length
@ -312,9 +347,7 @@ namespace Mist {
///\param messageType The type of message.
///\param streamId The ID of the AMF stream.
void OutRTMP::sendCommand(AMF::Object & amfReply, int messageType, int streamId) {
#if DEBUG >= 8
std::cerr << amfReply.Print() << std::endl;
#endif
HIGH_MSG("Sending: %s", amfReply.Print().c_str());
if (messageType == 17) {
myConn.SendNow(RTMPStream::SendChunk(3, messageType, streamId, (char)0 + amfReply.Pack()));
} else {
@ -327,38 +360,13 @@ namespace Mist {
///\param messageType The type of message.
///\param streamId The ID of the AMF stream.
void OutRTMP::parseAMFCommand(AMF::Object & amfData, int messageType, int streamId) {
#if DEBUG >= 5
fprintf(stderr, "Received command: %s\n", amfData.Print().c_str());
#endif
#if DEBUG >= 8
fprintf(stderr, "AMF0 command: %s\n", amfData.getContentP(0)->StrValue().c_str());
#endif
MEDIUM_MSG("Received command: %s", amfData.Print().c_str());
HIGH_MSG("AMF0 command: %s", amfData.getContentP(0)->StrValue().c_str());
if (amfData.getContentP(0)->StrValue() == "connect") {
double objencoding = 0;
if (amfData.getContentP(2)->getContentP("objectEncoding")) {
objencoding = amfData.getContentP(2)->getContentP("objectEncoding")->NumValue();
}
#if DEBUG >= 6
int tmpint;
if (amfData.getContentP(2)->getContentP("videoCodecs")) {
tmpint = (int)amfData.getContentP(2)->getContentP("videoCodecs")->NumValue();
if (tmpint & 0x04) {
fprintf(stderr, "Sorensen video support detected\n");
}
if (tmpint & 0x80) {
fprintf(stderr, "H264 video support detected\n");
}
}
if (amfData.getContentP(2)->getContentP("audioCodecs")) {
tmpint = (int)amfData.getContentP(2)->getContentP("audioCodecs")->NumValue();
if (tmpint & 0x04) {
fprintf(stderr, "MP3 audio support detected\n");
}
if (tmpint & 0x400) {
fprintf(stderr, "AAC audio support detected\n");
}
}
#endif
app_name = amfData.getContentP(2)->getContentP("tcUrl")->StrValue();
app_name = app_name.substr(app_name.find('/', 7) + 1);
RTMPStream::chunk_snd_max = 4096;
@ -467,7 +475,7 @@ namespace Mist {
} //getStreamLength
if ((amfData.getContentP(0)->StrValue() == "publish")) {
if (amfData.getContentP(3)) {
streamName = amfData.getContentP(3)->StrValue();
streamName = Encodings::URL::decode(amfData.getContentP(3)->StrValue());
if (streamName.find('/')){
streamName = streamName.substr(0, streamName.find('/'));
@ -485,32 +493,33 @@ namespace Mist {
Util::sanitizeName(streamName);
//pull the server configuration
IPC::sharedPage serverCfg("!mistConfig", DEFAULT_CONF_PAGE_SIZE); ///< Contains server configuration and capabilities
IPC::semaphore configLock("!mistConfLock", O_CREAT | O_RDWR, ACCESSPERMS, 1);
IPC::sharedPage serverCfg(SHM_CONF, DEFAULT_CONF_PAGE_SIZE); ///< Contains server configuration and capabilities
IPC::semaphore configLock(SEM_CONF, O_CREAT | O_RDWR, ACCESSPERMS, 1);
configLock.wait();
DTSC::Scan streamCfg = DTSC::Scan(serverCfg.mapped, serverCfg.len).getMember("streams").getMember(streamName);
if (streamCfg){
if (streamCfg.getMember("source").asString().substr(0, 7) != "push://"){
DEBUG_MSG(DLVL_FAIL, "Push rejected - stream %s not a push-able stream. (%s != push://*)", streamName.c_str(), streamCfg.getMember("source").asString().c_str());
FAIL_MSG("Push rejected - stream %s not a push-able stream. (%s != push://*)", streamName.c_str(), streamCfg.getMember("source").asString().c_str());
myConn.close();
}else{
std::string source = streamCfg.getMember("source").asString().substr(7);
std::string IP = source.substr(0, source.find('@'));
if (IP != ""){
if (!myConn.isAddress(IP)){
DEBUG_MSG(DLVL_FAIL, "Push from %s to %s rejected - source host not whitelisted", getConnectedHost().c_str(), streamName.c_str());
FAIL_MSG("Push from %s to %s rejected - source host not whitelisted", getConnectedHost().c_str(), streamName.c_str());
myConn.close();
}
}
}
}else{
DEBUG_MSG(DLVL_FAIL, "Push from %s rejected - stream '%s' not configured.", getConnectedHost().c_str(), streamName.c_str());
FAIL_MSG("Push from %s rejected - stream '%s' not configured.", getConnectedHost().c_str(), streamName.c_str());
myConn.close();
}
configLock.post();
configLock.close();
if (!myConn){return;}//do not initialize if rejected
isPushing = true;
initialize();
}
//send a _result reply
@ -698,16 +707,22 @@ namespace Mist {
}
return;
} //seek
if (amfData.getContentP(0)->StrValue() == "_error") {
WARN_MSG("Received error response: %s", amfData.Print().c_str());
return;
}
if ((amfData.getContentP(0)->StrValue() == "_result") || (amfData.getContentP(0)->StrValue() == "onFCPublish") || (amfData.getContentP(0)->StrValue() == "onStatus")) {
//Results are ignored. We don't really care.
return;
}
#if DEBUG >= 2
fprintf(stderr, "AMF0 command not processed!\n%s\n", amfData.Print().c_str());
#endif
WARN_MSG("AMF0 command not processed: %s", amfData.Print().c_str());
//send a _result reply
AMF::Object amfReply("container", AMF::AMF0_DDV_CONTAINER);
amfReply.addContent(AMF::Object("", "_error")); //result success
amfReply.addContent(amfData.getContent(1)); //same transaction ID
amfReply.addContent(AMF::Object("", (double)0, AMF::AMF0_NULL)); //null - command info
amfReply.addContent(AMF::Object("Command not implemented or recognized", "")); //stream ID?
amfReply.addContent(AMF::Object("", amfData.getContentP(0)->StrValue())); //null - command info
amfReply.addContent(AMF::Object("", "Command not implemented or recognized")); //stream ID?
sendCommand(amfReply, messageType, streamId);
} //parseAMFCommand
@ -734,9 +749,7 @@ namespace Mist {
switch (next.msg_type_id) {
case 0: //does not exist
#if DEBUG >= 2
fprintf(stderr, "UNKN: Received a zero-type message. Possible data corruption? Aborting!\n");
#endif
WARN_MSG("UNKN: Received a zero-type message. Possible data corruption? Aborting!");
while (inputBuffer.size()) {
inputBuffer.get().clear();
}
@ -745,20 +758,14 @@ namespace Mist {
break; //happens when connection breaks unexpectedly
case 1: //set chunk size
RTMPStream::chunk_rec_max = ntohl(*(int *)next.data.c_str());
#if DEBUG >= 5
fprintf(stderr, "CTRL: Set chunk size: %i\n", RTMPStream::chunk_rec_max);
#endif
MEDIUM_MSG("CTRL: Set chunk size: %i", RTMPStream::chunk_rec_max);
break;
case 2: //abort message - we ignore this one
#if DEBUG >= 5
fprintf(stderr, "CTRL: Abort message\n");
#endif
MEDIUM_MSG("CTRL: Abort message");
//4 bytes of stream id to drop
break;
case 3: //ack
#if DEBUG >= 8
fprintf(stderr, "CTRL: Acknowledgement\n");
#endif
VERYHIGH_MSG("CTRL: Acknowledgement");
RTMPStream::snd_window_at = ntohl(*(int *)next.data.c_str());
RTMPStream::snd_window_at = RTMPStream::snd_cnt;
break;
@ -773,49 +780,43 @@ namespace Mist {
//6 = pingrequest, 4 bytes data
//7 = pingresponse, 4 bytes data
//we don't need to process this
#if DEBUG >= 5
short int ucmtype = ntohs(*(short int *)next.data.c_str());
switch (ucmtype) {
case 0:
fprintf(stderr, "CTRL: UCM StreamBegin %i\n", ntohl(*((int *)(next.data.c_str() + 2))));
MEDIUM_MSG("CTRL: UCM StreamBegin %i", ntohl(*((int *)(next.data.c_str() + 2))));
break;
case 1:
fprintf(stderr, "CTRL: UCM StreamEOF %i\n", ntohl(*((int *)(next.data.c_str() + 2))));
MEDIUM_MSG("CTRL: UCM StreamEOF %i", ntohl(*((int *)(next.data.c_str() + 2))));
break;
case 2:
fprintf(stderr, "CTRL: UCM StreamDry %i\n", ntohl(*((int *)(next.data.c_str() + 2))));
MEDIUM_MSG("CTRL: UCM StreamDry %i", ntohl(*((int *)(next.data.c_str() + 2))));
break;
case 3:
fprintf(stderr, "CTRL: UCM SetBufferLength %i %i\n", ntohl(*((int *)(next.data.c_str() + 2))), ntohl(*((int *)(next.data.c_str() + 6))));
MEDIUM_MSG("CTRL: UCM SetBufferLength %i %i", ntohl(*((int *)(next.data.c_str() + 2))), ntohl(*((int *)(next.data.c_str() + 6))));
break;
case 4:
fprintf(stderr, "CTRL: UCM StreamIsRecorded %i\n", ntohl(*((int *)(next.data.c_str() + 2))));
MEDIUM_MSG("CTRL: UCM StreamIsRecorded %i", ntohl(*((int *)(next.data.c_str() + 2))));
break;
case 6:
fprintf(stderr, "CTRL: UCM PingRequest %i\n", ntohl(*((int *)(next.data.c_str() + 2))));
MEDIUM_MSG("CTRL: UCM PingRequest %i", ntohl(*((int *)(next.data.c_str() + 2))));
break;
case 7:
fprintf(stderr, "CTRL: UCM PingResponse %i\n", ntohl(*((int *)(next.data.c_str() + 2))));
MEDIUM_MSG("CTRL: UCM PingResponse %i", ntohl(*((int *)(next.data.c_str() + 2))));
break;
default:
fprintf(stderr, "CTRL: UCM Unknown (%hi)\n", ucmtype);
MEDIUM_MSG("CTRL: UCM Unknown (%hi)", ucmtype);
break;
}
#endif
}
break;
case 5: //window size of other end
#if DEBUG >= 5
fprintf(stderr, "CTRL: Window size\n");
#endif
MEDIUM_MSG("CTRL: Window size");
RTMPStream::rec_window_size = ntohl(*(int *)next.data.c_str());
RTMPStream::rec_window_at = RTMPStream::rec_cnt;
myConn.SendNow(RTMPStream::SendCTL(3, RTMPStream::rec_cnt)); //send ack (msg 3)
break;
case 6:
#if DEBUG >= 5
fprintf(stderr, "CTRL: Set peer bandwidth\n");
#endif
MEDIUM_MSG("CTRL: Set peer bandwidth");
//4 bytes window size, 1 byte limit type (ignored)
RTMPStream::snd_window_size = ntohl(*(int *)next.data.c_str());
myConn.SendNow(RTMPStream::SendCTL(5, RTMPStream::snd_window_size)); //send window acknowledgement size (msg 5)
@ -838,32 +839,31 @@ namespace Mist {
}
JSON::Value pack_out = F.toJSON(myMeta, *amf_storage, next.cs_id*3 + (F.data[0] == 0x09 ? 0 : (F.data[0] == 0x08 ? 1 : 2) ));
if ( !pack_out.isNull()){
if (!userClient.getData()){
if (!nProxy.userClient.getData()){
char userPageName[NAME_BUFFER_SIZE];
snprintf(userPageName, NAME_BUFFER_SIZE, SHM_USERS, streamName.c_str());
userClient = IPC::sharedClient(userPageName, 30, true);
nProxy.userClient = IPC::sharedClient(userPageName, PLAY_EX_SIZE, true);
}
continueNegotiate(pack_out["trackid"].asInt());
nProxy.streamName = streamName;
bufferLivePacket(pack_out);
}
break;
}
case 15:
DEBUG_MSG(DLVL_MEDIUM, "Received AMF3 data message");
MEDIUM_MSG("Received AMF3 data message");
break;
case 16:
DEBUG_MSG(DLVL_MEDIUM, "Received AMF3 shared object");
MEDIUM_MSG("Received AMF3 shared object");
break;
case 17: {
DEBUG_MSG(DLVL_MEDIUM, "Received AMF3 command message");
MEDIUM_MSG("Received AMF3 command message");
if (next.data[0] != 0) {
next.data = next.data.substr(1);
amf3data = AMF::parse3(next.data);
#if DEBUG >= 5
amf3data.Print();
#endif
MEDIUM_MSG("AMF3: %s", amf3data.Print().c_str());
} else {
DEBUG_MSG(DLVL_MEDIUM, "Received AMF3-0 command message");
MEDIUM_MSG("Received AMF3-0 command message");
next.data = next.data.substr(1);
amfdata = AMF::parse(next.data);
parseAMFCommand(amfdata, 17, next.msg_stream_id);
@ -871,7 +871,7 @@ namespace Mist {
}
break;
case 19:
DEBUG_MSG(DLVL_MEDIUM, "Received AMF0 shared object");
MEDIUM_MSG("Received AMF0 shared object");
break;
case 20: { //AMF0 command message
amfdata = AMF::parse(next.data);
@ -879,10 +879,10 @@ namespace Mist {
}
break;
case 22:
DEBUG_MSG(DLVL_MEDIUM, "Received aggregate message");
MEDIUM_MSG("Received aggregate message");
break;
default:
DEBUG_MSG(DLVL_FAIL, "Unknown chunk received! Probably protocol corruption, stopping parsing of incoming data.");
FAIL_MSG("Unknown chunk received! Probably protocol corruption, stopping parsing of incoming data.");
break;
}
}

4
src/output/output_rtmp.h
View file

@ -9,17 +9,19 @@ namespace Mist {
class OutRTMP : public Output {
public:
OutRTMP(Socket::Connection & conn);
~OutRTMP();
static void init(Util::Config * cfg);
void onRequest();
void sendNext();
void sendHeader();
bool isReadyForPlay();
protected:
bool isPushing;
void parseVars(std::string data);
std::string app_name;
void parseChunk(Socket::Buffer & inputBuffer);
void parseAMFCommand(AMF::Object & amfData, int messageType, int streamId);
void sendCommand(AMF::Object & amfReply, int messageType, int streamId);
virtual std::string getStatsName();
};
}

7
src/output/output_ts.cpp
View file

@ -40,17 +40,16 @@ namespace Mist {
capa["required"]["streamname"]["help"] = "What streamname to serve. For multiple streams, add this protocol multiple times using different ports.";
capa["required"]["streamname"]["type"] = "str";
capa["required"]["streamname"]["option"] = "--stream";
capa["required"]["streamname"]["short"] = "s";
capa["optional"]["tracks"]["name"] = "Tracks";
capa["optional"]["tracks"]["help"] = "The track IDs of the stream that this connector will transmit separated by spaces";
capa["optional"]["tracks"]["type"] = "str";
capa["optional"]["tracks"]["option"] = "--tracks";
capa["optional"]["tracks"]["short"] = "t";
capa["optional"]["tracks"]["default"] = "";
capa["codecs"][0u][0u].append("H264");
capa["codecs"][0u][1u].append("AAC");
capa["codecs"][0u][1u].append("MP3");
cfg->addOption("streamname",
JSON::fromString("{\"arg\":\"string\",\"short\":\"s\",\"long\":\"stream\",\"help\":\"The name of the stream that this connector will transmit.\"}"));
cfg->addOption("tracks",
JSON::fromString("{\"arg\":\"string\",\"value\":[\"\"],\"short\": \"t\",\"long\":\"tracks\",\"help\":\"The track IDs of the stream that this connector will transmit separated by spaces.\"}"));
cfg->addConnectorOptions(8888, capa);
config = cfg;
}

4
src/output/output_ts_base.cpp
View file

@ -32,7 +32,7 @@ namespace Mist {
if (packData.getBytesFree() == 184){
packData.clear();
packData.setPID(0x100 - 1 + thisPacket.getTrackId());
packData.setPID(thisPacket.getTrackId());
packData.setContinuityCounter(++contCounters[packData.getPID()]);
if (first[thisPacket.getTrackId()]){
packData.setUnitStart(1);
@ -125,7 +125,7 @@ namespace Mist {
break;
}
if (alreadySent + 4 > watKunnenWeIn1Ding){
nalLead = 4 - watKunnenWeIn1Ding-alreadySent;
nalLead = 4 - (watKunnenWeIn1Ding-alreadySent);
fillPacket("\000\000\000\001",watKunnenWeIn1Ding-alreadySent);
i += watKunnenWeIn1Ding-alreadySent;
alreadySent += watKunnenWeIn1Ding-alreadySent;