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

Fixes for everyone! *strooit met fixes als zwarte piet*

Browse source
This commit is contained in:
Thulinma 2012-09-17 16:47:17 +02:00
parent 8ba5823e00
commit 802f8a22b4
6 changed files with 233 additions and 200 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
src/conn_http.cpp
View file

@ -91,14 +91,14 @@ namespace Connector_HTTP{
H.Clean(); H.Clean();
H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver); H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver);
H.SetBody("<!DOCTYPE html><html><head><title>Unsupported Media Type</title></head><body><h1>Unsupported Media Type</h1>The server isn't quite sure what you wanted to receive from it.</body></html>"); H.SetBody("<!DOCTYPE html><html><head><title>Unsupported Media Type</title></head><body><h1>Unsupported Media Type</h1>The server isn't quite sure what you wanted to receive from it.</body></html>");
conn->Send(H.BuildResponse("415", "Unsupported Media Type")); conn->SendNow(H.BuildResponse("415", "Unsupported Media Type"));
} }
void Handle_Timeout(HTTP::Parser & H, Socket::Connection * conn){ void Handle_Timeout(HTTP::Parser & H, Socket::Connection * conn){
H.Clean(); H.Clean();
H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver); H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver);
H.SetBody("<!DOCTYPE html><html><head><title>Gateway timeout</title></head><body><h1>Gateway timeout</h1>Though the server understood your request and attempted to handle it, somehow handling it took longer than it should. Your request has been cancelled - please try again later.</body></html>"); H.SetBody("<!DOCTYPE html><html><head><title>Gateway timeout</title></head><body><h1>Gateway timeout</h1>Though the server understood your request and attempted to handle it, somehow handling it took longer than it should. Your request has been cancelled - please try again later.</body></html>");
conn->Send(H.BuildResponse("504", "Gateway Timeout")); conn->SendNow(H.BuildResponse("504", "Gateway Timeout"));
} }
/// Handles internal requests. /// Handles internal requests.
@ -111,7 +111,7 @@ namespace Connector_HTTP{
H.SetHeader("Content-Type", "text/xml"); H.SetHeader("Content-Type", "text/xml");
H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver); H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver);
H.SetBody("<?xml version=\"1.0\"?><!DOCTYPE cross-domain-policy SYSTEM \"http://www.adobe.com/xml/dtds/cross-domain-policy.dtd\"><cross-domain-policy><allow-access-from domain=\"*\" /><site-control permitted-cross-domain-policies=\"all\"/></cross-domain-policy>"); H.SetBody("<?xml version=\"1.0\"?><!DOCTYPE cross-domain-policy SYSTEM \"http://www.adobe.com/xml/dtds/cross-domain-policy.dtd\"><cross-domain-policy><allow-access-from domain=\"*\" /><site-control permitted-cross-domain-policies=\"all\"/></cross-domain-policy>");
conn->Send(H.BuildResponse("200", "OK")); conn->SendNow(H.BuildResponse("200", "OK"));
return; return;
}//crossdomain.xml }//crossdomain.xml
@ -173,7 +173,7 @@ namespace Connector_HTTP{
response.append("(\"" + streamname + "\"));\n"); response.append("(\"" + streamname + "\"));\n");
} }
H.SetBody(response); H.SetBody(response);
conn->Send(H.BuildResponse("200", "OK")); conn->SendNow(H.BuildResponse("200", "OK"));
return; return;
}//embed code generator }//embed code generator
@ -235,7 +235,7 @@ namespace Connector_HTTP{
return; return;
} }
//forward the original request //forward the original request
connconn[uid]->conn->Send(request); connconn[uid]->conn->SendNow(request);
connconn[uid]->lastuse = 0; connconn[uid]->lastuse = 0;
unsigned int timeout = 0; unsigned int timeout = 0;
//wait for a response //wait for a response
@ -267,13 +267,13 @@ namespace Connector_HTTP{
//known length - simply re-send the request with added headers and continue //known length - simply re-send the request with added headers and continue
H.SetHeader("X-UID", uid); H.SetHeader("X-UID", uid);
H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver); H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver);
conn->Send(H.BuildResponse("200", "OK")); conn->SendNow(H.BuildResponse("200", "OK"));
conn->flush(); conn->flush();
}else{ }else{
//unknown length //unknown length
H.SetHeader("X-UID", uid); H.SetHeader("X-UID", uid);
H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver); H.SetHeader("Server", "mistserver/" PACKAGE_VERSION "/" + Util::Config::libver);
conn->Send(H.BuildResponse("200", "OK")); conn->SendNow(H.BuildResponse("200", "OK"));
//switch out the connection for an empty one - it makes no sense to keep these globally //switch out the connection for an empty one - it makes no sense to keep these globally
Socket::Connection * myConn = connconn[uid]->conn; Socket::Connection * myConn = connconn[uid]->conn;
connconn[uid]->conn = new Socket::Connection(); connconn[uid]->conn = new Socket::Connection();
@ -282,9 +282,8 @@ namespace Connector_HTTP{
while (myConn->connected() && conn->connected()){ while (myConn->connected() && conn->connected()){
if (myConn->Received().size() || myConn->spool()){ if (myConn->Received().size() || myConn->spool()){
//forward any and all incoming data directly without parsing //forward any and all incoming data directly without parsing
conn->Send(myConn->Received().get()); conn->SendNow(myConn->Received().get());
myConn->Received().get().clear(); myConn->Received().get().clear();
conn->flush();
}else{ }else{
usleep(30000); usleep(30000);
} }

186
src/conn_http_dynamic.cpp
View file

@ -26,21 +26,29 @@
/// Holds everything unique to HTTP Dynamic Connector. /// Holds everything unique to HTTP Dynamic Connector.
namespace Connector_HTTP{ namespace Connector_HTTP{
std::string GenerateBootstrap(std::string & MovieId, JSON::Value & metadata){ std::string GenerateBootstrap(std::string & MovieId, JSON::Value & metadata, int fragnum, int starttime){
MP4::AFRT afrt; MP4::AFRT afrt;
afrt.SetUpdate(false); if (starttime == 0){
afrt.SetUpdate(false);
}else{
afrt.SetUpdate(true);
}
afrt.SetTimeScale(1000); afrt.SetTimeScale(1000);
afrt.AddQualityEntry(""); afrt.AddQualityEntry("");
if (!metadata.isMember("video") || !metadata["video"].isMember("keyms") || metadata["video"]["keyms"].asInt() == 0){ if (!metadata.isMember("video") || !metadata["video"].isMember("keyms") || metadata["video"]["keyms"].asInt() == 0){
//metadata["lasttime"].asInt()? //metadata["lasttime"].asInt()?
afrt.AddFragmentRunEntry(1, 0, 2000); //FirstFragment, FirstFragmentTimestamp,Fragment Duration in milliseconds afrt.AddFragmentRunEntry(fragnum, starttime, 2000); //FirstFragment, FirstFragmentTimestamp,Fragment Duration in milliseconds
}else{ }else{
afrt.AddFragmentRunEntry(1, 0, metadata["video"]["keyms"].asInt()); //FirstFragment, FirstFragmentTimestamp,Fragment Duration in milliseconds afrt.AddFragmentRunEntry(fragnum, starttime, metadata["video"]["keyms"].asInt()); //FirstFragment, FirstFragmentTimestamp,Fragment Duration in milliseconds
} }
afrt.WriteContent(); afrt.WriteContent();
MP4::ASRT asrt; MP4::ASRT asrt;
asrt.SetUpdate(false); if (starttime == 0){
asrt.SetUpdate(false);
}else{
asrt.SetUpdate(true);
}
asrt.AddQualityEntry(""); asrt.AddQualityEntry("");
/// \todo Actually use correct number of fragments. /// \todo Actually use correct number of fragments.
asrt.AddSegmentRunEntry(1, 20000);//1 Segment, 20000 Fragments asrt.AddSegmentRunEntry(1, 20000);//1 Segment, 20000 Fragments
@ -58,7 +66,11 @@ namespace Connector_HTTP{
abst.SetLive(true); abst.SetLive(true);
abst.SetMediaTime(0xFFFFFFFF);//metadata["lasttime"].asInt()? abst.SetMediaTime(0xFFFFFFFF);//metadata["lasttime"].asInt()?
} }
abst.SetUpdate(false); if (starttime == 0){
abst.SetUpdate(false);
}else{
abst.SetUpdate(true);
}
abst.SetTimeScale(1000); abst.SetTimeScale(1000);
abst.SetSMPTE(0); abst.SetSMPTE(0);
abst.SetMovieIdentifier(MovieId); abst.SetMovieIdentifier(MovieId);
@ -67,13 +79,11 @@ namespace Connector_HTTP{
abst.AddServerEntry(""); abst.AddServerEntry("");
abst.AddQualityEntry(""); abst.AddQualityEntry("");
abst.WriteContent(); abst.WriteContent();
std::string Result; //#if DEBUG >= 8
Result.append((char*)abst.GetBoxedData(), (int)abst.GetBoxedDataSize());
#if DEBUG >= 8
std::cout << "Sending bootstrap:" << std::endl << abst.toPrettyString(0) << std::endl; std::cout << "Sending bootstrap:" << std::endl << abst.toPrettyString(0) << std::endl;
#endif //#endif
return Base64::encode(Result); return std::string((char*)abst.GetBoxedData(), (int)abst.GetBoxedDataSize());
} }
@ -91,7 +101,7 @@ namespace Connector_HTTP{
"<streamType>recorded</streamType>\n" "<streamType>recorded</streamType>\n"
"<deliveryType>streaming</deliveryType>\n" "<deliveryType>streaming</deliveryType>\n"
"<bestEffortFetchInfo segmentDuration=\""+metadata["length"].asString()+".000\" fragmentDuration=\""+st.str()+"\" />\n" "<bestEffortFetchInfo segmentDuration=\""+metadata["length"].asString()+".000\" fragmentDuration=\""+st.str()+"\" />\n"
"<bootstrapInfo profile=\"named\" id=\"bootstrap1\">" + GenerateBootstrap(MovieId, metadata) + "</bootstrapInfo>\n" "<bootstrapInfo profile=\"named\" id=\"bootstrap1\">" + Base64::encode(GenerateBootstrap(MovieId, metadata, 1, 0)) + "</bootstrapInfo>\n"
"<media streamId=\"1\" bootstrapInfoId=\"bootstrap1\" url=\"" + MovieId + "/\"></media>\n" "<media streamId=\"1\" bootstrapInfoId=\"bootstrap1\" url=\"" + MovieId + "/\"></media>\n"
"</manifest>\n"; "</manifest>\n";
}else{ }else{
@ -101,7 +111,7 @@ namespace Connector_HTTP{
"<mimeType>video/mp4</mimeType>\n" "<mimeType>video/mp4</mimeType>\n"
"<streamType>live</streamType>\n" "<streamType>live</streamType>\n"
"<deliveryType>streaming</deliveryType>\n" "<deliveryType>streaming</deliveryType>\n"
"<bootstrapInfo profile=\"named\" id=\"bootstrap1\">" + GenerateBootstrap(MovieId, metadata) + "</bootstrapInfo>\n" "<bootstrapInfo profile=\"named\" id=\"bootstrap1\">" + Base64::encode(GenerateBootstrap(MovieId, metadata, 1, 0)) + "</bootstrapInfo>\n"
"<media streamId=\"1\" bootstrapInfoId=\"bootstrap1\" url=\"" + MovieId + "/\"></media>\n" "<media streamId=\"1\" bootstrapInfoId=\"bootstrap1\" url=\"" + MovieId + "/\"></media>\n"
"</manifest>\n"; "</manifest>\n";
} }
@ -113,9 +123,10 @@ namespace Connector_HTTP{
/// Main function for Connector_HTTP_Dynamic /// Main function for Connector_HTTP_Dynamic
int Connector_HTTP_Dynamic(Socket::Connection conn){ int Connector_HTTP_Dynamic(Socket::Connection conn){
std::stringstream FlashBuf; std::deque<std::string> FlashBuf;
int flashbuf_nonempty = 0; int FlashBufSize = 0;
FLV::Tag tmp;//temporary tag, for init data long long int FlashBufTime = 0;
FLV::Tag tmp;//temporary tag
DTSC::Stream Strm;//Incoming stream buffer. DTSC::Stream Strm;//Incoming stream buffer.
HTTP::Parser HTTP_R, HTTP_S;//HTTP Receiver en HTTP Sender. HTTP::Parser HTTP_R, HTTP_S;//HTTP Receiver en HTTP Sender.
@ -126,7 +137,6 @@ namespace Connector_HTTP{
bool inited = false; bool inited = false;
Socket::Connection ss(-1); Socket::Connection ss(-1);
std::string streamname; std::string streamname;
FLV::Tag tag;///< Temporary tag buffer.
std::string recBuffer = ""; std::string recBuffer = "";
std::string Quality; std::string Quality;
@ -155,7 +165,7 @@ namespace Connector_HTTP{
ss.close(); ss.close();
HTTP_S.Clean(); HTTP_S.Clean();
HTTP_S.SetBody("No such stream is available on the system. Please try again.\n"); HTTP_S.SetBody("No such stream is available on the system. Please try again.\n");
conn.Send(HTTP_S.BuildResponse("404", "Not found")); conn.SendNow(HTTP_S.BuildResponse("404", "Not found"));
ready4data = false; ready4data = false;
continue; continue;
} }
@ -173,8 +183,7 @@ namespace Connector_HTTP{
#endif #endif
std::stringstream sstream; std::stringstream sstream;
sstream << "f " << ReqFragment << "\no \n"; sstream << "f " << ReqFragment << "\no \n";
ss.Send(sstream.str().c_str()); ss.SendNow(sstream.str().c_str());
ss.flush();
Flash_RequestPending++; Flash_RequestPending++;
}else{ }else{
streamname = HTTP_R.url.substr(1,HTTP_R.url.find("/",1)-1); streamname = HTTP_R.url.substr(1,HTTP_R.url.find("/",1)-1);
@ -185,7 +194,7 @@ namespace Connector_HTTP{
if (Strm.metadata.isMember("length")){receive_marks = true;} if (Strm.metadata.isMember("length")){receive_marks = true;}
std::string manifest = BuildManifest(streamname, Strm.metadata); std::string manifest = BuildManifest(streamname, Strm.metadata);
HTTP_S.SetBody(manifest); HTTP_S.SetBody(manifest);
conn.Send(HTTP_S.BuildResponse("200", "OK")); conn.SendNow(HTTP_S.BuildResponse("200", "OK"));
#if DEBUG >= 3 #if DEBUG >= 3
printf("Sent manifest\n"); printf("Sent manifest\n");
#endif #endif
@ -198,7 +207,11 @@ namespace Connector_HTTP{
HTTP_R.Clean(); //clean for any possible next requests HTTP_R.Clean(); //clean for any possible next requests
} }
}else{ }else{
usleep(10000);//sleep 10ms if (Flash_RequestPending){
usleep(1000);//sleep 1ms
}else{
usleep(10000);//sleep 10ms
}
} }
if (ready4data){ if (ready4data){
if (!inited){ if (!inited){
@ -211,7 +224,7 @@ namespace Connector_HTTP{
ss.close(); ss.close();
HTTP_S.Clean(); HTTP_S.Clean();
HTTP_S.SetBody("No such stream is available on the system. Please try again.\n"); HTTP_S.SetBody("No such stream is available on the system. Please try again.\n");
conn.Send(HTTP_S.BuildResponse("404", "Not found")); conn.SendNow(HTTP_S.BuildResponse("404", "Not found"));
ready4data = false; ready4data = false;
continue; continue;
} }
@ -225,10 +238,10 @@ namespace Connector_HTTP{
if (now != lastStats){ if (now != lastStats){
lastStats = now; lastStats = now;
ss.Send("S "); ss.Send("S ");
ss.Send(conn.getStats("HTTP_Dynamic").c_str()); ss.SendNow(conn.getStats("HTTP_Dynamic").c_str());
} }
if (ss.spool() || ss.Received().size()){ if (ss.spool()){
if (Strm.parsePacket(ss.Received())){ while (Strm.parsePacket(ss.Received())){
if (Strm.getPacket(0).isMember("time")){ if (Strm.getPacket(0).isMember("time")){
if (!Strm.metadata.isMember("firsttime")){ if (!Strm.metadata.isMember("firsttime")){
Strm.metadata["firsttime"] = Strm.getPacket(0)["time"]; Strm.metadata["firsttime"] = Strm.getPacket(0)["time"];
@ -239,9 +252,6 @@ namespace Connector_HTTP{
} }
Strm.metadata["lasttime"] = Strm.getPacket(0)["time"]; Strm.metadata["lasttime"] = Strm.getPacket(0)["time"];
} }
if (Strm.lastType() == DTSC::VIDEO || Strm.lastType() == DTSC::AUDIO){
tag.DTSCLoader(Strm);
}
if (pending_manifest){ if (pending_manifest){
HTTP_S.Clean(); HTTP_S.Clean();
HTTP_S.SetHeader("Content-Type","text/xml"); HTTP_S.SetHeader("Content-Type","text/xml");
@ -249,7 +259,7 @@ namespace Connector_HTTP{
if (Strm.metadata.isMember("length")){receive_marks = true;} if (Strm.metadata.isMember("length")){receive_marks = true;}
std::string manifest = BuildManifest(streamname, Strm.metadata); std::string manifest = BuildManifest(streamname, Strm.metadata);
HTTP_S.SetBody(manifest); HTTP_S.SetBody(manifest);
conn.Send(HTTP_S.BuildResponse("200", "OK")); conn.SendNow(HTTP_S.BuildResponse("200", "OK"));
#if DEBUG >= 3 #if DEBUG >= 3
printf("Sent manifest\n"); printf("Sent manifest\n");
#endif #endif
@ -257,64 +267,88 @@ namespace Connector_HTTP{
} }
if (!receive_marks && Strm.metadata.isMember("length")){receive_marks = true;} if (!receive_marks && Strm.metadata.isMember("length")){receive_marks = true;}
if ((Strm.getPacket(0).isMember("keyframe") && !receive_marks) || Strm.lastType() == DTSC::PAUSEMARK){ if ((Strm.getPacket(0).isMember("keyframe") && !receive_marks) || Strm.lastType() == DTSC::PAUSEMARK){
if (flashbuf_nonempty || Strm.lastType() == DTSC::PAUSEMARK){ #if DEBUG >= 4
#if DEBUG >= 4 fprintf(stderr, "Received a %s fragment of %i bytes.\n", Strm.getPacket(0)["datatype"].asString().c_str(), FlashBufSize);
fprintf(stderr, "Received a %s fragment of %i packets.\n", Strm.getPacket(0)["datatype"].asString().c_str(), flashbuf_nonempty); #endif
if (Flash_RequestPending > 0 && FlashBufSize){
#if DEBUG >= 3
fprintf(stderr, "Sending a fragment...");
#endif #endif
if (Flash_RequestPending > 0){ static std::string btstrp;
HTTP_S.Clean(); btstrp = GenerateBootstrap(streamname, Strm.metadata, ReqFragment, FlashBufTime);
HTTP_S.SetHeader("Content-Type","video/mp4"); HTTP_S.Clean();
HTTP_S.SetBody(MP4::mdatFold(FlashBuf.str())); HTTP_S.SetHeader("Content-Type", "video/mp4");
conn.Send(HTTP_S.BuildResponse("200", "OK")); HTTP_S.SetBody("");
Flash_RequestPending--; HTTP_S.SetHeader("Content-Length", FlashBufSize+32+33+btstrp.size());
#if DEBUG >= 3 conn.SendNow(HTTP_S.BuildResponse("200", "OK"));
fprintf(stderr, "Sending a fragment..."); conn.SendNow("\x00\x00\x00\x21" "afra\x00\x00\x00\x00\x00\x00\x00\x03\xE8\x00\x00\x00\x01", 21);
#endif unsigned long tmptime = htonl(FlashBufTime << 32);
conn.flush(); conn.SendNow((char*)&tmptime, 4);
#if DEBUG >= 3 tmptime = htonl(FlashBufTime & 0xFFFFFFFF);
fprintf(stderr, "Done\n"); conn.SendNow((char*)&tmptime, 4);
#endif tmptime = htonl(65);
conn.SendNow((char*)&tmptime, 4);
conn.SendNow(btstrp);
conn.SendNow("\x00\x00\x00\x18moof\x00\x00\x00\x10mfhd\x00\x00\x00\x00", 20);
unsigned long fragno = htonl(ReqFragment);
conn.SendNow((char*)&fragno, 4);
unsigned long size = htonl(FlashBufSize+8);
conn.SendNow((char*)&size, 4);
conn.SendNow("mdat", 4);
while (FlashBuf.size() > 0){
conn.SendNow(FlashBuf.front());
FlashBuf.pop_front();
} }
Flash_RequestPending--;
#if DEBUG >= 3
fprintf(stderr, "Done\n");
#endif
} }
FlashBuf.str(""); FlashBuf.clear();
flashbuf_nonempty = 0; FlashBufSize = 0;
//fill buffer with init data, if needed.
if (Strm.metadata.isMember("audio") && Strm.metadata["audio"].isMember("init")){
tmp.DTSCAudioInit(Strm);
FlashBuf.write(tmp.data, tmp.len);
}
if (Strm.metadata.isMember("video") && Strm.metadata["video"].isMember("init")){
tmp.DTSCVideoInit(Strm);
FlashBuf.write(tmp.data, tmp.len);
}
} }
if (Strm.lastType() == DTSC::VIDEO || Strm.lastType() == DTSC::AUDIO){ if (Strm.lastType() == DTSC::VIDEO || Strm.lastType() == DTSC::AUDIO){
++flashbuf_nonempty; if (FlashBufSize == 0){
FlashBuf.write(tag.data, tag.len); //fill buffer with init data, if needed.
} if (Strm.metadata.isMember("audio") && Strm.metadata["audio"].isMember("init")){
}else{ tmp.DTSCAudioInit(Strm);
if (pending_manifest && !Strm.metadata.isNull()){ FlashBuf.push_back(std::string(tmp.data, tmp.len));
HTTP_S.Clean(); FlashBufSize += tmp.len;
HTTP_S.SetHeader("Content-Type","text/xml"); }
HTTP_S.SetHeader("Cache-Control","no-cache"); if (Strm.metadata.isMember("video") && Strm.metadata["video"].isMember("init")){
if (Strm.metadata.isMember("length")){receive_marks = true;} tmp.DTSCVideoInit(Strm);
std::string manifest = BuildManifest(streamname, Strm.metadata); FlashBuf.push_back(std::string(tmp.data, tmp.len));
HTTP_S.SetBody(manifest); FlashBufSize += tmp.len;
conn.Send(HTTP_S.BuildResponse("200", "OK")); }
#if DEBUG >= 3 FlashBufTime = Strm.getPacket(0)["time"].asInt();
printf("Sent manifest\n"); }
#endif tmp.DTSCLoader(Strm);
pending_manifest = false; FlashBuf.push_back(std::string(tmp.data, tmp.len));
FlashBufSize += tmp.len;
} }
} }
if (pending_manifest && !Strm.metadata.isNull()){
HTTP_S.Clean();
HTTP_S.SetHeader("Content-Type","text/xml");
HTTP_S.SetHeader("Cache-Control","no-cache");
if (Strm.metadata.isMember("length")){receive_marks = true;}
std::string manifest = BuildManifest(streamname, Strm.metadata);
HTTP_S.SetBody(manifest);
conn.SendNow(HTTP_S.BuildResponse("200", "OK"));
#if DEBUG >= 3
printf("Sent manifest\n");
#endif
pending_manifest = false;
}
} }
if (!ss.connected()){break;} if (!ss.connected()){break;}
} }
} }
conn.close(); conn.close();
ss.Send("S "); ss.Send("S ");
ss.Send(conn.getStats("HTTP_Dynamic").c_str()); ss.SendNow(conn.getStats("HTTP_Dynamic").c_str());
ss.flush();
ss.close(); ss.close();
#if DEBUG >= 1 #if DEBUG >= 1
if (FLV::Parse_Error){fprintf(stderr, "FLV Parser Error: %s\n", FLV::Error_Str.c_str());} if (FLV::Parse_Error){fprintf(stderr, "FLV Parser Error: %s\n", FLV::Error_Str.c_str());}

32
src/conn_http_progressive.cpp
View file

@ -55,7 +55,7 @@ namespace Connector_HTTP{
HTTP_R.Clean(); //clean for any possible next requests HTTP_R.Clean(); //clean for any possible next requests
} }
}else{ }else{
usleep(10000);//sleep 10ms usleep(1000);//sleep 1ms
} }
if (ready4data){ if (ready4data){
if (!inited){ if (!inited){
@ -68,58 +68,57 @@ namespace Connector_HTTP{
ss.close(); ss.close();
HTTP_S.Clean(); HTTP_S.Clean();
HTTP_S.SetBody("No such stream is available on the system. Please try again.\n"); HTTP_S.SetBody("No such stream is available on the system. Please try again.\n");
conn.Send(HTTP_S.BuildResponse("404", "Not found")); conn.SendNow(HTTP_S.BuildResponse("404", "Not found"));
ready4data = false; ready4data = false;
continue; continue;
} }
if (seek_pos){ if (seek_pos){
std::stringstream cmd; std::stringstream cmd;
cmd << "s " << seek_pos << "\n"; cmd << "s " << seek_pos << "\n";
ss.Send(cmd.str().c_str()); ss.SendNow(cmd.str().c_str());
} }
#if DEBUG >= 3 #if DEBUG >= 3
fprintf(stderr, "Everything connected, starting to send video data...\n"); fprintf(stderr, "Everything connected, starting to send video data...\n");
#endif #endif
ss.Send("p\n"); ss.SendNow("p\n");
ss.flush();
inited = true; inited = true;
} }
unsigned int now = time(0); unsigned int now = time(0);
if (now != lastStats){ if (now != lastStats){
lastStats = now; lastStats = now;
ss.Send("S "); ss.Send("S ");
ss.Send(conn.getStats("HTTP_Progressive").c_str()); ss.SendNow(conn.getStats("HTTP_Progressive").c_str());
} }
if (ss.spool() || ss.Received().size()){ if (ss.spool()){
if (Strm.parsePacket(ss.Received())){ while (Strm.parsePacket(ss.Received())){
tag.DTSCLoader(Strm);
if (!progressive_has_sent_header){ if (!progressive_has_sent_header){
HTTP_S.Clean();//make sure no parts of old requests are left in any buffers HTTP_S.Clean();//make sure no parts of old requests are left in any buffers
HTTP_S.SetHeader("Content-Type", "video/x-flv");//Send the correct content-type for FLV files HTTP_S.SetHeader("Content-Type", "video/x-flv");//Send the correct content-type for FLV files
//HTTP_S.SetHeader("Transfer-Encoding", "chunked"); //HTTP_S.SetHeader("Transfer-Encoding", "chunked");
HTTP_S.protocol = "HTTP/1.0"; HTTP_S.protocol = "HTTP/1.0";
conn.Send(HTTP_S.BuildResponse("200", "OK"));//no SetBody = unknown length - this is intentional, we will stream the entire file conn.SendNow(HTTP_S.BuildResponse("200", "OK"));//no SetBody = unknown length - this is intentional, we will stream the entire file
conn.Send(FLV::Header, 13);//write FLV header conn.SendNow(FLV::Header, 13);//write FLV header
static FLV::Tag tmp; static FLV::Tag tmp;
//write metadata //write metadata
tmp.DTSCMetaInit(Strm); tmp.DTSCMetaInit(Strm);
conn.Send(tmp.data, tmp.len); conn.SendNow(tmp.data, tmp.len);
//write video init data, if needed //write video init data, if needed
if (Strm.metadata.isMember("video") && Strm.metadata["video"].isMember("init")){ if (Strm.metadata.isMember("video") && Strm.metadata["video"].isMember("init")){
tmp.DTSCVideoInit(Strm); tmp.DTSCVideoInit(Strm);
conn.Send(tmp.data, tmp.len); conn.SendNow(tmp.data, tmp.len);
} }
//write audio init data, if needed //write audio init data, if needed
if (Strm.metadata.isMember("audio") && Strm.metadata["audio"].isMember("init")){ if (Strm.metadata.isMember("audio") && Strm.metadata["audio"].isMember("init")){
tmp.DTSCAudioInit(Strm); tmp.DTSCAudioInit(Strm);
conn.Send(tmp.data, tmp.len); conn.SendNow(tmp.data, tmp.len);
} }
progressive_has_sent_header = true; progressive_has_sent_header = true;
#if DEBUG >= 1 #if DEBUG >= 1
fprintf(stderr, "Sent progressive FLV header\n"); fprintf(stderr, "Sent progressive FLV header\n");
#endif #endif
} }
conn.Send(tag.data, tag.len);//write the tag contents tag.DTSCLoader(Strm);
conn.SendNow(tag.data, tag.len);//write the tag contents
} }
} }
if (!ss.connected()){break;} if (!ss.connected()){break;}
@ -127,8 +126,7 @@ namespace Connector_HTTP{
} }
conn.close(); conn.close();
ss.Send("S "); ss.Send("S ");
ss.Send(conn.getStats("HTTP_Dynamic").c_str()); ss.SendNow(conn.getStats("HTTP_Dynamic").c_str());
ss.flush();
ss.close(); ss.close();
#if DEBUG >= 1 #if DEBUG >= 1
if (FLV::Parse_Error){fprintf(stderr, "FLV Parser Error: %s\n", FLV::Error_Str.c_str());} if (FLV::Parse_Error){fprintf(stderr, "FLV Parser Error: %s\n", FLV::Error_Str.c_str());}

7
src/conn_raw.cpp
View file

@ -38,15 +38,12 @@ int main(int argc, char ** argv) {
lastStats = now; lastStats = now;
std::stringstream st; std::stringstream st;
st << "S localhost RAW " << (time(0) - started) << " " << S.dataDown() << " " << S.dataUp() << "\n"; st << "S localhost RAW " << (time(0) - started) << " " << S.dataDown() << " " << S.dataUp() << "\n";
std::string tmp = st.str(); S.SendNow(st.str().c_str());
S.Send(tmp);
} }
} }
std::stringstream st; std::stringstream st;
st << "S localhost RAW " << (time(0) - started) << " " << S.dataDown() << " " << S.dataUp() << "\n"; st << "S localhost RAW " << (time(0) - started) << " " << S.dataDown() << " " << S.dataUp() << "\n";
std::string tmp = st.str(); S.SendNow(st.str().c_str());
S.Send(tmp);
S.flush();
S.close(); S.close();
return 0; return 0;
} }

39
src/conn_rtmp.cpp
View file

@ -57,7 +57,7 @@ int Connector_RTMP::Connector_RTMP(Socket::Connection conn){
RTMPStream::rec_cnt += 1537; RTMPStream::rec_cnt += 1537;
if (RTMPStream::doHandshake()){ if (RTMPStream::doHandshake()){
Socket.Send(RTMPStream::handshake_out); Socket.SendNow(RTMPStream::handshake_out);
while (!Socket.Received().available(1536) && Socket.connected()){Socket.spool(); usleep(5000);} while (!Socket.Received().available(1536) && Socket.connected()){Socket.spool(); usleep(5000);}
Socket.Received().remove(1536); Socket.Received().remove(1536);
RTMPStream::rec_cnt += 1536; RTMPStream::rec_cnt += 1536;
@ -74,10 +74,12 @@ int Connector_RTMP::Connector_RTMP(Socket::Connection conn){
unsigned int lastStats = 0; unsigned int lastStats = 0;
while (Socket.connected()){ while (Socket.connected()){
if (Socket.Received().size() || Socket.spool()){ if (Socket.spool() || Socket.Received().size()){
parseChunk(Socket.Received().get()); while (Socket.Received().size()){
parseChunk(Socket.Received().get());
}
}else{ }else{
usleep(10000);//sleep 10ms to prevent high CPU usage usleep(1000);//sleep 1ms to prevent high CPU usage
} }
if (ready4data){ if (ready4data){
if (!inited){ if (!inited){
@ -94,7 +96,7 @@ int Connector_RTMP::Connector_RTMP(Socket::Connection conn){
#if DEBUG >= 3 #if DEBUG >= 3
fprintf(stderr, "Everything connected, starting to send video data...\n"); fprintf(stderr, "Everything connected, starting to send video data...\n");
#endif #endif
SS.Send("p\n");SS.flush(); SS.SendNow("p\n");
inited = true; inited = true;
} }
if (inited && !nostats){ if (inited && !nostats){
@ -102,11 +104,11 @@ int Connector_RTMP::Connector_RTMP(Socket::Connection conn){
if (now != lastStats){ if (now != lastStats){
lastStats = now; lastStats = now;
SS.Send("S "); SS.Send("S ");
SS.Send(Socket.getStats("RTMP").c_str()); SS.SendNow(Socket.getStats("RTMP").c_str());
} }
} }
if (SS.spool() || SS.Received().size()){ if (SS.spool()){
if (Strm.parsePacket(SS.Received())){ while (Strm.parsePacket(SS.Received())){
if (play_trans != -1){ if (play_trans != -1){
//send a status reply //send a status reply
AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER); AMF::Object amfreply("container", AMF::AMF0_DDV_CONTAINER);
@ -148,20 +150,20 @@ int Connector_RTMP::Connector_RTMP(Socket::Connection conn){
//sent init data if needed //sent init data if needed
if (!stream_inited){ if (!stream_inited){
init_tag.DTSCMetaInit(Strm); init_tag.DTSCMetaInit(Strm);
Socket.Send(RTMPStream::SendMedia(init_tag)); Socket.SendNow(RTMPStream::SendMedia(init_tag));
if (Strm.metadata.isMember("audio") && Strm.metadata["audio"].isMember("init")){ if (Strm.metadata.isMember("audio") && Strm.metadata["audio"].isMember("init")){
init_tag.DTSCAudioInit(Strm); init_tag.DTSCAudioInit(Strm);
Socket.Send(RTMPStream::SendMedia(init_tag)); Socket.SendNow(RTMPStream::SendMedia(init_tag));
} }
if (Strm.metadata.isMember("video") && Strm.metadata["video"].isMember("init")){ if (Strm.metadata.isMember("video") && Strm.metadata["video"].isMember("init")){
init_tag.DTSCVideoInit(Strm); init_tag.DTSCVideoInit(Strm);
Socket.Send(RTMPStream::SendMedia(init_tag)); Socket.SendNow(RTMPStream::SendMedia(init_tag));
} }
stream_inited = true; stream_inited = true;
} }
//sent a tag //sent a tag
tag.DTSCLoader(Strm); tag.DTSCLoader(Strm);
Socket.Send(RTMPStream::SendMedia(tag)); Socket.SendNow(RTMPStream::SendMedia(tag));
#if DEBUG >= 8 #if DEBUG >= 8
fprintf(stderr, "Sent tag to %i: [%u] %s\n", Socket.getSocket(), tag.tagTime(), tag.tagType().c_str()); fprintf(stderr, "Sent tag to %i: [%u] %s\n", Socket.getSocket(), tag.tagTime(), tag.tagType().c_str());
#endif #endif
@ -171,8 +173,7 @@ int Connector_RTMP::Connector_RTMP(Socket::Connection conn){
} }
Socket.close(); Socket.close();
SS.Send("S "); SS.Send("S ");
SS.Send(Socket.getStats("RTMP").c_str()); SS.SendNow(Socket.getStats("RTMP").c_str());
SS.flush();
SS.close(); SS.close();
#if DEBUG >= 1 #if DEBUG >= 1
if (FLV::Parse_Error){fprintf(stderr, "FLV Parse Error: %s\n", FLV::Error_Str.c_str());} if (FLV::Parse_Error){fprintf(stderr, "FLV Parse Error: %s\n", FLV::Error_Str.c_str());}
@ -281,15 +282,15 @@ void Connector_RTMP::parseChunk(std::string & inbuffer){
counter++; counter++;
if (counter > 8){ if (counter > 8){
sending = true; sending = true;
SS.Send(meta_out.toNetPacked()); SS.SendNow(meta_out.toNetPacked());
SS.Send(prebuffer.str().c_str());//write buffer SS.SendNow(prebuffer.str().c_str());//write buffer
prebuffer.str("");//clear buffer prebuffer.str("");//clear buffer
SS.Send(pack_out.toNetPacked()); SS.Send(pack_out.toNetPacked());
}else{ }else{
prebuffer << pack_out.toNetPacked(); prebuffer << pack_out.toNetPacked();
} }
}else{ }else{
SS.Send(pack_out.toNetPacked()); SS.SendNow(pack_out.toNetPacked());
} }
} }
}else{ }else{
@ -357,9 +358,9 @@ void Connector_RTMP::sendCommand(AMF::Object & amfreply, int messagetype, int st
std::cerr << amfreply.Print() << std::endl; std::cerr << amfreply.Print() << std::endl;
#endif #endif
if (messagetype == 17){ if (messagetype == 17){
Socket.Send(RTMPStream::SendChunk(3, messagetype, stream_id, (char)0+amfreply.Pack())); Socket.SendNow(RTMPStream::SendChunk(3, messagetype, stream_id, (char)0+amfreply.Pack()));
}else{ }else{
Socket.Send(RTMPStream::SendChunk(3, messagetype, stream_id, amfreply.Pack())); Socket.SendNow(RTMPStream::SendChunk(3, messagetype, stream_id, amfreply.Pack()));
} }
}//sendCommand }//sendCommand

152
src/player.cpp
View file

@ -70,7 +70,7 @@ int main(int argc, char** argv){
pausemark["time"] = (long long int)0; pausemark["time"] = (long long int)0;
Socket::Connection StatsSocket = Socket::Connection("/tmp/mist/statistics", true); Socket::Connection StatsSocket = Socket::Connection("/tmp/mist/statistics", true);
int lasttime = time(0); int lasttime = time(0);//time last packet was sent
//send the header //send the header
{ {
@ -81,95 +81,100 @@ int main(int argc, char** argv){
} }
JSON::Value meta = JSON::fromDTMI(meta_str); JSON::Value meta = JSON::fromDTMI(meta_str);
if (meta["video"]["keyms"].asInt() < 11){
meta["video"]["keyms"] = (long long int)1000;
}
JSON::Value last_pack; JSON::Value last_pack;
bool meta_sent = false; bool meta_sent = false;
long long now, timeDiff = 0, lastTime = 0; long long now, lastTime = 0;//for timing of sending packets
long long bench = 0;//for benchmarking
Stats sts; Stats sts;
while (in_out.connected() && std::cin.good() && std::cout.good() && (time(0) - lasttime < 60)){ while (in_out.connected() && std::cin.good() && std::cout.good() && (time(0) - lasttime < 60)){
if (in_out.Received().size() || in_out.spool()){ if (in_out.spool()){
//delete anything that doesn't end with a newline while (in_out.Received().size()){
if (!in_out.Received().get().empty() && *(in_out.Received().get().rbegin()) != '\n'){ //delete anything that doesn't end with a newline
in_out.Received().get().clear(); if (*(in_out.Received().get().rbegin()) != '\n'){
continue; in_out.Received().get().clear();
} continue;
in_out.Received().get().resize(in_out.Received().get().size() - 1); }
if (!in_out.Received().get().empty()){ in_out.Received().get().resize(in_out.Received().get().size() - 1);
switch (in_out.Received().get()[0]){ if (!in_out.Received().get().empty()){
case 'P':{ //Push switch (in_out.Received().get()[0]){
#if DEBUG >= 4 case 'P':{ //Push
std::cerr << "Received push - ignoring (" << in_out.Received().get() << ")" << std::endl; #if DEBUG >= 4
#endif std::cerr << "Received push - ignoring (" << in_out.Received().get() << ")" << std::endl;
in_out.close();//pushing to VoD makes no sense #endif
} break; in_out.close();//pushing to VoD makes no sense
case 'S':{ //Stats } break;
if (!StatsSocket.connected()){ case 'S':{ //Stats
StatsSocket = Socket::Connection("/tmp/mist/statistics", true); if (!StatsSocket.connected()){
} StatsSocket = Socket::Connection("/tmp/mist/statistics", true);
if (StatsSocket.connected()){ }
sts = Stats(in_out.Received().get().substr(2)); if (StatsSocket.connected()){
JSON::Value json_sts; sts = Stats(in_out.Received().get().substr(2));
json_sts["vod"]["down"] = (long long int)sts.down; JSON::Value json_sts;
json_sts["vod"]["up"] = (long long int)sts.up; json_sts["vod"]["down"] = (long long int)sts.down;
json_sts["vod"]["time"] = (long long int)sts.conntime; json_sts["vod"]["up"] = (long long int)sts.up;
json_sts["vod"]["host"] = sts.host; json_sts["vod"]["time"] = (long long int)sts.conntime;
json_sts["vod"]["connector"] = sts.connector; json_sts["vod"]["host"] = sts.host;
json_sts["vod"]["filename"] = conf.getString("filename"); json_sts["vod"]["connector"] = sts.connector;
json_sts["vod"]["now"] = (long long int)time(0); json_sts["vod"]["filename"] = conf.getString("filename");
json_sts["vod"]["start"] = (long long int)(time(0) - sts.conntime); json_sts["vod"]["now"] = (long long int)time(0);
if (!meta_sent){ json_sts["vod"]["start"] = (long long int)(time(0) - sts.conntime);
json_sts["vod"]["meta"] = meta; if (!meta_sent){
meta_sent = true; json_sts["vod"]["meta"] = meta;
} meta_sent = true;
StatsSocket.Send(json_sts.toString().c_str()); }
StatsSocket.Send("\n\n"); StatsSocket.Send(json_sts.toString().c_str());
StatsSocket.flush(); StatsSocket.Send("\n\n");
} StatsSocket.flush();
} break; }
case 's':{ //second-seek } break;
int ms = JSON::Value(in_out.Received().get().substr(2)).asInt(); case 's':{ //second-seek
bool ret = source.seek_time(ms); int ms = JSON::Value(in_out.Received().get().substr(2)).asInt();
} break; bool ret = source.seek_time(ms);
case 'f':{ //frame-seek } break;
bool ret = source.seek_frame(JSON::Value(in_out.Received().get().substr(2)).asInt()); case 'f':{ //frame-seek
} break; bool ret = source.seek_frame(JSON::Value(in_out.Received().get().substr(2)).asInt());
case 'p':{ //play } break;
playing = -1; case 'p':{ //play
in_out.setBlocking(false); playing = -1;
} break; in_out.setBlocking(false);
case 'o':{ //once-play } break;
if (playing <= 0){playing = 1;} case 'o':{ //once-play
++playing; if (playing <= 0){playing = 1;}
in_out.setBlocking(false); ++playing;
} break; in_out.setBlocking(false);
case 'q':{ //quit-playing #if DEBUG >= 4
playing = 0; std::cerr << "Playing one keyframe" << std::endl;
in_out.setBlocking(true); #endif
} break; bench = getNowMS();
} break;
case 'q':{ //quit-playing
playing = 0;
in_out.setBlocking(true);
} break;
}
in_out.Received().get().clear();
} }
in_out.Received().get().clear();
} }
} }
if (playing != 0){ if (playing != 0){
now = getNowMS(); now = getNowMS();
/// \todo This makes no sense. We're timing for packets here, but sending a whole keyframe. Fix. ASAP. if (playing > 0 || meta["video"]["keyms"].asInt() <= now-lastTime) {
if (playing > 0 || now - timeDiff >= lastTime || lastTime - (now - timeDiff) > 15000) {
source.seekNext(); source.seekNext();
lastTime = source.getJSON()["time"].asInt();
if ((now - timeDiff - lastTime) > 5000 || (now - timeDiff - lastTime < -5000)){
timeDiff = now - lastTime;
}
if (source.getJSON().isMember("keyframe")){ if (source.getJSON().isMember("keyframe")){
lastTime = now;
if (playing > 0){--playing;} if (playing > 0){--playing;}
if (playing == 0){ if (playing == 0){
#if DEBUG >= 4 #if DEBUG >= 4
std::cerr << "Sending pause_marker" << std::endl; std::cerr << "Sending pause_marker (" << (getNowMS() - bench) << "ms)" << std::endl;
#endif #endif
pausemark["time"] = (long long int)now; pausemark["time"] = (long long int)now;
pausemark.toPacked(); pausemark.toPacked();
in_out.Send(pausemark.toNetPacked()); in_out.SendNow(pausemark.toNetPacked());
in_out.flush();
in_out.setBlocking(true); in_out.setBlocking(true);
} }
} }
@ -180,11 +185,10 @@ int main(int argc, char** argv){
//insert the packet length //insert the packet length
unsigned int size = htonl(source.getPacket().size()); unsigned int size = htonl(source.getPacket().size());
in_out.Send((char*)&size, 4); in_out.Send((char*)&size, 4);
in_out.Send(source.getPacket()); in_out.SendNow(source.getPacket());
in_out.flush();
} }
} else { } else {
usleep(std::min(10000LL, lastTime - (now - timeDiff)) * 1000); usleep((meta["video"]["keyms"].asInt()-(now-lastTime))*1000);
} }
}else{ }else{
usleep(10000);//sleep 10ms usleep(10000);//sleep 10ms