Handle socket closing, shutdown, error & (upgrade) in HTTP handlers

src/AsyncSocket.h

@@ -9,6 +9,11 @@
 
 namespace uWS {
 
+// we need a variant of this where any failed write triggers a timeout!
+// AsyncSocket needs to derive from a uSockets base we can select like so:
+
+// HttpResponse -> AsyncSocket<T> -> TimeoutSocket or Socket (make StaticDispath Socket)
+
 template <bool SSL>
 struct AsyncSocket : StaticDispatch<SSL> {
     template <bool> friend struct HttpContext;
@@ -40,6 +45,11 @@ protected:
         return static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this);
     }
 
+    bool isFullyOpen() {
+        // todo:: not shutdown or closed
+        return true;
+    }
+
     /* Cork this socket. Only one socket may ever be corked per-loop at any given time */
     void cork() {
         //std::cout << "Cork called" << std::endl;

src/HttpContext.h

@@ -26,7 +26,7 @@ private:
     using StaticDispatch<SSL>::static_dispatch;
     HttpContext() = delete;
 
-    /* Maximum delay allowed until an HTTP connection is terminated due to outstanding request (slow loris protection) */
+    /* Maximum delay allowed until an HTTP connection is terminated due to outstanding request or rejected data (slow loris protection) */
     static const int HTTP_IDLE_TIMEOUT_S = 10;
 
     SOCKET_CONTEXT_TYPE *getSocketContext() {
@@ -95,22 +95,22 @@ private:
                 return s;
             }
 
-            // basically, when getting a new request we need to disable timeouts and enter paused mode!
+            HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s);
 
             /* Cork this socket */
             ((AsyncSocket<SSL> *) s)->cork();
 
-            // pass this pointer to pointer along with the routing and change it if upgraded
+            void *returnedSocket = httpResponseData->consumePostPadded(data, length, s, [httpContextData](void *s, uWS::HttpRequest *httpRequest) -> void * {
-            SOCKET_TYPE *returnedSocket = s;
 
-            HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s);
+                // we need HttpAsyncSocket to derive from AsyncSocket where any failed write will trigger the timeout?
-            httpResponseData->consumePostPadded(data, length, s, [httpContextData](void *s, uWS::HttpRequest *httpRequest) {
 
-                // whenever we get (the first) request we should disable timeout?
+                // http timeout logic in a nutshell:
+                // whenever a httpsocket writes and it fails, start a timer until onwritable, reset timer in next onwritable or next successful write
+                // if .end succeeds, then start a new timeout for the next request
 
-                // warning: if we are in shutdown state, resetting the timer is a security issue!
+                /* For every request we reset the timeout and hang until user makes action */
-                // todo: do not reset timer, disable it to allow hang requests!
+                /* Warning: if we are in shutdown state, resetting the timer is a security issue! */
-                static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) s, HTTP_IDLE_TIMEOUT_S);
+                static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) s, 0);
 
                 /* Reset httpResponse */
                 HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) s);
@@ -118,32 +118,64 @@ private:
                 httpResponseData->state = 0;
 
                 /* Route the method and URL */
+
+                // I guess upgrade will have to write to a global variable we check afterwards
                 httpContextData->router.route(httpRequest->getMethod(), httpRequest->getUrl(), {
                                                   (HttpResponse<SSL> *) s, httpRequest
                                               });
 
-                // here we can be closed and in shutdown?
 
-            }, [httpResponseData](void *user, std::string_view data) {
+                //if (isFullyOpen) return s otherwise return nullptr;
+
+
+                /* Was the socket closed? */
+                if (us_internal_socket_is_closed((struct us_socket *) s)) {
+                    return nullptr;
+                }
+
+                /* We absolutely have to terminate parsing if shutdown */
+                if (static_dispatch(us_ssl_socket_is_shut_down, us_socket_is_shut_down)((SOCKET_TYPE *) s)) {
+                    return nullptr;
+                }
+
+                /* Was the socket upgraded? */
+                // return new pointer to websocket
+
+                /* If we return anything other than user, then that means STOP PARSING and this new value should be returned all the way? */
+
+                /* Continue parsing */
+                return s;
+
+            }, [httpResponseData](void *user, std::string_view data) -> void * {
                 if (httpResponseData->inStream) {
                     httpResponseData->inStream(data);
                 }
-            }, [](void *user) {
-                // close any socket on HTTP errors
-                //static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) user);
 
+                /* Was the socket closed? */
+                if (us_internal_socket_is_closed((struct us_socket *) user)) {
+                    return nullptr;
+                }
+
+                /* We absolutely have to terminate parsing if shutdown */
+                if (static_dispatch(us_ssl_socket_is_shut_down, us_socket_is_shut_down)((SOCKET_TYPE *) user)) {
+                    return nullptr;
+                }
+
+                return user;
+            }, [](void *user) {
+                 /* Close any socket on HTTP errors */
+                static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) user);
+                return nullptr;
             });
 
