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

Convert to autotools build system for cleanness (part1).

Browse source
This commit is contained in:
Thulinma 2012-05-08 19:19:42 +02:00
parent 3ebfe1b693
commit 36e086e0e2
125 changed files with 125 additions and 4475 deletions
Download patch file Download diff file Expand all files Collapse all files

13
src/Makefile.am Normal file
View file

@ -0,0 +1,13 @@
outdir=../bin
out_PROGRAMS=MistBuffer MistController MistConnRAW MistConnRTMP MistConnHTTP
AM_LDFLAGS=-L../lib
MistBuffer_SOURCES=buffer.cpp buffer_stats.cpp buffer_user.cpp buffer_stream.cpp
MistBuffer_LDADD=-ljson -lsocket -ldtsc -ltinythread -lpthread
MistController_SOURCES=controller.cpp
MistController_LDADD=-ljson -lsocket -lprocs -lmd5 -lconfig -lhttp_parser -lauth -lbase64 -lssl -lcrypto
MistConnRAW_SOURCES=conn_raw.cpp
MistConnRAW_LDADD=-lsocket
MistConnRTMP_SOURCES=conn_rtmp.cpp
MistConnRTMP_LDADD=-lsocket -ldtsc -lrtmpchunks -lflv_tag -lconfig -lamf -lssl -lcrypto
MistConnHTTP_SOURCES=conn_http.cpp
MistConnHTTP_LDADD=-lsocket

244
src/analysers/abst_analyser.cpp Normal file
View file

@ -0,0 +1,244 @@
/// \file ABST_Parser/main.cpp
/// Debugging tool for ABST boxes.
/// Expects ABST data through stdin, outputs human-readable information to stderr.
/// \todo Erik, update, delete or properly document this file.
#include <stdint.h>
#include <iostream>
#include <string>
#include "../../util/MP4/box_includes.h"
void Parse( Box * source ,std::string PrintOffset ) {
if( source->header.BoxType == 0x61627374 ) {
uint8_t Version = source->Payload[0];
uint32_t Flags = (source->Payload[1] << 16) + (source->Payload[2] << 8) + (source->Payload[3]); //uint24_t
uint32_t BootstrapInfoVersion = (source->Payload[4] << 24) + (source->Payload[5] << 16) +(source->Payload[6] << 8) + (source->Payload[7]);
uint8_t Profile = (source->Payload[8] >> 6); //uint2_t
uint8_t Live = ((source->Payload[8] >> 5 ) & 0x1); //uint1_t
uint8_t Update = ((source->Payload[8] >> 4 ) & 0x1); //uint1_t
uint8_t Reserved = (source->Payload[8] & 0x4); //uint4_t
uint32_t Timescale = (source->Payload[9] << 24) + (source->Payload[10] << 16) +(source->Payload[11] << 8) + (source->Payload[12]);
uint32_t CurrentMediaTime_Upperhalf = (source->Payload[13] << 24) + (source->Payload[14] << 16) +(source->Payload[15] << 8) + (source->Payload[16]);
uint32_t CurrentMediaTime_Lowerhalf = (source->Payload[17] << 24) + (source->Payload[18] << 16) +(source->Payload[19] << 8) + (source->Payload[20]);
uint32_t SmpteTimeCodeOffset_Upperhalf = (source->Payload[21] << 24) + (source->Payload[22] << 16) +(source->Payload[23] << 8) + (source->Payload[24]);
uint32_t SmpteTimeCodeOffset_Lowerhalf = (source->Payload[25] << 24) + (source->Payload[26] << 16) +(source->Payload[27] << 8) + (source->Payload[28]);
std::string MovieIdentifier;
uint8_t ServerEntryCount = -1;
std::vector<std::string> ServerEntryTable;
uint8_t QualityEntryCount = -1;
std::vector<std::string> QualityEntryTable;
std::string DrmData;
std::string MetaData;
uint8_t SegmentRunTableCount = -1;
std::vector<Box*> SegmentRunTableEntries;
uint8_t FragmentRunTableCount = -1;
std::vector<Box*> FragmentRunTableEntries;
uint32_t CurrentOffset = 29;
uint32_t TempSize;
Box* TempBox;
std::string temp;
while( source->Payload[CurrentOffset] != '\0' ) { MovieIdentifier += source->Payload[CurrentOffset]; CurrentOffset ++; }
CurrentOffset ++;
ServerEntryCount = source->Payload[CurrentOffset];
CurrentOffset ++;
for( uint8_t i = 0; i < ServerEntryCount; i++ ) {
temp = "";
while( source->Payload[CurrentOffset] != '\0' ) { temp += source->Payload[CurrentOffset]; CurrentOffset ++; }
ServerEntryTable.push_back(temp);
CurrentOffset++;
}
QualityEntryCount = source->Payload[CurrentOffset];
CurrentOffset ++;
for( uint8_t i = 0; i < QualityEntryCount; i++ ) {
temp = "";
while( source->Payload[CurrentOffset] != '\0' ) { temp += source->Payload[CurrentOffset]; CurrentOffset ++; }
QualityEntryTable.push_back(temp);
CurrentOffset++;
}
while( source->Payload[CurrentOffset] != '\0' ) { DrmData += source->Payload[CurrentOffset]; CurrentOffset ++; }
CurrentOffset ++;
while( source->Payload[CurrentOffset] != '\0' ) { MetaData += source->Payload[CurrentOffset]; CurrentOffset ++; }
CurrentOffset ++;
SegmentRunTableCount = source->Payload[CurrentOffset];
CurrentOffset ++;
for( uint8_t i = 0; i < SegmentRunTableCount; i++ ) {
TempSize = (source->Payload[CurrentOffset] << 24) + (source->Payload[CurrentOffset+1]<< 16) + (source->Payload[CurrentOffset+2]<< 8) + (source->Payload[CurrentOffset+3]);
TempBox = new Box( &source->Payload[CurrentOffset], TempSize );
SegmentRunTableEntries.push_back(TempBox);
CurrentOffset += TempSize;
}
FragmentRunTableCount = source->Payload[CurrentOffset];
CurrentOffset ++;
for( uint8_t i = 0; i < FragmentRunTableCount; i++ ) {
TempSize = (source->Payload[CurrentOffset] << 24) + (source->Payload[CurrentOffset+1]<< 16) + (source->Payload[CurrentOffset+2]<< 8) + (source->Payload[CurrentOffset+3]);
TempBox = new Box( &source->Payload[CurrentOffset], TempSize );
FragmentRunTableEntries.push_back(TempBox);
CurrentOffset += TempSize;
}
std::cerr << "Box_ABST:\n";
std::cerr << PrintOffset << " Version: " << (int)Version << "\n";
std::cerr << PrintOffset << " Flags: " << (int)Flags << "\n";
std::cerr << PrintOffset << " BootstrapInfoVersion: " << (int)BootstrapInfoVersion << "\n";
std::cerr << PrintOffset << " Profile: " << (int)Profile << "\n";
std::cerr << PrintOffset << " Live: " << (int)Live << "\n";
std::cerr << PrintOffset << " Update: " << (int)Update << "\n";
std::cerr << PrintOffset << " Reserved: " << (int)Reserved << "\n";
std::cerr << PrintOffset << " Timescale: " << (int)Timescale << "\n";
std::cerr << PrintOffset << " CurrentMediaTime: " << (int)CurrentMediaTime_Upperhalf << " " << CurrentMediaTime_Lowerhalf << "\n";
std::cerr << PrintOffset << " SmpteTimeCodeOffset: " << (int)SmpteTimeCodeOffset_Upperhalf << " " << SmpteTimeCodeOffset_Lowerhalf << "\n";
std::cerr << PrintOffset << " MovieIdentifier: " << MovieIdentifier << "\n";
std::cerr << PrintOffset << " ServerEntryCount: " << (int)ServerEntryCount << "\n";
std::cerr << PrintOffset << " ServerEntryTable:\n";
for( uint32_t i = 0; i < ServerEntryTable.size( ); i++ ) {
std::cerr << PrintOffset << " " << i+1 << ": " << ServerEntryTable[i] << "\n";
}
std::cerr << PrintOffset << " QualityEntryCount: " << (int)QualityEntryCount << "\n";
std::cerr << PrintOffset << " QualityEntryTable:\n";
for( uint32_t i = 0; i < QualityEntryTable.size( ); i++ ) {
std::cerr << PrintOffset << " " << i+1 << ": " << QualityEntryTable[i] << "\n";
}
std::cerr << PrintOffset << " DrmData: " << DrmData << "\n";
std::cerr << PrintOffset << " MetaData: " << MetaData << "\n";
std::cerr << PrintOffset << " SegmentRunTableCount: " << (int)SegmentRunTableCount << "\n";
std::cerr << PrintOffset << " SegmentRunTableEntries:\n";
for( uint32_t i = 0; i < SegmentRunTableEntries.size( ); i++ ) {
std::cerr << PrintOffset << " " << i+1 << ": ";
Parse( SegmentRunTableEntries[i], PrintOffset+" " );
}
std::cerr << PrintOffset << " FragmentRunTableCount: " << (int)FragmentRunTableCount << "\n";
std::cerr << PrintOffset << " FragmentRunTableEntries:\n";
for( uint32_t i = 0; i < FragmentRunTableEntries.size( ); i++ ) {
std::cerr << PrintOffset << " " << i+1 << ": ";
Parse( FragmentRunTableEntries[i], PrintOffset+" " );
}
} else if ( source->header.BoxType == 0x61737274 ) {
uint8_t Version = source->Payload[0];
uint32_t Flags = (source->Payload[1] << 16) + (source->Payload[2] << 8) + (source->Payload[3]); //uint24_t
uint8_t QualityEntryCount;
std::vector<std::string> QualitySegmentUrlModifiers;
uint32_t SegmentRunEntryCount;
std::vector< std::pair<uint32_t,uint32_t> > SegmentRunEntryTable;
uint32_t CurrentOffset = 4;
std::string temp;
std::pair<uint32_t,uint32_t> TempPair;
QualityEntryCount = source->Payload[CurrentOffset];
CurrentOffset ++;
for( uint8_t i = 0; i < QualityEntryCount; i++ ) {
temp = "";
while( source->Payload[CurrentOffset] != '\0' ) { temp += source->Payload[CurrentOffset]; CurrentOffset ++; }
QualitySegmentUrlModifiers.push_back(temp);
CurrentOffset++;
}
SegmentRunEntryCount = (source->Payload[CurrentOffset] << 24) + (source->Payload[CurrentOffset+1] << 16) + (source->Payload[CurrentOffset+2] << 8) + (source->Payload[CurrentOffset+3]);
CurrentOffset +=4;
for( uint8_t i = 0; i < SegmentRunEntryCount; i++ ) {
TempPair.first = (source->Payload[CurrentOffset] << 24) + (source->Payload[CurrentOffset+1] << 16) + (source->Payload[CurrentOffset+2] << 8) + (source->Payload[CurrentOffset+3]);
CurrentOffset+=4;
TempPair.second = (source->Payload[CurrentOffset] << 24) + (source->Payload[CurrentOffset+1] << 16) + (source->Payload[CurrentOffset+2] << 8) + (source->Payload[CurrentOffset+3]);
CurrentOffset+=4;
SegmentRunEntryTable.push_back(TempPair);
}
std::cerr << "Box_ASRT:\n";
std::cerr << PrintOffset << " Version: " << (int)Version << "\n";
std::cerr << PrintOffset << " Flags: " << (int)Flags << "\n";
std::cerr << PrintOffset << " QualityEntryCount: " << (int)QualityEntryCount << "\n";
std::cerr << PrintOffset << " QualitySegmentUrlModifiers:\n";
for( uint32_t i = 0; i < QualitySegmentUrlModifiers.size( ); i++ ) {
std::cerr << PrintOffset << " " << i+1 << ": " << QualitySegmentUrlModifiers[i] << "\n";
}
std::cerr << PrintOffset << " SegmentRunEntryCount: " << (int)SegmentRunEntryCount << "\n";
std::cerr << PrintOffset << " SegmentRunEntryTable:\n";
for( uint32_t i = 0; i < SegmentRunEntryTable.size( ); i++ ) {
std::cerr << PrintOffset << " " << i+1 << ":\n";
std::cerr << PrintOffset << " FirstSegment: " << SegmentRunEntryTable[i].first << "\n";
std::cerr << PrintOffset << " FragmentsPerSegment: " << SegmentRunEntryTable[i].second << "\n";
}
} else if ( source->header.BoxType == 0x61667274 ) {
uint8_t Version = source->Payload[0];
uint32_t Flags = (source->Payload[1] << 16) + (source->Payload[2] << 8) + (source->Payload[3]); //uint24_t
uint32_t TimeScale = (source->Payload[4] << 24) + (source->Payload[5] << 16) + (source->Payload[6] << 8) + (source->Payload[7]);
uint8_t QualityEntryCount;
std::vector<std::string> QualitySegmentUrlModifiers;
uint32_t FragmentRunEntryCount;
std::vector<afrt_fragmentrunentry> FragmentRunEntryTable;
uint32_t CurrentOffset = 8;
std::string temp;
afrt_fragmentrunentry TempEntry;
QualityEntryCount = source->Payload[CurrentOffset];
CurrentOffset ++;
for( uint8_t i = 0; i < QualityEntryCount; i++ ) {
temp = "";
while( source->Payload[CurrentOffset] != '\0' ) { temp += source->Payload[CurrentOffset]; CurrentOffset ++; }
QualitySegmentUrlModifiers.push_back(temp);
CurrentOffset++;
}
FragmentRunEntryCount = (source->Payload[CurrentOffset] << 24) + (source->Payload[CurrentOffset+1] << 16) + (source->Payload[CurrentOffset+2] << 8) + (source->Payload[CurrentOffset+3]);
CurrentOffset +=4;
for( uint8_t i = 0; i < FragmentRunEntryCount; i ++ ) {
TempEntry.FirstFragment = (source->Payload[CurrentOffset] << 24) + (source->Payload[CurrentOffset+1] << 16) + (source->Payload[CurrentOffset+2] << 8) + (source->Payload[CurrentOffset+3]);
CurrentOffset +=4;
CurrentOffset +=4;
TempEntry.FirstFragmentTimestamp = (source->Payload[CurrentOffset] << 24) + (source->Payload[CurrentOffset+1] << 16) + (source->Payload[CurrentOffset+2] << 8) + (source->Payload[CurrentOffset+3]);
CurrentOffset +=4;
TempEntry.FragmentDuration = (source->Payload[CurrentOffset] << 24) + (source->Payload[CurrentOffset+1] << 16) + (source->Payload[CurrentOffset+2] << 8) + (source->Payload[CurrentOffset+3]);
CurrentOffset +=4;
if( TempEntry.FragmentDuration == 0 ) {
TempEntry.DiscontinuityIndicator = source->Payload[CurrentOffset];
CurrentOffset++;
}
FragmentRunEntryTable.push_back(TempEntry);
}
std::cerr << "Box_AFRT:\n";
std::cerr << PrintOffset << " Version: " << (int)Version << "\n";
std::cerr << PrintOffset << " Flags: " << (int)Flags << "\n";
std::cerr << PrintOffset << " Timescale: " << (int)TimeScale << "\n";
std::cerr << PrintOffset << " QualityEntryCount: " << (int)QualityEntryCount << "\n";
std::cerr << PrintOffset << " QualitySegmentUrlModifiers:\n";
for( uint32_t i = 0; i < QualitySegmentUrlModifiers.size( ); i++ ) {
std::cerr << PrintOffset << " " << i+1 << ": " << QualitySegmentUrlModifiers[i] << "\n";
}
std::cerr << PrintOffset << " FragmentRunEntryCount: " << (int)FragmentRunEntryCount << "\n";
std::cerr << PrintOffset << " FragmentRunEntryTable:\n";
for( uint32_t i = 0; i < FragmentRunEntryTable.size( ); i++ ) {
std::cerr << PrintOffset << " " << i+1 << ":\n";
std::cerr << PrintOffset << " FirstFragment: " << FragmentRunEntryTable[i].FirstFragment << "\n";
std::cerr << PrintOffset << " FirstFragmentTimestamp: " << FragmentRunEntryTable[i].FirstFragmentTimestamp << "\n";
std::cerr << PrintOffset << " FragmentDuration: " << FragmentRunEntryTable[i].FragmentDuration << "\n";
if( FragmentRunEntryTable[i].FragmentDuration == 0 ) {
std::cerr << PrintOffset << " DiscontinuityIndicator: " << (int)FragmentRunEntryTable[i].DiscontinuityIndicator << "\n";
}
}
} else {
std::cerr << "BoxType '"
<< (char)(source->header.BoxType >> 24)
<< (char)((source->header.BoxType << 8) >> 24)
<< (char)((source->header.BoxType << 16) >> 24)
<< (char)((source->header.BoxType << 24) >> 24)
<< "' not yet implemented!\n";
}
}
int main( ) {
std::string temp;
bool validinp = true;
char thischar;
while(validinp) {
thischar = std::cin.get( );
if(std::cin.good( ) ) {
temp += thischar;
} else {
validinp = false;
}
}
Box * TestBox = new Box((uint8_t*)temp.c_str( ), temp.size( ));
Parse( TestBox, "" );
delete TestBox;
}

22
src/analysers/amf_analyser.cpp Normal file
View file

@ -0,0 +1,22 @@
/// \file AMF_Tester/main.cpp
/// Debugging tool for AMF data.
/// Expects AMF data through stdin, outputs human-readable information to stderr.
#define DEBUG 10 //maximum debugging level
#include <cstdlib>
#include <iostream>
#include <fstream>
#include <string>
#include "../../util/amf.h"
/// Debugging tool for AMF data.
/// Expects AMF data through stdin, outputs human-readable information to stderr.
int main() {
std::string temp;
while (std::cin.good()){temp += std::cin.get();}//read all of std::cin to temp
temp.erase(temp.size()-1, 1);//strip the invalid last character
AMF::Object amfdata = AMF::parse(temp);//parse temp into an AMF::Object
amfdata.Print();//pretty-print the object
return 0;
}

38
src/analysers/dtsc_analyser.cpp Normal file
View file

@ -0,0 +1,38 @@
/// \file DTSC_Analyser/main.cpp
/// Contains the code for the DTSC Analysing tool.
#include <fcntl.h>
#include <iostream>
#include <string>
#include <vector>
#include <cstdlib>
#include <cstdio>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include "../../util/dtsc.h" //DTSC support
/// Reads DTSC from stdin and outputs human-readable information to stderr.
int main() {
DTSC::Stream Strm;
std::string inBuffer;
char charBuffer[1024*10];
unsigned int charCount;
bool doneheader = false;
while(std::cin.good()){
//invalidate the current buffer
std::cin.read(charBuffer, 1024*10);
charCount = std::cin.gcount();
inBuffer.append(charBuffer, charCount);
if (Strm.parsePacket(inBuffer)){
if (!doneheader){
doneheader = true;
Strm.metadata.Print();
}
Strm.getPacket().Print();
}
}
return 0;
}

