From b514da14ca2fe3f935637a9e921582b2fd1d592e Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Sat, 29 Sep 2018 21:35:28 +0200 Subject: [PATCH] Almost working async file streamer --- examples/helpers/AsyncFileReader.h | 7 +- misc/15.pro | 4 +- misc/main.cpp | 135 ++++++++++++++--------------- src/AsyncSocket.h | 12 +-- src/HttpResponse.h | 7 +- 5 files changed, 83 insertions(+), 82 deletions(-) diff --git a/examples/helpers/AsyncFileReader.h b/examples/helpers/AsyncFileReader.h index b64c643..001127e 100644 --- a/examples/helpers/AsyncFileReader.h +++ b/examples/helpers/AsyncFileReader.h @@ -55,9 +55,9 @@ public: /* Cache hit */ //std::cout << "Cache hit!" << std::endl; - if (fileSize - offset < cache.length()) { + /*if (fileSize - offset < cache.length()) { std::cout << "LESS THAN WHAT WE HAVE!" << std::endl; - } + }*/ int chunkSize = std::min(fileSize - offset, cache.length() - offset + cacheOffset); @@ -92,7 +92,7 @@ public: // den har stängts! öppna igen! if (!fin.good()) { fin.close(); - std::cout << "Reopening fin!" << std::endl; + //std::cout << "Reopening fin!" << std::endl; fin.open(fileName, std::ios::binary); } fin.seekg(offset, fin.beg); @@ -104,6 +104,7 @@ public: int chunkSize = std::min(cache.length(), fileSize - offset); + // båda dessa sker, wtf? if (chunkSize == 0) { std::cout << "Zero size!?" << std::endl; } diff --git a/misc/15.pro b/misc/15.pro index 81d4f0d..44a4063 100644 --- a/misc/15.pro +++ b/misc/15.pro @@ -32,5 +32,5 @@ HEADERS += \ ../src/Utilities.h INCLUDEPATH += ../uSockets/src ../src -#QMAKE_CXXFLAGS += -fsanitize=address -LIBS += -pthread -lssl -lcrypto +QMAKE_CXXFLAGS += -fsanitize=address +LIBS += -lasan -pthread -lssl -lcrypto -lstdc++fs diff --git a/misc/main.cpp b/misc/main.cpp index a1672ab..58c225b 100644 --- a/misc/main.cpp +++ b/misc/main.cpp @@ -2,89 +2,86 @@ #include "../examples/helpers/AsyncFileReader.h" -AsyncFileReader asyncFileReader("/home/alexhultman/sintel_small.mp4"); +#include -/*inline std::string slurp(const std::string &path) { - std::ostringstream buf; - std::ifstream input (path.c_str()); - buf << input.rdbuf(); - return buf.str(); -} +struct AsyncFileStreamer { -std::string sintelMovie = slurp("/home/alexhultman/sintel_small.mp4");*/ + std::map asyncFileReaders; + std::string root; -void streamFile(uWS::HttpResponse *res) { - //int offset = res->getWriteOffset(); - - std::cout << "streamFile called with offset: " << res->getWriteOffset() << std::endl; - - /* Peek from cache */ - std::string_view chunk = asyncFileReader.peek(res->getWriteOffset()); - if (!chunk.length() || res->tryEnd(chunk, asyncFileReader.getFileSize())) { - // request new chunk - std::cout << "Requesting new chunk!" << std::endl; - - asyncFileReader.request(res->getWriteOffset(), [res](std::string_view chunk) { - /* We were aborted */ - if (!chunk.length()) { - std::cout << "Async File Read request was aborted!" << std::endl; - // close the socket here? - // we need a way to NOT resume a paused socket! essentially close! - } else { - - // just call streamIt! - std::cout << "Got requested data, calling streamFile now!" << std::endl; - streamFile(res); - } - }); - } else { - std::cout << "Could not write everything, setting onWritable!" << std::endl; - // we have chunk but we could not write everything, so retry this in onWritable - res->onWritable([res](int offset) { - std::cout << "We are writable now, calling StreamFile" << std::endl; - streamFile(res); - return false; - }); + AsyncFileStreamer(std::string root) : root(root) { + // for all files in this path, init the map of AsyncFileReaders + updateRootCache(); } -} + + void updateRootCache() { + // todo: if the root folder changes, we want to reload the cache + for(auto &p : std::experimental::filesystem::recursive_directory_iterator(root)) { + std::string url = p.path().string().substr(root.length()); + if (url == "/index.html") { + url = "/"; + } + + char *key = new char[url.length()]; + memcpy(key, url.data(), url.length()); + asyncFileReaders[std::string_view(key, url.length())] = new AsyncFileReader(p.path().string()); + } + } + + void streamFile(uWS::HttpResponse *res, std::string_view url) { + auto it = asyncFileReaders.find(url); + if (it == asyncFileReaders.end()) { + std::cout << "Did not find file: " << url << std::endl; + } else { + streamFile(res, it->second); + } + } + + static void streamFile(uWS::HttpResponse *res, AsyncFileReader *asyncFileReader) { + /* Peek from cache */ + std::string_view chunk = asyncFileReader->peek(res->getWriteOffset()); + if (!chunk.length() || res->tryEnd(chunk, asyncFileReader->getFileSize())) { + /* Request new chunk */ + // todo: we need to abort this callback if peer closed! + // this also means Loop::defer needs to support aborting + asyncFileReader->request(res->getWriteOffset(), [res, asyncFileReader](std::string_view chunk) { + /* We were aborted for some reason */ + if (!chunk.length()) { + // todo: make sure to check for is_closed internally after all callbacks! + res->close(); + } else { + streamFile(res, asyncFileReader); + } + }); + } else { + /* We failed writing everything, so let's continue when we can */ + res->onWritable([res, asyncFileReader](int offset) { + + // här kan skiten avbrytas! + + streamFile(res, asyncFileReader); + // todo: I don't really know what this is supposed to mean? + return false; + })->onAborted([]() { + std::cout << "ABORTED!" << std::endl; + }); + } + } +}; int main(int argc, char **argv) { - //std::cout << "Sintel movie is " << sintelMovie.length() << " bytes" << std::endl; - - // läs in hela sintel-filmen här, testa strömmarna med den sen! + AsyncFileStreamer *asyncFileStreamer = new AsyncFileStreamer("/home/alexhultman/v0.15/public"); uWS::/*SSL*/App(/*{ .key_file_name = "/home/alexhultman/uWebSockets/misc/ssl/key.pem", .cert_file_name = "/home/alexhultman/uWebSockets/misc/ssl/cert.pem", .dh_params_file_name = "/home/alexhultman/dhparams.pem", .passphrase = "1234" - }*/).get("/", [](auto *res, auto *req) { + }*/).get("/*", [asyncFileStreamer](auto *res, auto *req) { - res->writeStatus(uWS::HTTP_200_OK); - res->writeHeader("Content-Type", "text/html;charset=utf-8"); - - // buffer it up and end by draining it - //res->end(sintelMovie); - - // stream it here - /*if (!res->tryEnd(sintelMovie)) { - res->onWritable([res](int offset) { - return res->tryEnd(std::string_view(sintelMovie).substr(offset)); - })->onAborted([]() { - // was it really aborted? - std::cout << "Streaming was aborted" << std::endl; - }); - }*/ - - // end can be called two times if the total length is longer then given - res->write("

Hallå!

Din user-agent är: "); - res->end(req->getHeader("user-agent")); - - }).get("/async/sintel.mkv", [](auto *res, auto *req) { - - // this should be enough, we can even delay this thing! - streamFile(res); + // depending on the file type we want to also add mime! + asyncFileStreamer->streamFile(res, req->getUrl()); }).listen(3000, [](auto *token) { if (token) { diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index 6c38b88..4bdce35 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -42,7 +42,7 @@ protected: /* Cork this socket. Only one socket may ever be corked per-loop at any given time */ void cork() { - std::cout << "Cork called" << std::endl; + //std::cout << "Cork called" << std::endl; LoopData *loopData = getLoopData(); loopData->corked = true; @@ -53,7 +53,7 @@ protected: int write(const char *src, int length, bool optionally = false, int nextLength = 0) { LoopData *loopData = getLoopData(); - std::cout << "Write called with length: " << length << ", optionally: " << optionally << std::endl; + //std::cout << "Write called with length: " << length << ", optionally: " << optionally << std::endl; AsyncSocketData *asyncSocketData = (AsyncSocketData *) getExt(); @@ -104,7 +104,7 @@ protected: /* Do nothing for a null sized chunk */ if (!length) { - std::cout << "Trying to write 0 length!" << std::endl; + //std::cout << "Trying to write 0 length!" << std::endl; return 0; } @@ -156,7 +156,7 @@ protected: } } - std::cout << "Write returned: " << length << std::endl; + //std::cout << "Write returned: " << length << std::endl; return length; } @@ -164,7 +164,7 @@ protected: /* It does NOT count bytes written from cork buffer (they are already accounted for in the write call responsible for its corking)! */ int uncork(const char *src = nullptr, int length = 0, bool optionally = false) { - std::cout << "Uncork called with length: " << length << std::endl; + //std::cout << "Uncork called with length: " << length << std::endl; LoopData *loopData = getLoopData(); @@ -180,7 +180,7 @@ protected: /* We should only return with new writes, not things written to cork already */ return write(src, length, optionally, 0); } else { - std::cout << "Not even corked!" << std::endl; + //std::cout << "Not even corked!" << std::endl; } return 0; diff --git a/src/HttpResponse.h b/src/HttpResponse.h index a2b2ad2..6ca37cc 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -39,6 +39,9 @@ private: } public: + + using Super::close; + /* Write the HTTP status */ HttpResponse *writeStatus(std::string_view status) { HttpResponseData *httpResponseData = getHttpResponseData(); @@ -167,13 +170,13 @@ public: int written = Super::write(data.data(), data.length(), true); httpResponseData->offset += written; - std::cout << "Offset is now: " << httpResponseData->offset << std::endl; + //std::cout << "Offset is now: " << httpResponseData->offset << std::endl; return written == data.length(); } // this path is completely wrong! - std::cout << "tryEnd returning " << (httpResponseData->offset == /*totalSize*/ data.length()) << std::endl; + //std::cout << "tryEnd returning " << (httpResponseData->offset == /*totalSize*/ data.length()) << std::endl; return httpResponseData->offset == /*totalSize*/ data.length(); }