Fix up async file streaming

misc/15.pro

@@ -5,32 +5,32 @@ CONFIG -= qt
 
 SOURCES += \
     main.cpp \
-    uSockets/src/eventing/epoll.c \
+    ../uSockets/src/eventing/epoll.c \
-    uSockets/src/context.c \
+    ../uSockets/src/context.c \
-    uSockets/src/socket.c \
+    ../uSockets/src/socket.c \
-    uSockets/src/eventing/libuv.c \
+    ../uSockets/src/eventing/libuv.c \
-    uSockets/src/ssl.c \
+    ../uSockets/src/ssl.c \
-    uSockets/src/loop.c
+    ../uSockets/src/loop.c
 
 HEADERS += \
-    src/HttpRouter.h \
+    ../src/HttpRouter.h \
-    src/HttpParser.h \
+    ../src/HttpParser.h \
-    src/websocket/libwshandshake.hpp \
+    ../src/websocket/libwshandshake.hpp \
-    src/websocket/WebSocketProtocol.h \
+    ../src/websocket/WebSocketProtocol.h \
-    src/websocket/WebSocket.h \
+    ../src/websocket/WebSocket.h \
-    src/websocket/WebSocketApp.h \
+    ../src/websocket/WebSocketApp.h \
-    src/HttpContext.h \
+    ../src/HttpContext.h \
-    src/HttpContextData.h \
+    ../src/HttpContextData.h \
-    src/HttpResponseData.h \
+    ../src/HttpResponseData.h \
-    src/HttpResponse.h \
+    ../src/HttpResponse.h \
-    src/StaticDispatch.h \
+    ../src/StaticDispatch.h \
-    src/LoopData.h \
+    ../src/LoopData.h \
-    src/AsyncSocket.h \
+    ../src/AsyncSocket.h \
-    src/AsyncSocketData.h \
+    ../src/AsyncSocketData.h \
-    src/Loop.h \
+    ../src/Loop.h \
-    src/App.h \
+    ../src/App.h \
-    src/Utilities.h
+    ../src/Utilities.h
 
-INCLUDEPATH += uSockets/src src
+INCLUDEPATH += ../uSockets/src ../src
 #QMAKE_CXXFLAGS += -fsanitize=address
 LIBS += -pthread -lssl -lcrypto

misc/main.cpp

@@ -1,22 +1,56 @@
 #include "App.h"
 
-#include "examples/helpers/AsyncFileReader.h"
+#include "../examples/helpers/AsyncFileReader.h"
 
 AsyncFileReader asyncFileReader("/home/alexhultman/sintel_small.mp4");
 
-
+/*inline std::string slurp(const std::string &path) {
-inline std::string slurp(const std::string &path) {
   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::string sintelMovie = slurp("/home/alexhultman/sintel_small.mp4");*/