52
src/analysers/flv_analyser.cpp Normal file
View file

@ -0,0 +1,52 @@
/// \file DTSC_Analyser/main.cpp
/// Contains the code for the DTSC Analysing tool.
#include <fcntl.h>
#include <iostream>
#include <string>
#include <vector>
#include <cstdlib>
#include <cstdio>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include "../../util/flv_tag.h" //FLV support
/// Reads DTSC from stdin and outputs human-readable information to stderr.
int main() {
FLV::Tag FLV_in; // Temporary storage for incoming FLV data.
while (!feof(stdin)){
if (FLV_in.FileLoader(stdin)){
std::cout << "Tag: " << FLV_in.tagType() << std::endl;
printf("%hhX %hhX %hhX %hhX %hhX %hhX %hhX %hhX %hhX %hhX\n", FLV_in.data[11], FLV_in.data[12], FLV_in.data[13], FLV_in.data[14], FLV_in.data[15], FLV_in.data[16], FLV_in.data[17], FLV_in.data[18], FLV_in.data[19], FLV_in.data[20]);
printf("%hhX %hhX %hhX %hhX %hhX %hhX %hhX %hhX %hhX %hhX\n", FLV_in.data[FLV_in.len-10], FLV_in.data[FLV_in.len-9], FLV_in.data[FLV_in.len-8], FLV_in.data[FLV_in.len-7], FLV_in.data[FLV_in.len-6], FLV_in.data[FLV_in.len-5], FLV_in.data[FLV_in.len-4], FLV_in.data[FLV_in.len-3], FLV_in.data[FLV_in.len-2], FLV_in.data[FLV_in.len-1]);
std::cout << std::endl;
}
}
DTSC::Stream Strm;
std::string inBuffer;
char charBuffer[1024*10];
unsigned int charCount;
bool doneheader = false;
while(std::cin.good()){
//invalidate the current buffer
std::cin.read(charBuffer, 1024*10);
charCount = std::cin.gcount();
inBuffer.append(charBuffer, charCount);
if (Strm.parsePacket(inBuffer)){
if (!doneheader){
doneheader = true;
Strm.metadata.Print();
}
Strm.getPacket().Print();
}
}
return 0;
}

45
src/analysers/httpbox_analyser.cpp Normal file
View file

@ -0,0 +1,45 @@
/// \file HTTP_Box_Parser/main.cpp
/// Debugging tool for F4M HTTP streaming data.
/// Expects raw TCP data through stdin, outputs human-readable information to stderr.
/// This will attempt to read either HTTP requests or responses from stdin, and if the body is more than
/// 10,000 bytes long will attempt to parse the data as a MP4 box. (Other cases show a message about the fragment being too small)
/// Then it will take the payload of this box, print the first four bytes, and attempt to parse the whole payload as FLV data.
/// The parsed FLV data is then pretty-printed, containing information about the codec parameters and types of tags it encounters.
#include <stdint.h>
#include <iostream>
#include <string>
#include <stdio.h>
#include "../../util/http_parser.h"
#include "../../util/MP4/box_includes.h"
#include "../../util/flv_tag.h"
/// Debugging tool for F4M HTTP streaming data.
/// Expects raw TCP data through stdin, outputs human-readable information to stderr.
/// This will attempt to read either HTTP requests or responses from stdin, and if the body is more than
/// 10,000 bytes long will attempt to parse the data as a MP4 box. (Other cases show a message about the fragment being too small)
/// Then it will take the payload of this box, print the first four bytes, and attempt to parse the whole payload as FLV data.
/// The parsed FLV data is then pretty-printed, containing information about the codec parameters and types of tags it encounters.
int main(){
HTTP::Parser H;
FLV::Tag F;
unsigned int P = 0;
char * Payload = 0;
while (H.Read(stdin) || H.CleanForNext()){
if (H.body.size() > 10000){
Box * TestBox = new Box((uint8_t*)H.body.c_str(), H.body.size());
Payload = (char*)TestBox->GetPayload();
printf("First bytes: %2hhu %2hhu %2hhu %2hhu\n", Payload[0], Payload[1], Payload[2], Payload[3]);
P = 0;
while (TestBox->GetPayloadSize() > P){
if (F.MemLoader(Payload, TestBox->GetPayloadSize(), P)){
std::cout << "Got a " << F.len << " bytes " << F.tagType() << " FLV tag of time " << F.tagTime() << "." << std::endl;
}
}
delete TestBox;
}else{
std::cout << "Skipped too small fragment of size " << H.body.size() << std::endl;
}
}
}//main

187
src/analysers/rtmp_analyser.cpp Normal file
View file

@ -0,0 +1,187 @@
/// \file RTMP_Parser/main.cpp
/// Debugging tool for RTMP data.
/// Expects RTMP data of one side of the conversion through stdin, outputs human-readable information to stderr.
/// Automatically skips 3073 bytes of handshake data.
/// Optionally reconstructs an FLV file
/// Singular argument is a bitmask indicating the following (defaulting to 0):
/// - 0 = Info: Output chunk meanings and fulltext commands to stderr.
/// - 1 = Reconstruct: Output valid .flv file to stdout.
/// - 2 = Explicit: Audio/video data details.
/// - 4 = Verbose: details about every whole chunk.
#define DEBUG 10 //maximum debugging level
#include <cstdlib>
#include <iostream>
#include <fstream>
#include <string>
#include <iostream>
#include "../../util/flv_tag.h"
#include "../../util/amf.h"
#include "../../util/rtmpchunks.h"
int Detail = 0;
#define DETAIL_RECONSTRUCT 1
#define DETAIL_EXPLICIT 2
#define DETAIL_VERBOSE 4
/// Debugging tool for RTMP data.
/// Expects RTMP data of one side of the conversion through stdin, outputs human-readable information to stderr.
/// Will output FLV file to stdout, if available
/// Automatically skips 3073 bytes of handshake data.
int main(int argc, char ** argv){
if (argc > 1){
Detail = atoi(argv[1]);
fprintf(stderr, "Detail level set:\n");
if ((Detail & DETAIL_RECONSTRUCT) == DETAIL_RECONSTRUCT){
fprintf(stderr, " - Will reconstuct FLV file to stdout\n");
std::cout.write(FLV::Header, 13);
}
if ((Detail & DETAIL_EXPLICIT) == DETAIL_EXPLICIT){
fprintf(stderr, " - Will list explicit video/audio data information\n");
}
if ((Detail & DETAIL_VERBOSE) == DETAIL_VERBOSE){
fprintf(stderr, " - Will list verbose chunk information\n");
}
}
std::string inbuffer;
while (std::cin.good()){inbuffer += std::cin.get();}//read all of std::cin to temp
inbuffer.erase(0, 3073);//strip the handshake part
RTMPStream::Chunk next;
FLV::Tag F;//FLV holder
AMF::Object amfdata("empty", AMF::AMF0_DDV_CONTAINER);
AMF::Object3 amf3data("empty", AMF::AMF3_DDV_CONTAINER);
while (next.Parse(inbuffer)){
if ((Detail & DETAIL_VERBOSE) == DETAIL_VERBOSE){
fprintf(stderr, "Chunk info: [%#2X] CS ID %u, timestamp %u, len %u, type ID %u, Stream ID %u\n", next.headertype, next.cs_id, next.timestamp, next.len, next.msg_type_id, next.msg_stream_id);
}
switch (next.msg_type_id){
case 0://does not exist
fprintf(stderr, "Error chunk - %i, %i, %i, %i, %i\n", next.cs_id, next.timestamp, next.real_len, next.len_left, next.msg_stream_id);
//return 0;
break;//happens when connection breaks unexpectedly
case 1://set chunk size
RTMPStream::chunk_rec_max = ntohl(*(int*)next.data.c_str());
fprintf(stderr, "CTRL: Set chunk size: %i\n", RTMPStream::chunk_rec_max);
break;
case 2://abort message - we ignore this one
fprintf(stderr, "CTRL: Abort message: %i\n", ntohl(*(int*)next.data.c_str()));
//4 bytes of stream id to drop
break;
case 3://ack
RTMPStream::snd_window_at = ntohl(*(int*)next.data.c_str());
fprintf(stderr, "CTRL: Acknowledgement: %i\n", RTMPStream::snd_window_at);
break;
case 4:{
short int ucmtype = ntohs(*(short int*)next.data.c_str());
switch (ucmtype){
case 0:
fprintf(stderr, "CTRL: User control message: stream begin %u\n", ntohl(*(unsigned int*)(next.data.c_str()+2)));
break;
case 1:
fprintf(stderr, "CTRL: User control message: stream EOF %u\n", ntohl(*(unsigned int*)(next.data.c_str()+2)));
break;
case 2:
fprintf(stderr, "CTRL: User control message: stream dry %u\n", ntohl(*(unsigned int*)(next.data.c_str()+2)));
break;
case 3:
fprintf(stderr, "CTRL: User control message: setbufferlen %u\n", ntohl(*(unsigned int*)(next.data.c_str()+2)));
break;
case 4:
fprintf(stderr, "CTRL: User control message: streamisrecorded %u\n", ntohl(*(unsigned int*)(next.data.c_str()+2)));
break;
case 6:
fprintf(stderr, "CTRL: User control message: pingrequest %u\n", ntohl(*(unsigned int*)(next.data.c_str()+2)));
break;
case 7:
fprintf(stderr, "CTRL: User control message: pingresponse %u\n", ntohl(*(unsigned int*)(next.data.c_str()+2)));
break;
default:
fprintf(stderr, "CTRL: User control message: UNKNOWN %hu - %u\n", ucmtype, ntohl(*(unsigned int*)(next.data.c_str()+2)));
break;
}
} break;
case 5://window size of other end
RTMPStream::rec_window_size = ntohl(*(int*)next.data.c_str());
RTMPStream::rec_window_at = RTMPStream::rec_cnt;
fprintf(stderr, "CTRL: Window size: %i\n", RTMPStream::rec_window_size);
break;
case 6:
RTMPStream::snd_window_size = ntohl(*(int*)next.data.c_str());
//4 bytes window size, 1 byte limit type (ignored)
fprintf(stderr, "CTRL: Set peer bandwidth: %i\n", RTMPStream::snd_window_size);
break;
case 8:
if (Detail & (DETAIL_EXPLICIT | DETAIL_RECONSTRUCT)){
F.ChunkLoader(next);
if ((Detail & DETAIL_EXPLICIT) == DETAIL_EXPLICIT){
fprintf(stderr, "Received %i bytes audio data\n", next.len);
std::cerr << "Got a " << F.len << " bytes " << F.tagType() << " FLV tag of time " << F.tagTime() << "." << std::endl;
}
if ((Detail & DETAIL_RECONSTRUCT) == DETAIL_RECONSTRUCT){
std::cout.write(F.data, F.len);
}
}
break;
case 9:
if (Detail & (DETAIL_EXPLICIT | DETAIL_RECONSTRUCT)){
F.ChunkLoader(next);
if ((Detail & DETAIL_EXPLICIT) == DETAIL_EXPLICIT){
fprintf(stderr, "Received %i bytes video data\n", next.len);
std::cerr << "Got a " << F.len << " bytes " << F.tagType() << " FLV tag of time " << F.tagTime() << "." << std::endl;
}
if ((Detail & DETAIL_RECONSTRUCT) == DETAIL_RECONSTRUCT){
std::cout.write(F.data, F.len);
}
}
break;
case 15:
fprintf(stderr, "Received AFM3 data message\n");
break;
case 16:
fprintf(stderr, "Received AFM3 shared object\n");
break;
case 17:{
fprintf(stderr, "Received AFM3 command message:\n");
char soort = next.data[0];
next.data = next.data.substr(1);
if (soort == 0){
amfdata = AMF::parse(next.data);
std::cerr << amfdata.Print() << std::endl;
}else{
amf3data = AMF::parse3(next.data);
std::cerr << amf3data.Print() << std::endl;
}
} break;
case 18:{
fprintf(stderr, "Received AFM0 data message (metadata):\n");
amfdata = AMF::parse(next.data);
amfdata.Print();
if ((Detail & DETAIL_RECONSTRUCT) == DETAIL_RECONSTRUCT){
F.ChunkLoader(next);
std::cout.write(F.data, F.len);
}
} break;
case 19:
fprintf(stderr, "Received AFM0 shared object\n");
break;
case 20:{//AMF0 command message
fprintf(stderr, "Received AFM0 command message:\n");
amfdata = AMF::parse(next.data);
std::cerr << amfdata.Print() << std::endl;
} break;
case 22:
fprintf(stderr, "Received aggregate message\n");
break;
default:
fprintf(stderr, "Unknown chunk received! Probably protocol corruption, stopping parsing of incoming data.\n");
return 1;
break;
}//switch for type of chunk
}//while chunk parsed
fprintf(stderr, "No more readable data\n");
return 0;
}//main

63
src/analysers/rtp_analyser.cpp Normal file
View file

@ -0,0 +1,63 @@
#include <iostream>
#include <cstdio>
///A struct that will contain all data stored in a RTP Header
struct RTP_Header {
char Version;
bool Padding;
bool Extension;
char CSRC_Count;
bool Marker;
char Payload_Type;
int Sequence_Number;
int Timestamp;
int SSRC;
};//RTP_Header
///Fills a RTP Header
///\param hdr A RTP Header structure
///\param Header A characterpointer to an RTP packet
///\param HeaderSize the expected length of the header
void Read_Header( RTP_Header & hdr, char * Header, int HeaderSize ) {
hdr.Version = ( Header[0] & 0xC0 ) >> 6;
hdr.Padding = ( Header[0] & 0x20 ) >> 5;
hdr.Extension = ( Header[0] & 0x10 ) >> 4;
hdr.CSRC_Count = ( Header[0] & 0x0F );
hdr.Marker = ( Header[1] & 0x80 ) >> 7;
hdr.Payload_Type = ( Header[1] & 0x7F );
hdr.Sequence_Number = ( ( ( Header[2] ) << 8 ) + ( Header[3] ) ) & 0x0000FFFF;
hdr.Timestamp = ( ( Header[4] ) << 24 ) + ( ( Header[5] ) << 16 ) + ( ( Header[6] ) << 8 ) + ( Header[7] );
hdr.SSRC = ( ( Header[8] ) << 24 ) + ( ( Header[9] ) << 16 ) + ( ( Header[10] ) << 8 ) + ( Header[11] );
}
///Prints a RTP header
///\param hdr The RTP Header
void Print_Header( RTP_Header hdr ) {
printf( "RTP Header:\n" );
printf( "\tVersion:\t\t%d\n", hdr.Version );
printf( "\tPadding:\t\t%d\n", hdr.Padding );
printf( "\tExtension:\t\t%d\n", hdr.Extension );
printf( "\tCSRC Count:\t\t%d\n", hdr.CSRC_Count );
printf( "\tMarker:\t\t\t%d\n", hdr.Marker );
printf( "\tPayload Type:\t\t%d\n", hdr.Payload_Type );
printf( "\tSequence Number:\t%d\n", hdr.Sequence_Number );
printf( "\tTimestamp:\t\t%u\n", hdr.Timestamp );
printf( "\tSSRC:\t\t\t%u\n", hdr.SSRC );
}
int main( ) {
int HeaderSize = 12;
char Header[ HeaderSize ];
for( int i = 0; i < HeaderSize; i++ ) {
if( !std::cin.good() ) { break; }
Header[ i ] = std::cin.get();
}
RTP_Header hdr;
Read_Header( hdr, Header, HeaderSize );
Print_Header( hdr );
return 0;
}

501
src/analysers/ts_analyser.cpp Normal file
View file

@ -0,0 +1,501 @@
/// \file TS_Analyser/main.cpp
/// Contains the code for the TS Analyser
#include <iostream>
#include <cstdlib>
#include <cstdio>
#include <string>
#include <vector>
#include <fstream>
/// Contains all data unique to a single entry in the PAT
struct program_association_table_entry {
unsigned int Program_Number;///< Number of the program adressed
unsigned char Reserved;
unsigned int Program_Map_PID;///< PID of the map associated with this program
};
/// The program association table ( PAT )
struct program_association_table {
unsigned char Pointer_Field;///< A single padding character
unsigned char Table_ID;///< ID of this table
bool Section_Syntax_Indicator;///< Indicates whether the payload confirms to specification, or is private
bool Zero;
unsigned char Reserved_1;
unsigned int Section_Length;///< Length of this section of the PAT
unsigned int Transport_Stream_ID;///< ID of the stream
unsigned char Reserved_2;
unsigned char Version_Number;///< Version of this section
bool Current_Next_Indicator;///< Currently applicable
unsigned char Section_Number;///< Number of this section
unsigned char Last_Section_Number;///< Amount of sections in the complete table
std::vector<program_association_table_entry> Entries;
unsigned int CRC_32;
};
/// An entry of the PMT
struct program_mapping_table_entry {
unsigned char Stream_Type;///< Type of stream we encounter
unsigned char Reserved_1;
unsigned int Elementary_PID;///< PID of the packages carying the elementary stream for this entry
unsigned char Reserved_2;
unsigned int ES_Info_Length;///< Length of extra info. Not needed for understanding the file
};
/// The program mapping table ( PMT )
struct program_mapping_table {
unsigned char Pointer_Field;///< A single padding character
unsigned char Table_ID;///< ID of this table
bool Section_Syntax_Indicator;///< Indicates whether the payload confirms to specification, or is private
bool Zero;
unsigned char Reserved_1;
unsigned int Section_Length;///< Length of this section
unsigned int Program_Number;///< Program number in stream
unsigned char Reserved_2;
unsigned char Version_Number;///< Version of this section
bool Current_Next_Indicator;///< Currently applicable
unsigned char Section_Number;///< Number of this section
unsigned char Last_Section_Number;///< Amount of sections in PMT
unsigned char Reserved_3;
unsigned int PCR_PID;///< PID of the packets that contain Program Counter References
unsigned char Reserved_4;
unsigned int Program_Info_Length;///< Length of the program descriptors. Skip for analysis
//vector Descriptors
std::vector<program_mapping_table_entry> Entries;
unsigned int CRC_32;
};
/// The adaptation field
struct adaptation_field {
unsigned char Adaptation_Field_Length;///Lenght of the complete field, greater or equal to 0
bool Discontinuity_Indicator;
bool Random_Access_Indicator;
bool Elementary_Stream_Priority_Indicator;
bool PCR_Flag;///< PCR Field existent
bool OPCR_Flag;///< OPCR Field existent
bool Splicing_Point_Flag;
bool Transport_Private_Data_Flag;
bool Adaptation_Field_Extension_Flag;
unsigned char Program_Clock_Reference_Base_MSB;///< Most significant bit for the Base value of the PCR
unsigned int Program_Clock_Reference_Base;///< Least significant 32 bits for the Base value of the PCR
unsigned char PCR_Reserved;
unsigned int Program_Clock_Reference_Extension;///< Extension of the PCR
unsigned char Original_Program_Clock_Reference_Base_MSB;///< Most significant bit for the Base value of the OPCR
unsigned int Original_Program_Clock_Reference_Base;///< Least significant 32 bits for the Base value of the OPCR
unsigned char OPCR_Reserved;
unsigned int Original_Program_Clock_Reference_Extension;///< Extension of the OPCR
};
/// The general structure of a PES packet
struct pes_packet {
unsigned int Packet_Start_Code_Prefix;///< Prefix, should be 0x000001
unsigned char Stream_ID;///< ID of the current stream
unsigned int PES_Packet_Length;///< Length of the PES packet
unsigned char Two;
unsigned char PES_Scrambling_Control;
bool PES_Priority;
bool Data_Alignment_Indicator;
bool Copyright;
bool Original_Or_Copy;
unsigned char PTS_DTS_Flags;///Presentation Time Stamp and/or Display Time Stamp available
bool ESCR_Flag;
bool ES_Rate_Flag;
bool DSM_Trick_Mode_Flag;
bool Additional_Copy_Info_Flag;
bool PES_CRC_Flag;
bool PES_Extension_Flag;
unsigned char PES_Header_Data_Length;///< Length of the header
unsigned char PTS_Spacer;///< Spacer, value depends on the flag
unsigned char PTS_MSB;///< Most significant bit of the PTS
unsigned int PTS;///< Least significant 32 bits of the PTS
unsigned char DTS_Spacer;///< Spacer, value depends on the flag
unsigned char DTS_MSB;///< Most significant bit of the DTS
unsigned int DTS;///< Least significant 32 bits of the DTS
std::vector<unsigned char> Header_Stuffing;///< Header stuffing, if present
std::vector<unsigned char> First_Bytes;///< Storage capacity for the first few bytes
};
/// Fills a PES Packet
/// \param PES The packet in which the data should be stored
/// \param TempChar The current TS packet data
/// \param Offset Offset of the PES data, changed if adaptation field exists
void fill_pes( pes_packet & PES, unsigned char * TempChar, int Offset = 4 ) {
PES.Packet_Start_Code_Prefix = ( TempChar[Offset] << 16 ) + ( TempChar[Offset+1] << 8 ) + TempChar[Offset+2];
PES.Stream_ID = TempChar[Offset+3];
PES.PES_Packet_Length = ( TempChar[Offset+4] << 8 ) + TempChar[Offset+5];
Offset += 6;
if( true ) { //Always for streams yet encountered
PES.Two = ( TempChar[Offset] & 0xC0 ) >> 6;
PES.PES_Scrambling_Control = ( TempChar[Offset] & 0x30 ) >> 4;
PES.PES_Priority = ( TempChar[Offset] & 0x08 ) >> 3;
PES.Data_Alignment_Indicator = ( TempChar[Offset] & 0x04 ) >> 2;
PES.Copyright = ( TempChar[Offset] & 0x02 ) >> 1;
PES.Original_Or_Copy = ( TempChar[Offset] & 0x01 );
Offset ++;
PES.PTS_DTS_Flags = ( TempChar[Offset] & 0xC0 ) >> 6;
PES.ESCR_Flag = ( TempChar[Offset] & 0x20 ) >> 5;
PES.ES_Rate_Flag = ( TempChar[Offset] & 0x10 ) >> 4;
PES.DSM_Trick_Mode_Flag = ( TempChar[Offset] & 0x08 ) >> 3;
PES.Additional_Copy_Info_Flag = ( TempChar[Offset] & 0x04 ) >> 2;
PES.PES_CRC_Flag = ( TempChar[Offset] & 0x02 ) >> 1;
PES.PES_Extension_Flag = ( TempChar[Offset] & 0x01 );
Offset ++;
PES.PES_Header_Data_Length = TempChar[Offset];
Offset ++;
int HeaderStart = Offset;
if( PES.PTS_DTS_Flags >= 2 ) {
PES.PTS_Spacer = ( TempChar[Offset] & 0xF0 ) >> 4;
PES.PTS_MSB = ( TempChar[Offset] & 0x08 ) >> 3;
PES.PTS = ( ( TempChar[Offset] & 0x06 ) << 29 );
PES.PTS += ( TempChar[Offset+1] ) << 22;
PES.PTS += ( ( TempChar[Offset+2] ) & 0xFE ) << 14;
PES.PTS += ( TempChar[Offset+3] ) << 7;
PES.PTS += ( ( TempChar[Offset+4] & 0xFE ) >> 1 );
Offset += 5;
}
if( PES.PTS_DTS_Flags == 3 ) {
PES.DTS_Spacer = ( TempChar[Offset] & 0xF0 ) >> 4;
PES.DTS_MSB = ( TempChar[Offset] & 0x08 ) >> 3;
PES.DTS = ( ( TempChar[Offset] & 0x06 ) << 29 );
PES.DTS += ( TempChar[Offset+1] ) << 22;
PES.DTS += ( ( TempChar[Offset+2] ) & 0xFE ) << 14;
PES.DTS += ( TempChar[Offset+3] ) << 7;
PES.DTS += ( ( TempChar[Offset+4] & 0xFE ) >> 1 );
Offset += 5;
}
PES.Header_Stuffing.clear();
while( Offset < HeaderStart + PES.PES_Header_Data_Length ) {
PES.Header_Stuffing.push_back( TempChar[Offset] );
Offset ++;
}
PES.First_Bytes.clear();
for( int i = 0; i < 30; i ++ ) {
PES.First_Bytes.push_back( TempChar[Offset+i] );
}
}
}
/// Prints a PES packet to STDOUT
/// \param PES The packet to be print
/// \param offset A string indicating the indentation of the outputed data
void print_pes( pes_packet PES, std::string offset="\t" ) {
printf( "%sPES Header\n", offset.c_str() );
printf( "%s\tPacket Start Code Prefix\t%.6X\n", offset.c_str(), PES.Packet_Start_Code_Prefix );
printf( "%s\tStream ID\t\t\t%X\n", offset.c_str(), PES.Stream_ID );
printf( "%s\tPES Packet Length\t\t%X\n", offset.c_str(), PES.PES_Packet_Length );
if( true ) { //Always for streams yet encountered
printf( "%s\tTwo:\t\t\t\t%d\n", offset.c_str(), PES.Two );
printf( "%s\tPES Scrambling Control:\t\t%d\n", offset.c_str(), PES.PES_Scrambling_Control );
printf( "%s\tPES Priority:\t\t\t%d\n", offset.c_str(), PES.PES_Priority );
printf( "%s\tData Alignment Indicator:\t%d\n", offset.c_str(), PES.Data_Alignment_Indicator );
printf( "%s\tCopyright:\t\t\t%d\n", offset.c_str(), PES.Copyright );
printf( "%s\tOriginal Or Copy:\t\t%d\n", offset.c_str(), PES.Original_Or_Copy );
printf( "%s\tPTS DTS Flags:\t\t\t%d\n", offset.c_str(), PES.PTS_DTS_Flags );
printf( "%s\tESCR Flag:\t\t\t%d\n", offset.c_str(), PES.ESCR_Flag );
printf( "%s\tES Rate Flag:\t\t\t%d\n", offset.c_str(), PES.ES_Rate_Flag );
printf( "%s\tDSM Trick Mode Flag:\t\t%d\n", offset.c_str(), PES.DSM_Trick_Mode_Flag );
printf( "%s\tAdditional Copy Info Flag:\t%d\n", offset.c_str(), PES.Additional_Copy_Info_Flag );
printf( "%s\tPES CRC Flag:\t\t\t%d\n", offset.c_str(), PES.PES_CRC_Flag );
printf( "%s\tPES Extension Flag:\t\t%d\n", offset.c_str(), PES.PES_Extension_Flag );
printf( "%s\tPES Header Data Length:\t\t%d\n", offset.c_str(), PES.PES_Header_Data_Length );
if( PES.PTS_DTS_Flags >= 2 ) {
printf( "%s\tPTS Spacer\t\t\t%d\n", offset.c_str(), PES.PTS_Spacer );
printf( "%s\tPTS\t\t\t\t%X%.8X\n", offset.c_str(), PES.PTS_MSB, PES.PTS );
}
if( PES.PTS_DTS_Flags == 3 ) {
printf( "%s\tDTS Spacer\t\t\t%d\n", offset.c_str(), PES.DTS_Spacer );
printf( "%s\tDTS\t\t\t\t%X%.8X\n", offset.c_str(), PES.DTS_MSB, PES.DTS );
}
printf( "%s\tHeader Stuffing\t\t\t", offset.c_str() );
for( int i = 0; i < PES.Header_Stuffing.size(); i++ ) {
printf( "%.2X ", PES.Header_Stuffing[i] );
}
printf( "\n" );
printf( "%s\tFirst_Bytes\t\t\t", offset.c_str() );
for( int i = 0; i < PES.First_Bytes.size(); i++ ) {
printf( "%.2X ", PES.First_Bytes[i] );
}
printf( "\n" );
}
}
/// Fills a PAT structure with the right data
/// \param PAT the structure to be filled
/// \param TempChar The TS packet data
void fill_pat( program_association_table & PAT, unsigned char * TempChar ) {
PAT.Pointer_Field = TempChar[4];
PAT.Table_ID = TempChar[5];
PAT.Section_Syntax_Indicator = ((TempChar[6] & 0x80 ) != 0 );
PAT.Zero = (( TempChar[6] & 0x40 ) != 0 );
PAT.Reserved_1 = (( TempChar[6] & 0x30 ) >> 4 );
PAT.Section_Length = (( TempChar[6] & 0x0F ) << 8 ) + TempChar[7];
PAT.Transport_Stream_ID = (( TempChar[8] << 8 ) + TempChar[9] );
PAT.Reserved_2 = (( TempChar[10] & 0xC0 ) >> 6 );
PAT.Version_Number = (( TempChar[10] & 0x01 ) >> 1 );
PAT.Current_Next_Indicator = (( TempChar[10] & 0x01 ) != 0 );
PAT.Section_Number = TempChar[11];
PAT.Last_Section_Number = TempChar[12];
PAT.Entries.clear( );
for( int i = 0; i < PAT.Section_Length - 9; i += 4 ) {
program_association_table_entry PAT_Entry;
PAT_Entry.Program_Number = ( TempChar[13+i] << 8 ) + TempChar[14+i];
PAT_Entry.Reserved = ( TempChar[15+i] & 0xE0 ) >> 5;
PAT_Entry.Program_Map_PID = (( TempChar[15+i] & 0x1F ) << 8 ) + TempChar[16+i];
PAT.Entries.push_back( PAT_Entry );
}
PAT.CRC_32 = ( TempChar[8+PAT.Section_Length-4] << 24 ) + ( TempChar[8+PAT.Section_Length-3] << 16 ) + ( TempChar[8+PAT.Section_Length-2] << 8 ) + ( TempChar[8+PAT.Section_Length-1] );
}
/// Prints a PAT to STDOUT
/// \param PAT The table to be print
/// \param offset A string indicating the indentation of the outputed data
void print_pat( program_association_table PAT, bool Pointer_Field = false, std::string offset="\t" ) {
printf( "%sProgram Association Table\n", offset.c_str() );
if( Pointer_Field ) {
printf( "%s\tPointer Field:\t\t\t%X\n", offset.c_str(), PAT.Pointer_Field );
}
printf( "%s\tTable ID:\t\t\t%X\n", offset.c_str(), PAT.Table_ID );
printf( "%s\tSection Syntax Indicator:\t%d\n", offset.c_str(), PAT.Section_Syntax_Indicator );
printf( "%s\t0:\t\t\t\t%d\n", offset.c_str(), PAT.Zero );
printf( "%s\tReserved:\t\t\t%d\n", offset.c_str(), PAT.Reserved_1 );
printf( "%s\tSection Length:\t\t\t%X\n", offset.c_str(), PAT.Section_Length );
printf( "%s\tTransport Stream ID\t\t%X\n", offset.c_str(), PAT.Transport_Stream_ID );
printf( "%s\tReserved:\t\t\t%d\n", offset.c_str(), PAT.Reserved_2 );
printf( "%s\tVersion Number:\t\t\t%X\n", offset.c_str(), PAT.Version_Number );
printf( "%s\tCurrent Next Indicator:\t\t%d\n", offset.c_str(), PAT.Current_Next_Indicator );
printf( "%s\tSection Number:\t\t\t%X\n", offset.c_str(), PAT.Section_Number );
printf( "%s\tLast Section Number:\t\t%d\n\n", offset.c_str(), PAT.Last_Section_Number );
for( int i = 0; i < PAT.Entries.size(); i++ ) {
printf( "%s\tEntry %d\n", offset.c_str(), i );
printf( "%s\t\tProgram Number:\t\t%X\n", offset.c_str(), PAT.Entries[i].Program_Number );
printf( "%s\t\tReserved:\t\t%X\n", offset.c_str(), PAT.Entries[i].Reserved );
printf( "%s\t\tProgram Map PID:\t%X\n", offset.c_str(), PAT.Entries[i].Program_Map_PID );
}
printf( "\n%s\tCRC_32:\t\t\t\t%X\n", offset.c_str(), PAT.CRC_32 );
}
/// Fills a PMT structure with the right data
/// \param PMT the structure to be filled
/// \param TempChar The TS packet data
void fill_pmt( program_mapping_table & PMT, unsigned char * TempChar ) {
int CurrentOffset;
PMT.Pointer_Field = TempChar[4];
PMT.Table_ID = TempChar[5];
PMT.Section_Syntax_Indicator = (( TempChar[6] & 0x80 ) != 0 );
PMT.Zero = (( TempChar[6] & 0x40 ) != 0 );
PMT.Reserved_1 = (( TempChar[6] & 0x30 ) >> 4 );
PMT.Section_Length = (( TempChar[6] & 0x0F ) << 8 ) + TempChar[7];
PMT.Program_Number = (TempChar[8] << 8) + TempChar[9];
PMT.Reserved_2 = (( TempChar[10] & 0xC0 ) >> 6 );
PMT.Version_Number = (( TempChar[10] & 0x1E ) >> 1 );
PMT.Current_Next_Indicator = ( TempChar[10] & 0x01 );
PMT.Section_Number = TempChar[11];
PMT.Last_Section_Number = TempChar[12];
PMT.Reserved_3 = (( TempChar[13] & 0xE0 ) >> 5 );
PMT.PCR_PID = (( TempChar[13] & 0x1F ) << 8 ) + TempChar[14];
PMT.Reserved_4 = (( TempChar[15] & 0xF0 ) >> 4 );
PMT.Program_Info_Length = ((TempChar[15] & 0x0F ) << 8 ) + TempChar[16];
CurrentOffset = 17 + PMT.Program_Info_Length;
PMT.Entries.clear( );
while( CurrentOffset < PMT.Section_Length - 8 ) {
program_mapping_table_entry PMT_Entry;
PMT_Entry.Stream_Type = TempChar[CurrentOffset];
PMT_Entry.Reserved_1 = (( TempChar[CurrentOffset+1] & 0xE0 ) >> 5 );
PMT_Entry.Elementary_PID = (( TempChar[CurrentOffset+1] & 0x1F ) << 8 ) + TempChar[CurrentOffset+2];
PMT_Entry.Reserved_2 = (( TempChar[CurrentOffset+3] & 0xF0 ) >> 4 );
PMT_Entry.ES_Info_Length = (( TempChar[CurrentOffset+3] & 0x0F ) << 8 ) + TempChar[CurrentOffset+4];
PMT.Entries.push_back( PMT_Entry );
CurrentOffset += 4 + PMT_Entry.ES_Info_Length;
}
PMT.CRC_32 = ( TempChar[CurrentOffset] << 24 ) + ( TempChar[CurrentOffset+1] << 16 ) + ( TempChar[CurrentOffset+2] << 8 ) + ( TempChar[CurrentOffset+3] );
}
/// Prints a PMT to STDOUT
/// \param PMT The table to be print
/// \param offset A string indicating the indentation of the outputed data
void print_pmt( program_mapping_table PMT, bool Pointer_Field = false, std::string offset="\t" ) {
if( Pointer_Field ) {
printf( "%s\tPointer Field:\t\t\t%X\n", offset.c_str(), PMT.Pointer_Field );
}
printf( "%s\tTable ID:\t\t\t%X\n", offset.c_str(), PMT.Table_ID );
printf( "%s\tSection Syntax Indicator:\t%d\n", offset.c_str(), PMT.Section_Syntax_Indicator);
printf( "%s\t0:\t\t\t\t%d\n", offset.c_str(), PMT.Zero );
printf( "%s\tReserved:\t\t\t%d\n", offset.c_str(), PMT.Reserved_1 );
printf( "%s\tSection Length:\t\t\t%X\n", offset.c_str(), PMT.Section_Length );
printf( "%s\tProgram Number:\t\t\t%X\n", offset.c_str(), PMT.Program_Number );
printf( "%s\tReserved:\t\t\t%d\n", offset.c_str(), PMT.Reserved_2 );
printf( "%s\tVersion Number:\t\t\t%d\n", offset.c_str(), PMT.Version_Number );
printf( "%s\tCurrent_Next_Indicator:\t\t%d\n", offset.c_str(), PMT.Current_Next_Indicator );
printf( "%s\tSection Number:\t\t\t%d\n", offset.c_str(), PMT.Section_Number );
printf( "%s\tLast Section Number:\t\t%d\n", offset.c_str(), PMT.Last_Section_Number );
printf( "%s\tReserved:\t\t\t%d\n", offset.c_str(), PMT.Reserved_3 );
printf( "%s\tPCR PID:\t\t\t%X\n", offset.c_str(), PMT.PCR_PID );
printf( "%s\tReserved:\t\t\t%d\n", offset.c_str(), PMT.Reserved_4 );
printf( "%s\tProgram Info Length:\t\t%d\n", offset.c_str(), PMT.Program_Info_Length );
printf( "%s\tProgram Descriptors Go Here\n\n" );
for( int i = 0; i < PMT.Entries.size(); i++ ) {
printf( "%s\tEntry %d:\n", offset.c_str(), i );
printf( "%s\t\tStream Type\t\t%d\n", offset.c_str(), PMT.Entries[i].Stream_Type );
printf( "%s\t\tReserved\t\t%d\n", offset.c_str(), PMT.Entries[i].Reserved_1 );
printf( "%s\t\tElementary PID\t\t%X\n", offset.c_str(), PMT.Entries[i].Elementary_PID );
printf( "%s\t\tReserved\t\t%d\n", offset.c_str(), PMT.Entries[i].Reserved_2 );
printf( "%s\t\tES Info Length\t\t%d\n", offset.c_str(), PMT.Entries[i].ES_Info_Length );
}
printf( "%s\tCRC 32\t\t%8X\n", offset.c_str(), PMT.CRC_32 );
}
/// Fills an AF structure with the right data
/// \param AF the structure to be filled
/// \param TempChar The TS packet data
void fill_af( adaptation_field & AF, unsigned char * TempChar ) {
AF.Adaptation_Field_Length = TempChar[4];
AF.Discontinuity_Indicator = (( TempChar[5] & 0x80 ) >> 7 );
AF.Random_Access_Indicator = (( TempChar[5] & 0x40 ) >> 6 );
AF.Elementary_Stream_Priority_Indicator = (( TempChar[5] & 0x20 ) >> 5 );
AF.PCR_Flag = (( TempChar[5] & 0x10 ) >> 4 );
AF.OPCR_Flag = (( TempChar[5] & 0x08 ) >> 3 );
AF.Splicing_Point_Flag = (( TempChar[5] & 0x04 ) >> 2 );
AF.Transport_Private_Data_Flag = (( TempChar[5] & 0x02 ) >> 1 );
AF.Adaptation_Field_Extension_Flag = (( TempChar[5] & 0x01 ) );
int CurrentOffset = 6;
if( AF.PCR_Flag ) {
AF.Program_Clock_Reference_Base_MSB = ( ( ( TempChar[CurrentOffset] ) & 0x80 ) >> 7 );
AF.Program_Clock_Reference_Base = ( ( ( TempChar[CurrentOffset] ) & 0x7F ) << 25 );
AF.Program_Clock_Reference_Base += ( ( TempChar[CurrentOffset+1] ) << 17 );
AF.Program_Clock_Reference_Base += ( ( TempChar[CurrentOffset+2] ) << 9 );
AF.Program_Clock_Reference_Base += ( ( TempChar[CurrentOffset+3] ) << 1 );
AF.Program_Clock_Reference_Base += ( ( ( TempChar[CurrentOffset+4] ) & 0x80 ) >> 7 );
AF.PCR_Reserved = ( ( TempChar[CurrentOffset+4] ) & 0x7E ) >> 1;
AF.Program_Clock_Reference_Extension = ( ( TempChar[CurrentOffset+4] ) & 0x01 ) << 8 + TempChar[CurrentOffset+5];
CurrentOffset += 6;
}
}
/// Prints an AF to STDOUT
/// \param AF The Adaptation Field to be print
/// \param offset A string indicating the indentation of the outputed data
void print_af( adaptation_field AF, std::string offset="\t" ) {
printf( "%sAdaptation Field\n", offset.c_str() );
printf( "%s\tAdaptation Field Length\t\t\t%X\n", offset.c_str(), AF.Adaptation_Field_Length );
printf( "%s\tDiscontinuity Indicator\t\t\t%X\n", offset.c_str(), AF.Discontinuity_Indicator );
printf( "%s\tRandom Access Indicator\t\t\t%X\n", offset.c_str(), AF.Random_Access_Indicator );
printf( "%s\tElementary Stream Priority Indicator\t%X\n", offset.c_str(), AF.Elementary_Stream_Priority_Indicator );
printf( "%s\tPCR Flag\t\t\t\t%X\n", offset.c_str(), AF.PCR_Flag );
printf( "%s\tOPCR Flag\t\t\t\t%X\n", offset.c_str(), AF.OPCR_Flag );
printf( "%s\tSplicing Point Flag\t\t\t%X\n", offset.c_str(), AF.Splicing_Point_Flag );
printf( "%s\tTransport Private Data Flag\t\t%X\n", offset.c_str(), AF.Transport_Private_Data_Flag );
printf( "%s\tAdaptation Field Extension Flag\t\t%X\n", offset.c_str(), AF.Adaptation_Field_Extension_Flag );
if( AF.PCR_Flag ) {
printf( "\n%s\tProgram Clock Reference Base\t\t%X%.8X\n", offset.c_str(), AF.Program_Clock_Reference_Base_MSB, AF.Program_Clock_Reference_Base );
printf( "%s\tReserved\t\t\t\t%d\n", offset.c_str(), AF.PCR_Reserved );
printf( "%s\tProgram Clock Reference Extension\t%X\n", offset.c_str(), AF.Program_Clock_Reference_Extension );
}
}
/// Locates a Packet ID in the PAT
/// \param PAT The PAT to look in
/// \param PID The PID to check for existense
/// \return The program number of the PAT, or -1 if not found
int find_pid_in_pat( program_association_table PAT, unsigned int PID ) {
for( int i = 0; i < PAT.Entries.size(); i++ ) {
if( PAT.Entries[i].Program_Map_PID == PID ) {
return PAT.Entries[i].Program_Number;
}
}
return -1;
}
/// Checks whether a packet is part of an elementary stream
/// \param PMT The program mapping table
/// \param PID The PID of the packet
/// \return PID is found in the elementary streams of PMT
bool is_elementary_pid( program_mapping_table PMT, unsigned int PID ) {
for( int i = 0; i < PMT.Entries.size(); i++ ) {
if( PMT.Entries[i].Elementary_PID == PID ) {
return true;
}
}
return false;
}
/// The main function of the analyser
int main( ) {
std::string File;
unsigned int BlockNo = 1;
unsigned int EmptyBlocks = 0;
unsigned char TempChar[188];
unsigned char Skip;
unsigned int SkippedBytes = 0;
unsigned int Adaptation;
program_association_table PAT;
program_mapping_table PMT;
adaptation_field AF;
pes_packet PES;
int ProgramNum;
std::ofstream outfile;
outfile.open( "out.ts" );
while( std::cin.good( ) && BlockNo <= 10000 ) {
for( int i = 0; i < 188; i++ ) {
if( std::cin.good( ) ){ TempChar[i] = std::cin.get(); }
}
int PID = ( ( TempChar[1] & 0x1F ) << 8 ) + ( TempChar[2] );
if( true ) {
printf( "Block %d:\n", BlockNo );
printf( "\tSync Byte:\t\t\t%X\n", TempChar[0] );
printf( "\tTransport Error Indicator:\t%d\n", ( ( TempChar[1] & 0x80 ) != 0 ) );
printf( "\tPayload Unit Start Indicator:\t%d\n", ( ( TempChar[1] & 0x40 ) != 0 ) );
printf( "\tTransport Priority:\t\t%d\n", ( ( TempChar[1] & 0x20 ) != 0 ) );
printf( "\tPID:\t\t\t\t%X\n", ( ( TempChar[1] & 0x1F ) << 8 ) + ( TempChar[2] ) );
printf( "\tScrambling control:\t\t%d\n", ( ( TempChar[3] & 0xC0 ) >> 6 ) );
printf( "\tAdaptation Field Exists:\t%d\n", ( ( TempChar[3] & 0x30 ) >> 4 ) );
printf( "\tContinuity Counter:\t\t%X\n", ( TempChar[3] & 0x0F ) );
Adaptation = ( ( TempChar[3] & 0x30 ) >> 4 );
//Adaptation Field Exists
if( Adaptation == 2 || Adaptation == 3 ) {
fprintf( stderr, "Block: %d -> Adaptation == %d\n", BlockNo, Adaptation );
fill_af( AF, TempChar );
print_af( AF );
}
if( ( ( ( TempChar[1] & 0x1F ) << 8 ) + TempChar[2] ) == 0 ) {
fill_pat( PAT, TempChar );
print_pat( PAT, true );
}
ProgramNum = find_pid_in_pat( PAT, ( ( TempChar[1] & 0x1F ) << 8 ) + ( TempChar[2] ) );
if( ProgramNum != -1 ) {
printf( "\tProgram Mapping Table for program %X\n", ProgramNum );
fill_pmt( PMT, TempChar );
print_pmt( PMT, true );
}
if( ( ( TempChar[1] & 0x40 ) ) && ( ( TempChar[1] & 0x1F ) << 8 ) + ( TempChar[2] ) ) {
fill_pes( PES, TempChar, ( Adaptation == 3 ? 5 + TempChar[4] : 4 ) );
print_pes( PES );
}
BlockNo ++;
} else {
EmptyBlocks ++;
}
//Find Next Sync Byte
SkippedBytes = 0;
while( (int)std::cin.peek( ) != 0x47 ) {
std::cin >> Skip;
SkippedBytes ++;
}
}
return 0;
}

