Almost working async file streamer

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<int>(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<int>(cache.length(), fileSize - offset);
 
+                // båda dessa sker, wtf?
                 if (chunkSize == 0) {
                     std::cout << "Zero size!?" << std::endl;
                 }

misc/15.pro

@@ -32,5 +32,5 @@ HEADERS += \
     ../src/Utilities.h
 
 INCLUDEPATH += ../uSockets/src ../src
-#QMAKE_CXXFLAGS += -fsanitize=address
+QMAKE_CXXFLAGS += -fsanitize=address
-LIBS += -pthread -lssl -lcrypto
+LIBS += -lasan -pthread -lssl -lcrypto -lstdc++fs

misc/main.cpp

@@ -2,89 +2,86 @@
 
 #include "../examples/helpers/AsyncFileReader.h"
 
-AsyncFileReader asyncFileReader("/home/alexhultman/sintel_small.mp4");
+#include <experimental/filesystem>
 
-/*inline std::string slurp(const std::string &path) {
+struct AsyncFileStreamer {
-  std::ostringstream buf;
-  std::ifstream input (path.c_str());
-  buf << input.rdbuf();
-  return buf.str();
-}
 
-std::string sintelMovie = slurp("/home/alexhultman/sintel_small.mp4");*/
+    std::map<std::string_view, AsyncFileReader *> asyncFileReaders;
+    std::string root;
 
-void streamFile(uWS::HttpResponse<false> *res) {
+    AsyncFileStreamer(std::string root) : root(root) {
-    //int offset = res->getWriteOffset();
+        // for all files in this path, init the map of AsyncFileReaders
-
+        updateRootCache();
-    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;
-        });
     }
-}
+
+    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<false> *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<false> *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;
+    AsyncFileStreamer *asyncFileStreamer = new AsyncFileStreamer("/home/alexhultman/v0.15/public");
-
-    // läs in hela sintel-filmen här, testa strömmarna med den sen!
 
     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);
+        // depending on the file type we want to also add mime!
-        res->writeHeader("Content-Type", "text/html;charset=utf-8");
+        asyncFileStreamer->streamFile(res, req->getUrl());
-
-        // 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("<h1>Hallå!</h1>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);
 
     }).listen(3000, [](auto *token) {
         if (token) {

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<SSL> *asyncSocketData = (AsyncSocketData<SSL> *) 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;

src/HttpResponse.h

@@ -39,6 +39,9 @@ private:
     }
 
 public:
+
+    using Super::close;
+
     /* Write the HTTP status */
     HttpResponse *writeStatus(std::string_view status) {
         HttpResponseData<SSL> *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();
     }