talksik/uWebSockets
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Begin work on AsyncSocket

Browse Source
This commit is contained in:
Alex Hultman
2018-09-11 19:41:32 +02:00
parent baa58c9e90
commit 3af62041a1
11 changed files with 233 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
15.pro
+3 -2
View File
@@ -15,7 +15,6 @@ SOURCES += \
HEADERS += \
src/new_design/HttpRouter.h \
src/Loop.h \
src/http/HttpSocket.h \
src/new_design/HttpParser.h \
src/websocket/libwshandshake.hpp \
src/websocket/WebSocketProtocol.h \
@@ -25,7 +24,9 @@ HEADERS += \
src/new_design/HttpContextData.h \
src/new_design/HttpResponseData.h \
src/new_design/HttpResponse.h \
src/new_design/StaticDispatch.h
src/new_design/StaticDispatch.h \
src/new_design/LoopData.h \
src/new_design/AsyncSocket.h
INCLUDEPATH += uSockets/src src
#QMAKE_CXXFLAGS += -fsanitize=address
design.dia
BIN
View File
Binary file not shown.
main.cpp
+7 -7
View File
@@ -59,18 +59,18 @@ int main(int argc, char **argv) {
// req, res?
httpContext->onGet("/", [](auto *res, auto *req) { // maybe use the terminology of HttpRequest for both?
std::cout << "URL: <" << req->getUrl() << ">" << std::endl;
/*std::cout << "URL: <" << req->getUrl() << ">" << std::endl;
std::cout << "Query: <" << req->getQuery() << ">" << std::endl;
std::cout << "User-Agent: <" << req->getHeader("user-agent") << ">" << std::endl;
*/
// read some data being passed
res->read([](std::string_view chunk) {
/*res->read([](std::string_view chunk) {
std::cout << "Reading some streamed in data:" << chunk << std::endl;
});
});*/
// res->writeHeader()->write();
// req
res->writeStatus(uWS::HTTP_200_OK)->write([](int offset) {
return std::string_view("Hello world!");
}, 12);
});
src/Loop.h
+9 -23
View File
@@ -1,31 +1,16 @@
#ifndef HUB_H
#define HUB_H
#include "libusockets.h"
#include <functional>
#include <new>
#include <string_view>
#include <iostream>
#include <unistd.h>
#include "new_design/LoopData.h"
#include <libusockets.h>
// Loop is a bit different, it holds the loop pointer, not as "this"
namespace uWS {
struct Loop {
us_loop *loop;
static const int CORK_BUFFER_SIZE = 16 * 1024;
static const int MAX_COPY_DISTANCE = 4096;
struct Data {
char *corkBuffer = new char[CORK_BUFFER_SIZE];
int corkOffset = 0;
Data() {
}
} *data;
static void wakeupCb(us_loop *loop) {
}
@@ -38,8 +23,9 @@ struct Loop {
}
Loop() : loop(us_create_loop(1, wakeupCb, preCb, postCb, sizeof(Data))) {
new (data = (Data *) us_loop_ext(loop)) Data();
Loop() : loop(us_create_loop(1, wakeupCb, preCb, postCb, sizeof(LoopData))) {
new (us_loop_ext(loop)) LoopData();
}
void run() {
@@ -47,7 +33,7 @@ struct Loop {
}
~Loop() {
// deconstruct the loopdata
}
};
src/http/HttpSocket.h
-176
View File
@@ -1,176 +0,0 @@
#ifdef HTTP_H
#define HTTP_H
#include "Socket.h"
#include "../new_design/HttpParser.h"
#include <cstring>
#include <algorithm>
#include <string>
template <bool SSL>
struct HttpSocket : Socket<SSL> {
const size_t MAX_FALLBACK_SIZE = 4096;
typedef typename Socket<SSL>::SOCKET_TYPE SOCKET_TYPE;
using Socket<SSL>::static_dispatch;
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;
}
// chunked response will be tricky with this buffering scheme
// if we do not fit, we can always use the header buffer for this (both in and out!)
// put first 8kb chunk in the http buffer, then from there it's the stream's job!
// httpheaders should only have 1 stream in and 1 stream out, but we can have helper wrappers
struct Data {
uWS::HttpParser httpParser;
std::function<void(std::string_view)> inStream;
// out streaming (.end should be a wrapper of this!)
int offset = 0;
std::function<std::string_view(int)> outStream;
};
// only this one should be used!
void writeToCorkBuffer(const char *src, int length) {
uWS::Loop::Data *loopData = (uWS::Loop::Data *) us_loop_ext(us_socket_context_loop(us_socket_get_context((us_socket *) this)));
memcpy(loopData->corkBuffer + loopData->corkOffset, src, length);
loopData->corkOffset += length;
}
// never rely on this one!
int writeToCorkBufferAndReset(const char *src, int length, int contentLength, bool expectMore) {
uWS::Loop::Data *loopData = (uWS::Loop::Data *) us_loop_ext(us_socket_context_loop(us_socket_get_context((us_socket *) this)));
memcpy(loopData->corkBuffer + loopData->corkOffset, "Content-Length: ", 16);
loopData->corkOffset += 16;
loopData->corkOffset += u32toa(contentLength, loopData->corkBuffer + loopData->corkOffset);
memcpy(loopData->corkBuffer + loopData->corkOffset, "\r\n\r\n", 4);
loopData->corkOffset += 4;
memcpy(loopData->corkBuffer + loopData->corkOffset, src, length);
loopData->corkOffset += length;
int written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, loopData->corkBuffer, loopData->corkOffset, expectMore);
loopData->corkOffset = 0;
return written;
}
HttpSocket *writeStatus(std::string_view status) {
writeToCorkBuffer("HTTP/1.1 ", 9);
writeToCorkBuffer(status.data(), status.length());
writeToCorkBuffer("\r\n", 2);
return this;
}
HttpSocket *writeHeader(std::string_view key, std::string_view value) {
writeToCorkBuffer(key.data(), key.length());
writeToCorkBuffer(": ", 2);
writeToCorkBuffer(value.data(), value.length());
writeToCorkBuffer("\r\n", 2);
return this;
}
// this should not be anything other than a simple convenience wrapper of streams!
void end(std::string_view data) {
// end should not explicitly flush the cork buffer! delay to when done with all http data!
writeToCorkBufferAndReset(data.data(), data.length(), data.length(), false);
}
// stream out (todo: fix up large sends and benchmark it again)
void write(std::function<std::string_view(int)> cb, int length) {
std::string_view chunk = cb(0);
// kopiera upp till (SSL eller icke-ssl) max copy distance
// om mer än detta, fortsätt skicka
// this strategy can be simplified to one, we can even have MAX_COPY_DISTANCE_SSL and MAX_COPY_DISTANCE
if (length < uWS::Loop::MAX_COPY_DISTANCE) {
// what if the streamer cannot return any data?
// then it should return something to pause write, and then start it again
// basically we need throttling
writeToCorkBufferAndReset(chunk.data(), chunk.length(), length, false);
} else {
// copying some data with the headers is a good idea for SSL but probably not for non-SSL
writeToCorkBufferAndReset(chunk.data(), uWS::Loop::MAX_COPY_DISTANCE, length, true);
// just assume this went fine
Data *httpData = (Data *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
// write that off!
static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, chunk.data() + uWS::Loop::MAX_COPY_DISTANCE, chunk.length() - uWS::Loop::MAX_COPY_DISTANCE, 0);
// if offset is at the end, we are done
if (httpData->offset < length) {
httpData->outStream = cb;
}
}
}
// this thing should only be reachable from App!
void onWritable() {
Data *httpData = (Data *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
// now we start streaming as much as possible in each call!
std::string_view chunk = httpData->outStream(httpData->offset);
// write that off!
static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, chunk.data(), chunk.length(), 0);
}
void onData(char *data, int length, std::function<void(HttpSocket<SSL> *, uWS::HttpRequest *)> &onHttpRequest) {
Data *httpData = (Data *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
// todo: this is where the HttpSocket binds together HttpParser and HttpRouter into one
httpData->httpParser.consumePostPadded(data, length, this, [&onHttpRequest](void *user, HttpRequest *httpRequest) {
onHttpRequest((HttpSocket<SSL> *) user, httpRequest);
}, [httpData](void *user, std::string_view data) {
if (httpData->inStream) {
httpData->inStream(data);
}
}, [](void *user) {
std::cout << "INVALID HTTP!" << std::endl;
});
}
void read(decltype(Data::inStream) stream) {
Data *httpData = (Data *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
httpData->inStream = stream;
}
// typical shared function?
void close() {
static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this);
}
HttpSocket() = delete;
};
#endif // HTTP_H
src/new_design/AsyncSocket.h
+121
View File
@@ -0,0 +1,121 @@
#ifndef ASYNCSOCKET_H
#define ASYNCSOCKET_H
#include "StaticDispatch.h"
#include "LoopData.h"
// todo: this is where the magic happens
namespace uWS {
template <bool SSL>
struct AsyncSocket : StaticDispatch<SSL> {
using SOCKET_TYPE = typename StaticDispatch<SSL>::SOCKET_TYPE;
using StaticDispatch<SSL>::static_dispatch;
// control everything with write, cork, uncork, close
// have HttpResponseData derive from AsyncSocketData?
// HttpResponse will only need one buffer for outgoing - the corked up
// maybe better name would be CorkableSocket?
// this does not belong here!
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;
}
void cork() {
LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(us_socket_get_context((us_socket *) this)));
loopData->corked = true;
}
void uncork() {
LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(us_socket_get_context((us_socket *) this)));
loopData->corked = false;
// send it off now!
int written = static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, loopData->corkBuffer, loopData->corkOffset, false);
loopData->corkOffset = 0;
// buffer the rest up in the outbuffer of this asynsocket!
}
void writeUnsigned(unsigned int value) {
LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(us_socket_get_context((us_socket *) this)));
loopData->corkOffset += u32toa(value, loopData->corkBuffer + loopData->corkOffset);
}
void write(const char *src, int length) {
LoopData *loopData = (LoopData *) us_loop_ext(us_socket_context_loop(us_socket_get_context((us_socket *) this)));
memcpy(loopData->corkBuffer + loopData->corkOffset, src, length);
loopData->corkOffset += length;
}
int writeOptionally(const char *src, int length) {
// not optional for now
AsyncSocket<SSL>::write(src, length);
return length;
/*
// kopiera upp till (SSL eller icke-ssl) max copy distance
// om mer än detta, fortsätt skicka
// this entire strategy should be made entirely in AsyncSocket!
// this strategy can be simplified to one, we can even have MAX_COPY_DISTANCE_SSL and MAX_COPY_DISTANCE
if (length < LoopData::MAX_COPY_DISTANCE) {
AsyncSocket<SSL>::write("Content-Length: ", 16);
AsyncSocket<SSL>::writeUnsigned(chunk.length());
AsyncSocket<SSL>::write("\r\n\r\n", 4);
AsyncSocket<SSL>::write(chunk.data(), chunk.length());
} else {
// copying some data with the headers is a good idea for SSL but probably not for non-SSL
//writeToCorkBufferAndReset(chunk.data(), LoopData::MAX_COPY_DISTANCE, length, true);
// just assume this went fine
HttpResponseData<SSL> *httpData = (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
// write that off! (should never happen here!)
//static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, chunk.data() + LoopData::MAX_COPY_DISTANCE, chunk.length() - LoopData::MAX_COPY_DISTANCE, 0);
// if offset is at the end, we are done
if (httpData->offset < length) {
httpData->outStream = cb;
}
}*/
}
void close() {
static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this);
}
};
}
#endif // ASYNCSOCKET_H
src/new_design/HttpContext.h
+27 -11
View File
@@ -4,6 +4,7 @@
#include "Loop.h"
#include "HttpContextData.h"
#include "HttpResponseData.h"
#include "AsyncSocket.h"
#include "StaticDispatch.h"
#include <string_view>
@@ -69,6 +70,9 @@ public:
static_dispatch(us_ssl_socket_context_on_data, us_socket_context_on_data)(getSocketContext(), [](auto *s, char *data, int length) {
HttpContextData<SSL> *httpContextData = getSocketContextData(s);
// cork this socket (move this to loop?)
((AsyncSocket<SSL> *) s)->cork();
HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s);
httpResponseData->consumePostPadded(data, length, s, [httpContextData](void *s, uWS::HttpRequest *httpRequest) {
@@ -82,8 +86,8 @@ public:
httpContextData->router.route("get", 3, httpRequest->getUrl().data(), httpRequest->getUrl().length(), &userData);
}, [httpResponseData](void *user, std::string_view data) {
if (httpResponseData->readHandler) {
httpResponseData->readHandler(data);
if (httpResponseData->inStream) {
httpResponseData->inStream(data);
}
}, [](void *user) {
std::cout << "INVALID HTTP!" << std::endl;
@@ -91,6 +95,9 @@ public:
// close it down
});
// uncork
((AsyncSocket<SSL> *) s)->uncork();
return s;
});
@@ -105,6 +112,18 @@ public:
// why? WE should emit this event via the socket data that we access!
//((HttpSocket<SSL> *) s)->onWritable();
// this thing should only be reachable from App!
/*void onWritable() {
Data *httpData = (Data *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
// now we start streaming as much as possible in each call!
std::string_view chunk = httpData->outStream(httpData->offset);
// write that off!
static_dispatch(us_ssl_socket_write, us_socket_write)((SOCKET_TYPE *) this, chunk.data(), chunk.length(), 0);
}*/
return s;
});
@@ -133,33 +152,30 @@ public:
}
static HttpContext *create(us_loop *loop) {
HttpContext *httpContext = (HttpContext *) us_create_socket_context(loop, sizeof(HttpContextData<SSL>));
new ((HttpContextData<SSL> *) static_dispatch(us_ssl_socket_context_ext, us_socket_context_ext)((SOCKET_CONTEXT_TYPE *) httpContext)) HttpContextData<SSL>();
return httpContext->init();
}
void free() {
// call destructor!
static_dispatch(us_ssl_socket_context_free, us_socket_context_free)(getSocketContext());
}
void onGet(std::string pattern, std::function<void(uWS::HttpResponse<SSL> *, uWS::HttpRequest *)> handler) {
HttpContextData<SSL> *data = getSocketContextData();
HttpContextData<SSL> *httpContextData = getSocketContextData();
// add things to the router
data->router.add("get", pattern.c_str(), [handler](typename HttpContextData<SSL>::UserData *user, auto *args) {
httpContextData->router.add("get", pattern.c_str(), [handler](typename HttpContextData<SSL>::UserData *user, auto *args) {
handler(user->httpResponse, user->httpRequest);
});
}
void listen(const char *host, int port, int options) {
static_dispatch(us_ssl_socket_context_listen, us_socket_context_listen)(getSocketContext(), host, port, options, sizeof(HttpContextData<SSL>));
static_dispatch(us_ssl_socket_context_listen, us_socket_context_listen)(getSocketContext(), host, port, options, sizeof(HttpResponseData<SSL>));
}
};
}
src/new_design/HttpResponse.h
+37 -9
View File
@@ -1,40 +1,68 @@
#ifndef HTTPRESPONSE_H
#define HTTPRESPONSE_H
#include "AsyncSocket.h"
#include "HttpResponseData.h"
#include "StaticDispatch.h"
// we will most probably depend on the LoopData to do corking and such
namespace uWS {
const char *HTTP_200_OK = "200 OK";
template <bool SSL>
struct HttpResponse : StaticDispatch<SSL> {
struct HttpResponse : public AsyncSocket<SSL> {
private:
using SOCKET_TYPE = typename StaticDispatch<SSL>::SOCKET_TYPE;
using StaticDispatch<SSL>::static_dispatch;
// helpers
HttpResponseData<SSL> *getHttpResponseData() {
return (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) this);
}
public:
void writeHeader() {
HttpResponse *writeStatus(std::string_view status) {
AsyncSocket<SSL>::write("HTTP/1.1 ", 9);
AsyncSocket<SSL>::write(status.data(), status.length());
AsyncSocket<SSL>::write("\r\n", 2);
return this;
}
HttpResponse *writeHeader(std::string_view key, std::string_view value) {
AsyncSocket<SSL>::write(key.data(), key.length());
AsyncSocket<SSL>::write(": ", 2);
AsyncSocket<SSL>::write(value.data(), value.length());
AsyncSocket<SSL>::write("\r\n", 2);
return this;
}
void write(std::function<std::string_view(int)> cb, int length) {
// what if the streamer cannot return any data?
// then it should return something to pause write, and then start it again
// basically we need throttling
std::string_view chunk = cb(0);
AsyncSocket<SSL>::write("Content-Length: ", 16);
AsyncSocket<SSL>::writeUnsigned(chunk.length());
AsyncSocket<SSL>::write("\r\n\r\n", 4);
if (AsyncSocket<SSL>::writeOptionally(chunk.data(), chunk.length()) < length) {
getHttpResponseData()->outStream = cb;
}
}
// this will probably not be this clean: it will most probably want to do some active pulling of data?
void read(std::function<void(std::string_view)> handler) {
HttpResponseData<SSL> *data = getHttpResponseData();
data->readHandler = handler;
data->inStream = handler;
}
// read and write streams most definitely will be called by the context, thus the context absolutely depends on the httpresponse!
// or, we both depend on each others data structures only?
// this should not be anything other than a simple convenience wrapper of streams!
void end(std::string_view data) {
writeStatus("200 OK")->write([data](int offset) {
return std::string_view(data.data() + offset, data.length() - offset);
}, data.length());
}
};
}
src/new_design/HttpResponseData.h
+6 -3
View File
@@ -1,8 +1,6 @@
#ifndef HTTPRESPONSEDATA_H
#define HTTPRESPONSEDATA_H
// so what do we depend on?
#include "HttpParser.h"
#include <functional>
@@ -11,7 +9,12 @@ namespace uWS {
template <bool SSL>
struct HttpResponseData : HttpParser {
std::function<void(std::string_view)> readHandler;
// inStream, outStream
std::function<void(std::string_view)> inStream;
std::function<std::string_view(int)> outStream;
int offset = 0;
// writeHandler
};
src/new_design/LoopData.h
+19
View File
@@ -0,0 +1,19 @@
#ifndef LOOPDATA_H
#define LOOPDATA_H
struct LoopData {
private:
public:
static const int CORK_BUFFER_SIZE = 16 * 1024;
static const int MAX_COPY_DISTANCE = 4096;
char *corkBuffer = new char[CORK_BUFFER_SIZE];
int corkOffset = 0;
bool corked = false;
};
#endif // LOOPDATA_H
src/new_design/StaticDispatch.h
+4
View File
@@ -6,6 +6,8 @@
#include <type_traits>
#include <libusockets.h>
namespace uWS {
template <bool SSL>
struct StaticDispatch {
template <class A, class B>
@@ -21,4 +23,6 @@ struct StaticDispatch {
typedef typename std::conditional<SSL, us_ssl_socket_context, us_socket_context>::type SOCKET_CONTEXT_TYPE;
};
}
#endif // STATICDISPATCH_H