226
src/buffer.cpp Normal file
View file

@ -0,0 +1,226 @@
/// \file Buffer/main.cpp
/// Contains the main code for the Buffer.
#include <fcntl.h>
#include <iostream>
#include <string>
#include <vector>
#include <cstdlib>
#include <cstdio>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <sstream>
#include <sys/time.h>
#include "buffer_stream.h"
/// Holds all code unique to the Buffer.
namespace Buffer{
volatile bool buffer_running = true; ///< Set to false when shutting down.
Stream * thisStream = 0;
Socket::Server SS; ///< The server socket.
/// Gets the current system time in milliseconds.
unsigned int getNowMS(){
timeval t;
gettimeofday(&t, 0);
return t.tv_sec + t.tv_usec/1000;
}//getNowMS
///A simple signal handler that ignores all signals.
void termination_handler (int signum){
switch (signum){
case SIGKILL: buffer_running = false; break;
case SIGPIPE: return; break;
default: return; break;
}
}
void handleStats(void * empty){
if (empty != 0){return;}
Socket::Connection StatsSocket = Socket::Connection("/tmp/mist/statistics", true);
while (buffer_running){
usleep(1000000); //sleep one second
if (!StatsSocket.connected()){
StatsSocket = Socket::Connection("/tmp/mist/statistics", true);
}
if (StatsSocket.connected()){
StatsSocket.write(Stream::get()->getStats()+"\n\n");
}
}
}
void handleUser(void * v_usr){
user * usr = (user*)v_usr;
std::cerr << "Thread launched for user " << usr->MyStr << ", socket number " << usr->S.getSocket() << std::endl;
usr->myRing = thisStream->getRing();
if (!usr->S.write(thisStream->getHeader())){
usr->Disconnect("failed to receive the header!");
return;
}
while (usr->S.connected()){
usleep(5000); //sleep 5ms
if (usr->S.canRead()){
usr->inbuffer.clear();
char charbuf;
while ((usr->S.iread(&charbuf, 1) == 1) && charbuf != '\n' ){
usr->inbuffer += charbuf;
}
if (usr->inbuffer != ""){
if (usr->inbuffer[0] == 'P'){
std::cout << "Push attempt from IP " << usr->inbuffer.substr(2) << std::endl;
if (thisStream->checkWaitingIP(usr->inbuffer.substr(2))){
if (thisStream->setInput(usr->S)){
std::cout << "Push accepted!" << std::endl;
usr->S = Socket::Connection(-1);
return;
}else{
usr->Disconnect("Push denied - push already in progress!");
}
}else{
usr->Disconnect("Push denied - invalid IP address!");
}
}
if (usr->inbuffer[0] == 'S'){
usr->tmpStats = Stats(usr->inbuffer.substr(2));
unsigned int secs = usr->tmpStats.conntime - usr->lastStats.conntime;
if (secs < 1){secs = 1;}
usr->curr_up = (usr->tmpStats.up - usr->lastStats.up) / secs;
usr->curr_down = (usr->tmpStats.down - usr->lastStats.down) / secs;
usr->lastStats = usr->tmpStats;
thisStream->saveStats(usr->MyStr, usr->tmpStats);
}
}
}
usr->Send();
}
thisStream->cleanUsers();
std::cerr << "User " << usr->MyStr << " disconnected, socket number " << usr->S.getSocket() << std::endl;
}
/// Loop reading DTSC data from stdin and processing it at the correct speed.
void handleStdin(void * empty){
if (empty != 0){return;}
unsigned int lastPacketTime = 0;//time in MS last packet was parsed
unsigned int currPacketTime = 0;//time of the last parsed packet (current packet)
unsigned int prevPacketTime = 0;//time of the previously parsed packet (current packet - 1)
std::string inBuffer;
char charBuffer[1024*10];
unsigned int charCount;
unsigned int now;
while (std::cin.good() && buffer_running){
//slow down packet receiving to real-time
now = getNowMS();
if ((now - lastPacketTime >= currPacketTime - prevPacketTime) || (currPacketTime <= prevPacketTime)){
std::cin.read(charBuffer, 1024*10);
charCount = std::cin.gcount();
inBuffer.append(charBuffer, charCount);
thisStream->getWriteLock();
if (thisStream->getStream()->parsePacket(inBuffer)){
thisStream->getStream()->outPacket(0);
lastPacketTime = now;
prevPacketTime = currPacketTime;
currPacketTime = thisStream->getStream()->getTime();
thisStream->dropWriteLock(true);
}else{
thisStream->dropWriteLock(false);
}
}else{
if (((currPacketTime - prevPacketTime) - (now - lastPacketTime)) > 999){
usleep(999000);
}else{
usleep(((currPacketTime - prevPacketTime) - (now - lastPacketTime)) * 999);
}
}
}
buffer_running = false;
SS.close();
}
/// Loop reading DTSC data from an IP push address.
/// No changes to the speed are made.
void handlePushin(void * empty){
if (empty != 0){return;}
std::string inBuffer;
while (buffer_running){
if (thisStream->getIPInput().connected()){
if (thisStream->getIPInput().iread(inBuffer)){
thisStream->getWriteLock();
if (thisStream->getStream()->parsePacket(inBuffer)){
thisStream->getStream()->outPacket(0);
thisStream->dropWriteLock(true);
}else{
thisStream->dropWriteLock(false);
}
}
}else{
usleep(1000000);
}
}
SS.close();
}
/// Starts a loop, waiting for connections to send data to.
int Start(int argc, char ** argv) {
//first make sure no segpipe signals will kill us
struct sigaction new_action;
new_action.sa_handler = termination_handler;
sigemptyset (&new_action.sa_mask);
new_action.sa_flags = 0;
sigaction (SIGPIPE, &new_action, NULL);
sigaction (SIGKILL, &new_action, NULL);
//then check and parse the commandline
if (argc < 2) {
std::cout << "usage: " << argv[0] << " streamName [awaiting_IP]" << std::endl;
return 1;
}
std::string name = argv[1];
SS = Socket::makeStream(name);
thisStream = Stream::get();
thisStream->setName(name);
Socket::Connection incoming;
Socket::Connection std_input(fileno(stdin));
tthread::thread StatsThread = tthread::thread(handleStats, 0);
tthread::thread * StdinThread = 0;
if (argc < 3){
StdinThread = new tthread::thread(handleStdin, 0);
}else{
thisStream->setWaitingIP(argv[2]);
StdinThread = new tthread::thread(handlePushin, 0);
}
while (buffer_running && SS.connected()){
//check for new connections, accept them if there are any
//starts a thread for every accepted connection
incoming = SS.accept(false);
if (incoming.connected()){
user * usr_ptr = new user(incoming);
thisStream->addUser(usr_ptr);
usr_ptr->Thread = new tthread::thread(handleUser, (void *)usr_ptr);
}
}//main loop
// disconnect listener
buffer_running = false;
std::cout << "End of input file - buffer shutting down" << std::endl;
SS.close();
StatsThread.join();
StdinThread->join();
delete thisStream;
return 0;
}
};//Buffer namespace
/// Entry point for Buffer, simply calls Buffer::Start().
int main(int argc, char ** argv){
return Buffer::Start(argc, argv);
}//main

32
src/buffer_stats.cpp Normal file
View file

@ -0,0 +1,32 @@
#include "buffer_stats.h"
#include <stdlib.h> //for atoi()
Buffer::Stats::Stats(){
up = 0;
down = 0;
conntime = 0;
}
Buffer::Stats::Stats(std::string s){
size_t f = s.find(' ');
if (f != std::string::npos){
host = s.substr(0, f);
s.erase(0, f+1);
}
f = s.find(' ');
if (f != std::string::npos){
connector = s.substr(0, f);
s.erase(0, f+1);
}
f = s.find(' ');
if (f != std::string::npos){
conntime = atoi(s.substr(0, f).c_str());
s.erase(0, f+1);
}
f = s.find(' ');
if (f != std::string::npos){
up = atoi(s.substr(0, f).c_str());
s.erase(0, f+1);
down = atoi(s.c_str());
}
}

16
src/buffer_stats.h Normal file
View file

@ -0,0 +1,16 @@
#pragma once
#include <string>
namespace Buffer{
/// Converts a stats line to up, down, host, connector and conntime values.
class Stats{
public:
unsigned int up;
unsigned int down;
std::string host;
std::string connector;
unsigned int conntime;
Stats();
Stats(std::string s);
};
}

217
src/buffer_stream.cpp Normal file
View file

