From 932b0b8944a32ad9b41eab3b6644fa1adb76e90a Mon Sep 17 00:00:00 2001 From: Alex Hultman Date: Thu, 20 Sep 2018 21:51:54 +0200 Subject: [PATCH] Begin work on new streams API --- main.cpp | 33 ++++++++- src/HttpContext.h | 3 +- src/HttpResponse.h | 154 +++++++++++++++++++++++++++++++++++++---- src/HttpResponseData.h | 16 ++++- 4 files changed, 191 insertions(+), 15 deletions(-) diff --git a/main.cpp b/main.cpp index 2f44351..db1403d 100644 --- a/main.cpp +++ b/main.cpp @@ -35,14 +35,45 @@ std::string_view getFile(std::string_view file) { } } +#include + +std::set *> delayedResponses; + 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; + + // resume should take a string_view! + x->resume(); + } + + delayedResponses.clear(); + + }, 1000, 1000); + uWS::SSLApp({ .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("/hello", [](auto *res, auto *req) { + }).get("/", [](auto *res, auto *req) { res->writeStatus(uWS::HTTP_200_OK)->write("Hello world!"); + }).get("/delayed", [](auto *res, auto *req) { + /* This route streams back chunks of data in delayed fashion */ + res->writeStatus(uWS::HTTP_200_OK)->write([res](int offset) { + + std::cout << "Delaying stream now" << std::endl; + delayedResponses.insert(res); + + //asyncFetchData(res) + return uWS::HTTP_STREAM_PAUSE; + + }, 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))); }).listen(3000, [](auto *token) { diff --git a/src/HttpContext.h b/src/HttpContext.h index 3fed80b..45d195c 100644 --- a/src/HttpContext.h +++ b/src/HttpContext.h @@ -86,6 +86,7 @@ private: HttpResponseData *httpResponseData = (HttpResponseData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) s); httpResponseData->offset = 0; + httpResponseData->state = 0; // route it! typename uWS::HttpContextData::UserData userData = { @@ -122,7 +123,7 @@ private: HttpResponseData *httpResponseData = (HttpResponseData *) asyncSocket->getExt(); if (httpResponseData->outStream) { - std::string_view chunk = httpResponseData->outStream(httpResponseData->offset); + auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset); // send, including any buffered up httpResponseData->offset += asyncSocket->mergeDrain(chunk); diff --git a/src/HttpResponse.h b/src/HttpResponse.h index e8e0074..e42df5e 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -11,6 +11,12 @@ namespace uWS { /* Some pre-defined status constants to use with writeStatus */ const char *HTTP_200_OK = "200 OK"; +/* Return this from a stream callback to signal pause */ +const std::pair HTTP_STREAM_PAUSE = {false, std::string_view(nullptr, 0)}; + +/* Return this from a stream callback to signal FIN */ +const std::pair HTTP_STREAM_FIN = {false, std::string_view((const char *) 1, 0)}; + template struct HttpResponse : public AsyncSocket { private: @@ -47,6 +53,16 @@ private: public: /* Write the HTTP status */ HttpResponse *writeStatus(std::string_view status) { + HttpResponseData *httpResponseData = getHttpResponseData(); + + /* Do not allow writing more than one status */ + if (httpResponseData->state & HttpResponseData::HTTP_STATUS_SENT) { + return this; + } + + /* Update status */ + httpResponseData->state |= HttpResponseData::HTTP_STATUS_SENT; + AsyncSocket::write("HTTP/1.1 ", 9); AsyncSocket::write(status.data(), status.length()); AsyncSocket::write("\r\n", 2); @@ -62,24 +78,138 @@ public: return this; } + /* Write an HTTP header with unsigned int value */ + HttpResponse *writeHeader(std::string_view key, unsigned int value) { + AsyncSocket::write(key.data(), key.length()); + AsyncSocket::write(": ", 2); + writeUnsigned(value); + AsyncSocket::write("\r\n", 2); + return this; + } + + /* Resume response streaming as far as possible */ + void resume(std::string_view chunk = {}) { + HttpResponseData *httpResponseData = getHttpResponseData(); + + /* Do nothing if not even paused */ + if (!(httpResponseData->state & HttpResponseData::HTTP_PAUSED_STREAM_OUT)) { + return; + } + + /* Remove paused status */ + httpResponseData->state &= ~HttpResponseData::HTTP_PAUSED_STREAM_OUT; + + int written = AsyncSocket::write(chunk.data(), chunk.length(), true); + + if (written == chunk.length()) { + // pull a new chunk from the callback (basically call onWritable) + } + + // no, basically just write this off and if all written, call streamOut callback + } + /* Attach a write handler for sending data. Length must be specified up front and chunks might be read more than once */ - void write(std::function cb, int length) { - std::string_view chunk = cb(0); - AsyncSocket::write("Content-Length: ", 16); - writeUnsigned(chunk.length()); - AsyncSocket::write("\r\n\r\n", 4); + + // chunk = stream(int offset) is current interface + + // we need more information such as: + + // 1. (RETURN) was the short response of data meant to be msg_more-d? basically: should the stream be called again? needed to allow expose of msg_more to streamer! + + // 2. (RETURN) what data are we returning, or are we returning no data (pause signal) + + // 3. (RETURN) how do we signal FIN for length-less data? return a different string_view with 0 length? HTTP_STREAM_FIN? + + // 4. (INPUT ARGS) how much has been acked of what we send? int offset is an indication? what about indicating ONLY the ack? + + + void write(std::function(int)> cb, int length = 0) { + HttpResponseData *httpResponseData = getHttpResponseData(); + + /* Do not allow write if already called */ + if (httpResponseData->state & HttpResponseData::HTTP_WRITE_CALLED) { + return; + } + + /* Write 200 OK if not already written any status */ + writeStatus(HTTP_200_OK); + + /* Update status */ + httpResponseData->state |= HttpResponseData::HTTP_WRITE_CALLED; + if (length) { + httpResponseData->state |= HttpResponseData::HTTP_KNOWN_STREAM_OUT_SIZE; + } + + // int tail, int head. tail is what has been sent off, head is where we read from. they can differ - does it matter? + + // just do resume() and pause() to throttle onwritable calling + // also add bool isAck to writable stream to be treated as ack signalling (return is never mind) + + // we need resume() to resume sending (very simple to just call the onWritable callback again) + // returning less than the kernel buffer will pause the streaming, so a return of 0 length is a pause + // returning less than but still something should call the callback again! + + // with no length, it makes sense to use chunked transfer? keep alive + + // if we have 0 length we fall back to connection close! + // return uWS::HTTP_STREAM_FIN to signal closure + + + + + // this check should be its own shared function? + auto [msg_more, chunk] = cb(0); + + + if (length) { + /* Rely on FIN to signal end if we do not pass any length */ + AsyncSocket::write("Content-Length: ", 16); + writeUnsigned(chunk.length()); + AsyncSocket::write("\r\n", 2); + } + + /* HTTP body separator */ + AsyncSocket::write("\r\n", 2); + + /* Did the user pause ? */ + if (chunk.length() == 0) { + + // skip sending optional, yet keep refusing to call onWritable while in paused mode (SSL may poll for writable!) + + // we thus need a status: paused to check for before requesting more data (also check for this in resume call!) + + std::cout << "Paused stream!" << std::endl; + return; + } + + /* Write as much as possible, optionally */ if (int written; (written = AsyncSocket::write(chunk.data(), chunk.length(), true)) < length) { std::cout << "HttpResponse::write failed to write everything" << std::endl; - getHttpResponseData()->offset = written; - getHttpResponseData()->outStream = cb; + httpResponseData->offset = written; + httpResponseData->outStream = cb; } } - /* Convenience function for static data (I don't like this one!) */ - void write(std::string_view data) { - write([data](int offset) { - return data.substr(offset); - }, data.length()); + /* Convenience function for static data */ + void write(std::string_view data, std::function cb = nullptr) { + if (cb) { + // todo: think about how the stream will signal done (streams API challenge overall) + write([data](int offset, int length) { + + // if offset == length then we know it is end + + // what if we want to return both fin and data? can't do that + + + + return {false, data.substr(offset)}; + }, data.length()); + } else { + // requires no extra alloc + write([data](int offset) { + return {false, data.substr(offset)}; + }, data.length()); + } } /* Attach a read handler for data sent. Will be called with a chunk of size 0 when FIN */ diff --git a/src/HttpResponseData.h b/src/HttpResponseData.h index a0968be..c3dc3e7 100644 --- a/src/HttpResponseData.h +++ b/src/HttpResponseData.h @@ -11,10 +11,24 @@ namespace uWS { template struct HttpResponseData : HttpParser, AsyncSocketData { + template friend struct HttpResponse; + template friend struct HttpContext; +private: + /* Bits of status */ + enum { + HTTP_STATUS_SENT = 1, + HTTP_WRITE_CALLED = 2, + HTTP_KNOWN_STREAM_OUT_SIZE = 4, + HTTP_PAUSED_STREAM_OUT = 8 + }; + std::function inStream; - std::function outStream; + std::function(int)> outStream; /* Outgoing offset */ int offset = 0; + + /* Current state (content-length sent, status sent, write called, etc */ + int state = 0; }; }