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

Werkende RTMP connector! WHEEEE! Also, RTMPf weggegooit. Er is geen ruimte voor faal in dit bedrijf!

Browse source
This commit is contained in:
Thulinma 2010-11-17 01:05:36 +01:00
parent fbd0520264
commit 273f30784b
22 changed files with 123 additions and 2015 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Connector_RTMP/Makefile
View file

@ -11,7 +11,7 @@ LIBS = -lssl -lcrypto
.PHONY: clean default
default: $(OUT)
.cpp.o:
$(CC) $(INCLUDES) $(CCFLAGS) $(LIBS) -c $< -o $@
$(CC) $(INCLUDES) $(CCFLAGS) -c $< -o $@
$(OUT): $(OBJ) chunkstream.cpp parsechunks.cpp handshake.cpp crypto.cpp amf.cpp
$(CC) $(LIBS) -o $(OUT) $(OBJ)
clean:

2
Connector_RTMP/amf.cpp
View file

@ -270,7 +270,7 @@ AMFType parseOneAMF(const unsigned char *& data, unsigned int &len, unsigned int
return ret;
} break;
}
#ifdef DEBUG
#if DEBUG >= 2
fprintf(stderr, "Error: Unimplemented AMF type %hhx - returning.\n", data[i]);
#endif
return AMFType("error", (unsigned char)0xFF);

2
Connector_RTMP/chunkstream.cpp
View file