@ -0,0 +1,217 @@
#include "buffer_stream.h"
/// Stores the globally equal reference.
Buffer::Stream * Buffer::Stream::ref = 0;
/// Returns a globally equal reference to this class.
Buffer::Stream * Buffer::Stream::get(){
static tthread::mutex creator;
if (ref == 0){
//prevent creating two at the same time
creator.lock();
if (ref == 0){ref = new Stream();}
creator.unlock();
}
return ref;
}
/// Creates a new DTSC::Stream object, private function so only one instance can exist.
Buffer::Stream::Stream(){
Strm = new DTSC::Stream(5);
}
/// Do cleanup on delete.
Buffer::Stream::~Stream(){
delete Strm;
while (users.size() > 0){
stats_mutex.lock();
for (usersIt = users.begin(); usersIt != users.end(); usersIt++){
if ((**usersIt).S.connected()){
if ((**usersIt).myRing->waiting){
(**usersIt).S.close();
printf("Closing user %s\n", (**usersIt).MyStr.c_str());
}
}
}
stats_mutex.unlock();
moreData.notify_all();
cleanUsers();
}
}
/// Calculate and return the current statistics in JSON format.
std::string Buffer::Stream::getStats(){
unsigned int now = time(0);
unsigned int tot_up = 0, tot_down = 0, tot_count = 0;
stats_mutex.lock();
if (users.size() > 0){
for (usersIt = users.begin(); usersIt != users.end(); usersIt++){
tot_down += (**usersIt).curr_down;
tot_up += (**usersIt).curr_up;
tot_count++;
}
}
Storage["totals"]["down"] = tot_down;
Storage["totals"]["up"] = tot_up;
Storage["totals"]["count"] = tot_count;
Storage["totals"]["now"] = now;
Storage["totals"]["buffer"] = name;
std::string ret = Storage.toString();
Storage["log"].null();
stats_mutex.unlock();
return ret;
}
/// Get a new DTSC::Ring object for a user.
DTSC::Ring * Buffer::Stream::getRing(){
return Strm->getRing();
}
/// Drop a DTSC::Ring object.
void Buffer::Stream::dropRing(DTSC::Ring * ring){
Strm->dropRing(ring);
}
/// Get the (constant) header data of this stream.
std::string & Buffer::Stream::getHeader(){
return Strm->outHeader();
}
/// Set the IP address to accept push data from.
void Buffer::Stream::setWaitingIP(std::string ip){
waiting_ip = ip;
}
/// Check if this is the IP address to accept push data from.
bool Buffer::Stream::checkWaitingIP(std::string ip){
if (ip == waiting_ip || ip == "::ffff:"+waiting_ip){
return true;
}else{
std::cout << ip << " != " << waiting_ip << std::endl;
return false;
}
}
/// Sets the current socket for push data.
bool Buffer::Stream::setInput(Socket::Connection S){
if (ip_input.connected()){
return false;
}else{
ip_input = S;
return true;
}
}
/// Gets the current socket for push data.
Socket::Connection & Buffer::Stream::getIPInput(){
return ip_input;
}
/// Stores intermediate statistics.
void Buffer::Stream::saveStats(std::string username, Stats & stats){
stats_mutex.lock();
Storage["curr"][username]["connector"] = stats.connector;
Storage["curr"][username]["up"] = stats.up;
Storage["curr"][username]["down"] = stats.down;
Storage["curr"][username]["conntime"] = stats.conntime;
Storage["curr"][username]["host"] = stats.host;
Storage["curr"][username]["start"] = (unsigned int) time(0) - stats.conntime;
stats_mutex.unlock();
}
/// Stores final statistics.
void Buffer::Stream::clearStats(std::string username, Stats & stats, std::string reason){
stats_mutex.lock();
Storage["curr"].removeMember(username);
Storage["log"][username]["connector"] = stats.connector;
Storage["log"][username]["up"] = stats.up;
Storage["log"][username]["down"] = stats.down;
Storage["log"][username]["conntime"] = stats.conntime;
Storage["log"][username]["host"] = stats.host;
Storage["log"][username]["start"] = (unsigned int)time(0) - stats.conntime;
std::cout << "Disconnected user " << username << ": " << reason << ". " << stats.connector << " transferred " << stats.up << " up and " << stats.down << " down in " << stats.conntime << " seconds to " << stats.host << std::endl;
stats_mutex.unlock();
cleanUsers();
}
/// Cleans up broken connections
void Buffer::Stream::cleanUsers(){
bool repeat = false;
stats_mutex.lock();
do{
repeat = false;
if (users.size() > 0){
for (usersIt = users.begin(); usersIt != users.end(); usersIt++){
if ((**usersIt).Thread == 0 && !(**usersIt).S.connected()){
delete *usersIt;
users.erase(usersIt);
repeat = true;
break;
}
}
}
}while(repeat);
stats_mutex.unlock();
}
/// Blocks until writing is safe.
void Buffer::Stream::getWriteLock(){
rw_mutex.lock();
writers++;
while (writers != 1 && readers != 0){
rw_change.wait(rw_mutex);
}
rw_mutex.unlock();
}
/// Drops a previously gotten write lock.
void Buffer::Stream::dropWriteLock(bool newpackets_available){
rw_mutex.lock();
writers--;
rw_mutex.unlock();
rw_change.notify_all();
if (newpackets_available){moreData.notify_all();}
}
/// Blocks until reading is safe.
void Buffer::Stream::getReadLock(){
rw_mutex.lock();
while (writers > 0){
rw_change.wait(rw_mutex);
}
readers++;
rw_mutex.unlock();
}
/// Drops a previously gotten read lock.
void Buffer::Stream::dropReadLock(){
rw_mutex.lock();
readers--;
rw_mutex.unlock();
rw_change.notify_all();
}
/// Retrieves a reference to the DTSC::Stream
DTSC::Stream * Buffer::Stream::getStream(){
return Strm;
}
/// Sets the buffer name.
void Buffer::Stream::setName(std::string n){
name = n;
}
/// Add a user to the userlist.
void Buffer::Stream::addUser(user * new_user){
stats_mutex.lock();
users.push_back(new_user);
stats_mutex.unlock();
}
/// Blocks the thread until new data is available.
void Buffer::Stream::waitForData(){
stats_mutex.lock();
moreData.wait(stats_mutex);
stats_mutex.unlock();
}

69
src/buffer_stream.h Normal file
View file

@ -0,0 +1,69 @@
#pragma once
#include <string>
#include "../lib/tinythread.h"
#include "../lib/json.h"
#include "buffer_user.h"
namespace Buffer{
class Stream{
public:
/// Get a reference to this Stream object.
static Stream * get();
/// Get the current statistics in JSON format.
std::string getStats();
/// Get a new DTSC::Ring object for a user.
DTSC::Ring * getRing();
/// Drop a DTSC::Ring object.
void dropRing(DTSC::Ring * ring);
/// Get the (constant) header data of this stream.
std::string & getHeader();
/// Set the IP address to accept push data from.
void setWaitingIP(std::string ip);
/// Check if this is the IP address to accept push data from.
bool checkWaitingIP(std::string ip);
/// Sets the current socket for push data.
bool setInput(Socket::Connection S);
/// Gets the current socket for push data.
Socket::Connection & getIPInput();
/// Stores intermediate statistics.
void saveStats(std::string username, Stats & stats);
/// Stores final statistics.
void clearStats(std::string username, Stats & stats, std::string reason);
/// Cleans up broken connections
void cleanUsers();
/// Blocks until writing is safe.
void getWriteLock();
/// Drops a previously gotten write lock.
void dropWriteLock(bool newpackets_available);
/// Blocks until reading is safe.
void getReadLock();
/// Drops a previously gotten read lock.
void dropReadLock();
/// Retrieves a reference to the DTSC::Stream
DTSC::Stream * getStream();
/// Sets the buffer name.
void setName(std::string n);
/// Add a user to the userlist.
void addUser(user * new_user);
/// Blocks the thread until new data is available.
void waitForData();
/// Cleanup function
~Stream();
private:
volatile int readers;///< Current count of active readers;
volatile int writers;///< Current count of waiting/active writers.
tthread::mutex rw_mutex; ///< Mutex for read/write locking.
tthread::condition_variable rw_change; ///< Triggered when reader/writer count changes.
static Stream * ref;
Stream();
JSON::Value Storage; ///< Global storage of data.
DTSC::Stream * Strm;
std::string waiting_ip; ///< IP address for media push.
Socket::Connection ip_input; ///< Connection used for media push.
tthread::mutex stats_mutex; ///< Mutex for stats/users modifications.
std::vector<user*> users; ///< All connected users.
std::vector<user*>::iterator usersIt; ///< Iterator for all connected users.
std::string name; ///< Name for this buffer.
tthread::condition_variable moreData; ///< Triggered when more data becomes available.
};
};

76
src/buffer_user.cpp Normal file
View file

@ -0,0 +1,76 @@
#include "buffer_user.h"
#include "buffer_stream.h"
#include <sstream>
int Buffer::user::UserCount = 0;
/// Creates a new user from a newly connected socket.
/// Also prints "User connected" text to stdout.
Buffer::user::user(Socket::Connection fd){
S = fd;
MyNum = UserCount++;
std::stringstream st;
st << MyNum;
MyStr = st.str();
curr_up = 0;
curr_down = 0;
currsend = 0;
myRing = 0;
Thread = 0;
std::cout << "User " << MyNum << " connected" << std::endl;
}//constructor
/// Drops held DTSC::Ring class, if one is held.
Buffer::user::~user(){
Stream::get()->dropRing(myRing);
}//destructor
/// Disconnects the current user. Doesn't do anything if already disconnected.
/// Prints "Disconnected user" to stdout if disconnect took place.
void Buffer::user::Disconnect(std::string reason) {
if (S.connected()){S.close();}
if (Thread != 0){
if (Thread->joinable()){Thread->join();}
Thread = 0;
}
Stream::get()->clearStats(MyStr, lastStats, reason);
}//Disconnect
/// Tries to send the current buffer, returns true if success, false otherwise.
/// Has a side effect of dropping the connection if send will never complete.
bool Buffer::user::doSend(const char * ptr, int len){
int r = S.iwrite(ptr+currsend, len-currsend);
if (r <= 0){
if (errno == EWOULDBLOCK){return false;}
Disconnect(S.getError());
return false;
}
currsend += r;
return (currsend == len);
}//doSend
/// Try to send data to this user. Disconnects if any problems occur.
void Buffer::user::Send(){
if (!myRing){return;}//no ring!
if (!S.connected()){return;}//cancel if not connected
if (myRing->waiting){
Stream::get()->waitForData();
return;
}//still waiting for next buffer?
if (myRing->starved){
//if corrupt data, warn and get new DTSC::Ring
std::cout << "Warning: User was send corrupt video data and send to the next keyframe!" << std::endl;
Stream::get()->dropRing(myRing);
myRing = Stream::get()->getRing();
return;
}
//try to complete a send
Stream::get()->getReadLock();
if (doSend(Stream::get()->getStream()->outPacket(myRing->b).c_str(), Stream::get()->getStream()->outPacket(myRing->b).length())){
//switch to next buffer
currsend = 0;
if (myRing->b <= 0){myRing->waiting = true; return;}//no next buffer? go in waiting mode.
myRing->b--;
}//completed a send
Stream::get()->dropReadLock();
}//send

41
src/buffer_user.h Normal file
View file

@ -0,0 +1,41 @@
#pragma once
#include <string>
#include "buffer_stats.h"
#include "../lib/dtsc.h"
#include "../lib/socket.h"
#include "../lib/tinythread.h"
namespace Buffer{
/// Holds connected users.
/// Keeps track of what buffer users are using and the connection status.
class user{
public:
tthread::thread * Thread; ///< Holds the thread dealing with this user.
DTSC::Ring * myRing; ///< Ring of the buffer for this user.
int MyNum; ///< User ID of this user.
std::string MyStr; ///< User ID of this user as a string.
std::string inbuffer; ///< Used to buffer input data.
int currsend; ///< Current amount of bytes sent.
Stats lastStats; ///< Holds last known stats for this connection.
Stats tmpStats; ///< Holds temporary stats for this connection.
unsigned int curr_up; ///< Holds the current estimated transfer speed up.
unsigned int curr_down; ///< Holds the current estimated transfer speed down.
bool gotproperaudio; ///< Whether the user received proper audio yet.
void * lastpointer; ///< Pointer to data part of current buffer.
static int UserCount; ///< Global user counter.
Socket::Connection S; ///< Connection to user
/// Creates a new user from a newly connected socket.
/// Also prints "User connected" text to stdout.
user(Socket::Connection fd);
/// Drops held DTSC::Ring class, if one is held.
~user();
/// Disconnects the current user. Doesn't do anything if already disconnected.
/// Prints "Disconnected user" to stdout if disconnect took place.
void Disconnect(std::string reason);
/// Tries to send the current buffer, returns true if success, false otherwise.
/// Has a side effect of dropping the connection if send will never complete.
bool doSend(const char * ptr, int len);
/// Try to send data to this user. Disconnects if any problems occur.
void Send();
};
}

336
src/conn_http.cpp Normal file
View file

@ -0,0 +1,336 @@
/// \file Connector_HTTP/main.cpp
/// Contains the main code for the HTTP Connector
#include <iostream>
#include <queue>
#include <cstdlib>
#include <cstdio>
#include <cmath>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <getopt.h>
#include <ctime>
#include "../lib/socket.h"
#include "../lib/http_parser.h"
#include "../lib/json.h"
#include "../lib/dtsc.h"
#include "../lib/flv_tag.h"
#include "../lib/base64.h"
#include "../lib/amf.h"
/// Holds everything unique to HTTP Connector.
namespace Connector_HTTP{
/// Defines the type of handler used to process this request.
enum {HANDLER_NONE, HANDLER_PROGRESSIVE, HANDLER_FLASH, HANDLER_APPLE, HANDLER_MICRO, HANDLER_JSCRIPT};
std::queue<std::string> Flash_FragBuffer;///<Fragment buffer for F4V
DTSC::Stream Strm;///< Incoming stream buffer.
HTTP::Parser HTTP_R, HTTP_S;///<HTTP Receiver en HTTP Sender.
/// Folds data into a mdat container and returns it.
std::string mdatFold(std::string data){
std::string Result;
unsigned int t_int;
t_int = htonl(data.size()+8);
Result.append((char*)&t_int, 4);
t_int = htonl(0x6D646174);
Result.append((char*)&t_int, 4);
Result.append(data);
return Result;
}
/// Returns AMF-format metadata for Adobe HTTP Dynamic Streaming.
std::string GetMetaData( ) {
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("onMetaData",AMF::AMF0_STRING));
amfreply.addContent(AMF::Object("",AMF::AMF0_ECMA_ARRAY));
amfreply.getContentP(1)->addContent(AMF::Object("trackinfo", AMF::AMF0_STRICT_ARRAY));
amfreply.getContentP(1)->getContentP(0)->addContent(AMF::Object("arrVal"));
//amfreply.getContentP(1)->getContentP(0)->getContentP(0)->addContent(AMF::Object("timescale",(double)1000));
//amfreply.getContentP(1)->getContentP(0)->getContentP(0)->addContent(AMF::Object("length",(double)59641700));
amfreply.getContentP(1)->getContentP(0)->getContentP(0)->addContent(AMF::Object("language","eng"));
amfreply.getContentP(1)->getContentP(0)->getContentP(0)->addContent(AMF::Object("sampledescription", AMF::AMF0_STRICT_ARRAY));
amfreply.getContentP(1)->getContentP(0)->getContentP(0)->getContentP(1)->addContent(AMF::Object("arrVal"));
amfreply.getContentP(1)->getContentP(0)->getContentP(0)->getContentP(1)->getContentP(0)->addContent(AMF::Object("sampletype","avc1"));
amfreply.getContentP(1)->getContentP(0)->addContent(AMF::Object("arrVal"));
//amfreply.getContentP(1)->getContentP(0)->getContentP(1)->addContent(AMF::Object("timescale",(double)44100));
//amfreply.getContentP(1)->getContentP(0)->getContentP(1)->addContent(AMF::Object("length",(double)28630000));
amfreply.getContentP(1)->getContentP(0)->getContentP(1)->addContent(AMF::Object("language","eng"));
amfreply.getContentP(1)->getContentP(0)->getContentP(1)->addContent(AMF::Object("sampledescription", AMF::AMF0_STRICT_ARRAY));
amfreply.getContentP(1)->getContentP(0)->getContentP(1)->getContentP(1)->addContent(AMF::Object("arrVal"));
amfreply.getContentP(1)->getContentP(0)->getContentP(1)->getContentP(1)->getContentP(0)->addContent(AMF::Object("sampletype","mp4a"));
amfreply.getContentP(1)->addContent(AMF::Object("audiochannels",(double)2));
amfreply.getContentP(1)->addContent(AMF::Object("audiosamplerate",(double)44100));
amfreply.getContentP(1)->addContent(AMF::Object("videoframerate",(double)25));
amfreply.getContentP(1)->addContent(AMF::Object("aacaot",(double)2));
amfreply.getContentP(1)->addContent(AMF::Object("avclevel",(double)12));
amfreply.getContentP(1)->addContent(AMF::Object("avcprofile",(double)77));
amfreply.getContentP(1)->addContent(AMF::Object("audiocodecid","mp4a"));
amfreply.getContentP(1)->addContent(AMF::Object("videocodecid","avc1"));
amfreply.getContentP(1)->addContent(AMF::Object("width",(double)1280));
amfreply.getContentP(1)->addContent(AMF::Object("height",(double)720));
amfreply.getContentP(1)->addContent(AMF::Object("frameWidth",(double)1280));
amfreply.getContentP(1)->addContent(AMF::Object("frameHeight",(double)720));
amfreply.getContentP(1)->addContent(AMF::Object("displayWidth",(double)1280));
amfreply.getContentP(1)->addContent(AMF::Object("displayHeight",(double)720));
//amfreply.getContentP(1)->addContent(AMF::Object("moovposition",(double)35506700));
//amfreply.getContentP(1)->addContent(AMF::Object("duration",(double)596.458));
return amfreply.Pack( );
}//getMetaData
/// Returns a F4M-format manifest file for Adobe HTTP Dynamic Streaming.
std::string BuildManifest(std::string MovieId) {
Interface * temp = new Interface;
std::string Result="<?xml version=\"1.0\" encoding=\"utf-8\"?>\n<manifest xmlns=\"http://ns.adobe.com/f4m/1.0\">\n";
Result += "<id>";
Result += MovieId;
Result += "</id>\n<mimeType>video/mp4</mimeType>\n";
Result += "<streamType>live</streamType>\n";
Result += "<deliveryType>streaming</deliveryType>\n";
Result += "<bootstrapInfo profile=\"named\" id=\"bootstrap1\">";
Result += Base64::encode(temp->GenerateLiveBootstrap(1));
Result += "</bootstrapInfo>\n";
Result += "<media streamId=\"1\" bootstrapInfoId=\"bootstrap1\" url=\"";
Result += MovieId;
Result += "/\">\n";
Result += "<metadata>";
Result += Base64::encode(GetMetaData());
Result += "</metadata>\n";
Result += "</media>\n";
Result += "</manifest>\n";
delete temp;
return Result;
}//BuildManifest
/// Handles Progressive download streaming requests
void Progressive(FLV::Tag & tag, HTTP::Parser HTTP_S, Socket::Connection & conn, DTSC::Stream & Strm){
static bool progressive_has_sent_header = false;
if (!progressive_has_sent_header){
HTTP_S.Clean();//troep opruimen die misschien aanwezig is...
HTTP_S.SetHeader("Content-Type", "video/x-flv");//FLV files hebben altijd dit content-type.
//HTTP_S.SetHeader("Transfer-Encoding", "chunked");
HTTP_S.protocol = "HTTP/1.0";
HTTP_S.SendResponse(conn, "200", "OK");//geen SetBody = unknown length! Dat willen we hier.
//HTTP_S.SendBodyPart(CONN_fd, FLVHeader, 13);//schrijf de FLV header
conn.write(FLV::Header, 13);
static FLV::Tag tmp;
tmp.DTSCMetaInit(Strm);
conn.write(tmp.data, tmp.len);
if (Strm.metadata.getContentP("audio") && Strm.metadata.getContentP("audio")->getContentP("init")){
tmp.DTSCAudioInit(Strm);
conn.write(tmp.data, tmp.len);
}
if (Strm.metadata.getContentP("video") && Strm.metadata.getContentP("video")->getContentP("init")){
tmp.DTSCVideoInit(Strm);
conn.write(tmp.data, tmp.len);
}
progressive_has_sent_header = true;
#if DEBUG >= 1
fprintf(stderr, "Sent progressive FLV header\n");
#endif
}
//HTTP_S.SendBodyPart(CONN_fd, tag->data, tag->len);//schrijf deze FLV tag onbewerkt weg
conn.write(tag.data, tag.len);
}
/// Handles Flash Dynamic HTTP streaming requests
void FlashDynamic(FLV::Tag & tag, DTSC::Stream & Strm){
static std::string FlashBuf;
static FLV::Tag tmp;
if (Strm.getPacket(0).getContentP("keyframe")){
if (FlashBuf != ""){
Flash_FragBuffer.push(FlashBuf);
#if DEBUG >= 4
fprintf(stderr, "Received a fragment. Now %i in buffer.\n", (int)Flash_FragBuffer.size());
#endif
}
FlashBuf.clear();
//fill buffer with init data, if needed.
if (Strm.metadata.getContentP("audio") && Strm.metadata.getContentP("audio")->getContentP("init")){
tmp.DTSCAudioInit(Strm);
FlashBuf.append(tmp.data, tmp.len);
}
if (Strm.metadata.getContentP("video") && Strm.metadata.getContentP("video")->getContentP("init")){
tmp.DTSCVideoInit(Strm);
FlashBuf.append(tmp.data, tmp.len);
}
}
FlashBuf.append(tag.data, tag.len);
}
/// Main function for Connector_HTTP
int Connector_HTTP(Socket::Connection conn){
int handler = HANDLER_PROGRESSIVE;///< The handler used for processing this request.
bool ready4data = false;///< Set to true when streaming is to begin.
bool inited = false;
Socket::Connection ss(-1);
std::string streamname;
FLV::Tag tag;///< Temporary tag buffer.
std::string recBuffer = "";
std::string Movie = "";
std::string Quality = "";
int Segment = -1;
int ReqFragment = -1;
int temp;
int Flash_RequestPending = 0;
bool Flash_ManifestSent = false;
unsigned int lastStats = 0;
//int CurrentFragment = -1; later herbruiken?
while (conn.connected()){
//only parse input if available or not yet init'ed
if (HTTP_R.Read(conn, ready4data)){
handler = HANDLER_PROGRESSIVE;
std::cout << "Received request: " << HTTP_R.url << std::endl;
if ((HTTP_R.url.find("Seg") != std::string::npos) && (HTTP_R.url.find("Frag") != std::string::npos)){handler = HANDLER_FLASH;}
if (HTTP_R.url.find("f4m") != std::string::npos){handler = HANDLER_FLASH;}
if (HTTP_R.url == "/crossdomain.xml"){
handler = HANDLER_NONE;
HTTP_S.Clean();
HTTP_S.SetHeader("Content-Type", "text/xml");
HTTP_S.SetBody("<?xml version=\"1.0\"?><!DOCTYPE cross-domain-policy SYSTEM \"http://www.adobe.com/xml/dtds/cross-domain-policy.dtd\"><cross-domain-policy><allow-access-from domain=\"*\" /><site-control permitted-cross-domain-policies=\"all\"/></cross-domain-policy>");
HTTP_S.SendResponse(conn, "200", "OK");
#if DEBUG >= 3
printf("Sending crossdomain.xml file\n");
#endif
}
if (HTTP_R.url.substr(0, 7) == "/embed_" && HTTP_R.url.substr(HTTP_R.url.length() - 3, 3) == ".js"){
streamname = HTTP_R.url.substr(7, HTTP_R.url.length() - 10);
JSON::Value ServConf = JSON::fromFile("/tmp/mist/streamlist");
std::string response;
handler = HANDLER_NONE;
HTTP_S.Clean();
HTTP_S.SetHeader("Content-Type", "application/javascript");
response = "// Generating embed code for stream " + streamname + "\n\n";
if (ServConf["streams"].isMember(streamname)){
std::string streamurl = "http://" + HTTP_S.GetHeader("Host") + "/" + streamname + ".flv";
response += "// Stream URL: " + streamurl + "\n\n";
response += "document.write('<object width=\"600\" height=\"409\"><param name=\"movie\" value=\"http://fpdownload.adobe.com/strobe/FlashMediaPlayback.swf\"></param><param name=\"flashvars\" value=\"src="+HTTP::Parser::urlencode(streamurl)+"&controlBarMode=floating\"></param><param name=\"allowFullScreen\" value=\"true\"></param><param name=\"allowscriptaccess\" value=\"always\"></param><embed src=\"http://fpdownload.adobe.com/strobe/FlashMediaPlayback.swf\" type=\"application/x-shockwave-flash\" allowscriptaccess=\"always\" allowfullscreen=\"true\" width=\"600\" height=\"409\" flashvars=\"src="+HTTP::Parser::urlencode(streamurl)+"&controlBarMode=floating\"></embed></object>');\n";
}else{
response += "// Stream not available at this server.\nalert(\"This stream is currently not available at this server.\\\\nPlease try again later!\");";
}
response += "";
HTTP_S.SetBody(response);
HTTP_S.SendResponse(conn, "200", "OK");
#if DEBUG >= 3
printf("Sending embed code for %s\n", streamname.c_str());
#endif
}
if (handler == HANDLER_FLASH){
if (HTTP_R.url.find("f4m") == std::string::npos){
Movie = HTTP_R.url.substr(1);
Movie = Movie.substr(0,Movie.find("/"));
Quality = HTTP_R.url.substr( HTTP_R.url.find("/",1)+1 );
Quality = Quality.substr(0, Quality.find("Seg"));
temp = HTTP_R.url.find("Seg") + 3;
Segment = atoi( HTTP_R.url.substr(temp,HTTP_R.url.find("-",temp)-temp).c_str());
temp = HTTP_R.url.find("Frag") + 4;
ReqFragment = atoi( HTTP_R.url.substr(temp).c_str() );
#if DEBUG >= 4
printf( "URL: %s\n", HTTP_R.url.c_str());
printf( "Movie: %s, Quality: %s, Seg %d Frag %d\n", Movie.c_str(), Quality.c_str(), Segment, ReqFragment);
#endif
Flash_RequestPending++;
}else{
Movie = HTTP_R.url.substr(1);
Movie = Movie.substr(0,Movie.find("/"));
}
streamname = Movie;
if( !Flash_ManifestSent ) {
HTTP_S.Clean();
HTTP_S.SetHeader("Content-Type","text/xml");
HTTP_S.SetHeader("Cache-Control","no-cache");
HTTP_S.SetBody(BuildManifest(Movie));
HTTP_S.SendResponse(conn, "200", "OK");
Flash_ManifestSent = true;//stop manifest from being sent multiple times
std::cout << "Sent manifest" << std::endl;
}
ready4data = true;
}//FLASH handler
if (handler == HANDLER_PROGRESSIVE){
//we assume the URL is the stream name with a 3 letter extension
std::string extension = HTTP_R.url.substr(HTTP_R.url.size()-4);
streamname = HTTP_R.url.substr(0, HTTP_R.url.size()-4);//strip the extension
/// \todo VoD streams will need support for position reading from the URL parameters
ready4data = true;
}//PROGRESSIVE handler
HTTP_R.CleanForNext(); //clean for any possinble next requests
}
if (ready4data){
if (!inited){
//we are ready, connect the socket!
ss = Socket::getStream(streamname);
if (!ss.connected()){
#if DEBUG >= 1
fprintf(stderr, "Could not connect to server!\n");
#endif
conn.close();
break;
}
#if DEBUG >= 3
fprintf(stderr, "Everything connected, starting to send video data...\n");
#endif
inited = true;
}
if ((Flash_RequestPending > 0) && !Flash_FragBuffer.empty()){
HTTP_S.Clean();
HTTP_S.SetHeader("Content-Type","video/mp4");
HTTP_S.SetBody(mdatFold(Flash_FragBuffer.front()));
Flash_FragBuffer.pop();
HTTP_S.SendResponse(conn, "200", "OK");//schrijf de HTTP response header
Flash_RequestPending--;
#if DEBUG >= 3
fprintf(stderr, "Sending a video fragment. %i left in buffer, %i requested\n", (int)Flash_FragBuffer.size(), Flash_RequestPending);
#endif
}
if (inited){
unsigned int now = time(0);
if (now != lastStats){
lastStats = now;
std::string stat = "S "+conn.getStats("HTTP");
ss.write(stat);
}
}
if (ss.canRead()){
ss.spool();
if (Strm.parsePacket(ss.Received())){
tag.DTSCLoader(Strm);
if (handler == HANDLER_FLASH){
FlashDynamic(tag, Strm);
}
if (handler == HANDLER_PROGRESSIVE){
Progressive(tag, HTTP_S, conn, Strm);
}
}
}
}
}
conn.close();
if (inited) ss.close();
#if DEBUG >= 1
if (FLV::Parse_Error){fprintf(stderr, "FLV Parser Error: %s\n", FLV::Error_Str.c_str());}
fprintf(stderr, "User %i disconnected.\n", conn.getSocket());
if (inited){
fprintf(stderr, "Status was: inited\n");
}else{
if (ready4data){
fprintf(stderr, "Status was: ready4data\n");
}else{
fprintf(stderr, "Status was: connected\n");
}
}
#endif
return 0;
}//Connector_HTTP main function
};//Connector_HTTP namespace
// Load main server setup file, default port 8080, handler is Connector_HTTP::Connector_HTTP
#define DEFAULT_PORT 8080
#define MAINHANDLER Connector_HTTP::Connector_HTTP
#define CONFIGSECT HTTP
#include "server_setup.cpp"

29
src/conn_raw.cpp Normal file
View file

@ -0,0 +1,29 @@
/// \file Connector_RAW/main.cpp
/// Contains the main code for the RAW connector.
#include <iostream>
#include "../lib/socket.h"
/// Contains the main code for the RAW connector.
/// Expects a single commandline argument telling it which stream to connect to,
/// then outputs the raw stream to stdout.
int main(int argc, char ** argv) {
if (argc < 2){
std::cout << "Usage: " << argv[0] << " stream_name" << std::endl;
return 1;
}
std::string input = "/tmp/shared_socket_";
input += argv[1];
//connect to the proper stream
Socket::Connection S(input);
if (!S.connected()){
std::cout << "Could not open stream " << argv[1] << std::endl;
return 1;
}
//transport ~50kb at a time
//this is a nice tradeoff between CPU usage and speed
const char buffer[50000] = {0};
while(std::cout.good() && S.read(buffer,50000)){std::cout.write(buffer,50000);}
S.close();
return 0;
}

508
src/conn_rtmp.cpp Normal file
View file