-            if (us_internal_socket_is_closed((struct us_socket *) s)) {
+            /* Only uncork still valid sockets */
-                // do you really return s? I guess so?
+            if (returnedSocket == s) {
-                return s;
+                ((AsyncSocket<SSL> *) s)->uncork();
+            } else {
+                // was this socket upgraded?
+                std::cout << "Socket was closed or shut down in handler or maybe upgraded" << std::endl;
             }
 
-            // uncork only if not closed
-            ((AsyncSocket<SSL> *) s)->uncork();
-
-            // how do we return a new socket here, from the http route?
-            // maybe hold upgradedSocket in the loopData and return that?
             return s;
         });
 
@@ -152,8 +184,8 @@ private:
 
             std::cout << "HttpContext::onWritable event fired!" << std::endl;
 
-            /* Writing data should reset the timeout */
+            /* We are now writable, so hang timeout again */
-            static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S);
+            static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, 0);
 
             AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
             HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) asyncSocket->getExt();
@@ -176,7 +208,7 @@ 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) {
 
-            std::cout << "FIN sent" << std::endl;
+            //std::cout << "FIN sent" << std::endl;
 
             /* We do not care for half closed sockets */
             AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
@@ -226,15 +258,12 @@ public:
         static_dispatch(us_ssl_socket_context_free, us_socket_context_free)(getSocketContext());
     }
 
-    /* Register an HTTP GET route handler acording to URL pattern */
+    /* Register an HTTP route handler acording to URL pattern */
     void onHttp(std::string method, std::string pattern, std::function<void(uWS::HttpResponse<SSL> *, uWS::HttpRequest *)> handler) {
         HttpContextData<SSL> *httpContextData = getSocketContextData();
 
         httpContextData->router.add(method, pattern, [handler](typename HttpContextData<SSL>::RouterData user, std::pair<int, std::string_view *> params) {
-
-            // todo: attach params to the req here!
             user.httpRequest->setParameters(params);
-
             handler(user.httpResponse, user.httpRequest);
         });
     }

src/HttpParser.h

@@ -110,7 +110,7 @@ private:
 
     // the only caller of getHeaders
     template <int CONSUME_MINIMALLY>
-    int fenceAndConsumePostPadded(char *data, int length, void *user, HttpRequest *req, std::function<void(void *, HttpRequest *)> &requestHandler, std::function<void(void *, std::string_view)> &dataHandler) {
+    std::pair<int, void *> fenceAndConsumePostPadded(char *data, int length, void *user, HttpRequest *req, std::function<void *(void *, HttpRequest *)> &requestHandler, std::function<void *(void *, std::string_view)> &dataHandler) {
         int consumedTotal = 0;
         data[length] = '\r';
 
@@ -125,8 +125,19 @@ private:
             const char *querySeparatorPtr = (const char *) memchr(req->headers->value.data(), '?', req->headers->value.length());
             req->querySeparator = (querySeparatorPtr ? querySeparatorPtr : req->headers->value.data() + req->headers->value.length()) - req->headers->value.data();
 
-            requestHandler(user, req);
+            // this one should return socket and exit on closed
+            // what happens with data left for websockets?
+            void *returnedUser = requestHandler(user, req);
+            if (returnedUser != user) {
+                // upgraded socket, or otherwise broken
 
+                // return pair of consumed and user
+                return {consumedTotal, returnedUser};
+            }
+
+            // do not check this for GET!
+
+            // todo: also support reading chunked streams
             std::string_view contentLengthString = req->getHeader("content-length");
             if (contentLengthString.length()) {
                 remainingStreamingBytes = toUnsignedInteger(contentLengthString);
@@ -146,29 +157,36 @@ private:
                 break;
             }
         }
-        return consumedTotal;
+        return {consumedTotal, user};
     }
 
 public:
 
     // todo: what can we do with the socket inside the handlers? we need to check on return from any handler if we closed or terminated or upgraded the socket
