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

First compiling version of DTSC-powered Buffer. Merged DTSC and DTMI into DTSC for ease of use. Todo: FLV2DTSC executable, DTSC support in all connectors...

Browse source
This commit is contained in:
Thulinma 2011-09-13 00:32:35 +02:00
parent 7ee6500c3b
commit f9f13a1fa1
8 changed files with 520 additions and 500 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Buffer/Makefile
View file

@ -1,4 +1,4 @@
SRC = main.cpp ../util/json/json_reader.cpp ../util/json/json_value.cpp ../util/json/json_writer.cpp ../util/socket.cpp ../util/flv_tag.cpp
SRC = main.cpp ../util/json/json_reader.cpp ../util/json/json_value.cpp ../util/json/json_writer.cpp ../util/socket.cpp ../util/dtsc.cpp
OBJ = $(SRC:.cpp=.o)
OUT = DDV_Buffer
INCLUDES =

197
Buffer/main.cpp
View file

@ -10,8 +10,7 @@
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include <sstream>
#include "../util/flv_tag.h" //FLV format parser
#include "../util/dtsc.h" //DTSC support
#include "../util/socket.h" //Socket lib
#include "../util/json/json.h"
@ -28,11 +27,7 @@ namespace Buffer{
}
}
///holds FLV::Tag objects and their numbers
struct buffer{
int number;
FLV::Tag FLV;
};//buffer
DTSC::Stream * Strm = 0;
/// Converts a stats line to up, down, host, connector and conntime values.
class Stats{
@ -76,9 +71,7 @@ namespace Buffer{
/// Keeps track of what buffer users are using and the connection status.
class user{
public:
int MyBuffer; ///< Index of currently used buffer.
int MyBuffer_num; ///< Number of currently used buffer.
int MyBuffer_len; ///< Length in bytes of currently used buffer.
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.
int currsend; ///< Current amount of bytes sent.
@ -97,11 +90,16 @@ namespace Buffer{
std::stringstream st;
st << MyNum;
MyStr = st.str();
gotproperaudio = false;
curr_up = 0;
curr_down = 0;
currsend = 0;
myRing = Strm->getRing();
std::cout << "User " << MyNum << " connected" << std::endl;
}//constructor
/// Drops held DTSC::Ring class, if one is held.
~user(){
Strm->dropRing(myRing);
}//destructor
/// 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) {
@ -119,70 +117,41 @@ namespace Buffer{
}//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 doSend(){
int r = S.iwrite((char*)lastpointer+currsend, MyBuffer_len-currsend);
bool 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 == MyBuffer_len);
return (currsend == len);
}//doSend
/// Try to send data to this user. Disconnects if any problems occur.
/// \param ringbuf Array of buffers (FLV:Tag with ID attached)
/// \param buffers Count of elements in ringbuf
void Send(buffer ** ringbuf, int buffers){
/// \todo For MP3: gotproperaudio - if false, only send if first byte is 0xFF and set to true
void Send(){
if (!S.connected()){return;}//cancel if not connected
//still waiting for next buffer? check it
if (MyBuffer_num < 0){
MyBuffer_num = ringbuf[MyBuffer]->number;
if (MyBuffer_num < 0){
return; //still waiting? don't crash - wait longer.
}else{
MyBuffer_len = ringbuf[MyBuffer]->FLV.len;
lastpointer = ringbuf[MyBuffer]->FLV.data;
}
}
//do check for buffer resizes
if (lastpointer != ringbuf[MyBuffer]->FLV.data){
Disconnect("Buffer resize at wrong time... had to disconnect");
return;
}
if (myRing->waiting){return;}//still waiting for next buffer?
//try to complete a send
if (doSend()){
if (doSend(Strm->outPacket(myRing->b).c_str(), Strm->outPacket(myRing->b).length())){
//switch to next buffer
if ((ringbuf[MyBuffer]->number != MyBuffer_num)){
//if corrupt data, warn and find keyframe
std::cout << "Warning: User " << MyNum << " was send corrupt video data and send to the next keyframe!" << std::endl;
int nocrashcount = 0;
do{
MyBuffer++;
nocrashcount++;
MyBuffer %= buffers;
}while(!ringbuf[MyBuffer]->FLV.isKeyframe && (nocrashcount < buffers));
//if keyframe not available, try again later
if (nocrashcount >= buffers){
std::cout << "Warning: No keyframe found in buffers! Skipping search for now..." << std::endl;
return;
}
}else{
MyBuffer++;
MyBuffer %= buffers;
if (myRing->b <= 0){myRing->waiting = true; return;}//no next buffer? go in waiting mode.
myRing->b--;
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;
Strm->dropRing(myRing);
myRing = Strm->getRing();
}
MyBuffer_num = -1;
lastpointer = 0;
currsend = 0;
}//completed a send
}//send
};
int user::UserCount = 0;
/// Starts a loop, waiting for connections to send video data to.
/// 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;
@ -193,31 +162,26 @@ namespace Buffer{
//then check and parse the commandline
if (argc < 3) {
std::cout << "usage: " << argv[0] << " buffers_count streamname [awaiting_IP]" << std::endl;
std::cout << "usage: " << argv[0] << " streamname [awaiting_IP]" << std::endl;
return 1;
}
std::string waiting_ip = "";
bool ip_waiting = false;
Socket::Connection ip_input;
if (argc >= 4){
waiting_ip += argv[3];
waiting_ip += argv[2];
ip_waiting = true;
}
std::string shared_socket = "/tmp/shared_socket_";
shared_socket += argv[2];
shared_socket += argv[1];
Socket::Server SS(shared_socket, true);
FLV::Tag metadata;
FLV::Tag video_init;
FLV::Tag audio_init;
int buffers = atoi(argv[1]);
buffer ** ringbuf = (buffer**) calloc (buffers,sizeof(buffer*));
Strm = new DTSC::Stream(5);
std::vector<user> users;
std::vector<user>::iterator usersIt;
for (int i = 0; i < buffers; ++i) ringbuf[i] = new buffer;
int current_buffer = 0;
int lastproper = 0;//last properly finished buffer number
unsigned int loopcount = 0;
std::string inBuffer;
char charBuffer[1024*10];
unsigned int charCount;
unsigned int stattimer = 0;
Socket::Connection incoming;
Socket::Connection std_input(fileno(stdin));
@ -226,11 +190,7 @@ namespace Buffer{
Storage["log"] = Json::Value(Json::objectValue);
Storage["curr"] = Json::Value(Json::objectValue);
Storage["totals"] = Json::Value(Json::objectValue);
unsigned char packtype;
bool gotVideoInfo = false;
bool gotAudioInfo = false;
bool gotData = false;
while((!feof(stdin) || ip_waiting) && !FLV::Parse_Error){
usleep(1000); //sleep for 1 ms, to prevent 100% CPU time
@ -259,89 +219,21 @@ namespace Buffer{
}
}
//invalidate the current buffer
ringbuf[current_buffer]->number = -1;
if (
(!ip_waiting &&
(std_input.canRead()) && ringbuf[current_buffer]->FLV.FileLoader(stdin)
) || (ip_waiting && (ip_input.connected()) &&
ringbuf[current_buffer]->FLV.SockLoader(ip_input)
)
){
loopcount++;
packtype = ringbuf[current_buffer]->FLV.data[0];
//store metadata, if available
if (packtype == 0x12){
metadata = ringbuf[current_buffer]->FLV;
std::cout << "Received metadata!" << std::endl;
if (gotVideoInfo && gotAudioInfo){
FLV::Parse_Error = true;
std::cout << "... after proper video and audio? Cancelling broadcast!" << std::endl;
}
gotVideoInfo = false;
gotAudioInfo = false;
}
//store video init data, if available
if (!gotVideoInfo && ringbuf[current_buffer]->FLV.isKeyframe){
if ((ringbuf[current_buffer]->FLV.data[11] & 0x0f) == 7){//avc packet
if (ringbuf[current_buffer]->FLV.data[12] == 0){
ringbuf[current_buffer]->FLV.tagTime(0);//timestamp to zero
video_init = ringbuf[current_buffer]->FLV;
gotVideoInfo = true;
std::cout << "Received video configuration!" << std::endl;
}
}else{gotVideoInfo = true;}//non-avc = no config...
}
//store audio init data, if available
if (!gotAudioInfo && (packtype == 0x08)){
if (((ringbuf[current_buffer]->FLV.data[11] & 0xf0) >> 4) == 10){//aac packet
ringbuf[current_buffer]->FLV.tagTime(0);//timestamp to zero
audio_init = ringbuf[current_buffer]->FLV;
gotAudioInfo = true;
std::cout << "Received audio configuration!" << std::endl;
}else{gotAudioInfo = true;}//no aac = no config...
}
//on keyframe set possible start point
if (packtype == 0x09){
if (((ringbuf[current_buffer]->FLV.data[11] & 0xf0) >> 4) == 1){
lastproper = current_buffer;
}
}
if (loopcount > 5){gotData = true;}
//keep track of buffers
ringbuf[current_buffer]->number = loopcount;
current_buffer++;
current_buffer %= buffers;
if ( (!ip_waiting && std_input.canRead()) || (ip_waiting && ip_input.connected()) ){
std::cin.read(charBuffer, 1024*10);
charCount = std::cin.gcount();
inBuffer.append(charBuffer, charCount);
Strm->parsePacket(inBuffer);
}
//check for new connections, accept them if there are any
incoming = SS.accept(true);
if (incoming.connected()){
users.push_back(incoming);
//send the FLV header
users.back().currsend = 0;
users.back().MyBuffer = lastproper;
users.back().MyBuffer_num = -1;
/// \todo Do this more nicely?
if (gotData){
if (!users.back().S.write(FLV::Header, 13)){
users.back().Disconnect("failed to receive the header!");
}else{
if (metadata.len > 0){
if (!users.back().S.write(metadata.data, metadata.len)){
users.back().Disconnect("failed to receive metadata!");
}
}
if (audio_init.len > 0){
if (!users.back().S.write(audio_init.data, audio_init.len)){
users.back().Disconnect("failed to receive audio init!");
}
}
if (video_init.len > 0){
if (!users.back().S.write(video_init.data, video_init.len)){
users.back().Disconnect("failed to receive video init!");
}
}
}
//send the header
if (!users.back().S.write(Strm->outHeader())){
/// \todo Do this more nicely?
users.back().Disconnect("failed to receive the header!");
}
}
@ -391,18 +283,15 @@ namespace Buffer{
}
}
}
(*usersIt).Send(ringbuf, buffers);
(*usersIt).Send();
}
}
}
}//main loop
// disconnect listener
if (FLV::Parse_Error){
std::cout << "FLV parse error" << std::endl;
}else{
std::cout << "Reached EOF of input" << std::endl;
}
/// \todo Deal with EOF more nicely - doesn't send the end of the stream to all users!
std::cout << "Reached EOF of input" << std::endl;
SS.close();
while (users.size() > 0){
for (usersIt = users.begin(); usersIt != users.end(); usersIt++){