diff --git a/examples/helpers/AsyncFileReader.h b/examples/helpers/AsyncFileReader.h new file mode 100644 index 0000000..9537dcc --- /dev/null +++ b/examples/helpers/AsyncFileReader.h @@ -0,0 +1,82 @@ +#include +#include +#include +#include +#include + +/* This is just a very simple and inefficient demo of async responses, + * please do roll your own variant or use a database or Node.js's async + * features instead of this really bad demo */ +struct AsyncFileReader { +private: + /* The cache we have in memory for this file */ + std::string cache; + int cacheOffset = 0; + + /* The pending async file read (yes we only support one pending read) */ + std::function pendingReadCb; + + int fileSize; + std::ifstream fin; + uWS::Loop *loop; + +public: + /* Construct a demo async. file reader for fileName */ + AsyncFileReader(std::string fileName) { + fin.open(fileName, std::ios::binary); + + // get fileSize + fin.seekg(0, fin.end); + fileSize = fin.tellg(); + + //std::cout << "File size is: " << fileSize << std::endl; + + // cache up 1 mb! + cache.resize(1024 * 1024); + + //std::cout << "Caching 1 MB at offset = " << 0 << std::endl; + fin.seekg(0, fin.beg); + fin.read(cache.data(), cache.length()); + cacheOffset = 0; + + // get loop for thread + + loop = uWS::Loop::defaultLoop(); + } + + /* Returns any data already cached for this offset */ + std::string_view peek(int offset) { + /* Did we hit the cache? */ + if (offset >= cacheOffset && ((offset - cacheOffset) < cache.length())) { + /* Cache hit */ + //std::cout << "Cache hit!" << std::endl; + return std::string_view(cache.data() + offset - cacheOffset, cache.length() - offset + cacheOffset); + } else { + /* Cache miss */ + //std::cout << "Cache miss!" << std::endl; + return std::string_view(nullptr, 0); + } + } + + /* Asynchronously request more data at offset */ + void request(int offset, std::function cb) { + + // std::async this + //std::cout << "Caching 1 MB at offset = " << offset << std::endl; + fin.seekg(offset, fin.beg); + fin.read(cache.data(), cache.length()); + cacheOffset = offset; + + // Loop.defer this + cb(std::string_view(cache.data(), cache.length())); + } + + /* Abort any pending async. request */ + void abort() { + + } + + int getFileSize() { + return fileSize; + } +}; diff --git a/examples/helpers/FileCache.h b/examples/helpers/FileCache.h deleted file mode 100644 index d78b67e..0000000 --- a/examples/helpers/FileCache.h +++ /dev/null @@ -1,49 +0,0 @@ -#include -#include -#include - -struct FileCache { -private: - std::map cache; - - void cacheFile(std::string file, std::string root) { - std::cout << "Caching file: " << file << std::endl; - - std::ifstream fin(root + file, std::ios::binary); - std::ostringstream oss; - oss << fin.rdbuf(); - - if (file == "/index.html") { - file = "/"; - } - - char *cachedFile = (char *) malloc(oss.str().size()); - memcpy(cachedFile, oss.str().data(), oss.str().size()); - - char *key = (char *) malloc(file.length()); - memcpy(key, file.data(), file.length()); - - std::cout << "Size: " << oss.str().size() << std::endl; - - cache[std::string_view(key, file.length())] = std::string_view(cachedFile, oss.str().size()); - } - -public: - FileCache(std::string root) { - for(auto &p : std::experimental::filesystem::recursive_directory_iterator(root)) { - cacheFile(p.path().string().substr(root.length()), root); - } - } - - std::string_view getFile(std::string_view file) { - auto it = cache.find(file); - - if (it == cache.end()) { -//std::cout << "Did not find file: " << file << std::endl; - return "

Nope!