@ -0,0 +1,508 @@
/// \file Connector_RTMP/main.cpp
/// Contains the main code for the RTMP Connector
#include <iostream>
#include <cstdlib>
#include <cstdio>
#include <cmath>
#include <unistd.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <getopt.h>
#include <sstream>
#include "../lib/socket.h"
#include "../lib/flv_tag.h"
#include "../lib/amf.h"
#include "../lib/rtmpchunks.h"
/// Holds all functions and data unique to the RTMP Connector
namespace Connector_RTMP{
//for connection to server
bool ready4data = false; ///< Set to true when streaming starts.
bool inited = false; ///< Set to true when ready to connect to Buffer.
bool nostats = false; ///< Set to true if no stats should be sent anymore (push mode).
bool stopparsing = false; ///< Set to true when all parsing needs to be cancelled.
Socket::Connection Socket; ///< Socket connected to user
Socket::Connection SS; ///< Socket connected to server
std::string streamname; ///< Stream that will be opened
void parseChunk(std::string & buffer);///< Parses a single RTMP chunk.
void sendCommand(AMF::Object & amfreply, int messagetype, int stream_id);///< Sends a RTMP command either in AMF or AMF3 mode.
void parseAMFCommand(AMF::Object & amfdata, int messagetype, int stream_id);///< Parses a single AMF command message.
int Connector_RTMP(Socket::Connection conn);
};//Connector_RTMP namespace;
/// Main Connector_RTMP function
int Connector_RTMP::Connector_RTMP(Socket::Connection conn){
Socket = conn;
FLV::Tag tag, init_tag;
DTSC::Stream Strm;
bool stream_inited = false;//true if init data for audio/video was sent
//first timestamp set
RTMPStream::firsttime = RTMPStream::getNowMS();
RTMPStream::handshake_in.reserve(1537);
Socket.read((char*)RTMPStream::handshake_in.c_str(), 1537);
RTMPStream::rec_cnt += 1537;
if (RTMPStream::doHandshake()){
Socket.write(RTMPStream::handshake_out);
Socket.read((char*)RTMPStream::handshake_in.c_str(), 1536);
RTMPStream::rec_cnt += 1536;
#if DEBUG >= 4
fprintf(stderr, "Handshake succcess!\n");
#endif
}else{
#if DEBUG >= 1
fprintf(stderr, "Handshake fail!\n");
#endif
return 0;
}
unsigned int lastStats = 0;
conn.setBlocking(false);
while (Socket.connected()){
usleep(10000);//sleep 10ms to prevent high CPU usage
if (Socket.spool()){
parseChunk(Socket.Received());
}
if (ready4data){
if (!inited){
//we are ready, connect the socket!
SS = Socket::getStream(streamname);
if (!SS.connected()){
#if DEBUG >= 1
fprintf(stderr, "Could not connect to server!\n");
#endif
Socket.close();//disconnect user
break;
}
#if DEBUG >= 3
fprintf(stderr, "Everything connected, starting to send video data...\n");
#endif
inited = true;
}
if (inited && !nostats){
unsigned int now = time(0);
if (now != lastStats){
lastStats = now;
std::string stat = "S "+Socket.getStats("RTMP");
SS.write(stat);
}
}
if (SS.spool()){
if (Strm.parsePacket(SS.Received())){
//sent init data if needed
if (!stream_inited){
if (Strm.metadata.getContentP("audio") && Strm.metadata.getContentP("audio")->getContentP("init")){
init_tag.DTSCAudioInit(Strm);
Socket.write(RTMPStream::SendMedia(init_tag));
}
if (Strm.metadata.getContentP("video") && Strm.metadata.getContentP("video")->getContentP("init")){
init_tag.DTSCVideoInit(Strm);
Socket.write(RTMPStream::SendMedia(init_tag));
}
stream_inited = true;
}
//sent a tag
tag.DTSCLoader(Strm);
Socket.write(RTMPStream::SendMedia(tag));
#if DEBUG >= 8
fprintf(stderr, "Sent tag to %i: [%u] %s\n", Socket.getSocket(), tag.tagTime(), tag.tagType().c_str());
#endif
}
}
}
}
SS.close();
Socket.close();
#if DEBUG >= 1
if (FLV::Parse_Error){fprintf(stderr, "FLV Parse Error: %s\n", FLV::Error_Str.c_str());}
fprintf(stderr, "User %i disconnected.\n", conn.getSocket());
if (inited){
fprintf(stderr, "Status was: inited\n");
}else{
if (ready4data){
fprintf(stderr, "Status was: ready4data\n");
}else{
fprintf(stderr, "Status was: connected\n");
}
}
#endif
return 0;
}//Connector_RTMP
/// Tries to get and parse one RTMP chunk at a time.
void Connector_RTMP::parseChunk(std::string & inbuffer){
//for DTSC conversion
static DTSC::DTMI meta_out;
static std::stringstream prebuffer; // Temporary buffer before sending real data
static bool sending = false;
static unsigned int counter = 0;
//for chunk parsing
static RTMPStream::Chunk next;
FLV::Tag F;
static AMF::Object amfdata("empty", AMF::AMF0_DDV_CONTAINER);
static AMF::Object amfelem("empty", AMF::AMF0_DDV_CONTAINER);
static AMF::Object3 amf3data("empty", AMF::AMF3_DDV_CONTAINER);
static AMF::Object3 amf3elem("empty", AMF::AMF3_DDV_CONTAINER);
while (next.Parse(inbuffer)){
//send ACK if we received a whole window
if ((RTMPStream::rec_cnt - RTMPStream::rec_window_at > RTMPStream::rec_window_size)){
RTMPStream::rec_window_at = RTMPStream::rec_cnt;
Socket.write(RTMPStream::SendCTL(3, RTMPStream::rec_cnt));//send ack (msg 3)
}
switch (next.msg_type_id){
case 0://does not exist
#if DEBUG >= 2
fprintf(stderr, "UNKN: Received a zero-type message. This is an error.\n");
#endif
break;//happens when connection breaks unexpectedly
case 1://set chunk size
RTMPStream::chunk_rec_max = ntohl(*(int*)next.data.c_str());
#if DEBUG >= 4
fprintf(stderr, "CTRL: Set chunk size: %i\n", RTMPStream::chunk_rec_max);
#endif
break;
case 2://abort message - we ignore this one
#if DEBUG >= 4
fprintf(stderr, "CTRL: Abort message\n");
#endif
//4 bytes of stream id to drop
break;
case 3://ack
#if DEBUG >= 4
fprintf(stderr, "CTRL: Acknowledgement\n");
#endif
RTMPStream::snd_window_at = ntohl(*(int*)next.data.c_str());
RTMPStream::snd_window_at = RTMPStream::snd_cnt;
break;
case 4:{
#if DEBUG >= 4
short int ucmtype = ntohs(*(short int*)next.data.c_str());
fprintf(stderr, "CTRL: User control message %hi\n", ucmtype);
#endif
//2 bytes event type, rest = event data
//types:
//0 = stream begin, 4 bytes ID
//1 = stream EOF, 4 bytes ID
//2 = stream dry, 4 bytes ID
//3 = setbufferlen, 4 bytes ID, 4 bytes length
//4 = streamisrecorded, 4 bytes ID
//6 = pingrequest, 4 bytes data
//7 = pingresponse, 4 bytes data
//we don't need to process this
} break;
case 5://window size of other end
#if DEBUG >= 4
fprintf(stderr, "CTRL: Window size\n");
#endif
RTMPStream::rec_window_size = ntohl(*(int*)next.data.c_str());
RTMPStream::rec_window_at = RTMPStream::rec_cnt;
Socket.write(RTMPStream::SendCTL(3, RTMPStream::rec_cnt));//send ack (msg 3)
break;
case 6:
#if DEBUG >= 4
fprintf(stderr, "CTRL: Set peer bandwidth\n");
#endif
//4 bytes window size, 1 byte limit type (ignored)
RTMPStream::snd_window_size = ntohl(*(int*)next.data.c_str());
Socket.write(RTMPStream::SendCTL(5, RTMPStream::snd_window_size));//send window acknowledgement size (msg 5)
break;
case 8://audio data
case 9://video data
case 18://meta data
if (SS.connected()){
F.ChunkLoader(next);
DTSC::DTMI pack_out = F.toDTSC(meta_out);
if (!pack_out.isEmpty()){
if (!sending){
counter++;
if (counter > 8){
sending = true;
meta_out.Pack(true);//pack metadata
meta_out.packed.replace(0, 4, DTSC::Magic_Header);//prepare proper header
SS.write(meta_out.packed);//write header/metadata
SS.write(prebuffer.str());//write buffer
prebuffer.str("");//clear buffer
SS.write(pack_out.Pack(true));//simply write
}else{
prebuffer << pack_out.Pack(true);//buffer
}
}else{
SS.write(pack_out.Pack(true));//simple write
}
}
}else{
#if DEBUG >= 4
fprintf(stderr, "Received useless media data\n");
#endif
Socket.close();
}
break;
case 15:
#if DEBUG >= 4
fprintf(stderr, "Received AFM3 data message\n");
#endif
break;
case 16:
#if DEBUG >= 4
fprintf(stderr, "Received AFM3 shared object\n");
#endif
break;
case 17:{
#if DEBUG >= 4
fprintf(stderr, "Received AFM3 command message\n");
#endif
if (next.data[0] != 0){
next.data = next.data.substr(1);
amf3data = AMF::parse3(next.data);
#if DEBUG >= 4
amf3data.Print();
#endif
}else{
#if DEBUG >= 4
fprintf(stderr, "Received AFM3-0 command message\n");
#endif
next.data = next.data.substr(1);
amfdata = AMF::parse(next.data);
parseAMFCommand(amfdata, 17, next.msg_stream_id);
}//parsing AMF0-style
} break;
case 19:
#if DEBUG >= 4
fprintf(stderr, "Received AFM0 shared object\n");
#endif
break;
case 20:{//AMF0 command message
amfdata = AMF::parse(next.data);
parseAMFCommand(amfdata, 20, next.msg_stream_id);
} break;
case 22:
#if DEBUG >= 4
fprintf(stderr, "Received aggregate message\n");
#endif
break;
default:
#if DEBUG >= 1
fprintf(stderr, "Unknown chunk received! Probably protocol corruption, stopping parsing of incoming data.\n");
#endif
Connector_RTMP::stopparsing = true;
break;
}
}
}//parseChunk
void Connector_RTMP::sendCommand(AMF::Object & amfreply, int messagetype, int stream_id){
if (messagetype == 17){
Socket.write(RTMPStream::SendChunk(3, messagetype, stream_id, (char)0+amfreply.Pack()));
}else{
Socket.write(RTMPStream::SendChunk(3, messagetype, stream_id, amfreply.Pack()));
}
}//sendCommand
void Connector_RTMP::parseAMFCommand(AMF::Object & amfdata, int messagetype, int stream_id){
#if DEBUG >= 4
fprintf(stderr, "Received command: %s\n", amfdata.Print().c_str());
#endif
#if DEBUG >= 3
fprintf(stderr, "AMF0 command: %s\n", amfdata.getContentP(0)->StrValue().c_str());
#endif
if (amfdata.getContentP(0)->StrValue() == "connect"){
double objencoding = 0;
if (amfdata.getContentP(2)->getContentP("objectEncoding")){
objencoding = amfdata.getContentP(2)->getContentP("objectEncoding")->NumValue();
}
fprintf(stderr, "Object encoding set to %e\n", objencoding);
#if DEBUG >= 4
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
RTMPStream::chunk_snd_max = 4096;
Socket.write(RTMPStream::SendCTL(1, RTMPStream::chunk_snd_max));//send chunk size max (msg 1)
Socket.write(RTMPStream::SendCTL(5, RTMPStream::snd_window_size));//send window acknowledgement size (msg 5)
Socket.write(RTMPStream::SendCTL(6, RTMPStream::rec_window_size));//send rec window acknowledgement size (msg 6)
Socket.write(RTMPStream::SendUSR(0, 1));//send UCM StreamBegin (0), stream 1
//send a _result reply
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("", "_result"));//result success
amfreply.addContent(amfdata.getContent(1));//same transaction ID
amfreply.addContent(AMF::Object(""));//server properties
amfreply.getContentP(2)->addContent(AMF::Object("fmsVer", "FMS/3,0,1,123"));
amfreply.getContentP(2)->addContent(AMF::Object("capabilities", (double)31));
//amfreply.getContentP(2)->addContent(AMF::Object("mode", (double)1));
amfreply.addContent(AMF::Object(""));//info
amfreply.getContentP(3)->addContent(AMF::Object("level", "status"));
amfreply.getContentP(3)->addContent(AMF::Object("code", "NetConnection.Connect.Success"));
amfreply.getContentP(3)->addContent(AMF::Object("description", "Connection succeeded."));
amfreply.getContentP(3)->addContent(AMF::Object("clientid", 1337));
amfreply.getContentP(3)->addContent(AMF::Object("objectEncoding", objencoding));
//amfreply.getContentP(3)->addContent(AMF::Object("data", AMF::AMF0_ECMA_ARRAY));
//amfreply.getContentP(3)->getContentP(4)->addContent(AMF::Object("version", "3,5,4,1004"));
#if DEBUG >= 4
amfreply.Print();
#endif
sendCommand(amfreply, messagetype, stream_id);
//send onBWDone packet - no clue what it is, but real server sends it...
//amfreply = AMF::Object("container", AMF::AMF0_DDV_CONTAINER);
//amfreply.addContent(AMF::Object("", "onBWDone"));//result
//amfreply.addContent(amfdata.getContent(1));//same transaction ID
//amfreply.addContent(AMF::Object("", (double)0, AMF::AMF0_NULL));//null
//sendCommand(amfreply, messagetype, stream_id);
return;
}//connect
if (amfdata.getContentP(0)->StrValue() == "createStream"){
//send a _result reply
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("", "_result"));//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("", (double)1));//stream ID - we use 1
#if DEBUG >= 4
amfreply.Print();
#endif
sendCommand(amfreply, messagetype, stream_id);
Socket.write(RTMPStream::SendUSR(0, 1));//send UCM StreamBegin (0), stream 1
return;
}//createStream
if ((amfdata.getContentP(0)->StrValue() == "closeStream") || (amfdata.getContentP(0)->StrValue() == "deleteStream")){
if (SS.connected()){SS.close();}
return;
}
if ((amfdata.getContentP(0)->StrValue() == "getStreamLength") || (amfdata.getContentP(0)->StrValue() == "getMovLen")){
//send a _result reply
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("", "_result"));//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("", (double)0));//zero length
#if DEBUG >= 4
amfreply.Print();
#endif
sendCommand(amfreply, messagetype, stream_id);
return;
}//getStreamLength
if ((amfdata.getContentP(0)->StrValue() == "publish")){
if (amfdata.getContentP(3)){
streamname = amfdata.getContentP(3)->StrValue();
SS = Socket::getStream(streamname);
if (!SS.connected()){
#if DEBUG >= 1
fprintf(stderr, "Could not connect to server!\n");
#endif
Socket.close();//disconnect user
return;
}
SS.write("P "+Socket.getHost()+'\n');
nostats = true;
#if DEBUG >= 4
fprintf(stderr, "Connected to buffer, starting to send data...\n");
#endif
}
//send a _result reply
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("", "_result"));//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("", 1, AMF::AMF0_BOOL));//publish success?
#if DEBUG >= 4
amfreply.Print();
#endif
sendCommand(amfreply, messagetype, stream_id);
Socket.write(RTMPStream::SendUSR(0, 1));//send UCM StreamBegin (0), stream 1
//send a status reply
amfreply = AMF::Object("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("", "onStatus"));//status reply
amfreply.addContent(AMF::Object("", 0, AMF::AMF0_NUMBER));//same transaction ID
amfreply.addContent(AMF::Object("", (double)0, AMF::AMF0_NULL));//null - command info
amfreply.addContent(AMF::Object(""));//info
amfreply.getContentP(3)->addContent(AMF::Object("level", "status"));
amfreply.getContentP(3)->addContent(AMF::Object("code", "NetStream.Publish.Start"));
amfreply.getContentP(3)->addContent(AMF::Object("description", "Stream is now published!"));
amfreply.getContentP(3)->addContent(AMF::Object("clientid", (double)1337));
#if DEBUG >= 4
amfreply.Print();
#endif
sendCommand(amfreply, messagetype, stream_id);
return;
}//getStreamLength
if (amfdata.getContentP(0)->StrValue() == "checkBandwidth"){
//send a _result reply
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("", "_result"));//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("", (double)0, AMF::AMF0_NULL));//null - command info
#if DEBUG >= 4
amfreply.Print();
#endif
sendCommand(amfreply, messagetype, stream_id);
return;
}//checkBandwidth
if ((amfdata.getContentP(0)->StrValue() == "play") || (amfdata.getContentP(0)->StrValue() == "play2")){
//send streambegin
streamname = amfdata.getContentP(3)->StrValue();
Socket.write(RTMPStream::SendUSR(0, 1));//send UCM StreamBegin (0), stream 1
//send a status reply
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("", "onStatus"));//status reply
amfreply.addContent(amfdata.getContent(1));//same transaction ID
amfreply.addContent(AMF::Object("", (double)0, AMF::AMF0_NULL));//null - command info
amfreply.addContent(AMF::Object(""));//info
amfreply.getContentP(3)->addContent(AMF::Object("level", "status"));
amfreply.getContentP(3)->addContent(AMF::Object("code", "NetStream.Play.Reset"));
amfreply.getContentP(3)->addContent(AMF::Object("description", "Playing and resetting..."));
amfreply.getContentP(3)->addContent(AMF::Object("details", "DDV"));
amfreply.getContentP(3)->addContent(AMF::Object("clientid", (double)1337));
#if DEBUG >= 4
amfreply.Print();
#endif
sendCommand(amfreply, messagetype, stream_id);
amfreply = AMF::Object("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("", "onStatus"));//status reply
amfreply.addContent(amfdata.getContent(1));//same transaction ID
amfreply.addContent(AMF::Object("", (double)0, AMF::AMF0_NULL));//null - command info
amfreply.addContent(AMF::Object(""));//info
amfreply.getContentP(3)->addContent(AMF::Object("level", "status"));
amfreply.getContentP(3)->addContent(AMF::Object("code", "NetStream.Play.Start"));
amfreply.getContentP(3)->addContent(AMF::Object("description", "Playing!"));
amfreply.getContentP(3)->addContent(AMF::Object("details", "DDV"));
amfreply.getContentP(3)->addContent(AMF::Object("clientid", (double)1337));
#if DEBUG >= 4
amfreply.Print();
#endif
sendCommand(amfreply, messagetype, stream_id);
RTMPStream::chunk_snd_max = 102400;//100KiB
Socket.write(RTMPStream::SendCTL(1, RTMPStream::chunk_snd_max));//send chunk size max (msg 1)
Connector_RTMP::ready4data = true;//start sending video data!
return;
}//createStream
#if DEBUG >= 2
fprintf(stderr, "AMF0 command not processed! :(\n");
#endif
}//parseAMFCommand
// Load main server setup file, default port 1935, handler is Connector_RTMP::Connector_RTMP
#define DEFAULT_PORT 1935
#define MAINHANDLER Connector_RTMP::Connector_RTMP
#define CONFIGSECT RTMP
#include "server_setup.cpp"

505
src/controller.cpp Normal file
View file

