diff --git a/15.pro b/15.pro index 283f0e6..ab6ef9e 100644 --- a/15.pro +++ b/15.pro @@ -28,8 +28,9 @@ HEADERS += \ src/AsyncSocket.h \ src/AsyncSocketData.h \ src/Loop.h \ - src/App.h + src/App.h \ + src/Utilities.h INCLUDEPATH += uSockets/src src -QMAKE_CXXFLAGS += -fsanitize=address -LIBS += -lasan -lssl -lcrypto +#QMAKE_CXXFLAGS += -fsanitize=address +LIBS += -pthread -lssl -lcrypto diff --git a/main.cpp b/main.cpp index 852f7ef..d1f91f8 100644 --- a/main.cpp +++ b/main.cpp @@ -2,45 +2,85 @@ #include "examples/helpers/AsyncFileReader.h" -AsyncFileReader asyncFileReader("/home/alexhultman/v0.15/sintel.mkv"); +AsyncFileReader asyncFileReader("/home/alexhultman/sintel_small.mp4"); + + +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"); int main(int argc, char **argv) { + std::cout << "Sintel movie is " << sintelMovie.length() << " bytes" << std::endl; + + // 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("/*", [](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"); - res->write("

Hallå!

Din user-agent är: "); - res->end(req->getHeader("user-agent")); + std::cout << "Endar nu" << std::endl; + // buffer it up and end by draining it + 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)); + }); + }*/ + + // 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")); }).get("/async/sintel.mkv", [](auto *res, auto *req) { - // asyncFileReader.getFileSize() - - /* Peek from cache */ - std::string_view chunk = asyncFileReader.peek(/*offset*/ 0); - if (chunk.length()) { - /* We had parts of this file cached already */ - res->write(chunk); - } else { - /* We had nothing readily available right now, request async chunk and pause the stream until we have */ - asyncFileReader.request(/*offset*/ 0, [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! + // 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 finally got the data, resume stream with this chunk */ - res->write(chunk); + /* 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) { diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index 5cbd3d6..07f4ebf 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -11,6 +11,7 @@ namespace uWS { template struct AsyncSocket : StaticDispatch { + template friend struct HttpContext; protected: using SOCKET_TYPE = typename StaticDispatch::SOCKET_TYPE; using StaticDispatch::static_dispatch; @@ -23,39 +24,68 @@ protected: } } -public: void *getExt() { return static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this); } + void timeout(unsigned int seconds) { + static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) this, seconds); + } + + void shutdown() { + static_dispatch(us_ssl_socket_shutdown, us_socket_shutdown)((SOCKET_TYPE *) this); + } + + SOCKET_TYPE *close() { + return static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this); + } + /* Cork this socket. Only one socket may ever be corked per-loop at any given time */ void cork() { + std::cout << "Cork called" << std::endl; + LoopData *loopData = getLoopData(); loopData->corked = true; } - /* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer */ + /* Write in three levels of prioritization: cork-buffer, syscall, socket-buffer. Always drain if possible. */ // todo: consider supporting nextLength = UNKNOWN as -1 (more but unknown size) 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 *asyncSocketData = (AsyncSocketData *) getExt(); /* Do nothing for a null sized chunk */ - if (length == 0) { + if (length == 0 && !asyncSocketData->buffer.length()) { //std::cout << "Write returned: 0" << std::endl; return 0; } - AsyncSocketData *asyncSocketData = (AsyncSocketData *) getExt(); - /* Do not write anything if we have a per-socket buffer */ if (asyncSocketData->buffer.length()) { if (optionally) { - //std::cout << "Write returned: 0" << std::endl; + + + // 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; + + + } else { - std::cout << "Buffering at top of write!" << std::endl; + 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 */ if (nextLength) { @@ -104,7 +134,7 @@ public: return written; } - std::cout << "Buffering at bottom of write!" << std::endl; + std::cout << "Buffering at bottom of write (okay)!" << std::endl; /* Fall back to worst possible case (should be very rare for HTTP) */ /* At least we can reserve room for next chunk if we know it up front */ @@ -124,6 +154,9 @@ public: /* Uncork this socket and flush or buffer any corked and/or passed data. It is essential to remember doing this. */ /* 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; + LoopData *loopData = getLoopData(); if (loopData->corked) { @@ -137,48 +170,12 @@ public: /* 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; } return 0; } - - /* Check if this socket has buffered data */ - bool hasBuffer() { - return ((AsyncSocketData *) getExt())->buffer.length() > 0; - } - - /* Drain any socket-buffer while also optionally sending a chunk */ - int mergeDrain(std::string_view optionalChunk = {}) { - - // strategy: if we have two parts and both will fit in cork buffer then cork them and recursively send them off - - // write any per-socket buffer and optionally more - - AsyncSocketData *asyncSocketData = (AsyncSocketData *) getExt(); - - // not handled yet - if (asyncSocketData->buffer.length()) { - std::cout << "ERROR! has socket buffer!" << std::endl; - exit(0); - } - - - /* Write the optional part */ - return write(optionalChunk.data(), optionalChunk.length(), true, 0); - } - - /* These should not be public to the user! */ - void timeout(unsigned int seconds) { - static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) this, seconds); - } - - void shutdown() { - static_dispatch(us_ssl_socket_shutdown, us_socket_shutdown)((SOCKET_TYPE *) this); - } - - SOCKET_TYPE *close() { - return static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this); - } }; } diff --git a/src/HttpContext.h b/src/HttpContext.h index e0fc479..554973e 100644 --- a/src/HttpContext.h +++ b/src/HttpContext.h @@ -135,60 +135,38 @@ 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; + 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(); - if (httpResponseData->state & HttpResponseData::HTTP_PAUSED_STREAM_OUT) { - return s; - } /* Writing data should reset the timeout */ static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S); - /* Are we already ended and just waiting for a drain / shutdown? */ - if (httpResponseData->state & HttpResponseData::HTTP_ENDED_STREAM_OUT) { + AsyncSocket *asyncSocket = (AsyncSocket *) s; + HttpResponseData *httpResponseData = (HttpResponseData *) asyncSocket->getExt(); - /* Try and send everything buffered up */ - asyncSocket->mergeDrain(); - - /* If we succeed with drainage we can finally shut down */ - if (!asyncSocket->hasBuffer()) { - asyncSocket->shutdown(); - } - - /* Nothing here for us */ - return s; - } - - if (httpResponseData->outStream) { - /* Regular path, request more data */ - - // todo: share this path with HttpResponse::write (it is exatly the same logic!) - while (true) { - auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset); - - if (chunk.length() == 0) { - //std::cout << "onwritable paused!" << std::endl; - httpResponseData->state |= HttpResponseData::HTTP_PAUSED_STREAM_OUT; - break; - } - - 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)! + // 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 { - /* We can come here if we only have socket buffers to drain yet no attached stream */ - asyncSocket->mergeDrain(); + // lets drain here + std::cout << "LEts drain!" << std::endl; + + // mergeDrain + 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; }); diff --git a/src/HttpResponse.h b/src/HttpResponse.h index 8c81712..2409394 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -5,6 +5,7 @@ #include "AsyncSocket.h" #include "HttpResponseData.h" +#include "Utilities.h" namespace uWS { @@ -13,62 +14,28 @@ const char *HTTP_200_OK = "200 OK"; template struct HttpResponse : public AsyncSocket { + typedef AsyncSocket Super; private: HttpResponseData *getHttpResponseData() { - return (HttpResponseData *) AsyncSocket::getExt(); - } - - int u32toaHex(uint32_t value, char *dst) { - char palette[] = "0123456789abcdef"; - char temp[10]; - char *p = temp; - do { - *p++ = palette[value % 16]; - value /= 16; - } while (value > 0); - - int ret = p - temp; - - do { - *dst++ = *--p; - } while (p != temp); - - return ret; + return (HttpResponseData *) Super::getExt(); } /* Write an unsigned 32-bit integer in hex */ void writeUnsignedHex(unsigned int value) { char buf[10]; - int length = u32toaHex(value, buf); + int length = utils::u32toaHex(value, buf); /* For now we do this copy */ - AsyncSocket::write(buf, length); - } - - int u32toa(uint32_t value, char *dst) { - char temp[10]; - char *p = temp; - do { - *p++ = (char) (value % 10) + '0'; - value /= 10; - } while (value > 0); - - int ret = p - temp; - - do { - *dst++ = *--p; - } while (p != temp); - - return ret; + Super::write(buf, length); } /* Write an unsigned 32-bit integer */ void writeUnsigned(unsigned int value) { char buf[10]; - int length = u32toa(value, buf); + int length = utils::u32toa(value, buf); /* For now we do this copy */ - AsyncSocket::write(buf, length); + Super::write(buf, length); } public: @@ -77,59 +44,61 @@ public: HttpResponseData *httpResponseData = getHttpResponseData(); /* Do not allow writing more than one status */ - if (httpResponseData->state & HttpResponseData::HTTP_STATUS_SENT) { + if (httpResponseData->state & HttpResponseData::HTTP_STATUS_CALLED) { return this; } /* Update status */ - httpResponseData->state |= HttpResponseData::HTTP_STATUS_SENT; + httpResponseData->state |= HttpResponseData::HTTP_STATUS_CALLED; - AsyncSocket::write("HTTP/1.1 ", 9); - AsyncSocket::write(status.data(), status.length()); - AsyncSocket::write("\r\n", 2); + Super::write("HTTP/1.1 ", 9); + Super::write(status.data(), status.length()); + Super::write("\r\n", 2); return this; } /* Write an HTTP header with string value */ HttpResponse *writeHeader(std::string_view key, std::string_view value) { - AsyncSocket::write(key.data(), key.length()); - AsyncSocket::write(": ", 2); - AsyncSocket::write(value.data(), value.length()); - AsyncSocket::write("\r\n", 2); + Super::write(key.data(), key.length()); + Super::write(": ", 2); + Super::write(value.data(), value.length()); + Super::write("\r\n", 2); 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); + Super::write(key.data(), key.length()); + Super::write(": ", 2); writeUnsigned(value); - AsyncSocket::write("\r\n", 2); + Super::write("\r\n", 2); return this; } /* End the response with an optional data chunk */ void end(std::string_view data = {}) { + writeStatus(HTTP_200_OK); + HttpResponseData *httpResponseData = getHttpResponseData(); if (httpResponseData->state & HttpResponseData::HTTP_WRITE_CALLED) { /* Do not allow sending 0 chunk here */ if (data.length()) { - AsyncSocket::write("\r\n", 2); + Super::write("\r\n", 2); writeUnsignedHex(data.length()); - AsyncSocket::write("\r\n", 2); - AsyncSocket::write(data.data(), data.length()); + Super::write("\r\n", 2); + Super::write(data.data(), data.length()); } /* Terminating 0 chunk */ - AsyncSocket::write("\r\n0\r\n\r\n", 7); + Super::write("\r\n0\r\n\r\n", 7); } else { /* We have a known send size */ - AsyncSocket::write("Content-Length: ", 16); + Super::write("Content-Length: ", 16); writeUnsigned(data.length()); - AsyncSocket::write("\r\n\r\n", 4); + Super::write("\r\n\r\n", 4); - AsyncSocket::write(data.data(), data.length()); + Super::write(data.data(), data.length()); } } @@ -147,94 +116,77 @@ public: httpResponseData->state |= HttpResponseData::HTTP_WRITE_CALLED; } - AsyncSocket::write("\r\n", 2); + Super::write("\r\n", 2); writeUnsignedHex(data.length()); - AsyncSocket::write("\r\n", 2); - AsyncSocket::write(data.data(), data.length()); + Super::write("\r\n", 2); + Super::write(data.data(), data.length()); // are we corked still? return true; } - // we really want tryEnd(data) integer to try and stream something with known size - - // write/tryWrite called first should enter into chunked? - - // tryWrite(char *, length) int - - // write(char *, length) bool - - - - /* Attach an output stream function. Chunks may be read more than once. Negative offset mean broken stream */ - void writeOldRemoveMe(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 */ + // todo: share this code in a function + bool tryEnd(std::string_view data, int totalSize = 0) { + /* Write status if not already done */ writeStatus(HTTP_200_OK); - /* Update status */ - httpResponseData->state |= HttpResponseData::HTTP_WRITE_CALLED; - if (length) { - httpResponseData->state |= HttpResponseData::HTTP_KNOWN_STREAM_OUT_SIZE; - - /* Rely on FIN to signal end if we do not pass any length */ - AsyncSocket::write("Content-Length: ", 16); - writeUnsigned(length); - AsyncSocket::write("\r\n", 2); + /* If no total size given then assume this chunk is everything */ + if (!totalSize) { + totalSize = data.length(); } - /* HTTP body separator */ - AsyncSocket::write("\r\n", 2); + HttpResponseData *httpResponseData = getHttpResponseData(); - for (int offset = 0; length == 0 || offset < length; ) { - /* Pull a chunk from stream */ - auto [msg_more, chunk] = cb(offset); + if (httpResponseData->state & HttpResponseData::HTTP_WRITE_CALLED) { + /* Do not allow sending 0 chunk here */ + if (data.length()) { + Super::write("\r\n", 2); + writeUnsignedHex(data.length()); + Super::write("\r\n", 2); - /* Handle PAUSE and FIN */ - if (chunk.length() == 0) { - /* FIN */ - if (chunk.data()) { - /* Try flush and shut down */ - AsyncSocket::uncork(); - if (!AsyncSocket::hasBuffer()) { - /* Uncork finished with no buffered data */ - AsyncSocket::shutdown(); - } else { - /* Let it shut down when drained */ - httpResponseData->state |= HttpResponseData::HTTP_ENDED_STREAM_OUT; - } - } else { - /* Disable timeout and mark this stream as paused (important to silence spurious onWritable events) */ - AsyncSocket::timeout(0); - httpResponseData->state |= HttpResponseData::HTTP_PAUSED_STREAM_OUT; - - // forgot about this path, if we sent things off and then ended up pausing! - httpResponseData->offset = offset; - httpResponseData->outStream = cb; - } - return; + // should be optional + Super::write(data.data(), data.length()); } - /* Send off the chunk */ - int written = AsyncSocket::write(chunk.data(), chunk.length(), true); + /* Terminating 0 chunk */ + Super::write("\r\n0\r\n\r\n", 7); + } else { + // if not already ended! should we call tryWrite after this? we need an extra flag! end called! - /* If we failed to send everything, exit */ - if (written < chunk.length()) { - httpResponseData->offset = offset + written; - httpResponseData->outStream = cb; - return; + if (!(httpResponseData->state & HttpResponseData::HTTP_END_CALLED)) { + /* We have a known send size */ + Super::write("Content-Length: ", 16); + writeUnsigned(data.length()); + Super::write("\r\n\r\n", 4); + + /* Mark end called */ + httpResponseData->state |= HttpResponseData::HTTP_END_CALLED; } - offset += written; + /* Write as much as possible without causing backpressure */ + httpResponseData->offset += Super::write(data.data(), data.length(), true); } + + return httpResponseData->offset == totalSize; } + /* Attach handler for writable HTTP response */ + HttpResponse *onWritable(std::function handler) { + HttpResponseData *httpResponseData = getHttpResponseData(); + + httpResponseData->onWritable = handler; + return this; + } + + /* Attach handler for aborted HTTP request */ + HttpResponse *onAborted(std::function handler) { + HttpResponseData *httpResponseData = getHttpResponseData(); + + httpResponseData->onAborted = handler; + return this; + } + + // onData(chunk, remaining == -1 or 0 or actual remaining)? /* Attach a read handler for data sent. Will be called with a chunk of size 0 when FIN */ void read(std::function handler) { HttpResponseData *data = getHttpResponseData(); diff --git a/src/HttpResponseData.h b/src/HttpResponseData.h index 1a4be6a..949e675 100644 --- a/src/HttpResponseData.h +++ b/src/HttpResponseData.h @@ -16,15 +16,15 @@ struct HttpResponseData : HttpParser, AsyncSocketData { private: /* Bits of status */ enum { - HTTP_STATUS_SENT = 1, // used + HTTP_STATUS_CALLED = 1, // used HTTP_WRITE_CALLED = 2, // used - HTTP_KNOWN_STREAM_OUT_SIZE = 4, // not used + HTTP_END_CALLED = 4, // used HTTP_PAUSED_STREAM_OUT = 8, // not used HTTP_ENDED_STREAM_OUT = 16 // not used }; /* Per socket event handlers */ - std::function onWritable; + std::function onWritable; //std::function onData; std::function inStream; diff --git a/src/Utilities.h b/src/Utilities.h new file mode 100644 index 0000000..0dd0390 --- /dev/null +++ b/src/Utilities.h @@ -0,0 +1,49 @@ +#ifndef UTILITIES_H +#define UTILITIES_H + +/* Various common utilities */ + +#include + +namespace uWS { +namespace utils { + +int u32toaHex(uint32_t value, char *dst) { + char palette[] = "0123456789abcdef"; + char temp[10]; + char *p = temp; + do { + *p++ = palette[value % 16]; + value /= 16; + } while (value > 0); + + int ret = p - temp; + + do { + *dst++ = *--p; + } while (p != temp); + + return ret; +} + +int u32toa(uint32_t value, char *dst) { + char temp[10]; + char *p = temp; + do { + *p++ = (char) (value % 10) + '0'; + value /= 10; + } while (value > 0); + + int ret = p - temp; + + do { + *dst++ = *--p; + } while (p != temp); + + return ret; +} + +} +} + +#endif // UTILITIES_H