@ -457,7 +457,7 @@ chunkpack * AddChunkPart(chunkpack newchunk){
p = it->second;
tmpdata = (unsigned char*)realloc(p->data, p->real_len + newchunk.real_len);
if (tmpdata == 0){
#ifdef DEBUG
#if DEBUG >= 1
fprintf(stderr, "Error allocating memory!\n");
#endif
return 0;

2
Connector_RTMP/crypto.cpp
View file

@ -497,7 +497,7 @@ bool ValidateClientScheme(uint8_t * pBuffer, uint8_t scheme) {
uint8_t *pTempHash = new uint8_t[512];
HMACsha256(pTempBuffer, 1536 - 32, genuineFPKey, 30, pTempHash);
bool result = (memcmp(pBuffer+clientDigestOffset, pTempHash, 32) == 0);
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Client scheme validation %hhi %s\n", scheme, result?"success":"failed");
#endif
delete[] pTempBuffer;

4
Connector_RTMP/handshake.cpp
View file

@ -69,14 +69,14 @@ bool doHandshake(){
for (int i = 8; i < 3072; ++i){Server[i] = versionstring[i%13];}//"random" data
bool encrypted = (Version == 6);
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Handshake version is %hhi\n", Version);
#endif
uint8_t _validationScheme = 5;
if (ValidateClientScheme(Client, 0)) _validationScheme = 0;
if (ValidateClientScheme(Client, 1)) _validationScheme = 1;
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Handshake type is %hhi, encryption is %s\n", _validationScheme, encrypted?"on":"off");
#endif

129
Connector_RTMP/main.cpp
View file

@ -1,4 +1,10 @@
#undef DEBUG
//debugging level 0 = nothing
//debugging level 1 = critical errors
//debugging level 2 = errors
//debugging level 3 = status information
//debugging level 4 = extremely verbose status information
#define DEBUG 3
#include <iostream>
#include <cstdlib>
#include <cstdio>
@ -27,6 +33,12 @@ int server_socket = 0;
void termination_handler (int signum){
if (server_socket == 0) return;
switch (signum){
case SIGINT: break;
case SIGHUP: break;
case SIGTERM: break;
default: return; break;
}
close(server_socket);
server_socket = 0;
}
@ -41,9 +53,16 @@ int main(int argc, char ** argv){
sigaction (SIGHUP, &new_action, NULL);
sigaction (SIGTERM, &new_action, NULL);
server_socket = DDV_Listen(1936);
if ((argc < 2) || (argv[1] == "nd")){
if (server_socket > 0){daemon(1, 0);}else{return 1;}
server_socket = DDV_Listen(1935);
fprintf(stderr, "Made a listening socket on port 1936...");
if ((argc < 2) || (strcmp(argv[1], "nd") != 0)){
if (server_socket > 0){
daemon(1, 0);
fprintf(stderr, "Going into background mode...");
}else{
fprintf(stderr, "Error: could not make listening socket");
return 1;
}
}
int status;
while (server_socket > 0){
@ -54,7 +73,7 @@ int main(int argc, char ** argv){
if (myid == 0){
break;
}else{
printf("Spawned new process %i for handling socket %i\n", (int)myid, CONN_fd);
fprintf(stderr, "Spawned new process %i for handling socket %i\n", (int)myid, CONN_fd);
}
}
}
@ -71,23 +90,16 @@ int main(int argc, char ** argv){
//first timestamp set
firsttime = getNowMS();
#ifdef DEBUG
fprintf(stderr, "Doing handshake...\n");
#endif
if (doHandshake()){
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Handshake succcess!\n");
#endif
}else{
#ifdef DEBUG
#if DEBUG >= 1
fprintf(stderr, "Handshake fail!\n");
#endif
return 0;
}
#ifdef DEBUG
fprintf(stderr, "Starting processing...\n");
#endif
int retval;
int poller = epoll_create(1);
@ -104,10 +116,17 @@ int main(int argc, char ** argv){
while (!socketError && !All_Hell_Broke_Loose){
//only parse input if available or not yet init'ed
//rightnow = getNowMS();
retval = epoll_wait(poller, events, 1, 0);
if ((retval > 0) || !ready4data || (snd_cnt - snd_window_at >= snd_window_size)){
if (DDV_ready(CONN_fd)){
parseChunk();
retval = epoll_wait(poller, events, 1, 1);
if ((retval > 0) || !ready4data){// || (snd_cnt - snd_window_at >= snd_window_size)
switch (DDV_ready(CONN_fd)){
case 0:
socketError = true;
#if DEBUG >= 1
fprintf(stderr, "User socket is disconnected.\n");
#endif
break;
case -1: break;//not ready yet
default: parseChunk(); break;
}
}
if (ready4data){
@ -115,47 +134,57 @@ int main(int argc, char ** argv){
//we are ready, connect the socket!
ss = DDV_OpenUnix(streamname);
if (ss <= 0){
#ifdef DEBUG
#if DEBUG >= 1
fprintf(stderr, "Could not connect to server!\n");
#endif
return 0;
socketError = 1;
break;
}
ev.events = EPOLLIN;
ev.data.fd = ss;
epoll_ctl(sspoller, EPOLL_CTL_ADD, ss, &ev);
#ifdef DEBUG
#if DEBUG >= 3
fprintf(stderr, "Everything connected, starting to send video data...\n");
#endif
inited = true;
}
retval = epoll_wait(poller, events, 1, 50);
if (DDV_ready(ss)){
if (FLV_GetPacket(tag, ss)){//able to read a full packet?
ts = tag->data[7] * 256*256*256;
ts += tag->data[4] * 256*256;
ts += tag->data[5] * 256;
ts += tag->data[6];
if (ts != 0){
if (fts == 0){fts = ts;ftst = getNowMS();}
ts -= fts;
tag->data[7] = ts / (256*256*256);
tag->data[4] = ts / (256*256);
tag->data[5] = ts / 256;
tag->data[6] = ts % 256;
ts += ftst;
}else{
ftst = getNowMS();
tag->data[7] = ftst / (256*256*256);
tag->data[4] = ftst / (256*256);
tag->data[5] = ftst / 256;
tag->data[6] = ftst % 256;
}
SendMedia((unsigned char)tag->data[0], (unsigned char *)tag->data+11, tag->len-15, ts);
#ifdef DEBUG
fprintf(stderr, "Sent a tag to %i\n", CONN_fd);
retval = epoll_wait(poller, events, 1, 1);
switch (DDV_ready(ss)){
case 0:
socketError = true;
#if DEBUG >= 1
fprintf(stderr, "Source socket is disconnected.\n");
#endif
}
break;
case -1: break;//not ready yet
default:
if (FLV_GetPacket(tag, ss)){//able to read a full packet?
ts = tag->data[7] * 256*256*256;
ts += tag->data[4] * 256*256;
ts += tag->data[5] * 256;
ts += tag->data[6];
if (ts != 0){
if (fts == 0){fts = ts;ftst = getNowMS();}
ts -= fts;
tag->data[7] = ts / (256*256*256);
tag->data[4] = ts / (256*256);
tag->data[5] = ts / 256;
tag->data[6] = ts % 256;
ts += ftst;
}else{
ftst = getNowMS();
tag->data[7] = ftst / (256*256*256);
tag->data[4] = ftst / (256*256);
tag->data[5] = ftst / 256;
tag->data[6] = ftst % 256;
}
SendMedia((unsigned char)tag->data[0], (unsigned char *)tag->data+11, tag->len-15, ts);
#if DEBUG >= 4
fprintf(stderr, "Sent a tag to %i\n", CONN_fd);
#endif
}
break;
}
}
//send ACK if we received a whole window
@ -164,10 +193,10 @@ int main(int argc, char ** argv){
SendCTL(3, rec_cnt);//send ack (msg 3)
}
}
//#ifdef DEBUG
if (socketError){fprintf(stderr, "socketError\n");}
close(CONN_fd);
#if DEBUG >= 1
if (All_Hell_Broke_Loose){fprintf(stderr, "All Hell Broke Loose\n");}
fprintf(stderr, "User %i disconnected.\n", CONN_fd);
//#endif
#endif
return 0;
}//main

53
Connector_RTMP/parsechunks.cpp
View file

@ -13,25 +13,25 @@ void parseChunk(){
break;//happens when connection breaks unexpectedly
case 1://set chunk size
chunk_rec_max = ntohl(*(int*)next.data);
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "CTRL: Set chunk size: %i\n", chunk_rec_max);
#endif
break;
case 2://abort message - we ignore this one
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "CTRL: Abort message\n");
#endif
//4 bytes of stream id to drop
break;
case 3://ack
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "CTRL: Acknowledgement\n");
#endif
snd_window_at = ntohl(*(int*)next.data);
snd_window_at = snd_cnt;
break;
case 4:{
#ifdef DEBUG
#if DEBUG >= 4
short int ucmtype = ntohs(*(short int*)next.data);
fprintf(stderr, "CTRL: User control message %hi\n", ucmtype);
#endif
@ -47,7 +47,7 @@ void parseChunk(){
//we don't need to process this
} break;
case 5://window size of other end
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "CTRL: Window size\n");
#endif
rec_window_size = ntohl(*(int*)next.data);
@ -55,7 +55,7 @@ void parseChunk(){
SendCTL(3, rec_cnt);//send ack (msg 3)
break;
case 6:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "CTRL: Set peer bandwidth\n");
#endif
//4 bytes window size, 1 byte limit type (ignored)
@ -63,49 +63,45 @@ void parseChunk(){
SendCTL(5, snd_window_size);//send window acknowledgement size (msg 5)
break;
case 8:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received audio data\n");
#endif
break;
case 9:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received video data\n");
#endif
break;
case 15:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received AFM3 data message\n");
#endif
break;
case 16:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received AFM3 shared object\n");
#endif
break;
case 17:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received AFM3 command message\n");
#endif
break;
case 18:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received AFM0 data message\n");
#endif
break;
case 19:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received AFM0 shared object\n");
#endif
break;
case 20:{//AMF0 command message
bool parsed = false;
amfdata = parseAMF(next.data, next.real_len);
#ifdef DEBUG
fprintf(stderr, "Received AFM0 command message:\n");
amfdata.Print();
#endif
if (amfdata.getContentP(0)->StrValue() == "connect"){
#ifdef DEBUG
#if DEBUG >= 4
int tmpint;
tmpint = amfdata.getContentP(2)->getContentP("videoCodecs")->NumValue();
if (tmpint & 0x04){fprintf(stderr, "Sorensen video support detected\n");}
@ -149,9 +145,6 @@ void parseChunk(){
amfreply.addContent(AMFType("", (double)1));//stream ID - we use 1
SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
SendUSR(0, 0);//send UCM StreamBegin (0), stream 0
#ifdef DEBUG
fprintf(stderr, "AMF0 command: createStream result\n");
#endif
parsed = true;
}//createStream
if ((amfdata.getContentP(0)->StrValue() == "getStreamLength") || (amfdata.getContentP(0)->StrValue() == "getMovLen")){
@ -162,9 +155,6 @@ void parseChunk(){
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
amfreply.addContent(AMFType("", (double)0));//zero length
SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
#ifdef DEBUG
fprintf(stderr, "AMF0 command: getStreamLength result\n");
#endif
parsed = true;
}//getStreamLength
if (amfdata.getContentP(0)->StrValue() == "checkBandwidth"){
@ -175,9 +165,6 @@ void parseChunk(){
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
SendChunk(3, 20, 1, amfreply.Pack());
#ifdef DEBUG
fprintf(stderr, "AMF0 command: checkBandwidth result\n");
#endif
parsed = true;
}//checkBandwidth
if ((amfdata.getContentP(0)->StrValue() == "play") || (amfdata.getContentP(0)->StrValue() == "play2")){
@ -220,24 +207,24 @@ void parseChunk(){
chunk_snd_max = 1024*1024;
SendCTL(1, chunk_snd_max);//send chunk size max (msg 1)
ready4data = true;//start sending video data!
#ifdef DEBUG
fprintf(stderr, "AMF0 command: play result (%s)\n", streamname.c_str());
#endif
parsed = true;
}//createStream
#if DEBUG >= 3
fprintf(stderr, "AMF0 command: %s\n", amfdata.getContentP(0)->StrValue().c_str());
#endif
if (!parsed){
#ifdef DEBUG
#if DEBUG >= 2
fprintf(stderr, "AMF0 command not processed! :(\n");
#endif
}
} break;
case 22:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received aggregate message\n");
#endif
break;
default:
#ifdef DEBUG
#if DEBUG >= 1
fprintf(stderr, "Unknown chunk received! Probably protocol corruption, stopping parsing of incoming data.\n");
#endif
stopparsing = true;