-    void consumePostPadded(char *data, int length, void *user, std::function<void(void *, HttpRequest *)> &&requestHandler, std::function<void(void *, std::string_view)> &&dataHandler, std::function<void(void *)> &&errorHandler) {
+    void *consumePostPadded(char *data, int length, void *user, std::function<void *(void *, HttpRequest *)> &&requestHandler, std::function<void *(void *, std::string_view)> &&dataHandler, std::function<void *(void *)> &&errorHandler) {
 
         HttpRequest req;
 
         if (remainingStreamingBytes) {
+
+            // this is exactly the same as below!
             if (remainingStreamingBytes >= length) {
-                dataHandler(user, std::string_view(data, length));
+                void *returnedUser = dataHandler(user, std::string_view(data, length));
                 remainingStreamingBytes -= length;
-                return;
+                return returnedUser;
             } else {
-                dataHandler(user, std::string_view(data, remainingStreamingBytes));
+                void *returnedUser = dataHandler(user, std::string_view(data, remainingStreamingBytes));
 
                 data += remainingStreamingBytes;
                 length -= remainingStreamingBytes;
 
                 remainingStreamingBytes = 0;
+
+                if (returnedUser != user) {
+                    return returnedUser;
+                }
             }
+
         } else if (fallback.length()) {
             int had = fallback.length();
 
@@ -177,50 +195,67 @@ public:
             fallback.reserve(maxCopyDistance + 32); // padding should be same as libus
             fallback.append(data, maxCopyDistance);
 
-            int consumed = fenceAndConsumePostPadded<true>(fallback.data(), fallback.length(), user, &req, requestHandler, dataHandler);
+            // break here on break
-            if (consumed) {
+            std::pair<int, void *> consumed = fenceAndConsumePostPadded<true>(fallback.data(), fallback.length(), user, &req, requestHandler, dataHandler);
+            if (consumed.second != user) {
+                return consumed.second;
+            }
+
+            if (consumed.first) {
 
                 fallback.clear();
 
-                data += consumed - had;
+                data += consumed.first - had;
-                length -= consumed - had;
+                length -= consumed.first - had;
 
-                // this is exactly the same as above!
                 if (remainingStreamingBytes) {
+                    // this is exactly the same as above!
                     if (remainingStreamingBytes >= length) {
-                        dataHandler(user, std::string_view(data, length));
+                        void *returnedUser = dataHandler(user, std::string_view(data, length));
                         remainingStreamingBytes -= length;
-                        return;
+                        return returnedUser;
                     } else {
-                        dataHandler(user, std::string_view(data, remainingStreamingBytes));
+                        void *returnedUser = dataHandler(user, std::string_view(data, remainingStreamingBytes));
 
                         data += remainingStreamingBytes;
                         length -= remainingStreamingBytes;
 
                         remainingStreamingBytes = 0;
+
+                        if (returnedUser != user) {
+                            return returnedUser;
+                        }
                     }
                 }
 
             } else {
                 if (fallback.length() == MAX_FALLBACK_SIZE) {
-                    errorHandler(user);
+                    // you don't really need error handler, just return something strange!
+                    // we could have it return a constant pointer to denote error!
+                    return errorHandler(user);
                 }
-                return;
+                return user;
             }
         }
 
-        int consumed = fenceAndConsumePostPadded<false>(data, length, user, &req, requestHandler, dataHandler);
+        std::pair<int, void *> consumed = fenceAndConsumePostPadded<false>(data, length, user, &req, requestHandler, dataHandler);
+        if (consumed.second != user) {
+            return consumed.second;
+        }
 
-        data += consumed;
+        data += consumed.first;
-        length -= consumed;
+        length -= consumed.first;
 
         if (length) {
             if (length < MAX_FALLBACK_SIZE) {
                 fallback.append(data, length);
             } else {
-                errorHandler(user);
+                return errorHandler(user);
             }
         }
+
+        // added for now
+        return user;
     }
 };
 

src/HttpResponse.h

@@ -103,6 +103,8 @@ public:
 
             Super::write(data.data(), data.length());
         }
+
+        // todo: calling end should start a timeout of the socket!
     }
 
     /* Write parts of the response in chunking fashion */