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

Refactor and doxygen of all connectors finished. Internal documentation follows later.

Browse source
This commit is contained in:
Erik Zandvliet 2013-03-28 12:05:45 +01:00
parent 570aa95315
commit 9b41a07c2f
7 changed files with 809 additions and 763 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

200
src/connectors/conn_http.cpp
View file

@ -30,18 +30,18 @@ namespace Connector_HTTP {
class ConnConn{
public:
Socket::Connection * conn; ///< The socket of this connection
unsigned int lastuse; ///< Seconds since last use of this connection.
tthread::mutex in_use; ///< Mutex for this connection.
/// Constructor that sets the socket and lastuse to 0.
unsigned int lastUse; ///< Seconds since last use of this connection.
tthread::mutex inUse; ///< Mutex for this connection.
/// Constructor that sets the socket and lastUse to 0.
ConnConn(){
conn = 0;
lastuse = 0;
lastUse = 0;
}
;
/// Constructor that sets lastuse to 0, but socket to s.
/// Constructor that sets lastUse to 0, but socket to s.
ConnConn(Socket::Connection * s){
conn = s;
lastuse = 0;
lastUse = 0;
}
;
/// Destructor that deletes the socket if non-null.
@ -55,44 +55,51 @@ namespace Connector_HTTP {
;
};
std::map<std::string, ConnConn *> connconn; ///< Connections to connectors
std::set<tthread::thread *> active_threads; ///< Holds currently active threads
std::set<tthread::thread *> done_threads; ///< Holds threads that are done and ready to be joined.
tthread::mutex thread_mutex; ///< Mutex for adding/removing threads.
tthread::mutex conn_mutex; ///< Mutex for adding/removing connector connections.
tthread::mutex timeout_mutex; ///< Mutex for timeout thread.
std::map<std::string, ConnConn *> connectorConnections; ///< Connections to connectors
std::set<tthread::thread *> activeThreads; ///< Holds currently active threads
std::set<tthread::thread *> doneThreads; ///< Holds threads that are done and ready to be joined.
tthread::mutex threadMutex; ///< Mutex for adding/removing threads.
tthread::mutex connMutex; ///< Mutex for adding/removing connector connections.
tthread::mutex timeoutMutex; ///< Mutex for timeout thread.
tthread::thread * timeouter = 0; ///< Thread that times out connections to connectors.
///\brief Function run as a thread to timeout requests on the proxy.
///\param n A NULL-pointer
void proxyTimeoutThread(void * n){
n = 0; //prevent unused variable warning
tthread::lock_guard<tthread::mutex> guard(timeout_mutex);
tthread::lock_guard<tthread::mutex> guard(timeoutMutex);
while (true){
{
tthread::lock_guard<tthread::mutex> guard(conn_mutex);
if (connconn.empty()){
tthread::lock_guard<tthread::mutex> guard(connMutex);
if (connectorConnections.empty()){
return;
}
std::map<std::string, ConnConn*>::iterator it;
for (it = connconn.begin(); it != connconn.end(); it++){
if ( !it->second->conn->connected() || it->second->lastuse++ > 15){
if (it->second->in_use.try_lock()){
it->second->in_use.unlock();
for (it = connectorConnections.begin(); it != connectorConnections.end(); it++){
if ( !it->second->conn->connected() || it->second->lastUse++ > 15){
if (it->second->inUse.try_lock()){
it->second->inUse.unlock();
delete it->second;
connconn.erase(it);
it = connconn.begin(); //get a valid iterator
if (it == connconn.end()){
connectorConnections.erase(it);
it = connectorConnections.begin(); //get a valid iterator
if (it == connectorConnections.end()){
return;
}
}
}
}
conn_mutex.unlock();
connMutex.unlock();
}
usleep(1000000); //sleep 1 second and re-check
}
}
/// Handles requests without associated handler, displaying a nice friendly error message.
///\brief Handles requests without associated handler.
///
///Displays a friendly error message.
///\param H The request to be handled.
///\param conn The connection to the client that issued the request.
///\return A timestamp indicating when the request was parsed.
long long int proxyHandleUnsupported(HTTP::Parser & H, Socket::Connection * conn){
H.Clean();
H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver);
@ -103,6 +110,12 @@ namespace Connector_HTTP {
return ret;
}
///\brief Handles requests that have timed out.
///
///Displays a friendly error message.
///\param H The request that was being handled upon timeout.
///\param conn The connection to the client that issued the request.
///\return A timestamp indicating when the request was parsed.
long long int proxyHandleTimeout(HTTP::Parser & H, Socket::Connection * conn){
H.Clean();
H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver);
@ -113,7 +126,19 @@ namespace Connector_HTTP {
return ret;
}
/// Handles internal requests.
///\brief Handles requests within the proxy.
///
///Currently supported urls:
/// - /crossdomain.xml
/// - /clientaccesspolicy.xml
/// - *.ico (for favicon)
/// - /info_[streamname].js (stream info)
/// - /embed_[streamname].js (embed info)
///
///Unsupported urls default to proxyHandleUnsupported( ).
///\param H The request to be handled.
///\param conn The connection to the client that issued the request.
///\return A timestamp indicating when the request was parsed.
long long int proxyHandleInternal(HTTP::Parser & H, Socket::Connection * conn){
std::string url = H.getUrl();
@ -263,7 +288,11 @@ namespace Connector_HTTP {
return proxyHandleUnsupported(H, conn); //anything else doesn't get handled
}
/// Handles requests without associated handler, displaying a nice friendly error message.
///\brief Handles requests by dispatching them to the corresponding connector.
///\param H The request to be handled.
///\param conn The connection to the client that issued the request.
///\param connector The type of connector to be invoked.
///\return A timestamp indicating when the request was parsed.
long long int proxyHandleThroughConnector(HTTP::Parser & H, Socket::Connection * conn, std::string & connector){
//create a unique ID based on a hash of the user agent and host, followed by the stream name and connector
std::string uid = Secure::md5(H.GetHeader("User-Agent") + conn->getHost()) + "_" + H.GetVar("stream") + "_" + connector;
@ -275,13 +304,13 @@ namespace Connector_HTTP {
H.Clean();
//check if a connection exists, and if not create one
conn_mutex.lock();
if ( !connconn.count(uid) || !connconn[uid]->conn->connected()){
if (connconn.count(uid)){
connconn.erase(uid);
connMutex.lock();
if ( !connectorConnections.count(uid) || !connectorConnections[uid]->conn->connected()){
if (connectorConnections.count(uid)){
connectorConnections.erase(uid);
}
connconn[uid] = new ConnConn(new Socket::Connection("/tmp/mist/http_" + connector));
connconn[uid]->conn->setBlocking(false); //do not block on spool() with no data
connectorConnections[uid] = new ConnConn(new Socket::Connection("/tmp/mist/http_" + connector));
connectorConnections[uid]->conn->setBlocking(false); //do not block on spool() with no data
#if DEBUG >= 4
std::cout << "Created new connection " << uid << std::endl;
#endif
@ -291,56 +320,56 @@ namespace Connector_HTTP {
#endif
}
//start a new timeout thread, if neccesary
if (timeout_mutex.try_lock()){
if (timeoutMutex.try_lock()){
if (timeouter){
timeouter->join();
delete timeouter;
}
timeouter = new tthread::thread(Connector_HTTP::proxyTimeoutThread, 0);
timeout_mutex.unlock();
timeoutMutex.unlock();
}
conn_mutex.unlock();
connMutex.unlock();
//lock the mutex for this connection, and handle the request
tthread::lock_guard<tthread::mutex> guard(connconn[uid]->in_use);
tthread::lock_guard<tthread::mutex> guard(connectorConnections[uid]->inUse);
//if the server connection is dead, handle as timeout.
if ( !connconn.count(uid) || !connconn[uid]->conn->connected()){
connconn[uid]->conn->close();
if ( !connectorConnections.count(uid) || !connectorConnections[uid]->conn->connected()){
connectorConnections[uid]->conn->close();
return proxyHandleTimeout(H, conn);
}
//forward the original request
connconn[uid]->conn->SendNow(request);
connconn[uid]->lastuse = 0;
connectorConnections[uid]->conn->SendNow(request);
connectorConnections[uid]->lastUse = 0;
unsigned int timeout = 0;
unsigned int retries = 0;
//wait for a response
while (connconn.count(uid) && connconn[uid]->conn->connected() && conn->connected()){
while (connectorConnections.count(uid) && connectorConnections[uid]->conn->connected() && conn->connected()){
conn->spool();
if (connconn[uid]->conn->Received().size() || connconn[uid]->conn->spool()){
if (connectorConnections[uid]->conn->Received().size() || connectorConnections[uid]->conn->spool()){
//make sure we end in a \n
if ( *(connconn[uid]->conn->Received().get().rbegin()) != '\n'){
std::string tmp = connconn[uid]->conn->Received().get();
connconn[uid]->conn->Received().get().clear();
if (connconn[uid]->conn->Received().size()){
connconn[uid]->conn->Received().get().insert(0, tmp);
if ( *(connectorConnections[uid]->conn->Received().get().rbegin()) != '\n'){
std::string tmp = connectorConnections[uid]->conn->Received().get();
connectorConnections[uid]->conn->Received().get().clear();
if (connectorConnections[uid]->conn->Received().size()){
connectorConnections[uid]->conn->Received().get().insert(0, tmp);
}else{
connconn[uid]->conn->Received().append(tmp);
connectorConnections[uid]->conn->Received().append(tmp);
}
}
//check if the whole response was received
if (H.Read(connconn[uid]->conn->Received().get())){
if (H.Read(connectorConnections[uid]->conn->Received().get())){
//208 means the fragment is too new, retry in 3s
if (H.url == "208"){
retries++;
if (retries >= 5){
std::cout << "[5 retry-laters, cancelled]" << std::endl;
connconn[uid]->conn->close();
connectorConnections[uid]->conn->close();
return proxyHandleTimeout(H, conn);
}
connconn[uid]->lastuse = 0;
connectorConnections[uid]->lastUse = 0;
timeout = 0;
Util::sleep(3000);
connconn[uid]->conn->SendNow(request);
connectorConnections[uid]->conn->SendNow(request);
H.Clean();
continue;
}
@ -350,16 +379,16 @@ namespace Connector_HTTP {
//keep trying unless the timeout triggers
if (timeout++ > 4000){
std::cout << "[20s timeout triggered]" << std::endl;
connconn[uid]->conn->close();
connectorConnections[uid]->conn->close();
return proxyHandleTimeout(H, conn);
}else{
Util::sleep(5);
}
}
}
if ( !connconn.count(uid) || !connconn[uid]->conn->connected() || !conn->connected()){
if ( !connectorConnections.count(uid) || !connectorConnections[uid]->conn->connected() || !conn->connected()){
//failure, disconnect and sent error to user
connconn[uid]->conn->close();
connectorConnections[uid]->conn->close();
return proxyHandleTimeout(H, conn);
}else{
long long int ret = Util::getMS();
@ -375,9 +404,9 @@ namespace Connector_HTTP {
H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver);
conn->SendNow(H.BuildResponse("200", "OK"));
//switch out the connection for an empty one - it makes no sense to keep these globally
Socket::Connection * myConn = connconn[uid]->conn;
connconn[uid]->conn = new Socket::Connection();
connconn[uid]->in_use.unlock();
Socket::Connection * myConn = connectorConnections[uid]->conn;
connectorConnections[uid]->conn = new Socket::Connection();
connectorConnections[uid]->inUse.unlock();
//continue sending data from this socket and keep it permanently in use
while (myConn->connected() && conn->connected()){
if (myConn->Received().size() || myConn->spool()){
@ -396,12 +425,16 @@ namespace Connector_HTTP {
}
}
/// Returns the name of the HTTP connector the given request should be served by.
/// Can currently return:
/// - none (request not supported)
/// - internal (request fed from information internal to this connector)
/// - dynamic (request fed from http_dynamic connector)
/// - progressive (request fed from http_progressive connector)
///\brief Determines the type of connector to be used for handling a request.
///\param H The request to be handled..
///\return A string indicating the type of connector.
///Possible values are:
/// - "none" The request is not supported.
/// - "internal" The request should be handled by the proxy itself.
/// - "dynamic" The request should be dispatched to the HTTP Dynamic Connector
/// - "progressive" The request should be dispatched to the HTTP Progressive Connector
/// - "smooth" The request should be dispatched to the HTTP Smooth Connector
/// - "live" The request should be dispatched to the HTTP Live Connector
std::string proxyGetHandleType(HTTP::Parser & H){
std::string url = H.getUrl();
if (url.find("/dynamic/") != std::string::npos){
@ -449,7 +482,8 @@ namespace Connector_HTTP {
return "none";
}
/// Thread for handling a single HTTP connection
///\brief Function run as a thread to handle a single HTTP connection.
///\param pointer A Socket::Connection* indicating the connection to th client.
void proxyHandleHTTPConnection(void * pointer){
Socket::Connection * conn = (Socket::Connection *)pointer;
conn->setBlocking(false); //do not block on conn.spool() when no data is available
@ -504,17 +538,17 @@ namespace Connector_HTTP {
//close and remove the connection
conn->close();
delete conn;
//remove this thread from active_threads and add it to done_threads.
thread_mutex.lock();
for (std::set<tthread::thread *>::iterator it = active_threads.begin(); it != active_threads.end(); it++){
//remove this thread from activeThreads and add it to doneThreads.
threadMutex.lock();
for (std::set<tthread::thread *>::iterator it = activeThreads.begin(); it != activeThreads.end(); it++){
if (( *it)->get_id() == tthread::this_thread::get_id()){
tthread::thread * T = ( *it);
active_threads.erase(T);
done_threads.insert(T);
activeThreads.erase(T);
doneThreads.insert(T);
break;
}
}
thread_mutex.unlock();
threadMutex.unlock();
}
} //Connector_HTTP namespace
@ -533,17 +567,17 @@ int main(int argc, char ** argv){
Socket::Connection S = server_socket.accept();
if (S.connected()){ //check if the new connection is valid
//lock the thread mutex and spawn a new thread for this connection
Connector_HTTP::thread_mutex.lock();
Connector_HTTP::threadMutex.lock();
tthread::thread * T = new tthread::thread(Connector_HTTP::proxyHandleHTTPConnection, (void *)(new Socket::Connection(S)));
Connector_HTTP::active_threads.insert(T);
Connector_HTTP::activeThreads.insert(T);
//clean up any threads that may have finished
while ( !Connector_HTTP::done_threads.empty()){
T = *Connector_HTTP::done_threads.begin();
while ( !Connector_HTTP::doneThreads.empty()){
T = *Connector_HTTP::doneThreads.begin();
T->join();
Connector_HTTP::done_threads.erase(T);
Connector_HTTP::doneThreads.erase(T);
delete T;
}
Connector_HTTP::thread_mutex.unlock();
Connector_HTTP::threadMutex.unlock();
}else{
Util::sleep(10); //sleep 10ms
}
@ -553,16 +587,16 @@ int main(int argc, char ** argv){
//wait for existing connections to drop
bool repeat = true;
while (repeat){
Connector_HTTP::thread_mutex.lock();
repeat = !Connector_HTTP::active_threads.empty();
Connector_HTTP::threadMutex.lock();
repeat = !Connector_HTTP::activeThreads.empty();
//clean up any threads that may have finished
while ( !Connector_HTTP::done_threads.empty()){
tthread::thread * T = *Connector_HTTP::done_threads.begin();
while ( !Connector_HTTP::doneThreads.empty()){
tthread::thread * T = *Connector_HTTP::doneThreads.begin();
T->join();
Connector_HTTP::done_threads.erase(T);
Connector_HTTP::doneThreads.erase(T);
delete T;
}
Connector_HTTP::thread_mutex.unlock();
Connector_HTTP::threadMutex.unlock();
if (repeat){
Util::sleep(100); //sleep 100ms
}

22
src/connectors/conn_http_dynamic.cpp
View file

@ -26,11 +26,11 @@
/// Holds everything unique to HTTP Connectors.
namespace Connector_HTTP {
///\brief Builds a bootstrap for use in HTTP Dynamic streaming.
///\param MovieId The name of the movie.
///\param streamName The name of the stream.
///\param metadata The current metadata, used to generate the index.
///\param fragnum The index of the current fragment
///\return The generated bootstrap.
std::string dynamicBootstrap(std::string & MovieId, JSON::Value & metadata, int fragnum = 0){
std::string dynamicBootstrap(std::string & streamName, JSON::Value & metadata, int fragnum = 0){
std::string empty;
MP4::ASRT asrt;
@ -80,7 +80,7 @@ namespace Connector_HTTP {
abst.setLive(false);
abst.setCurrentMediaTime(metadata["lastms"].asInt());
abst.setSmpteTimeCodeOffset(0);
abst.setMovieIdentifier(MovieId);
abst.setMovieIdentifier(streamName);
abst.setSegmentRunTable(asrt, 0);
abst.setFragmentRunTable(afrt, 0);
@ -91,24 +91,24 @@ namespace Connector_HTTP {
}
///\brief Builds an index file for HTTP Dynamic streaming.
///\param MovieId The name of the movie.
///\param streamName The name of the stream.
///\param metadata The current metadata, used to generate the index.
///\return The index file for HTTP Dynamic Streaming.
std::string dynamicIndex(std::string & MovieId, JSON::Value & metadata){
std::string dynamicIndex(std::string & streamName, JSON::Value & metadata){
std::string Result;
if (metadata.isMember("vod")){
Result =
"<?xml version=\"1.0\" encoding=\"utf-8\"?>\n"
"<manifest xmlns=\"http://ns.adobe.com/f4m/1.0\">\n"
"<id>" + MovieId + "</id>\n"
"<id>" + streamName + "</id>\n"
"<width>" + metadata["video"]["width"].asString() + "</width>\n"
"<height>" + metadata["video"]["height"].asString() + "</height>\n"
"<duration>" + metadata["length"].asString() + ".000</duration>\n"
"<mimeType>video/mp4</mimeType>\n"
"<streamType>recorded</streamType>\n"
"<deliveryType>streaming</deliveryType>\n"
"<bootstrapInfo profile=\"named\" id=\"bootstrap1\">" + Base64::encode(dynamicBootstrap(MovieId, metadata)) + "</bootstrapInfo>\n"
"<media streamId=\"1\" bootstrapInfoId=\"bootstrap1\" url=\"" + MovieId + "/\">\n"
"<bootstrapInfo profile=\"named\" id=\"bootstrap1\">" + Base64::encode(dynamicBootstrap(streamName, metadata)) + "</bootstrapInfo>\n"
"<media streamId=\"1\" bootstrapInfoId=\"bootstrap1\" url=\"" + streamName + "/\">\n"
"<metadata>AgAKb25NZXRhRGF0YQMAAAk=</metadata>\n"
"</media>\n"
"</manifest>\n";
@ -116,15 +116,15 @@ namespace Connector_HTTP {
Result =
"<?xml version=\"1.0\" encoding=\"utf-8\"?>\n"
"<manifest xmlns=\"http://ns.adobe.com/f4m/1.0\">\n"
"<id>" + MovieId + "</id>\n"
"<id>" + streamName + "</id>\n"
"<dvrInfo windowDuration=\"" + metadata["buffer_window"].asString().substr(0, metadata["buffer_window"].asString().size() - 3) + "\"></dvrInfo>"
"<mimeType>video/mp4</mimeType>\n"
"<streamType>live</streamType>\n"
"<deliveryType>streaming</deliveryType>\n"
"<media url=\"" + MovieId + "/\">\n"
"<media url=\"" + streamName + "/\">\n"
"<metadata>AgAKb25NZXRhRGF0YQMAAAk=</metadata>\n"
"</media>\n"
"<bootstrapInfo profile=\"named\" url=\"" + MovieId + ".abst\" />\n"
"<bootstrapInfo profile=\"named\" url=\"" + streamName + ".abst\" />\n"
"</manifest>\n";
}
#if DEBUG >= 8

7
src/connectors/conn_http_live.cpp
View file

@ -26,10 +26,9 @@
/// Holds everything unique to HTTP Connectors.
namespace Connector_HTTP {
///\brief Builds an index file for HTTP Live streaming.
///\param MovieId The name of the movie.
///\param metadata The current metadata, used to generate the index.
///\return The index file for HTTP Live Streaming.
std::string liveIndex(std::string & MovieId, JSON::Value & metadata){
std::string liveIndex(JSON::Value & metadata){
std::stringstream Result;
if ( !metadata.isMember("live")){
int longestFragment = 0;
@ -62,7 +61,7 @@ namespace Connector_HTTP {
return Result.str();
} //liveIndex
///\brief Main function for the HTTP HLS Connector
///\brief Main function for the HTTP Live Connector
///\param conn A socket describing the connection the client.
///\return The exit code of the connector.
int liveConnector(Socket::Connection conn){
@ -175,7 +174,7 @@ namespace Connector_HTTP {
HTTP_S.Clean();
HTTP_S.SetHeader("Content-Type", manifestType);
HTTP_S.SetHeader("Cache-Control", "no-cache");
std::string manifest = liveIndex(streamname, Strm.metadata);
std::string manifest = liveIndex(Strm.metadata);
HTTP_S.SetBody(manifest);
conn.SendNow(HTTP_S.BuildResponse("200", "OK"));
}

5
src/connectors/conn_http_smooth.cpp
View file

@ -28,10 +28,9 @@
///\brief Holds everything unique to HTTP Connectors.
namespace Connector_HTTP {
///\brief Builds an index file for HTTP Smooth streaming.
///\param MovieId The name of the movie.
///\param metadata The current metadata, used to generate the index.
///\return The index file for HTTP Smooth Streaming.
std::string smoothIndex(std::string & MovieId, JSON::Value & metadata){
std::string smoothIndex(JSON::Value & metadata){
std::stringstream Result;
Result << "<?xml version=\"1.0\" encoding=\"utf-8\"?>\n";
Result << "<SmoothStreamingMedia "
@ -245,7 +244,7 @@ namespace Connector_HTTP {
HTTP_S.Clean();
HTTP_S.SetHeader("Content-Type", "text/xml");
HTTP_S.SetHeader("Cache-Control", "no-cache");
std::string manifest = smoothIndex(streamname, Strm.metadata);
std::string manifest = smoothIndex(Strm.metadata);
HTTP_S.SetBody(manifest);
conn.SendNow(HTTP_S.BuildResponse("200", "OK"));
}

7
src/connectors/conn_raw.cpp
View file

@ -8,9 +8,10 @@
#include <mist/stream.h>
#include <mist/timing.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.
///\brief 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){
Util::Config conf(argv[0], PACKAGE_VERSION);
conf.addOption("stream_name", JSON::fromString("{\"arg_num\":1, \"help\":\"Name of the stream to write to stdout.\"}"));

786
src/connectors/conn_rtmp.cpp
View file

@ -2,6 +2,8 @@
/// Contains the main code for the RTMP Connector
#include <iostream>
#include <sstream>
#include <cstdlib>
#include <cstdio>
#include <cmath>
@ -10,7 +12,7 @@
#include <sys/types.h>
#include <sys/wait.h>
#include <getopt.h>
#include <sstream>
#include <mist/socket.h>
#include <mist/config.h>
#include <mist/flv_tag.h>
@ -19,171 +21,261 @@
#include <mist/stream.h>
#include <mist/timing.h>
/// Holds all functions and data unique to the RTMP Connector
///\brief Holds everything 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.
bool ready4data = false; ///< Indicates whether streaming can start.
bool inited = false; ///< Indicates whether we are ready to connect to the Buffer.
bool noStats = false; ///< Indicates when no stats should be sent anymore. Used in push mode.
bool stopParsing = false; ///< Indicates when to stop all parsing.
//for reply to play command
int play_trans = -1;
int play_streamid = -1;
int play_msgtype = -1;
int playTransaction = -1;///<The transaction number of the reply.
int playStreamId = -1;///<The stream id of the reply.
int playMessageType = -1;///<The message type of the reply.
//generic state keeping
bool stream_inited = false; ///true if init data for audio/video was sent
bool streamInited = false;///<Indicates whether init data for audio/video was sent.
Socket::Connection Socket; ///< Socket connected to user
Socket::Connection SS; ///< Socket connected to server
std::string streamname; ///< Stream that will be opened
void parseChunk(Socket::Buffer & 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;
Socket::Connection Socket; ///< A copy of the user socket to allow helper functions to directly send data.
Socket::Connection ss; ///< Socket connected to server.
std::string streamName; ///< Stream that will be opened.
///\brief Main Connector_RTMP function
///\param conn A socket describing the connection the client.
///\return The exit code of the connector.
int Connector_RTMP::Connector_RTMP(Socket::Connection conn){
Socket = conn;
Socket.setBlocking(false);
FLV::Tag tag, init_tag;
DTSC::Stream Strm;
while ( !Socket.Received().available(1537) && Socket.connected()){
Socket.spool();
Util::sleep(5);
}
RTMPStream::handshake_in = Socket.Received().remove(1537);
RTMPStream::rec_cnt += 1537;
if (RTMPStream::doHandshake()){
Socket.SendNow(RTMPStream::handshake_out);
while ( !Socket.Received().available(1536) && Socket.connected()){
Socket.spool();
Util::sleep(5);
}
Socket.Received().remove(1536);
RTMPStream::rec_cnt += 1536;
#if DEBUG >= 5
fprintf(stderr, "Handshake succcess!\n");
#endif
///\brief Sends a RTMP command either in AMF or AMF3 mode.
///\param amfReply The data to be sent over RTMP.
///\param messageType The type of message.
///\param streamId The ID of the AMF stream.
void sendCommand(AMF::Object & amfReply, int messageType, int streamId){
#if DEBUG >= 8
std::cerr << amfReply.Print() << std::endl;
#endif
if (messageType == 17){
Socket.SendNow(RTMPStream::SendChunk(3, messageType, streamId, (char)0 + amfReply.Pack()));
}else{
#if DEBUG >= 5
fprintf(stderr, "Handshake fail!\n");
#endif
return 0;
Socket.SendNow(RTMPStream::SendChunk(3, messageType, streamId, amfReply.Pack()));
}
} //sendCommand
unsigned int lastStats = 0;
bool firsttime = true;
while (Socket.connected()){
if (Socket.spool() || firsttime){
parseChunk(Socket.Received());
firsttime = false;
}else{
Util::sleep(1); //sleep 1ms to prevent high CPU usage
///\brief Parses a single AMF command message, and sends a direct response through sendCommand().
///\param amfData The received request.
///\param messageType The type of message.
///\param streamId The ID of the AMF stream.
void parseAMFCommand(AMF::Object & amfData, int messageType, int streamId){
#if DEBUG >= 5
fprintf(stderr, "Received command: %s\n", amfData.Print().c_str());
#endif
#if DEBUG >= 8
fprintf(stderr, "AMF0 command: %s\n", amfData.getContentP(0)->StrValue().c_str());
#endif
if (amfData.getContentP(0)->StrValue() == "connect"){
double objencoding = 0;
if (amfData.getContentP(2)->getContentP("objectEncoding")){
objencoding = amfData.getContentP(2)->getContentP("objectEncoding")->NumValue();
}
if (ready4data){
if ( !inited){
//we are ready, connect the socket!
SS = Util::Stream::getStream(streamname);
if ( !SS.connected()){
#if DEBUG >= 1
fprintf(stderr, "Could not connect to server!\n");
#endif
Socket.close(); //disconnect user
break;
#if DEBUG >= 6
int tmpint;
if (amfData.getContentP(2)->getContentP("videoCodecs")){
tmpint = (int)amfData.getContentP(2)->getContentP("videoCodecs")->NumValue();
if (tmpint & 0x04){
fprintf(stderr, "Sorensen video support detected\n");
}
SS.setBlocking(false);
SS.SendNow("p\n");
inited = true;
}
if (inited && !nostats){
long long int now = Util::epoch();
if (now != lastStats){
lastStats = now;
SS.SendNow(Socket.getStats("RTMP").c_str());
if (tmpint & 0x80){
fprintf(stderr, "H264 video support detected\n");
}
}
if (SS.spool()){
while (Strm.parsePacket(SS.Received())){
if (play_trans != -1){
//send a status reply
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("", "onStatus")); //status reply
amfreply.addContent(AMF::Object("", (double)play_trans)); //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));
sendCommand(amfreply, play_msgtype, play_streamid);
//send streamisrecorded if stream, well, is recorded.
if (Strm.metadata.isMember("length") && Strm.metadata["length"].asInt() > 0){
Socket.Send(RTMPStream::SendUSR(4, 1)); //send UCM StreamIsRecorded (4), stream 1
if (amfData.getContentP(2)->getContentP("audioCodecs")){
tmpint = (int)amfData.getContentP(2)->getContentP("audioCodecs")->NumValue();
if (tmpint & 0x04){
fprintf(stderr, "MP3 audio support detected\n");
}
//send streambegin
Socket.Send(RTMPStream::SendUSR(0, 1)); //send UCM StreamBegin (0), stream 1
//and more reply
amfreply = AMF::Object("container", AMF::AMF0_DDV_CONTAINER);
amfreply.addContent(AMF::Object("", "onStatus")); //status reply
amfreply.addContent(AMF::Object("", (double)play_trans)); //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));
sendCommand(amfreply, play_msgtype, play_streamid);
RTMPStream::chunk_snd_max = 102400; //100KiB
if (tmpint & 0x400){
fprintf(stderr, "AAC audio support detected\n");
}
}
#endif
RTMPStream::chunk_snd_max = 4096;
Socket.Send(RTMPStream::SendCTL(1, RTMPStream::chunk_snd_max)); //send chunk size max (msg 1)
//send dunno?
Socket.Send(RTMPStream::SendUSR(32, 1)); //send UCM no clue?, stream 1
play_trans = -1;
Socket.Send(RTMPStream::SendCTL(5, RTMPStream::snd_window_size)); //send window acknowledgement size (msg 5)
Socket.Send(RTMPStream::SendCTL(6, RTMPStream::rec_window_size)); //send rec window acknowledgement size (msg 6)
Socket.Send(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,5,5,2004"));
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"));
sendCommand(amfReply, messageType, streamId);
//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, streamId);
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
sendCommand(amfReply, messageType, streamId);
Socket.Send(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() == "FCPublish") || (amfData.getContentP(0)->StrValue() == "FCUnpublish") || (amfData.getContentP(0)->StrValue() == "releaseStream")){
// ignored
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
sendCommand(amfReply, messageType, streamId);
return;
} //getStreamLength
if ((amfData.getContentP(0)->StrValue() == "publish")){
if (amfData.getContentP(3)){
streamName = amfData.getContentP(3)->StrValue();
/// \todo implement push for MistPlayer or restrict and change to getLive
ss = Util::Stream::getStream(streamName);
if ( !ss.connected()){
#if DEBUG >= 1
fprintf(stderr, "Could not connect to server!\n");
#endif
Socket.close(); //disconnect user
return;
}
ss.Send("P ");
ss.Send(Socket.getHost().c_str());
ss.Send("\n");
noStats = true;
}
//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?
sendCommand(amfReply, messageType, streamId);
Socket.Send(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));
sendCommand(amfReply, messageType, streamId);
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
sendCommand(amfReply, messageType, streamId);
return;
} //checkBandwidth
if ((amfData.getContentP(0)->StrValue() == "play") || (amfData.getContentP(0)->StrValue() == "play2")){
//set reply number and stream name, actual reply is sent up in the ss.spool() handler
playTransaction = amfData.getContentP(1)->NumValue();
playMessageType = messageType;
playStreamId = streamId;
streamName = amfData.getContentP(3)->StrValue();
Connector_RTMP::ready4data = true; //start sending video data!
return;
} //play
if ((amfData.getContentP(0)->StrValue() == "seek")){
//set reply number and stream name, actual reply is sent up in the ss.spool() handler
playTransaction = amfData.getContentP(1)->NumValue();
playMessageType = messageType;
playStreamId = streamId;
streamInited = false;
//sent init data if needed
if ( !stream_inited){
init_tag.DTSCMetaInit(Strm);
Socket.SendNow(RTMPStream::SendMedia(init_tag));
if (Strm.metadata.isMember("audio") && Strm.metadata["audio"].isMember("init")){
init_tag.DTSCAudioInit(Strm);
Socket.SendNow(RTMPStream::SendMedia(init_tag));
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.Seek.Notify"));
amfReply.getContentP(3)->addContent(AMF::Object("description", "Seeking to the specified time"));
amfReply.getContentP(3)->addContent(AMF::Object("details", "DDV"));
amfReply.getContentP(3)->addContent(AMF::Object("clientid", (double)1337));
sendCommand(amfReply, playMessageType, playStreamId);
ss.Send("s ");
ss.Send(JSON::Value((long long int)amfData.getContentP(3)->NumValue()).asString().c_str());
ss.Send("\n");
return;
} //seek
if ((amfData.getContentP(0)->StrValue() == "pauseRaw") || (amfData.getContentP(0)->StrValue() == "pause")){
if (amfData.getContentP(3)->NumValue()){
ss.Send("q\n"); //quit playing
//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.Pause.Notify"));
amfReply.getContentP(3)->addContent(AMF::Object("description", "Pausing playback"));
amfReply.getContentP(3)->addContent(AMF::Object("details", "DDV"));
amfReply.getContentP(3)->addContent(AMF::Object("clientid", (double)1337));
sendCommand(amfReply, playMessageType, playStreamId);
}else{
ss.Send("p\n"); //start playing
//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.Unpause.Notify"));
amfReply.getContentP(3)->addContent(AMF::Object("description", "Resuming playback"));
amfReply.getContentP(3)->addContent(AMF::Object("details", "DDV"));
amfReply.getContentP(3)->addContent(AMF::Object("clientid", (double)1337));
sendCommand(amfReply, playMessageType, playStreamId);
}
if (Strm.metadata.isMember("video") && Strm.metadata["video"].isMember("init")){
init_tag.DTSCVideoInit(Strm);
Socket.SendNow(RTMPStream::SendMedia(init_tag));
}
stream_inited = true;
}
//sent a tag
tag.DTSCLoader(Strm);
Socket.SendNow(RTMPStream::SendMedia(tag));
#if DEBUG >= 8
fprintf(stderr, "Sent tag to %i: [%u] %s\n", Socket.getSocket(), tag.tagTime(), tag.tagType().c_str());
#endif
}
}
}
}
Socket.close();
SS.SendNow(Socket.getStats("RTMP").c_str());
SS.close();
return 0;
} //Connector_RTMP
return;
} //seek
///\brief Tries to get and parse one RTMP chunk at a time.
///\param inbuffer A buffer filled with chunk data.
void Connector_RTMP::parseChunk(Socket::Buffer & inbuffer){
#if DEBUG >= 2
fprintf(stderr, "AMF0 command not processed!\n%s\n", amfData.Print().c_str());
#endif
} //parseAMFCommand
///\brief Gets and parses one RTMP chunk at a time.
///\param inputBuffer A buffer filled with chunk data.
void parseChunk(Socket::Buffer & inputBuffer){
//for DTSC conversion
static JSON::Value meta_out;
static std::stringstream prebuffer; // Temporary buffer before sending real data
@ -197,7 +289,7 @@ void Connector_RTMP::parseChunk(Socket::Buffer & inbuffer){
static AMF::Object3 amf3data("empty", AMF::AMF3_DDV_CONTAINER);
static AMF::Object3 amf3elem("empty", AMF::AMF3_DDV_CONTAINER);
while (next.Parse(inbuffer)){
while (next.Parse(inputBuffer)){
//send ACK if we received a whole window
if ((RTMPStream::rec_cnt - RTMPStream::rec_window_at > RTMPStream::rec_window_size)){
@ -207,31 +299,31 @@ void Connector_RTMP::parseChunk(Socket::Buffer & inbuffer){
switch (next.msg_type_id){
case 0: //does not exist
#if DEBUG >= 2
#if DEBUG >= 2
fprintf(stderr, "UNKN: Received a zero-type message. Possible data corruption? Aborting!\n");
#endif
while (inbuffer.size()){
inbuffer.get().clear();
#endif
while (inputBuffer.size()){
inputBuffer.get().clear();
}
SS.close();
ss.close();
Socket.close();
break; //happens when connection breaks unexpectedly
case 1: //set chunk size
RTMPStream::chunk_rec_max = ntohl(*(int*)next.data.c_str());
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "CTRL: Set chunk size: %i\n", RTMPStream::chunk_rec_max);
#endif
#endif
break;
case 2: //abort message - we ignore this one
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "CTRL: Abort message\n");
#endif
#endif
//4 bytes of stream id to drop
break;
case 3: //ack
#if DEBUG >= 8
#if DEBUG >= 8
fprintf(stderr, "CTRL: Acknowledgement\n");
#endif
#endif
RTMPStream::snd_window_at = ntohl(*(int*)next.data.c_str());
RTMPStream::snd_window_at = RTMPStream::snd_cnt;
break;
@ -246,7 +338,7 @@ void Connector_RTMP::parseChunk(Socket::Buffer & inbuffer){
//6 = pingrequest, 4 bytes data
//7 = pingresponse, 4 bytes data
//we don't need to process this
#if DEBUG >= 5
#if DEBUG >= 5
short int ucmtype = ntohs(*(short int*)next.data.c_str());
switch (ucmtype){
case 0:
@ -274,21 +366,21 @@ void Connector_RTMP::parseChunk(Socket::Buffer & inbuffer){
fprintf(stderr, "CTRL: UCM Unknown (%hi)\n", ucmtype);
break;
}
#endif
#endif
}
break;
case 5: //window size of other end
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "CTRL: Window size\n");
#endif
#endif
RTMPStream::rec_window_size = ntohl(*(int*)next.data.c_str());
RTMPStream::rec_window_at = RTMPStream::rec_cnt;
Socket.Send(RTMPStream::SendCTL(3, RTMPStream::rec_cnt)); //send ack (msg 3)
break;
case 6:
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "CTRL: Set peer bandwidth\n");
#endif
#endif
//4 bytes window size, 1 byte limit type (ignored)
RTMPStream::snd_window_size = ntohl(*(int*)next.data.c_str());
Socket.Send(RTMPStream::SendCTL(5, RTMPStream::snd_window_size)); //send window acknowledgement size (msg 5)
@ -296,7 +388,7 @@ void Connector_RTMP::parseChunk(Socket::Buffer & inbuffer){
case 8: //audio data
case 9: //video data
case 18: //meta data
if (SS.connected()){
if (ss.connected()){
F.ChunkLoader(next);
JSON::Value pack_out = F.toJSON(meta_out);
if ( !pack_out.isNull()){
@ -304,48 +396,48 @@ void Connector_RTMP::parseChunk(Socket::Buffer & inbuffer){
counter++;
if (counter > 8){
sending = true;
SS.SendNow(meta_out.toNetPacked());
SS.SendNow(prebuffer.str().c_str(), prebuffer.str().size()); //write buffer
ss.SendNow(meta_out.toNetPacked());
ss.SendNow(prebuffer.str().c_str(), prebuffer.str().size()); //write buffer
prebuffer.str(""); //clear buffer
SS.SendNow(pack_out.toNetPacked());
ss.SendNow(pack_out.toNetPacked());
}else{
prebuffer << pack_out.toNetPacked();
}
}else{
SS.SendNow(pack_out.toNetPacked());
ss.SendNow(pack_out.toNetPacked());
}
}
}else{
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "Received useless media data\n");
#endif
#endif
Socket.close();
}
break;
case 15:
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "Received AFM3 data message\n");
#endif
#endif
break;
case 16:
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "Received AFM3 shared object\n");
#endif
#endif
break;
case 17: {
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "Received AFM3 command message\n");
#endif
#endif
if (next.data[0] != 0){
next.data = next.data.substr(1);
amf3data = AMF::parse3(next.data);
#if DEBUG >= 5
#if DEBUG >= 5
amf3data.Print();
#endif
#endif
}else{
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "Received AFM3-0 command message\n");
#endif
#endif
next.data = next.data.substr(1);
amfdata = AMF::parse(next.data);
parseAMFCommand(amfdata, 17, next.msg_stream_id);
@ -353,9 +445,9 @@ void Connector_RTMP::parseChunk(Socket::Buffer & inbuffer){
}
break;
case 19:
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "Received AFM0 shared object\n");
#endif
#endif
break;
case 20: { //AMF0 command message
amfdata = AMF::parse(next.data);
@ -363,242 +455,156 @@ void Connector_RTMP::parseChunk(Socket::Buffer & inbuffer){
}
break;
case 22:
#if DEBUG >= 5
#if DEBUG >= 5
fprintf(stderr, "Received aggregate message\n");
#endif
#endif
break;
default:
#if DEBUG >= 1
#if DEBUG >= 1
fprintf(stderr, "Unknown chunk received! Probably protocol corruption, stopping parsing of incoming data.\n");
#endif
Connector_RTMP::stopparsing = true;
#endif
stopParsing = true;
break;
}
}
} //parseChunk
} //parseChunk
void Connector_RTMP::sendCommand(AMF::Object & amfreply, int messagetype, int stream_id){
#if DEBUG >= 8
std::cerr << amfreply.Print() << std::endl;
#endif
if (messagetype == 17){
Socket.SendNow(RTMPStream::SendChunk(3, messagetype, stream_id, (char)0 + amfreply.Pack()));
///\brief Main function for the RTMP Connector
///\param conn A socket describing the connection the client.
///\return The exit code of the connector.
int rtmpConnector(Socket::Connection conn){
Socket = conn;
Socket.setBlocking(false);
FLV::Tag tag, init_tag;
DTSC::Stream Strm;
while ( !Socket.Received().available(1537) && Socket.connected()){
Socket.spool();
Util::sleep(5);
}
RTMPStream::handshake_in = Socket.Received().remove(1537);
RTMPStream::rec_cnt += 1537;
if (RTMPStream::doHandshake()){
Socket.SendNow(RTMPStream::handshake_out);
while ( !Socket.Received().available(1536) && Socket.connected()){
Socket.spool();
Util::sleep(5);
}
Socket.Received().remove(1536);
RTMPStream::rec_cnt += 1536;
#if DEBUG >= 5
fprintf(stderr, "Handshake succcess!\n");
#endif
}else{
Socket.SendNow(RTMPStream::SendChunk(3, messagetype, stream_id, amfreply.Pack()));
#if DEBUG >= 5
fprintf(stderr, "Handshake fail!\n");
#endif
return 0;
}
} //sendCommand
void Connector_RTMP::parseAMFCommand(AMF::Object & amfdata, int messagetype, int stream_id){
#if DEBUG >= 5
fprintf(stderr, "Received command: %s\n", amfdata.Print().c_str());
#endif
#if DEBUG >= 8
fprintf(stderr, "AMF0 command: %s\n", amfdata.getContentP(0)->StrValue().c_str());
#endif
if (amfdata.getContentP(0)->StrValue() == "connect"){
double objencoding = 0;
if (amfdata.getContentP(2)->getContentP("objectEncoding")){
objencoding = amfdata.getContentP(2)->getContentP("objectEncoding")->NumValue();
unsigned int lastStats = 0;
bool firsttime = true;
while (Socket.connected()){
if (Socket.spool() || firsttime){
parseChunk(Socket.Received());
firsttime = false;
}else{
Util::sleep(1); //sleep 1ms to prevent high CPU usage
}
#if DEBUG >= 6
int tmpint;
if (amfdata.getContentP(2)->getContentP("videoCodecs")){
tmpint = (int)amfdata.getContentP(2)->getContentP("videoCodecs")->NumValue();
if (tmpint & 0x04){
fprintf(stderr, "Sorensen video support detected\n");
if (ready4data){
if ( !inited){
//we are ready, connect the socket!
ss = Util::Stream::getStream(streamName);
if ( !ss.connected()){
#if DEBUG >= 1
fprintf(stderr, "Could not connect to server!\n");
#endif
Socket.close(); //disconnect user
break;
}
if (tmpint & 0x80){
fprintf(stderr, "H264 video support detected\n");
ss.setBlocking(false);
ss.SendNow("p\n");
inited = true;
}
if (inited && !noStats){
long long int now = Util::epoch();
if (now != lastStats){
lastStats = now;
ss.SendNow(Socket.getStats("RTMP").c_str());
}
}
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.Send(RTMPStream::SendCTL(1, RTMPStream::chunk_snd_max)); //send chunk size max (msg 1)
Socket.Send(RTMPStream::SendCTL(5, RTMPStream::snd_window_size)); //send window acknowledgement size (msg 5)
Socket.Send(RTMPStream::SendCTL(6, RTMPStream::rec_window_size)); //send rec window acknowledgement size (msg 6)
Socket.Send(RTMPStream::SendUSR(0, 1)); //send UCM StreamBegin (0), stream 1
//send a _result reply
if (ss.spool()){
while (Strm.parsePacket(ss.Received())){
if (playTransaction != -1){
//send a status 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,5,5,2004"));
amfreply.getContentP(2)->addContent(AMF::Object("capabilities", (double)31));
amfreply.getContentP(2)->addContent(AMF::Object("mode", (double)1));
amfreply.addContent(AMF::Object("", "onStatus")); //status reply
amfreply.addContent(AMF::Object("", (double)playTransaction)); //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", "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"));
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
sendCommand(amfreply, messagetype, stream_id);
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));
sendCommand(amfreply, playMessageType, playStreamId);
//send streamisrecorded if stream, well, is recorded.
if (Strm.metadata.isMember("length") && Strm.metadata["length"].asInt() > 0){
Socket.Send(RTMPStream::SendUSR(4, 1)); //send UCM StreamIsRecorded (4), stream 1
}
//send streambegin
Socket.Send(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() == "FCPublish") || (amfdata.getContentP(0)->StrValue() == "FCUnpublish") || (amfdata.getContentP(0)->StrValue() == "releaseStream")){
// ignored
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
sendCommand(amfreply, messagetype, stream_id);
return;
} //getStreamLength
if ((amfdata.getContentP(0)->StrValue() == "publish")){
if (amfdata.getContentP(3)){
streamname = amfdata.getContentP(3)->StrValue();
/// \todo implement push for MistPlayer or restrict and change to getLive
SS = Util::Stream::getStream(streamname);
if ( !SS.connected()){
#if DEBUG >= 1
fprintf(stderr, "Could not connect to server!\n");
#endif
Socket.close(); //disconnect user
return;
}
SS.Send("P ");
SS.Send(Socket.getHost().c_str());
SS.Send("\n");
nostats = true;
}
//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?
sendCommand(amfreply, messagetype, stream_id);
Socket.Send(RTMPStream::SendUSR(0, 1)); //send UCM StreamBegin (0), stream 1
//send a status reply
//and more 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)playTransaction)); //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));
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
sendCommand(amfreply, messagetype, stream_id);
return;
} //checkBandwidth
if ((amfdata.getContentP(0)->StrValue() == "play") || (amfdata.getContentP(0)->StrValue() == "play2")){
//set reply number and stream name, actual reply is sent up in the SS.spool() handler
play_trans = amfdata.getContentP(1)->NumValue();
play_msgtype = messagetype;
play_streamid = stream_id;
streamname = amfdata.getContentP(3)->StrValue();
Connector_RTMP::ready4data = true; //start sending video data!
return;
} //play
if ((amfdata.getContentP(0)->StrValue() == "seek")){
//set reply number and stream name, actual reply is sent up in the SS.spool() handler
play_trans = amfdata.getContentP(1)->NumValue();
play_msgtype = messagetype;
play_streamid = stream_id;
stream_inited = false;
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.Seek.Notify"));
amfreply.getContentP(3)->addContent(AMF::Object("description", "Seeking to the specified time"));
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));
sendCommand(amfreply, play_msgtype, play_streamid);
SS.Send("s ");
SS.Send(JSON::Value((long long int)amfdata.getContentP(3)->NumValue()).asString().c_str());
SS.Send("\n");
return;
} //seek
if ((amfdata.getContentP(0)->StrValue() == "pauseRaw") || (amfdata.getContentP(0)->StrValue() == "pause")){
if (amfdata.getContentP(3)->NumValue()){
SS.Send("q\n"); //quit playing
//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.Pause.Notify"));
amfreply.getContentP(3)->addContent(AMF::Object("description", "Pausing playback"));
amfreply.getContentP(3)->addContent(AMF::Object("details", "DDV"));
amfreply.getContentP(3)->addContent(AMF::Object("clientid", (double)1337));
sendCommand(amfreply, play_msgtype, play_streamid);
}else{
SS.Send("p\n"); //start playing
//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.Unpause.Notify"));
amfreply.getContentP(3)->addContent(AMF::Object("description", "Resuming playback"));
amfreply.getContentP(3)->addContent(AMF::Object("details", "DDV"));
amfreply.getContentP(3)->addContent(AMF::Object("clientid", (double)1337));
sendCommand(amfreply, play_msgtype, play_streamid);
sendCommand(amfreply, playMessageType, playStreamId);
RTMPStream::chunk_snd_max = 102400; //100KiB
Socket.Send(RTMPStream::SendCTL(1, RTMPStream::chunk_snd_max)); //send chunk size max (msg 1)
//send dunno?
Socket.Send(RTMPStream::SendUSR(32, 1)); //send UCM no clue?, stream 1
playTransaction = -1;
}
return;
} //seek
#if DEBUG >= 2
fprintf(stderr, "AMF0 command not processed!\n%s\n", amfdata.Print().c_str());
#endif
} //parseAMFCommand
//sent init data if needed
if ( !streamInited){
init_tag.DTSCMetaInit(Strm);
Socket.SendNow(RTMPStream::SendMedia(init_tag));
if (Strm.metadata.isMember("audio") && Strm.metadata["audio"].isMember("init")){
init_tag.DTSCAudioInit(Strm);
Socket.SendNow(RTMPStream::SendMedia(init_tag));
}
if (Strm.metadata.isMember("video") && Strm.metadata["video"].isMember("init")){
init_tag.DTSCVideoInit(Strm);
Socket.SendNow(RTMPStream::SendMedia(init_tag));
}
streamInited = true;
}
//sent a tag
tag.DTSCLoader(Strm);
Socket.SendNow(RTMPStream::SendMedia(tag));
#if DEBUG >= 8
fprintf(stderr, "Sent tag to %i: [%u] %s\n", Socket.getSocket(), tag.tagTime(), tag.tagType().c_str());
#endif
}
}
}
}
Socket.close();
ss.SendNow(Socket.getStats("RTMP").c_str());
ss.close();
return 0;
} //Connector_RTMP
}
///\brief The standard process-spawning main function.
int main(int argc, char ** argv){
@ -616,7 +622,7 @@ int main(int argc, char ** argv){
if (S.connected()){ //check if the new connection is valid
pid_t myid = fork();
if (myid == 0){ //if new child, start MAINHANDLER
return Connector_RTMP::Connector_RTMP(S);
return Connector_RTMP::rtmpConnector(S);
}else{ //otherwise, do nothing or output debugging text
#if DEBUG >= 5
fprintf(stderr, "Spawned new process %i for socket %i\n", (int)myid, S.getSocket());

27
src/connectors/conn_ts.cpp
View file

@ -2,28 +2,34 @@
/// Contains the main code for the TS Connector
#include <queue>
#include <string>
#include <iostream>
#include <cmath>
#include <ctime>
#include <cstdio>
#include <string>
#include <cstdlib>
#include <cstring>
#include <unistd.h>
#include <getopt.h>
#include <iostream>
#include <sys/time.h>
#include <sys/wait.h>
#include <sys/types.h>
#include <mist/socket.h>
#include <mist/config.h>
#include <mist/stream.h>
#include <mist/ts_packet.h> //TS support
#include <mist/dtsc.h> //DTSC support
#include <mist/mp4.h> //For initdata conversion
/// The main function of the connector
/// \param conn A connection with the client
/// \param streamname The name of the stream
int TS_Handler(Socket::Connection conn, std::string streamname){
///\brief Holds everything unique to the TS Connector
namespace Connector_TS {
///\brief Main function for the TS Connector
///\param conn A socket describing the connection the client.
///\param streamName The stream to connect to.
///\return The exit code of the connector.
int tsConnector(Socket::Connection conn, std::string streamName){
std::string ToPack;
TS::Packet PackData;
std::string DTMIData;
@ -45,11 +51,11 @@ int TS_Handler(Socket::Connection conn, std::string streamname){
while (conn.connected()){
if ( !inited){
ss = Util::Stream::getStream(streamname);
ss = Util::Stream::getStream(streamName);
if ( !ss.connected()){
#if DEBUG >= 1
#if DEBUG >= 1
fprintf(stderr, "Could not connect to server!\n");
#endif
#endif
conn.close();
break;
}
@ -141,6 +147,7 @@ int TS_Handler(Socket::Connection conn, std::string streamname){
}
}
return 0;
}
}
int main(int argc, char ** argv){
@ -160,7 +167,7 @@ int main(int argc, char ** argv){
if (S.connected()){ //check if the new connection is valid
pid_t myid = fork();
if (myid == 0){ //if new child, start MAINHANDLER
return TS_Handler(S, conf.getString("streamname"));
return Connector_TS::tsConnector(S, conf.getString("streamname"));
}else{ //otherwise, do nothing or output debugging text
#if DEBUG >= 5
fprintf(stderr, "Spawned new process %i for socket %i\n", (int)myid, S.getSocket());