@ -0,0 +1,505 @@
#include <iostream>
#include <fstream>
#include <string>
#include <sstream>
#include <vector>
#include <map>
#include <set>
#include <cstdlib>
#include <queue>
#include <cmath>
#include <ctime>
#include <cstdio>
#include <climits>
#include <cstring>
#include <unistd.h>
#include <getopt.h>
#include <set>
#include <sys/time.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <signal.h>
#include <sstream>
#include "../lib/socket.h"
#include "../lib/http_parser.h"
#include "../lib/md5.h"
#include "../lib/json.h"
#include "../lib/procs.h"
#include "../lib/config.h"
#include "../lib/auth.h"
#define UPLINK_INTERVAL 30
Socket::Server API_Socket; ///< Main connection socket.
std::map<std::string, int> lastBuffer; ///< Last moment of contact with all buffers.
Auth keychecker; ///< Checks key authorization.
/// Basic signal handler. Disconnects the server_socket if it receives
/// a SIGINT, SIGHUP or SIGTERM signal, but does nothing for SIGPIPE.
/// Disconnecting the server_socket will terminate the main listening loop
/// and cleanly shut down the process.
void signal_handler (int signum){
switch (signum){
case SIGINT:
#if DEBUG >= 1
fprintf(stderr, "Received SIGINT - closing server socket.\n");
#endif
break;
case SIGHUP:
#if DEBUG >= 1
fprintf(stderr, "Received SIGHUP - closing server socket.\n");
#endif
break;
case SIGTERM:
#if DEBUG >= 1
fprintf(stderr, "Received SIGTERM - closing server socket.\n");
#endif
break;
default: return; break;
}
API_Socket.close();
}//signal_handler
JSON::Value Storage; ///< Global storage of data.
void WriteFile( std::string Filename, std::string contents ) {
std::ofstream File;
File.open( Filename.c_str( ) );
File << contents << std::endl;
File.close( );
}
class ConnectedUser{
public:
std::string writebuffer;
Socket::Connection C;
HTTP::Parser H;
bool Authorized;
bool clientMode;
int logins;
std::string Username;
ConnectedUser(Socket::Connection c){
C = c;
H.Clean();
logins = 0;
Authorized = false;
clientMode = false;
}
};
void Log(std::string kind, std::string message){
JSON::Value m;
m.append((long long int)time(0));
m.append(kind);
m.append(message);
Storage["log"].append(m);
std::cout << "[" << kind << "] " << message << std::endl;
}
void Authorize( JSON::Value & Request, JSON::Value & Response, ConnectedUser & conn ) {
time_t Time = time(0);
tm * TimeInfo = localtime(&Time);
std::stringstream Date;
std::string retval;
Date << TimeInfo->tm_mday << "-" << TimeInfo->tm_mon << "-" << TimeInfo->tm_year + 1900;
std::string Challenge = md5( Date.str().c_str() + conn.C.getHost() );
if( Request.isMember( "authorize" ) ) {
std::string UserID = Request["authorize"]["username"];
if (Storage["account"].isMember(UserID)){
if( md5( (std::string)(Storage["account"][UserID]["password"]) + Challenge ) == (std::string)Request["authorize"]["password"] ) {
Response["authorize"]["status"] = "OK";
conn.Username = UserID;
conn.Authorized = true;
return;
}
}
if (UserID != ""){
Log("AUTH", "Failed login attempt "+UserID+" @ "+conn.C.getHost());
}
conn.logins++;
}
conn.Username = "";
conn.Authorized = false;
Response["authorize"]["status"] = "CHALL";
Response["authorize"]["challenge"] = Challenge;
return;
}
void CheckProtocols(JSON::Value & p){
static std::map<std::string, std::string> current_connectors;
std::map<std::string, std::string> new_connectors;
std::map<std::string, std::string>::iterator iter;
std::string tmp;
JSON::Value counter = (long long int)0;
//collect object type
for (JSON::ObjIter jit = p.ObjBegin(); jit != p.ObjEnd(); jit++){
if (!jit->second.isMember("connector") || ((std::string)jit->second["connector"]) == ""){continue;}
if (!jit->second.isMember("port") || ((long long int)jit->second["port"]) == 0){continue;}
tmp = "MistConn";
tmp += (std::string)jit->second["connector"];
tmp += " -n -p ";
tmp += (std::string)jit->second["port"];
if (jit->second.isMember("interface") && ((std::string)jit->second["interface"]) != "" && ((std::string)jit->second["interface"]) != "0.0.0.0"){
tmp += " -i ";
tmp += (std::string)jit->second["interface"];
}
if (jit->second.isMember("username") && ((std::string)jit->second["username"]) != "" && ((std::string)jit->second["username"]) != "root"){
tmp += " -u ";
tmp += (std::string)jit->second["username"];
}
counter = (long long int)counter + 1;
new_connectors[std::string("Conn")+(std::string)counter] = tmp;
}
//collect array type
for (JSON::ArrIter ait = p.ArrBegin(); ait != p.ArrEnd(); ait++){
if (!(*ait).isMember("connector") || ((std::string)(*ait)["connector"]) == ""){continue;}
if (!(*ait).isMember("port") || ((long long int)(*ait)["port"]) == 0){continue;}
tmp = "MistConn";
tmp += (std::string)(*ait)["connector"];
tmp += " -n -p ";
tmp += (std::string)(*ait)["port"];
if ((*ait).isMember("interface") && ((std::string)(*ait)["interface"]) != "" && ((std::string)(*ait)["interface"]) != "0.0.0.0"){
tmp += " -i ";
tmp += (std::string)(*ait)["interface"];
}
if ((*ait).isMember("username") && ((std::string)(*ait)["username"]) != "" && ((std::string)(*ait)["username"]) != "root"){
tmp += " -u ";
tmp += (std::string)(*ait)["username"];
}
counter = (long long int)counter + 1;
new_connectors[std::string("Conn")+(std::string)counter] = tmp;
}
//shut down deleted/changed connectors
for (iter = current_connectors.begin(); iter != current_connectors.end(); iter++){
if (new_connectors.count(iter->first) != 1 || new_connectors[iter->first] != iter->second){
Util::Procs::Stop(iter->first);
}
}
//start up new/changed connectors
for (iter = new_connectors.begin(); iter != new_connectors.end(); iter++){
if (current_connectors.count(iter->first) != 1 || current_connectors[iter->first] != iter->second || !Util::Procs::isActive(iter->first)){
Util::Procs::Start(iter->first, iter->second);
}
}
//store new state
current_connectors = new_connectors;
}
void CheckConfig(JSON::Value & in, JSON::Value & out){
for (JSON::ObjIter jit = in.ObjBegin(); jit != in.ObjEnd(); jit++){
if (out.isMember(jit->first)){
if (jit->second != out[jit->first]){
Log("CONF", std::string("Updated configuration value ")+jit->first);
}
}else{
Log("CONF", std::string("New configuration value ")+jit->first);
}
}
for (JSON::ObjIter jit = out.ObjBegin(); jit != out.ObjEnd(); jit++){
if (!in.isMember(jit->first)){
Log("CONF", std::string("Deleted configuration value ")+jit->first);
}
}
out = in;
out["version"] = TOSTRING(PACKAGE_VERSION);
}
bool streamsEqual(JSON::Value & one, JSON::Value & two){
if (one["channel"]["URL"] != two["channel"]["URL"]){return false;}
if (one["preset"]["cmd"] != two["preset"]["cmd"]){return false;}
return true;
}
void startStream(std::string name, JSON::Value & data){
Log("BUFF", "(re)starting stream buffer "+name);
std::string URL = data["channel"]["URL"];
std::string preset = data["preset"]["cmd"];
std::string cmd1, cmd2, cmd3;
if (URL.substr(0, 4) == "push"){
std::string pusher = URL.substr(7);
cmd2 = "MistBuffer "+name+" "+pusher;
Util::Procs::Start(name, cmd2);
}else{
if (URL.substr(0, 1) == "/"){
cmd1 = "cat "+URL;
}else{
cmd1 = "ffmpeg -re -async 2 -i "+URL+" "+preset+" -f flv -";
cmd2 = "MistFLV2DTSC";
}
cmd3 = "MistBuffer "+name;
if (cmd2 != ""){
Util::Procs::Start(name, cmd1, cmd2, cmd3);
}else{
Util::Procs::Start(name, cmd1, cmd3);
}
}
}
void CheckAllStreams(JSON::Value & data){
unsigned int currTime = time(0);
bool changed = false;
for (JSON::ObjIter jit = data.ObjBegin(); jit != data.ObjEnd(); jit++){
if (!Util::Procs::isActive(jit->first)){
startStream(jit->first, jit->second);
}
if (currTime - lastBuffer[jit->first] > 5){
if ((long long int)jit->second["online"] != 0){changed = true;}
jit->second["online"] = 0;
}else{
if ((long long int)jit->second["online"] != 1){changed = true;}
jit->second["online"] = 1;
}
}
if (changed){
WriteFile("/tmp/mist/streamlist", Storage.toString());
}
}
void CheckStreams(JSON::Value & in, JSON::Value & out){
bool changed = false;
for (JSON::ObjIter jit = in.ObjBegin(); jit != in.ObjEnd(); jit++){
if (out.isMember(jit->first)){
if (!streamsEqual(jit->second, out[jit->first])){
Log("STRM", std::string("Updated stream ")+jit->first);
changed = true;
Util::Procs::Stop(jit->first);
startStream(jit->first, jit->second);
}
}else{
Log("STRM", std::string("New stream ")+jit->first);
changed = true;
startStream(jit->first, jit->second);
}
}
for (JSON::ObjIter jit = out.ObjBegin(); jit != out.ObjEnd(); jit++){
if (!in.isMember(jit->first)){
Log("STRM", std::string("Deleted stream ")+jit->first);
changed = true;
Util::Procs::Stop(jit->first);
}
}
out = in;
if (changed){
WriteFile("/tmp/mist/streamlist", Storage.toString());
}
}
int main(int argc, char ** argv){
//setup signal handler
struct sigaction new_action;
new_action.sa_handler = signal_handler;
sigemptyset (&new_action.sa_mask);
new_action.sa_flags = 0;
sigaction(SIGINT, &new_action, NULL);
sigaction(SIGHUP, &new_action, NULL);
sigaction(SIGTERM, &new_action, NULL);
sigaction(SIGPIPE, &new_action, NULL);
Storage = JSON::fromFile("config.json");
Util::Config C;
C.listen_port = (long long int)Storage["config"]["controller"]["port"];
if (C.listen_port < 1){C.listen_port = 4242;}
C.interface = (std::string)Storage["config"]["controller"]["interface"];
if (C.interface == ""){C.interface = "0.0.0.0";}
C.username = (std::string)Storage["config"]["controller"]["username"];
if (C.username == ""){C.username = "root";}
C.parseArgs(argc, argv);
time_t lastuplink = 0;
time_t processchecker = 0;
API_Socket = Socket::Server(C.listen_port, C.interface, true);
mkdir("/tmp/mist", S_IRWXU | S_IRWXG | S_IRWXO);//attempt to create /tmp/mist/ - ignore failures
Socket::Server Stats_Socket = Socket::Server("/tmp/mist/statistics", true);
Util::setUser(C.username);
if (C.daemon_mode){
Util::Daemonize();
}
Socket::Connection Incoming;
std::vector< ConnectedUser > users;
std::vector<Socket::Connection> buffers;
JSON::Value Request;
JSON::Value Response;
std::string jsonp;
ConnectedUser * uplink = 0;
while (API_Socket.connected()){
usleep(100000); //sleep for 100 ms - prevents 100% CPU time
if (time(0) - processchecker > 10){
processchecker = time(0);
CheckProtocols(Storage["config"]["protocols"]);
CheckAllStreams(Storage["streams"]);
}
if (time(0) - lastuplink > UPLINK_INTERVAL){
lastuplink = time(0);
bool gotUplink = false;
if (users.size() > 0){
for( std::vector< ConnectedUser >::iterator it = users.end() - 1; it >= users.begin(); it--) {
if (!it->C.connected()){
it->C.close();
users.erase(it);
break;
}
if (it->clientMode){uplink = &*it; gotUplink = true;}
}
}
if (!gotUplink){
Incoming = Socket::Connection("gearbox.ddvtech.com", 4242, true);
if (Incoming.connected()){
users.push_back(Incoming);
users.back().clientMode = true;
uplink = &users.back();
gotUplink = true;
}
}
if (gotUplink){
Response.null(); //make sure no data leaks from previous requests
Response["config"] = Storage["config"];
Response["streams"] = Storage["streams"];
Response["log"] = Storage["log"];
Response["statistics"] = Storage["statistics"];
Response["now"] = (unsigned int)lastuplink;
uplink->H.Clean();
uplink->H.SetBody("command="+HTTP::Parser::urlencode(Response.toString()));
uplink->H.BuildRequest();
uplink->writebuffer += uplink->H.BuildResponse("200", "OK");
uplink->H.Clean();
//Log("UPLK", "Sending server data to uplink.");
}else{
Log("UPLK", "Could not connect to uplink.");
}
}
Incoming = API_Socket.accept();
if (Incoming.connected()){users.push_back(Incoming);}
Incoming = Stats_Socket.accept();
if (Incoming.connected()){buffers.push_back(Incoming);}
if (buffers.size() > 0){
for( std::vector< Socket::Connection >::iterator it = buffers.begin(); it != buffers.end(); it++) {
if (!it->connected()){
it->close();
buffers.erase(it);
break;
}
it->spool();
if (it->Received() != ""){
size_t newlines = it->Received().find("\n\n");
while (newlines != std::string::npos){
Request = it->Received().substr(0, newlines);
if (Request.isMember("totals") && Request["totals"].isMember("buffer")){
std::string thisbuffer = Request["totals"]["buffer"];
lastBuffer[thisbuffer] = time(0);
Storage["statistics"][thisbuffer]["curr"] = Request["curr"];
std::stringstream st;
st << (long long int)Request["totals"]["now"];
std::string nowstr = st.str();
Storage["statistics"][thisbuffer]["totals"][nowstr] = Request["totals"];
for (JSON::ObjIter jit = Request["log"].ObjBegin(); jit != Request["log"].ObjEnd(); jit++){
Storage["statistics"][thisbuffer]["log"].append(jit->second);
}
}
it->Received().erase(0, newlines+2);
newlines = it->Received().find("\n\n");
}
}
}
}
if (users.size() > 0){
for( std::vector< ConnectedUser >::iterator it = users.begin(); it != users.end(); it++) {
if (!it->C.connected() || it->logins > 3){
it->C.close();
users.erase(it);
break;
}
if (it->writebuffer != ""){
it->C.iwrite(it->writebuffer);
}
if (it->H.Read(it->C)){
Response.null(); //make sure no data leaks from previous requests
if (it->clientMode){
// In clientMode, requests are reversed. These are connections we initiated to GearBox.
// They are assumed to be authorized, but authorization to gearbox is still done.
// This authorization uses the compiled-in username and password (account).
Request = JSON::fromString(it->H.body);
if (Request["authorize"]["status"] != "OK"){
if (Request["authorize"].isMember("challenge")){
it->logins++;
if (it->logins > 2){
Log("UPLK", "Max login attempts passed - dropping connection to uplink.");
it->C.close();
}else{
Response["config"] = Storage["config"];
Response["streams"] = Storage["streams"];
Response["log"] = Storage["log"];
Response["statistics"] = Storage["statistics"];
Response["authorize"]["username"] = TOSTRING(COMPILED_USERNAME);
Log("UPLK", "Responding to login challenge: " + (std::string)Request["authorize"]["challenge"]);
Response["authorize"]["password"] = md5(TOSTRING(COMPILED_PASSWORD) + (std::string)Request["authorize"]["challenge"]);
it->H.Clean();
it->H.SetBody("command="+HTTP::Parser::urlencode(Response.toString()));
it->H.BuildRequest();
it->writebuffer += it->H.BuildResponse("200", "OK");
it->H.Clean();
Log("UPLK", "Attempting login to uplink.");
}
}
}else{
if (Request.isMember("config")){CheckConfig(Request["config"], Storage["config"]);}
if (Request.isMember("streams")){CheckStreams(Request["streams"], Storage["streams"]);}
if (Request.isMember("clearstatlogs")){
Storage["log"].null();
Storage["statistics"].null();
}
}
}else{
Request = JSON::fromString(it->H.GetVar("command"));
std::cout << "Request: " << Request.toString() << std::endl;
Authorize(Request, Response, (*it));
if (it->Authorized){
//Parse config and streams from the request.
if (Request.isMember("config")){CheckConfig(Request["config"], Storage["config"]);}
if (Request.isMember("streams")){CheckStreams(Request["streams"], Storage["streams"]);}
//sent current configuration, no matter if it was changed or not
//Response["streams"] = Storage["streams"];
Response["config"] = Storage["config"];
Response["streams"] = Storage["streams"];
//add required data to the current unix time to the config, for syncing reasons
Response["config"]["time"] = (long long int)time(0);
if (!Response["config"].isMember("serverid")){Response["config"]["serverid"] = "";}
//sent any available logs and statistics
Response["log"] = Storage["log"];
Response["statistics"] = Storage["statistics"];
//clear log and statistics to prevent useless data transfer
Storage["log"].null();
Storage["statistics"].null();
}
jsonp = "";
if (it->H.GetVar("callback") != ""){jsonp = it->H.GetVar("callback");}
if (it->H.GetVar("jsonp") != ""){jsonp = it->H.GetVar("jsonp");}
it->H.Clean();
it->H.protocol = "HTTP/1.0";
it->H.SetHeader("Content-Type", "text/javascript");
if (jsonp == ""){
it->H.SetBody(Response.toString()+"\n\n");
}else{
it->H.SetBody(jsonp+"("+Response.toString()+");\n\n");
}
it->writebuffer += it->H.BuildResponse("200", "OK");
it->H.Clean();
}
}
}
}
}
Util::Procs::StopAll();
WriteFile("config.json", Storage.toString());
std::cout << "Killed all processes, wrote config to disk. Exiting." << std::endl;
return 0;
}

64
src/converters/dtsc2flv.cpp Normal file
View file

@ -0,0 +1,64 @@
/// \file DTSC2FLV/main.cpp
/// Contains the code that will transform any valid DTSC input into valid FLVs.
#include <iostream>
#include <sstream>
#include <string>
#include <vector>
#include <cstdlib>
#include <cstdio>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include "../../util/flv_tag.h" //FLV support
#include "../../util/dtsc.h" //DTSC support
#include "../../util/amf.h" //AMF support
/// Holds all code that converts filetypes to DTSC.
namespace Converters{
/// Reads DTSC from STDIN, outputs FLV to STDOUT.
int DTSC2FLV() {
FLV::Tag FLV_out; // Temporary storage for outgoing FLV data.
DTSC::Stream Strm;
std::string inBuffer;
char charBuffer[1024*10];
unsigned int charCount;
bool doneheader = false;
while (std::cin.good()){
std::cin.read(charBuffer, 1024*10);
charCount = std::cin.gcount();
inBuffer.append(charBuffer, charCount);
if (Strm.parsePacket(inBuffer)){
if (!doneheader){
doneheader = true;
std::cout.write(FLV::Header, 13);
FLV_out.DTSCMetaInit(Strm);
std::cout.write(FLV_out.data, FLV_out.len);
if (Strm.metadata.getContentP("video") && Strm.metadata.getContentP("video")->getContentP("init")){
FLV_out.DTSCVideoInit(Strm);
std::cout.write(FLV_out.data, FLV_out.len);
}
if (Strm.metadata.getContentP("audio") && Strm.metadata.getContentP("audio")->getContentP("init")){
FLV_out.DTSCAudioInit(Strm);
std::cout.write(FLV_out.data, FLV_out.len);
}
}
if (FLV_out.DTSCLoader(Strm)){
std::cout.write(FLV_out.data, FLV_out.len);
}
}
}
std::cerr << "Done!" << std::endl;
return 0;
}//FLV2DTSC
};//Converter namespace
/// Entry point for DTSC2FLV, simply calls Converters::DTSC2FLV().
int main(){
return Converters::DTSC2FLV();
}//main

70
src/converters/flv2dtsc.cpp Normal file
View file

@ -0,0 +1,70 @@
/// \file FLV2DTSC/main.cpp
/// Contains the code that will transform any valid FLV input into valid DTSC.
#include <iostream>
#include <sstream>
#include <string>
#include <vector>
#include <cstdlib>
#include <cstdio>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include "../../util/flv_tag.h" //FLV support
#include "../../util/dtsc.h" //DTSC support
#include "../../util/amf.h" //AMF support
/// Holds all code that converts filetypes to DTSC.
namespace Converters{
/// Reads FLV from STDIN, outputs DTSC to STDOUT.
int FLV2DTSC() {
FLV::Tag FLV_in; // Temporary storage for incoming FLV data.
DTSC::DTMI meta_out; // Storage for outgoing DTMI header data.
DTSC::DTMI pack_out; // Storage for outgoing DTMI data.
std::stringstream prebuffer; // Temporary buffer before sending real data
bool sending = false;
unsigned int counter = 0;
while (!feof(stdin)){
if (FLV_in.FileLoader(stdin)){
pack_out = FLV_in.toDTSC(meta_out);
if (pack_out.isEmpty()){continue;}
if (!sending){
counter++;
if (counter > 8){
sending = true;
meta_out.Pack(true);
meta_out.packed.replace(0, 4, DTSC::Magic_Header);
std::cout << meta_out.packed;
std::cout << prebuffer.rdbuf();
prebuffer.str("");
std::cerr << "Buffer done, starting real-time output..." << std::endl;
}else{
prebuffer << pack_out.Pack(true);//buffer
continue;//don't also write
}
}
std::cout << pack_out.Pack(true);//simply write
}
}
// if the FLV input is very short, do output it correctly...
if (!sending){
std::cerr << "EOF - outputting buffer..." << std::endl;
meta_out.Pack(true);
meta_out.packed.replace(0, 4, DTSC::Magic_Header);
std::cout << meta_out.packed;
std::cout << prebuffer.rdbuf();
}
std::cerr << "Done!" << std::endl;
return 0;
}//FLV2DTSC
};//Buffer namespace
/// Entry point for FLV2DTSC, simply calls Converters::FLV2DTSC().
int main(){
return Converters::FLV2DTSC();
}//main

115
src/server_setup.cpp Normal file
View file

@ -0,0 +1,115 @@
/// \file server_setup.cpp
/// Contains generic functions for setting up a DDVTECH Connector.
#ifndef MAINHANDLER
/// Handler that is called for accepted incoming connections.
#define MAINHANDLER NoHandler
#error "No handler was set!"
#endif
#ifndef DEFAULT_PORT
/// Default port for this server.
#define DEFAULT_PORT 0
#error "No default port was set!"
#endif
#include "../lib/socket.h" //Socket library
#include "../lib/config.h" //utilities for config management
#include <signal.h>
#include <sys/types.h>
#include <pwd.h>
#include <fstream>
Socket::Server server_socket; ///< Placeholder for the server socket
/// Basic signal handler. Disconnects the server_socket if it receives
/// a SIGINT, SIGHUP or SIGTERM signal, but does nothing for SIGPIPE.
/// Disconnecting the server_socket will terminate the main listening loop
/// and cleanly shut down the process.
void signal_handler (int signum){
switch (signum){
case SIGINT:
#if DEBUG >= 1
fprintf(stderr, "Received SIGINT - closing server socket.\n");
#endif
break;
case SIGHUP:
#if DEBUG >= 1
fprintf(stderr, "Received SIGHUP - closing server socket.\n");
#endif
break;
case SIGTERM:
#if DEBUG >= 1
fprintf(stderr, "Received SIGTERM - closing server socket.\n");
#endif
break;
case SIGCHLD:
wait(0);
return;
break;
default: return; break;
}
if (!server_socket.connected()) return;
server_socket.close();
}//signal_handler
/// Generic main entry point and loop for DDV Connectors.
/// This sets up the proper termination handler, checks commandline options,
/// parses config files and opens a listening socket on the requested port.
/// Any incoming connections will be accepted and start up the function #MAINHANDLER,
/// which should be defined before including server_setup.cpp.
/// The default port is set by define #DEFAULT_PORT.
/// The configuration file section is set by define #CONFIGSECT.
int main(int argc, char ** argv){
Socket::Connection S;//placeholder for incoming connections
//setup signal handler
struct sigaction new_action;
new_action.sa_handler = signal_handler;
sigemptyset (&new_action.sa_mask);
new_action.sa_flags = 0;
sigaction(SIGINT, &new_action, NULL);
sigaction(SIGHUP, &new_action, NULL);
sigaction(SIGTERM, &new_action, NULL);
sigaction(SIGPIPE, &new_action, NULL);
sigaction(SIGCHLD, &new_action, NULL);
//set and parse configuration
Util::Config C;
C.listen_port = DEFAULT_PORT;
C.parseArgs(argc, argv);
//setup a new server socket, for the correct interface and port
server_socket = Socket::Server(C.listen_port, C.interface);
#if DEBUG >= 3
fprintf(stderr, "Made a listening socket on %s:%i...\n", C.interface.c_str(), C.listen_port);
#endif
if (!server_socket.connected()){
#if DEBUG >= 1
fprintf(stderr, "Error: could not make listening socket\n");
#endif
return 1;
}
Util::setUser(C.username);
if (C.daemon_mode){Util::Daemonize();}
while (server_socket.connected()){
S = server_socket.accept();
if (S.connected()){//check if the new connection is valid
pid_t myid = fork();
if (myid == 0){//if new child, start MAINHANDLER
return MAINHANDLER(S);
}else{//otherwise, do nothing or output debugging text
#if DEBUG >= 3
fprintf(stderr, "Spawned new process %i for socket %i\n", (int)myid, S.getSocket());
#endif
}
}
}//while connected
#if DEBUG >= 1
fprintf(stderr, "Server socket closed, exiting.\n");
#endif
return 0;
}//main