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

Various comments and fixes

Browse Source
This commit is contained in:
Alex Hultman
2018-09-23 18:00:18 +02:00
parent ede5fb304a
commit 7f3753e9db
4 changed files with 111 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.cpp
+8 -4
View File
@@ -47,8 +47,6 @@ int main(int argc, char **argv) {
for (auto *x : delayedResponses) {
std::cout << "Resuming a response now!" << std::endl;
// resume should take a string_view!
x->resume();
}
@@ -76,10 +74,16 @@ int main(int argc, char **argv) {
/* This route streams back chunks of data in delayed fashion */
res->writeStatus(uWS::HTTP_200_OK)->write([res](int offset) {
/* Handle aborted stream out */
if (offset == -1) {
std::cout << "Removing closed stream!" << std::endl;
delayedResponses.erase(res);
return uWS::HTTP_STREAM_FIN;
}
/* Delay stream out */
std::cout << "Delaying stream now" << std::endl;
delayedResponses.insert(res);
//asyncFetchData(res)
return uWS::HTTP_STREAM_PAUSE;
}, 100);
src/AsyncSocket.h
+12 -3
View File
@@ -148,7 +148,7 @@ public:
}
/* Drain any socket-buffer while also optionally sending a chunk */
int mergeDrain(std::string_view optionalChunk) {
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
@@ -167,8 +167,17 @@ public:
return write(optionalChunk.data(), optionalChunk.length(), true, 0);
}
void close() {
static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this);
/* 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);
}
};
src/HttpContext.h
+75 -40
View File
@@ -44,16 +44,12 @@ private:
/* Init the HttpContext by registering libusockets event handlers */
HttpContext<SSL> *init() {
//new (data = (Data *) static_dispatch(us_ssl_socket_context_ext, us_socket_context_ext)(httpServerContext)) Data();
/* Handle socket connections */
static_dispatch(us_ssl_socket_context_on_open, us_socket_context_on_open)(getSocketContext(), [](auto *s, int is_client) {
HttpContextData<SSL> *httpContextData = getSocketContextData(s);
std::cout << "Opened http connection" << std::endl;
/* Any connected socket should timeout until it has a request */
static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S);
/* Init socket ext */
new (static_dispatch(us_ssl_socket_ext, us_socket_ext)(s)) HttpResponseData<SSL>;
return s;
@@ -61,9 +57,21 @@ private:
/* Handle socket disconnections */
static_dispatch(us_ssl_socket_context_on_close, us_socket_context_on_close)(getSocketContext(), [](auto *s) {
HttpContextData<SSL> *httpContextData = getSocketContextData(s);
/* Get socket ext */
HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s);
((HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s))->~HttpResponseData<SSL>();
/* Signal broken out stream */
if (httpResponseData->outStream) {
httpResponseData->outStream(-1);
}
/* Signal broken in stream */
if (httpResponseData->inStream) {
httpResponseData->inStream(std::string_view(nullptr, 0));
}
/* Destruct socket ext */
httpResponseData->~HttpResponseData<SSL>();
return s;
});
@@ -72,7 +80,14 @@ private:
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?)
/* Do not accept any data while in shutdown state */
if (static_dispatch(us_ssl_socket_is_shut_down, us_socket_is_shut_down)((SOCKET_TYPE *) s)) {
return s;
}
// basically, when getting a new request we need to disable timeouts and enter paused mode!
/* Cork this socket */
((AsyncSocket<SSL> *) s)->cork();
// pass this pointer to pointer along with the routing and change it if upgraded
@@ -81,9 +96,13 @@ private:
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) {
// whenever we get (the first) request we should disable timeout?
// warning: if we are in shutdown state, resetting the timer is a security issue!
// todo: do not reset timer, disable it to allow hang requests!
static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) s, HTTP_IDLE_TIMEOUT_S);
/* Reset httpResponse */
HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) s);
httpResponseData->offset = 0;
httpResponseData->state = 0;
@@ -94,6 +113,8 @@ private:
};
httpContextData->router.route("get", 3, httpRequest->getUrl().data(), httpRequest->getUrl().length(), &userData);
// here we can be closed and in shutdown?
}, [httpResponseData](void *user, std::string_view data) {
if (httpResponseData->inStream) {
httpResponseData->inStream(data);
@@ -114,25 +135,41 @@ private:
/* Handle HTTP write out */
static_dispatch(us_ssl_socket_context_on_writable, us_socket_context_on_writable)(getSocketContext(), [](auto *s) {
// I think it's fair to never mind this one -> if we keep writing data after shutting down then that's an issue for us
static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S);
/* Silence any spurious writable events due to SSL_read failing to write */
AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
// get next chunk to send
HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) asyncSocket->getExt();
if (httpResponseData->outStream) {
auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset);
// send, including any buffered up
httpResponseData->offset += asyncSocket->mergeDrain(chunk);
} else {
std::cout << "We did not have any outStream!" << std::endl;
asyncSocket->mergeDrain(std::string_view(nullptr, 0));
if (httpResponseData->state & HttpResponseData<SSL>::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<SSL>::HTTP_ENDED_STREAM_OUT) {
/* 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 */
auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset);
httpResponseData->offset += asyncSocket->mergeDrain(chunk);
// todo: we should loop until we cannot send anymore just like we do in HttpResponse::write(stream)!
} else {
/* We can come here if we only have socket buffers to drain yet no attached stream */
asyncSocket->mergeDrain();
}
return s;
});
@@ -140,26 +177,18 @@ private:
/* 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) {
// static_dispatch(us_ssl_socket_close, us_socket_close)(s);
/* We do not care for half closed sockets */
AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
asyncSocket->close();
return asyncSocket->close();
return s;
});
/* Handle socket timeouts */
/* Handle socket timeouts, simply close them so to not confuse client with FIN */
static_dispatch(us_ssl_socket_context_on_timeout, us_socket_context_on_timeout)(getSocketContext(), [](auto *s) {
if (static_dispatch(us_ssl_socket_is_shut_down, us_socket_is_shut_down)(s)) {
std::cout << "Forcefully closing socket since shutdown was not answered in time" << std::endl;
static_dispatch(us_ssl_socket_close, us_socket_close)(s);
} else {
std::cout << "Shutting down socket now" << std::endl;
static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S);
static_dispatch(us_ssl_socket_shutdown, us_socket_shutdown)(s);
}
return s;
/* Force close rather than gracefully shutdown and risk confusing the client with a complete download */
AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
return asyncSocket->close();
});
@@ -177,16 +206,22 @@ public:
httpContext = (HttpContext *) us_create_socket_context((us_loop *) loop, sizeof(HttpContextData<SSL>));
}
if (!httpContext) {
return nullptr;
}
/* Init socket context data */
new ((HttpContextData<SSL> *) static_dispatch(us_ssl_socket_context_ext, us_socket_context_ext)((SOCKET_CONTEXT_TYPE *) httpContext)) HttpContextData<SSL>();
return httpContext->init();
}
/* Destruct the HttpContext, it does not follow RAII */
void free() {
/* Destruct socket context data */
HttpContextData<SSL> *httpContextData = getSocketContextData();
httpContextData->~HttpContextData<SSL>();
// call destructor!
/* Free the socket context in whole */
static_dispatch(us_ssl_socket_context_free, us_socket_context_free)(getSocketContext());
}
src/HttpResponse.h
+16 -10
View File
@@ -96,19 +96,25 @@ public:
return;
}
std::cout << "Resume called and we really are paused" << std::endl;
/* Remove paused status */
httpResponseData->state &= ~HttpResponseData<SSL>::HTTP_PAUSED_STREAM_OUT;
int written = AsyncSocket<SSL>::write(chunk.data(), chunk.length(), true);
if (chunk.length()) {
int written = AsyncSocket<SSL>::write(chunk.data(), chunk.length(), true);
if (written == chunk.length()) {
// pull a new chunk from the callback (basically call onWritable)
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. Chunks might be read more than once */
/* Attach an output stream function. Chunks may be read more than once. Negative offset mean broken stream */
void write(std::function<std::pair<bool, std::string_view>(int)> cb, int length = 0) {
HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
@@ -140,21 +146,21 @@ public:
/* Handle PAUSE and FIN */
if (chunk.length() == 0) {
if (chunk == HTTP_STREAM_FIN.second) {
/* FIN */
if (chunk.data()) {
/* Try flush and shut down */
AsyncSocket<SSL>::uncork();
if (!AsyncSocket<SSL>::hasBuffer()) {
/* Uncork finished with no buffered data */
us_socket_shutdown((us_socket *) this);
AsyncSocket<SSL>::shutdown();
} else {
/* Let it shut down when drained */
httpResponseData->state |= HttpResponseData<SSL>::HTTP_ENDED_STREAM_OUT;
}
} else {
std::cout << "Paused stream is not implemented!" << std::endl;
// 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!)
/* Disable timeout and mark this stream as paused (important to silence spurious onWritable events) */
AsyncSocket<SSL>::timeout(0);
httpResponseData->state |= HttpResponseData<SSL>::HTTP_PAUSED_STREAM_OUT;
}
return;
}