+
+void streamFile(uWS::HttpResponse<false> *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;
+        });
+    }
+}
 
 int main(int argc, char **argv) {
 
-    std::cout << "Sintel movie is " << sintelMovie.length() << " bytes" << std::endl;
+    //std::cout << "Sintel movie is " << sintelMovie.length() << " bytes" << std::endl;
 
     // läs in hela sintel-filmen här, testa strömmarna med den sen!
 
@@ -25,63 +59,32 @@ int main(int argc, char **argv) {
         .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("/", [](auto *res, auto *req) {
 
-        //res->writeStatus(uWS::HTTP_200_OK);
+        res->writeStatus(uWS::HTTP_200_OK);
-        //res->writeHeader("Content-Type", "text/html;charset=utf-8");
+        res->writeHeader("Content-Type", "text/html;charset=utf-8");
 
-        std::cout << "Endar nu" << std::endl;
         // buffer it up and end by draining it
-        res->end(sintelMovie);
+        //res->end(sintelMovie);
 
         // stream it here
         /*if (!res->tryEnd(sintelMovie)) {
             res->onWritable([res](int offset) {
-                std::cout << "Streaming data at offset " << offset << std::endl;
+                return res->tryEnd(std::string_view(sintelMovie).substr(offset));
-                res->tryEnd(std::string_view(sintelMovie).substr(offset));
+            })->onAborted([]() {
+                // was it really aborted?
+                std::cout << "Streaming was aborted" << std::endl;
             });
         }*/
 
-        // tryWrite / tryEnd
+        // end can be called two times if the total length is longer then given
-
+        res->write("<h1>Hallå!</h1>Din user-agent är: ");
-        // vad om man skriver tryEnd("<h1>Hallå!</h1>Din user-agent är: ", totalLength)
+        res->end(req->getHeader("user-agent"));
-
-        //res->write("<h1>Hallå!</h1>Din user-agent är: ");
-        //res->end(req->getHeader("user-agent"));
 
     }).get("/async/sintel.mkv", [](auto *res, auto *req) {
 
-        // I guess it should return wherer or not it wants to be called again?
+        // this should be enough, we can even delay this thing!
-        auto streamIt = [res](int offset) {
+        streamFile(res);
-            while(true) {
-                /* Peek from cache */
-                std::string_view chunk = asyncFileReader.peek(offset);
-                if (chunk.length()) {
-                    /* We had parts of this file cached already */
-                    res->tryEnd(chunk, asyncFileReader.getFileSize());
-                } else {
-                    /* We had nothing readily available right now, request async chunk and pause the stream until we have */
-                    asyncFileReader.request(offset, [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->tryEnd(chunk, asyncFileReader.getFileSize());
-                        }
-                    });
-                }
-                return true;
-            }
-        };
-
-        // basically do this
-        if (!streamIt(0)) {
-            // what should it return? will onWritable be called again depending on what it returns?
-            res->onWritable(streamIt);
-        }
 
     }).listen(3000, [](auto *token) {
         if (token) {

src/AsyncSocket.h

@@ -57,34 +57,37 @@ protected:
 
         AsyncSocketData<SSL> *asyncSocketData = (AsyncSocketData<SSL> *) getExt();
 
-        /* Do nothing for a null sized chunk */
+        /* We are limited if we have a per-socket buffer */
-        if (length == 0 && !asyncSocketData->buffer.length()) {
-            //std::cout << "Write returned: 0" << std::endl;
-            return 0;
-        }
-
-        /* Do not write anything if we have a per-socket buffer */
         if (asyncSocketData->buffer.length()) {
+
+            std::cout << "WHAT THE FUCK WE HAVE BUFFER!" << std::endl;
+
+            // probably want to swap the hierarchy here to: if buffer, if length, if optionally
+
+            /* We only try and drain if we are in optional mode */
             if (optionally) {
 
 
-                // we have buffer and we are optionally, if drain then drain else quit
+                /* Is this a merge drain or not? */
-
+                if (length) {
-                // drain here
+                    // merge drain may happen if the developer continues to write data despite having buffered up content
-                std::cout << "Drain path" << std::endl;
-
-                // will just end up in a loop!
-                int written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, asyncSocketData->buffer.data(), asyncSocketData->buffer.length(), nextLength != 0);//write(asyncSocketData->buffer.data(), asyncSocketData->buffer.length(), optionally, 0, true);
-
-                // removeBuffer
-                asyncSocketData->buffer = asyncSocketData->buffer.substr(written);
-
-                // should we really return this here? should be 0 as we took 0 new data!
-                return 0;
 
 
+                    // for now we do not support this yet
+                    return 0;
 
+                } else {
+                    /* Write off as much as we can */
+                    int written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, asyncSocketData->buffer.data(), asyncSocketData->buffer.length(), nextLength != 0);
+
+                    /* Update buffering (should probably have different allocation strategies here) */
+                    asyncSocketData->buffer = asyncSocketData->buffer.substr(written);
+
+                    /* We consumed no new data */
+                    return 0;
+                }
             } else {
+                /* Currently there is no drainage for non-optional writes */
                 std::cout << "Buffering at top of write (really bad)!" << std::endl;
 
                 /* At least we can reserve room for next chunk if we know it up front */
@@ -99,6 +102,12 @@ protected:
             }
         }
 
+        /* Do nothing for a null sized chunk */
+        if (!length) {
+            std::cout << "Trying to write 0 length!" << std::endl;
+            return 0;
+        }
+
         if (loopData->corked) {
             /* We are corked */
             if (LoopData::CORK_BUFFER_SIZE - loopData->corkOffset >= length) {
@@ -130,7 +139,7 @@ protected:
             if (written < length) {
                 /* If the write was optional then just bail out */
                 if (optionally) {
-                    //std::cout << "Write returned: " << written << std::endl;
+                    std::cout << "Write returned: " << written << std::endl;
                     return written;
                 }
 
@@ -147,7 +156,7 @@ protected:
             }
         }
 
-        //std::cout << "Write returned: " << length << std::endl;
+        std::cout << "Write returned: " << length << std::endl;
         return length;
     }
 

src/HttpContext.h

@@ -60,15 +60,15 @@ private:
             /* Get socket ext */
             HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s);
 
-            /* Signal broken out stream */
+            std::cout << "Close event!" << std::endl;
-            if (httpResponseData->outStream) {
+
-                httpResponseData->outStream(-1);
+            /* Signal broken HTTP request */
+            if (httpResponseData->onAborted) {
+                httpResponseData->onAborted();
             }
 
-            /* Signal broken in stream */
+            // we might want to also signal the read stream?
-            if (httpResponseData->inStream) {
+            // smash onAborted together with read?
-                httpResponseData->inStream(std::string_view(nullptr, 0));
-            }
 
             /* Destruct socket ext */
             httpResponseData->~HttpResponseData<SSL>();
@@ -132,11 +132,10 @@ private:
             return s;
         });
 
-        /* Handle HTTP write out */
+        /* Handle HTTP write out (note: SSL_read may trigger this spuriously, the app need to handle spurious calls) */
         static_dispatch(us_ssl_socket_context_on_writable, us_socket_context_on_writable)(getSocketContext(), [](auto *s) {
 
-            std::cout << "Writable event!" << std::endl;
+            std::cout << "HttpContext::onWritable event fired!" << std::endl;
-
 
             /* Writing data should reset the timeout */
             static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S);
@@ -146,33 +145,24 @@ private:
 
             // if this, then it means it finished with no issues so we need to empty any buffers?
             if (httpResponseData->onWritable) {
-                httpResponseData->onWritable(httpResponseData->offset);
+                /* We expect the developer to return whether or not write was successful (true) */
-            } else {
+                bool success = httpResponseData->onWritable(httpResponseData->offset);
-                // lets drain here
-                std::cout << "LEts drain!" << std::endl;
 
-                // mergeDrain
+                // on writable should return whether it wants more data or not
+                // but we don't need to know that here? we cannot drain because a sucessful write should mean there is no buffer to drain
+            } else {
+                /* This is used to drain any buffers we might have */
                 asyncSocket->write(nullptr, 0, true, 0);
             }
 
-
-
-            // bascially just empty the buffer and if successful also call onWritable (bad strategy!)
-            //asyncSocket->mergeDrain();
-
-            // what we want is to immediately call onWritable and have AsyncSocket::write calls always try and empty any buffers at the same time?
-            // AsyncSocket::write can take boolean drain = true to know it should try and drain the buffers according to whatever strategy
-            // mergeDrain is basically AsyncSocket::write with boolean drain = true!
-
-
-            // on writable should return whether it wants more data or not?
-
             return s;
         });
 
         /* Handle FIN, HTTP does not support half-closed sockets, so simply close */
         static_dispatch(us_ssl_socket_context_on_end, us_socket_context_on_end)(getSocketContext(), [](auto *s) {
 
+            std::cout << "FIN sent" << std::endl;
+
             /* We do not care for half closed sockets */
             AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
             return asyncSocket->close();

src/HttpResponse.h

@@ -156,7 +156,7 @@ public:
             if (!(httpResponseData->state & HttpResponseData<SSL>::HTTP_END_CALLED)) {
                 /* We have a known send size */
                 Super::write("Content-Length: ", 16);
-                writeUnsigned(data.length());
+                writeUnsigned(/*data.length()*/totalSize);
                 Super::write("\r\n\r\n", 4);
 
                 /* Mark end called */
@@ -164,14 +164,27 @@ public:
             }
 
             /* Write as much as possible without causing backpressure */
-            httpResponseData->offset += Super::write(data.data(), data.length(), true);
+            int written = Super::write(data.data(), data.length(), true);
+
+            httpResponseData->offset += written;
+            std::cout << "Offset is now: " << httpResponseData->offset << std::endl;
+
+            return written == data.length();
         }
 
-        return httpResponseData->offset == totalSize;
+        // this path is completely wrong!
+        std::cout << "tryEnd returning " << (httpResponseData->offset == /*totalSize*/ data.length()) << std::endl;
+        return httpResponseData->offset == /*totalSize*/ data.length();
+    }
+
+    int getWriteOffset() {
+        HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
+
+        return httpResponseData->offset;
     }
 
     /* Attach handler for writable HTTP response */
-    HttpResponse *onWritable(std::function<void(int)> handler) {
+    HttpResponse *onWritable(std::function<bool(int)> handler) {
         HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
 
         httpResponseData->onWritable = handler;

src/HttpResponseData.h

@@ -24,7 +24,8 @@ private:
     };
 
     /* Per socket event handlers */
-    std::function<void(int)> onWritable;
+    std::function<bool(int)> onWritable;
+    std::function<void()> onAborted;
     //std::function<void()> onData;
 
     std::function<void(std::string_view)> inStream;