"; - } else { -//std::cout << "Did find file: " << file << std::endl; - return it->second; - } - } -}; diff --git a/main.cpp b/main.cpp index 313b17a..a7c666f 100644 --- a/main.cpp +++ b/main.cpp @@ -1,59 +1,11 @@ #include "App.h" -#include -#include -#include -#include -#include +#include "examples/helpers/AsyncFileReader.h" -// should probably fix this one up a bit some time -std::string_view getFile(std::string_view file) { - static std::map cache; - - auto it = cache.find(file); - - if (it == cache.end()) { - std::cout << "Did not have file: " << file << std::endl; - - std::ifstream fin(std::string(file), std::ios::binary); - std::ostringstream oss; - oss << fin.rdbuf(); - - char *cachedFile = (char *) malloc(oss.str().size()); - memcpy(cachedFile, oss.str().data(), oss.str().size()); - - char *key = (char *) malloc(file.length()); - memcpy(key, file.data(), file.length()); - - std::cout << "Size: " << oss.str().size() << std::endl; - - cache[std::string_view(key, file.length())] = std::string_view(cachedFile, oss.str().size()); - - return getFile(file); - } else { - return it->second; - } -} - -#include - -std::set *> delayedResponses; +AsyncFileReader asyncFileReader("/home/alexhultman/v0.15/sintel.mkv"); int main(int argc, char **argv) { - // create a timer that resumes sockets - auto *timer = us_create_timer((us_loop *) uWS::Loop::defaultLoop(), 1, 0); - us_timer_set(timer, [](auto *timer) { - - for (auto *x : delayedResponses) { - std::cout << "Resuming a response now!" << std::endl; - x->resume(); - } - - delayedResponses.clear(); - - }, 1000, 1000); - uWS::/*SSL*/App(/*{ .key_file_name = "/home/alexhultman/uWebSockets/misc/ssl/key.pem", .cert_file_name = "/home/alexhultman/uWebSockets/misc/ssl/cert.pem", @@ -70,25 +22,53 @@ int main(int argc, char **argv) { } return uWS::HTTP_STREAM_FIN; }); - }).get("/delayed", [](auto *res, auto *req) { + }).get("/async/sintel.mkv", [](auto *res, auto *req) { + + // res->write(asyncFileReader.stream(res)) + // asyncFileReader.getAsStream() + /* This route streams back chunks of data in delayed fashion */ res->writeStatus(uWS::HTTP_200_OK)->write([res](int offset) { - /* Handle aborted stream out */ + /* Handle broken stream */ if (offset == -1) { - std::cout << "Removing closed stream!" << std::endl; - delayedResponses.erase(res); - return uWS::HTTP_STREAM_FIN; + std::cout << "Stream was closed by peer!" << std::endl; + return uWS::HTTP_STREAM_IGNORE; } - /* Delay stream out */ - std::cout << "Delaying stream now" << std::endl; - delayedResponses.insert(res); - return uWS::HTTP_STREAM_PAUSE; + /* Peek from cache */ + std::string_view chunk = asyncFileReader.peek(offset); + if (chunk.length()) { + /* We had parts of this file cached already */ + return std::pair(false, chunk); + } else { - }, 100); - }).get("/:folder/:file", [](auto *res, auto *req) { - res->writeStatus(uWS::HTTP_200_OK)->write(getFile((req->getUrl() == "/" ? "/rocket_files/rocket.html" : req->getUrl()).substr(1))); + std::string_view outerChunk; + + /* We had nothing readily available right now, request async chunk and pause the stream until we have */ + asyncFileReader.request(offset, [&outerChunk, 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 { + /* We finally got the data, resume stream with this chunk */ + //res->resume(chunk); + outerChunk = chunk; + } + }); + + std::cout << "Returning chunk of size: " << outerChunk.length() << std::endl; + return std::pair(false, outerChunk); + + // what if we resumed before we paused! we cannot do that! + + std::cout << "PAusing stream out due to empty cache!" << std::endl; + return uWS::HTTP_STREAM_PAUSE; + } + }, asyncFileReader.getFileSize()); }).listen(3000, [](auto *token) { if (token) { std::cout << "Listening on port " << 3000 << std::endl; diff --git a/src/HttpContext.h b/src/HttpContext.h index 52cb70f..c2574f2 100644 --- a/src/HttpContext.h +++ b/src/HttpContext.h @@ -135,6 +135,8 @@ private: /* Handle HTTP write out */ static_dispatch(us_ssl_socket_context_on_writable, us_socket_context_on_writable)(getSocketContext(), [](auto *s) { + std::cout << "Writable event!" << std::endl; + /* Silence any spurious writable events due to SSL_read failing to write */ AsyncSocket *asyncSocket = (AsyncSocket *) s; HttpResponseData *httpResponseData = (HttpResponseData *) asyncSocket->getExt(); @@ -162,8 +164,17 @@ private: if (httpResponseData->outStream) { /* Regular path, request more data */ - auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset); - httpResponseData->offset += asyncSocket->mergeDrain(chunk); + + // todo: share this path with HttpResponse::write (it is exatly the same logic!) + while (true) { + auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset); + int written = asyncSocket->mergeDrain(chunk); + httpResponseData->offset += written; + // this is not correct, we can reach the end! + if (written < chunk.length()) { + break; + } + } // todo: we should loop until we cannot send anymore just like we do in HttpResponse::write(stream)! } else { diff --git a/src/HttpResponse.h b/src/HttpResponse.h index 602fca6..a18a1ca 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -17,6 +17,9 @@ const std::pair HTTP_STREAM_PAUSE = {false, std::string_ /* Return this from a stream callback to signal FIN */ const std::pair HTTP_STREAM_FIN = {false, std::string_view((const char *) 1, 0)}; +/* Nobody cares what value this one has */ +const auto HTTP_STREAM_IGNORE = HTTP_STREAM_FIN; + template struct HttpResponse : public AsyncSocket { private: