Removing epoll in favor of more cross-platform poll - also adding RTMP push support and Buffer push support with IP security
This commit is contained in:
Thulinma
parent
880b19f465
commit
f45d02124a
4 changed files with 191 additions and 34 deletions
|
|
@ -25,6 +25,7 @@ namespace Connector_RTMP{
|
|||
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 = "/tmp/shared_socket"; ///< Stream that will be opened
|
||||
void parseChunk();
|
||||
int Connector_RTMP(Socket::Connection conn);
|
||||
|
|
@ -34,7 +35,6 @@ namespace Connector_RTMP{
|
|||
/// Main Connector_RTMP function
|
||||
int Connector_RTMP::Connector_RTMP(Socket::Connection conn){
|
||||
Socket = conn;
|
||||
Socket::Connection SS;
|
||||
FLV::Tag tag, viddata, auddata;
|
||||
bool viddone = false, auddone = false;
|
||||
|
||||
|
|
@ -168,6 +168,7 @@ int Connector_RTMP::Connector_RTMP(Socket::Connection conn){
|
|||
void Connector_RTMP::parseChunk(){
|
||||
static RTMPStream::Chunk next;
|
||||
static std::string inbuffer;
|
||||
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);
|
||||
|
|
@ -243,11 +244,19 @@ void Connector_RTMP::parseChunk(){
|
|||
#if DEBUG >= 4
|
||||
fprintf(stderr, "Received audio data\n");
|
||||
#endif
|
||||
F.ChunkLoader(next);
|
||||
if (SS.connected()){
|
||||
SS.write(std::string(F.data, F.len));
|
||||
}
|
||||
break;
|
||||
case 9:
|
||||
#if DEBUG >= 4
|
||||
fprintf(stderr, "Received video data\n");
|
||||
#endif
|
||||
F.ChunkLoader(next);
|
||||
if (SS.connected()){
|
||||
SS.write(std::string(F.data, F.len));
|
||||
}
|
||||
break;
|
||||
case 15:
|
||||
#if DEBUG >= 4
|
||||
|
|
@ -352,6 +361,51 @@ void Connector_RTMP::parseChunk(){
|
|||
Socket.write(RTMPStream::SendChunk(3, 17, next.msg_stream_id, (char)0+amfreply.Pack()));
|
||||
parsed3 = true;
|
||||
}//getStreamLength
|
||||
if ((amfdata.getContentP(0)->StrValue() == "publish")){
|
||||
if (amfdata.getContentP(3)){
|
||||
streamname = amfdata.getContentP(3)->StrValue();
|
||||
bool stoptokens = false;
|
||||
for (std::string::iterator i=streamname.end()-1; i>=streamname.begin(); --i){
|
||||
if (*i == '?'){stoptokens = true;}
|
||||
if (stoptokens || (!isalpha(*i) && !isdigit(*i))){streamname.erase(i);}else{*i=tolower(*i);}
|
||||
}
|
||||
streamname = "/tmp/shared_socket_" + streamname;
|
||||
SS = Socket::Connection(streamname);
|
||||
if (!SS.connected()){
|
||||
#if DEBUG >= 1
|
||||
fprintf(stderr, "Could not connect to server!\n");
|
||||
#endif
|
||||
Socket.close();//disconnect user
|
||||
break;
|
||||
}
|
||||
}
|
||||
//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
|
||||
Socket.write(RTMPStream::SendChunk(3, 17, next.msg_stream_id, (char)0+amfreply.Pack()));
|
||||
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
|
||||
Socket.write(RTMPStream::SendChunk(3, 17, next.msg_stream_id, (char)0+amfreply.Pack()));
|
||||
parsed3 = true;
|
||||
}//getStreamLength
|
||||
if (amfdata.getContentP(0)->StrValue() == "checkBandwidth"){
|
||||
//send a _result reply
|
||||
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
|
||||
|
|
@ -421,6 +475,10 @@ void Connector_RTMP::parseChunk(){
|
|||
#if DEBUG >= 4
|
||||
fprintf(stderr, "Received AFM0 data message (metadata)\n");
|
||||
#endif
|
||||
F.ChunkLoader(next);
|
||||
if (SS.connected()){
|
||||
SS.write(std::string(F.data, F.len));
|
||||
}
|
||||
break;
|
||||
case 19:
|
||||
#if DEBUG >= 4
|
||||
|
|
@ -441,12 +499,16 @@ void Connector_RTMP::parseChunk(){
|
|||
fprintf(stderr, "Object encoding set to %e\n", objencoding);
|
||||
#if DEBUG >= 4
|
||||
int tmpint;
|
||||
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");}
|
||||
tmpint = (int)amfdata.getContentP(2)->getContentP("audioCodecs")->NumValue();
|
||||
if (tmpint & 0x04){fprintf(stderr, "MP3 audio support detected\n");}
|
||||
if (tmpint & 0x400){fprintf(stderr, "AAC video support detected\n");}
|
||||
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 video support detected\n");}
|
||||
}
|
||||
#endif
|
||||
RTMPStream::chunk_snd_max = 4096;
|
||||
Socket.write(RTMPStream::SendCTL(1, RTMPStream::chunk_snd_max));//send chunk size max (msg 1)
|
||||
|
|
@ -508,6 +570,52 @@ void Connector_RTMP::parseChunk(){
|
|||
Socket.write(RTMPStream::SendChunk(3, 20, next.msg_stream_id, amfreply.Pack()));
|
||||
parsed = true;
|
||||
}//getStreamLength
|
||||
if ((amfdata.getContentP(0)->StrValue() == "publish")){
|
||||
if (amfdata.getContentP(3)){
|
||||
streamname = amfdata.getContentP(3)->StrValue();
|
||||
bool stoptokens = false;
|
||||
for (std::string::iterator i=streamname.end()-1; i>=streamname.begin(); --i){
|
||||
if (*i == '?'){stoptokens = true;}
|
||||
if (stoptokens || (!isalpha(*i) && !isdigit(*i))){streamname.erase(i);}else{*i=tolower(*i);}
|
||||
}
|
||||
streamname = "/tmp/shared_socket_" + streamname;
|
||||
SS = Socket::Connection(streamname);
|
||||
if (!SS.connected()){
|
||||
#if DEBUG >= 1
|
||||
fprintf(stderr, "Could not connect to server!\n");
|
||||
#endif
|
||||
Socket.close();//disconnect user
|
||||
break;
|
||||
}
|
||||
SS.write(Socket.getHost()+'\n');
|
||||
}
|
||||
//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
|
||||
Socket.write(RTMPStream::SendChunk(3, 20, next.msg_stream_id, amfreply.Pack()));
|
||||
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
|
||||
Socket.write(RTMPStream::SendChunk(4, 20, next.msg_stream_id, amfreply.Pack()));
|
||||
parsed = true;
|
||||
}//getStreamLength
|
||||
if (amfdata.getContentP(0)->StrValue() == "checkBandwidth"){
|
||||
//send a _result reply
|
||||
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue