diff --git a/misc/15.pro b/misc/15.pro index ab6ef9e..81d4f0d 100644 --- a/misc/15.pro +++ b/misc/15.pro @@ -5,32 +5,32 @@ CONFIG -= qt SOURCES += \ main.cpp \ - uSockets/src/eventing/epoll.c \ - uSockets/src/context.c \ - uSockets/src/socket.c \ - uSockets/src/eventing/libuv.c \ - uSockets/src/ssl.c \ - uSockets/src/loop.c + ../uSockets/src/eventing/epoll.c \ + ../uSockets/src/context.c \ + ../uSockets/src/socket.c \ + ../uSockets/src/eventing/libuv.c \ + ../uSockets/src/ssl.c \ + ../uSockets/src/loop.c HEADERS += \ - src/HttpRouter.h \ - src/HttpParser.h \ - src/websocket/libwshandshake.hpp \ - src/websocket/WebSocketProtocol.h \ - src/websocket/WebSocket.h \ - src/websocket/WebSocketApp.h \ - src/HttpContext.h \ - src/HttpContextData.h \ - src/HttpResponseData.h \ - src/HttpResponse.h \ - src/StaticDispatch.h \ - src/LoopData.h \ - src/AsyncSocket.h \ - src/AsyncSocketData.h \ - src/Loop.h \ - src/App.h \ - src/Utilities.h + ../src/HttpRouter.h \ + ../src/HttpParser.h \ + ../src/websocket/libwshandshake.hpp \ + ../src/websocket/WebSocketProtocol.h \ + ../src/websocket/WebSocket.h \ + ../src/websocket/WebSocketApp.h \ + ../src/HttpContext.h \ + ../src/HttpContextData.h \ + ../src/HttpResponseData.h \ + ../src/HttpResponse.h \ + ../src/StaticDispatch.h \ + ../src/LoopData.h \ + ../src/AsyncSocket.h \ + ../src/AsyncSocketData.h \ + ../src/Loop.h \ + ../src/App.h \ + ../src/Utilities.h -INCLUDEPATH += uSockets/src src +INCLUDEPATH += ../uSockets/src ../src #QMAKE_CXXFLAGS += -fsanitize=address LIBS += -pthread -lssl -lcrypto diff --git a/misc/main.cpp b/misc/main.cpp index d1f91f8..a1672ab 100644 --- a/misc/main.cpp +++ b/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 *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->writeHeader("Content-Type", "text/html;charset=utf-8"); + res->writeStatus(uWS::HTTP_200_OK); + 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; - res->tryEnd(std::string_view(sintelMovie).substr(offset)); + return res->tryEnd(std::string_view(sintelMovie).substr(offset)); + })->onAborted([]() { + // was it really aborted? + std::cout << "Streaming was aborted" << std::endl; }); }*/ - // tryWrite / tryEnd - - // vad om man skriver tryEnd("

Hallå!

Din user-agent är: ", totalLength) - - //res->write("

Hallå!

Din user-agent är: "); - //res->end(req->getHeader("user-agent")); + // 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) { - // I guess it should return wherer or not it wants to be called again? - auto streamIt = [res](int offset) { - 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); - } + // this should be enough, we can even delay this thing! + streamFile(res); }).listen(3000, [](auto *token) { if (token) { diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index 8b4db93..6c38b88 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -57,34 +57,37 @@ protected: AsyncSocketData *asyncSocketData = (AsyncSocketData *) getExt(); - /* Do nothing for a null sized chunk */ - 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 */ + /* We are limited 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 - - // drain here - 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; + /* Is this a merge drain or not? */ + if (length) { + // merge drain may happen if the developer continues to write data despite having buffered up content + // 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; } diff --git a/src/HttpContext.h b/src/HttpContext.h index 554973e..bf77c1e 100644 --- a/src/HttpContext.h +++ b/src/HttpContext.h @@ -60,15 +60,15 @@ private: /* Get socket ext */ HttpResponseData *httpResponseData = (HttpResponseData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s); - /* Signal broken out stream */ - if (httpResponseData->outStream) { - httpResponseData->outStream(-1); + std::cout << "Close event!" << std::endl; + + /* Signal broken HTTP request */ + if (httpResponseData->onAborted) { + httpResponseData->onAborted(); } - /* Signal broken in stream */ - if (httpResponseData->inStream) { - httpResponseData->inStream(std::string_view(nullptr, 0)); - } + // we might want to also signal the read stream? + // smash onAborted together with read? /* Destruct socket ext */ httpResponseData->~HttpResponseData(); @@ -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); - } else { - // lets drain here - std::cout << "LEts drain!" << std::endl; + /* We expect the developer to return whether or not write was successful (true) */ + bool success = httpResponseData->onWritable(httpResponseData->offset); - // 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 *asyncSocket = (AsyncSocket *) s; return asyncSocket->close(); diff --git a/src/HttpResponse.h b/src/HttpResponse.h index 2409394..a2b2ad2 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -156,7 +156,7 @@ public: if (!(httpResponseData->state & HttpResponseData::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 *httpResponseData = getHttpResponseData(); + + return httpResponseData->offset; } /* Attach handler for writable HTTP response */ - HttpResponse *onWritable(std::function handler) { + HttpResponse *onWritable(std::function handler) { HttpResponseData *httpResponseData = getHttpResponseData(); httpResponseData->onWritable = handler; diff --git a/src/HttpResponseData.h b/src/HttpResponseData.h index 949e675..fb8a5fa 100644 --- a/src/HttpResponseData.h +++ b/src/HttpResponseData.h @@ -24,7 +24,8 @@ private: }; /* Per socket event handlers */ - std::function onWritable; + std::function onWritable; + std::function onAborted; //std::function onData; std::function inStream;