diff --git a/examples/Crc32.cpp b/examples/Crc32.cpp index d27350c2e..35539c183 100644 --- a/examples/Crc32.cpp +++ b/examples/Crc32.cpp @@ -44,23 +44,23 @@ int main() { std::cout << key << ": " << value << std::endl; } - auto isAborted = std::make_shared(false); uint32_t crc = 0xFFFFFFFF; - res->onData([res, isAborted, crc](std::string_view chunk, bool isFin) mutable { + res->onData([crc](auto *res, std::string_view chunk, uint64_t maxRemainingBodyLength) mutable { + /* Aborted, there is no response anymore */ + if (!res) { + return; + } + if (chunk.length()) { crc = crc32(chunk.data(), chunk.length(), crc); } - if (isFin && !*isAborted) { + if (maxRemainingBodyLength == 0) { std::stringstream s; s << std::hex << (~crc) << std::endl; res->end(s.str()); } }); - - res->onAborted([isAborted]() { - *isAborted = true; - }); }).listen(3000, [](auto *listen_socket) { if (listen_socket) { std::cerr << "Listening on port " << 3000 << std::endl; diff --git a/fuzzing/EpollHelloWorld.cpp b/fuzzing/EpollHelloWorld.cpp index f7ef33917..991e41c39 100644 --- a/fuzzing/EpollHelloWorld.cpp +++ b/fuzzing/EpollHelloWorld.cpp @@ -35,14 +35,15 @@ void test() { res->end("Hello world!"); } }).post("/*", [](auto *res, auto *req) { - res->onAborted([]() { - /* We might as well use this opportunity to stress the loop a bit */ - uWS::Loop::get()->defer([]() { + res->onData([](auto *res, std::string_view chunk, uint64_t maxRemainingBodyLength) { + if (!res) { + /* We might as well use this opportunity to stress the loop a bit */ + uWS::Loop::get()->defer([]() { - }); - }); - res->onData([res](std::string_view chunk, bool isEnd) { - if (isEnd) { + }); + return; + } + if (maxRemainingBodyLength == 0) { res->cork([res, chunk]() { res->write("something ahead"); res->end(chunk); diff --git a/misc/READMORE.md b/misc/READMORE.md index 8c5822541..26841e441 100644 --- a/misc/READMORE.md +++ b/misc/READMORE.md @@ -58,7 +58,7 @@ uWS::App().get("/hello", [](auto *res, auto *req) { Important for all routes is that "req", the `uWS::HttpRequest *` dies with return. In other words, req is stack allocated so don't keep it in your pocket. -res, the `uWS::HttpResponse *` will be alive and accessible until either its .onAborted callback emits, or you've responded to the request via res.end or res.tryEnd. +res, the `uWS::HttpResponse *` will be alive and accessible until either its .onAborted callback emits, or you've responded to the request via res.end or res.tryEnd. Instead of .onAborted you can use the .onData and .onWritable overloads that take the response as first argument: they get the abort as a null response. In other words, you either respond to the request immediately and return, or you attach lambdas to the res (which may hold captured data), and respond later on in some other async callback. diff --git a/src/HttpContext.h b/src/HttpContext.h index 79de776b8..2b18b2b5c 100644 --- a/src/HttpContext.h +++ b/src/HttpContext.h @@ -115,6 +115,17 @@ struct HttpContext { httpResponseData->onAborted(); } + /* Also to the onData and onWritable overloads that get the response, in this order. markDone removes them, + * so they are only set while pending */ + MoveOnlyFunction *, std::string_view, uint64_t)> inStreamWithAbort = std::move(httpResponseData->inStreamWithAbort); + MoveOnlyFunction *, uintmax_t)> onWritableWithAbort = std::move(httpResponseData->onWritableWithAbort); + if (inStreamWithAbort) { + inStreamWithAbort(nullptr, {}, 0); + } + if (onWritableWithAbort) { + onWritableWithAbort(nullptr, 0); + } + /* Destruct socket ext */ httpResponseData->~HttpResponseData(); @@ -212,8 +223,9 @@ struct HttpContext { return nullptr; } - /* Returning from a request handler without responding or attaching an onAborted handler is ill-use */ - if (!((HttpResponse *) s)->hasResponded() && !httpResponseData->onAborted) { + /* Returning from a request handler without responding or attaching an onAborted handler is ill-use. + * The onData and onWritable overloads that get the response also get the abort */ + if (!((HttpResponse *) s)->hasResponded() && !httpResponseData->onAborted && !httpResponseData->inStreamWithAbort && !httpResponseData->onWritableWithAbort) { /* Throw exception here? */ std::cerr << "Error: Returning from a request handler without responding or attaching an abort handler is forbidden!" << std::endl @@ -223,7 +235,7 @@ struct HttpContext { } /* If we have not responded and we have a data handler, we need to timeout to enfore client sending the data */ - if (!((HttpResponse *) s)->hasResponded() && httpResponseData->inStream) { + if (!((HttpResponse *) s)->hasResponded() && (httpResponseData->inStream || httpResponseData->inStreamWithAbort)) { us_socket_timeout(SSL, (us_socket_t *) s, HTTP_IDLE_TIMEOUT_S); } @@ -232,7 +244,7 @@ struct HttpContext { }, [httpResponseData](void *user, std::string_view data, uint64_t maxRemainingBodyLength) -> void * { /* We always get an empty chunk even if there is no data */ - if (httpResponseData->inStream) { + if (httpResponseData->inStream || httpResponseData->inStreamWithAbort) { /* Todo: can this handle timeout for non-post as well? */ if (maxRemainingBodyLength == 0) { @@ -249,7 +261,11 @@ struct HttpContext { } /* We might respond in the handler, so do not change timeout after this */ - httpResponseData->inStream(data, maxRemainingBodyLength); + if (httpResponseData->inStream) { + httpResponseData->inStream(data, maxRemainingBodyLength); + } else { + httpResponseData->callInStreamWithAbort((HttpResponse *) user, data, maxRemainingBodyLength); + } /* Was the socket closed? */ if (us_socket_is_closed(SSL, (struct us_socket_t *) user)) { @@ -262,7 +278,8 @@ struct HttpContext { } /* If we were given the last data chunk, reset data handler to ensure following - * requests on the same socket won't trigger any previously registered behavior */ + * requests on the same socket won't trigger any previously registered behavior. + * The onData that gets the response stays until the response ends, markDone removes it */ if (maxRemainingBodyLength == 0) { httpResponseData->inStream = nullptr; } @@ -347,13 +364,14 @@ struct HttpContext { HttpResponseData *httpResponseData = (HttpResponseData *) asyncSocket->getAsyncSocketData(); /* Ask the developer to write data and return success (true) or failure (false), OR skip sending anything and return success (true). */ - if (httpResponseData->onWritable) { + if (httpResponseData->onWritable || httpResponseData->onWritableWithAbort) { /* We are now writable, so hang timeout again, the user does not have to do anything so we should hang until end or tryEnd rearms timeout */ us_socket_timeout(SSL, s, 0); /* We expect the developer to return whether or not write was successful (true). * If write was never called, the developer should still return true so that we may drain. */ - bool success = httpResponseData->callOnWritable(httpResponseData->offset); + bool success = httpResponseData->onWritable ? httpResponseData->callOnWritable(httpResponseData->offset) + : httpResponseData->callOnWritableWithAbort((HttpResponse *) s, httpResponseData->offset); /* The developer indicated that their onWritable failed. */ if (!success) { diff --git a/src/HttpResponse.h b/src/HttpResponse.h index 8b46a24fc..1bcce4c44 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -218,7 +218,7 @@ struct HttpResponse : public AsyncSocket { Super::timeout(HTTP_TIMEOUT_S); } - /* Remove onAborted, onWritable function and mark done if we reach the end, or if we were given no data (faked size like in HEAD response) */ + /* Remove onAborted, onWritable, the onData that gets the response and mark done if we reach the end, or if we were given no data (faked size like in HEAD response) */ /* I need to figure out if this line should rather be simply httpResponseData->offset == data.length() */ /* No that can't be right, tryEnd with fake length should not complete the response even if the smaller chunk wrote in one go */ /* Possibly need to separate endWithoutBody and tryEnd with fake length into two separate calls with a boolean that explicitly marks isHeadOnly */ @@ -589,15 +589,38 @@ struct HttpResponse : public AsyncSocket { return this; } - /* Attach handler for writable HTTP response */ + /* Attach handler for writable HTTP response. + * Deprecated: the overload that gets the response also gets the abort */ HttpResponse *onWritable(MoveOnlyFunction &&handler) { HttpResponseData *httpResponseData = getHttpResponseData(); httpResponseData->onWritable = std::move(handler); + httpResponseData->onWritableWithAbort = nullptr; return this; } - /* Attach handler for aborted HTTP request */ + /* Attach handler for writable HTTP response, called with the response and the offset. If the request is aborted + * before the response ends, it is called once with a null response (what it returns is then ignored), so no + * onAborted is needed. With the onData that gets the response too, both get the abort, onData first. */ + HttpResponse *onWritable(MoveOnlyFunction &&handler) { + HttpResponseData *httpResponseData = getHttpResponseData(); + + httpResponseData->onWritableWithAbort = std::move(handler); + httpResponseData->onWritable = nullptr; + return this; + } + + /* Remove the writable handler, of either kind */ + HttpResponse *onWritable(std::nullptr_t) { + HttpResponseData *httpResponseData = getHttpResponseData(); + + httpResponseData->onWritable = nullptr; + httpResponseData->onWritableWithAbort = nullptr; + return this; + } + + /* Attach handler for aborted HTTP request. + * Deprecated: the onData and onWritable overloads that get the response also get the abort */ HttpResponse *onAborted(MoveOnlyFunction &&handler) { HttpResponseData *httpResponseData = getHttpResponseData(); @@ -605,7 +628,8 @@ struct HttpResponse : public AsyncSocket { return this; } - /* Attach a read handler for data sent. Will be called with FIN set true if last segment. */ + /* Attach a read handler for data sent. Will be called with FIN set true if last segment. + * Deprecated: the overload that gets the response also gets the abort */ void onData(MoveOnlyFunction &&handler) { if (handler) { onDataV2([handler = std::move(handler)](std::string_view chunk, uint64_t maxRemainingBodyLength) mutable { @@ -616,10 +640,30 @@ struct HttpResponse : public AsyncSocket { } } - /* Attach a read handler for data sent. Will be called with maxRemainingBodyLength. maxRemainingBodyLength == 0 is the same as isLast. */ + /* Attach a read handler for data sent. Will be called with the response, the chunk and maxRemainingBodyLength, + * maxRemainingBodyLength == 0 is the same as isLast. If the request is aborted before the response ends, also + * after the last chunk, it is called once with a null response, so no onAborted is needed. After the response + * ends it gets nothing more. */ + void onData(MoveOnlyFunction &&handler) { + HttpResponseData *data = getHttpResponseData(); + data->inStreamWithAbort = std::move(handler); + data->inStream = nullptr; + + /* Always reset this counter here */ + data->received_bytes_per_timeout = 0; + } + + /* Remove the read handler, of either kind */ + void onData(std::nullptr_t) { + onDataV2(nullptr); + } + + /* Attach a read handler for data sent. Will be called with maxRemainingBodyLength. maxRemainingBodyLength == 0 is the same as isLast. + * Deprecated: the onData overload that gets the response also gets the abort */ void onDataV2(MoveOnlyFunction &&handler) { HttpResponseData *data = getHttpResponseData(); data->inStream = std::move(handler); + data->inStreamWithAbort = nullptr; /* Always reset this counter here */ data->received_bytes_per_timeout = 0; diff --git a/src/HttpResponseData.h b/src/HttpResponseData.h index f17690de2..02eb699d8 100644 --- a/src/HttpResponseData.h +++ b/src/HttpResponseData.h @@ -26,8 +26,12 @@ #include "MoveOnlyFunction.h" +#include + namespace uWS { +template struct HttpResponse; + template struct HttpResponseData : AsyncSocketData, HttpParser { template friend struct HttpResponse; @@ -38,6 +42,9 @@ struct HttpResponseData : AsyncSocketData, HttpParser { onAborted = nullptr; /* Also remove onWritable so that we do not emit when draining behind the scenes. */ onWritable = nullptr; + /* The handlers that get the abort get nothing more, not even the rest of the body */ + onWritableWithAbort = nullptr; + inStreamWithAbort = nullptr; /* We are done with this request */ state &= ~HttpResponseData::HTTP_RESPONSE_PENDING; @@ -46,8 +53,22 @@ struct HttpResponseData : AsyncSocketData, HttpParser { /* Caller of onWritable. It is possible onWritable calls markDone so we need to borrow it. * It is also possible user code sets a new onWritable while running user registered onWritable. */ bool callOnWritable(uintmax_t offset) { + return callBorrowed(onWritable, offset); + } + + /* Callers of the handlers that get the abort, borrowed for the same reasons */ + bool callOnWritableWithAbort(HttpResponse *res, uintmax_t offset) { + return callBorrowed(onWritableWithAbort, res, offset); + } + + void callInStreamWithAbort(HttpResponse *res, std::string_view data, uint64_t maxRemainingBodyLength) { + callBorrowed(inStreamWithAbort, res, data, maxRemainingBodyLength); + } +private: + template + static R callBorrowed(MoveOnlyFunction &slot, Args... args) { /* 1. Borrow the real callback */ - MoveOnlyFunction borrowedOnWritable = std::move(onWritable); + MoveOnlyFunction borrowed = std::move(slot); /* 2. Setup the stack-based detection flag */ bool placeholderReplaced = false; @@ -72,25 +93,28 @@ struct HttpResponseData : AsyncSocketData, HttpParser { }; /* 3. Set placeholder with the captured Sentinel */ - onWritable = [tracker = Sentinel(&placeholderReplaced)](uintmax_t) { - return true; + slot = [tracker = Sentinel(&placeholderReplaced)](Args...) -> R { + if constexpr (!std::is_void_v) { + return true; + } }; - /* 4. Run the borrowed callback */ - bool ret = borrowedOnWritable(offset); - - /* - 5. If placeholderReplaced is STILL false, it means the lambda (and its Sentinel) - is still sitting inside 'onWritable'. If it's true, the lambda was destroyed - to make room for a new one. - */ - if (!placeholderReplaced) { - onWritable = std::move(borrowedOnWritable); - } - - return ret; + /* 4. Run the borrowed callback. 5. When done, if placeholderReplaced is STILL false, it means the lambda (and + its Sentinel) is still sitting inside the slot, so put the callback back. If it's true, the lambda was + destroyed to make room for a new one. */ + struct PutBack { + MoveOnlyFunction &slot, &borrowed; + bool &placeholderReplaced; + ~PutBack() { + if (!placeholderReplaced) { + slot = std::move(borrowed); + } + } + } putBack{slot, borrowed, placeholderReplaced}; + + return borrowed(args...); } -private: + /* Bits of status */ enum { HTTP_STATUS_CALLED = 1, // used @@ -104,6 +128,10 @@ struct HttpResponseData : AsyncSocketData, HttpParser { MoveOnlyFunction onWritable; MoveOnlyFunction onAborted; MoveOnlyFunction inStream; // onData + /* The onWritable and onData overloads that get the response first, and the abort as a null response until + * markDone. Each one replaces the old handler of the same kind, and the other way around */ + MoveOnlyFunction *, uintmax_t)> onWritableWithAbort; + MoveOnlyFunction *, std::string_view, uint64_t)> inStreamWithAbort; /* Outgoing offset */ uintmax_t offset = 0; diff --git a/tests/HttpResponse.cpp b/tests/HttpResponse.cpp index 619bd8eb1..2e0d56219 100644 --- a/tests/HttpResponse.cpp +++ b/tests/HttpResponse.cpp @@ -260,6 +260,49 @@ static void expectChunked(int port, const char *path, std::string_view expectedB std::cout << "OK " << path << std::endl; } +/* Sends a raw request and closes the socket after delayMs without reading, or after reading the + * response when readResponse is set: the server sees an abort, or a close after the response. */ +static void sendThenClose(int port, const std::string &raw, int delayMs, bool readResponse) { + int fd = (int) ::socket(AF_INET, SOCK_STREAM, 0); + sockaddr_in addr{}; + addr.sin_family = AF_INET; + addr.sin_port = htons((uint16_t) port); + addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + if (fd < 0 || ::connect(fd, (sockaddr *) &addr, sizeof(addr)) != 0) { + std::cerr << "connect failed: " << std::strerror(errno) << std::endl; + std::abort(); + } + if (::send(fd, raw.data(), raw.size(), 0) != (ssize_t) raw.size()) { + std::cerr << "send failed: " << std::strerror(errno) << std::endl; + std::abort(); + } + if (readResponse) { + timeval tv{}; + tv.tv_sec = 2; + setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + char buf[2048]; + if (::recv(fd, buf, sizeof(buf), 0) <= 0) { + std::cerr << "no response for " << raw << std::endl; + std::abort(); + } + } + usleep(delayMs * 1000); + ::close(fd); +} + +static void expectCount(const char *what, std::atomic &value, int expected) { + for (int i = 0; i < 200 && value.load() < expected; i++) { + usleep(10000); + } + /* Give a wrong extra call the time to show up */ + usleep(50000); + if (value.load() != expected) { + std::cerr << what << ": expected " << expected << ", got " << value.load() << std::endl; + std::abort(); + } + std::cout << "OK " << what << std::endl; +} + int main() { uWS::App app; uWS::Loop *loop = uWS::Loop::get(); @@ -267,6 +310,18 @@ int main() { std::atomic backpressureWritten{false}; constexpr int chunkSize = 64 * 1024; constexpr int maxFillChunks = 512; + /* The onData and onWritable overloads that get the response also get the abort, as a null response, until the response ends */ + std::atomic waitEmptyChunks{0}, waitAborts{0}; + std::atomic bodyBytes{0}, bodyAborts{0}; + std::atomic allOrder{0}, allCalls{0}; + std::atomic endedAborts{0}; + std::atomic countedBytes{0}; + std::atomic oldChunks{0}, oldAborts{0}, replacedCalls{0}; + std::atomic newChunks{0}, newAborts{0}, replacedOldCalls{0}; + std::atomic writableCalls{0}, writableAborts{0}; + std::atomic removedCalls{0}, removedAborts{0}; + std::atomic writableAttached{false}; + static const std::string big((size_t) 4 * 1024 * 1024, 'b'); app.get("/write-end", [](auto *res, auto * /*req*/) { res->onAborted([]() {}); @@ -355,6 +410,112 @@ int main() { res->end(); }); backpressureWritten.store(true); + }).get("/abort-wait", [&](auto *res, auto * /*req*/) { + /* Only waiting, no onAborted: a GET gets the empty last chunk, then the abort */ + res->onData([&](auto *res, std::string_view chunk, uint64_t maxRemainingBodyLength) { + if (!res) { + waitAborts++; + } else if (chunk.empty() && maxRemainingBodyLength == 0) { + waitEmptyChunks++; + } + }); + }).post("/abort-after-body", [&](auto *res, auto * /*req*/) { + /* The whole body arrives, the response does not: the abort comes after the last chunk */ + res->onData([&](auto *res, std::string_view chunk, uint64_t) { + if (!res) { + bodyAborts++; + } else { + bodyBytes += (int) chunk.length(); + } + }); + }).get("/abort-all", [&](auto *res, auto * /*req*/) { + /* onAborted and the onData and onWritable that get the response all get the abort, in this order */ + res->onAborted([&]() { + allCalls++; + allOrder = allOrder * 10 + 1; + }); + res->onData([&](auto *res, std::string_view, uint64_t) { + if (!res) { + allCalls++; + allOrder = allOrder * 10 + 2; + } + }); + res->onWritable([&](auto *res, uintmax_t) { + if (!res) { + allCalls++; + allOrder = allOrder * 10 + 3; + } + return true; + }); + }).post("/no-abort-after-end", [&](auto *res, auto * /*req*/) { + /* Answered on the last chunk: closing afterwards is no abort */ + res->onData([&](auto *res, std::string_view, uint64_t maxRemainingBodyLength) { + if (!res) { + endedAborts++; + } else if (maxRemainingBodyLength == 0) { + res->end("done"); + } + }); + }).post("/count-body", [&](auto *res, auto * /*req*/) { + /* Ended on the last chunk: the next request on the socket must not reach this handler */ + res->onData([&](auto *res, std::string_view chunk, uint64_t maxRemainingBodyLength) { + if (res) { + countedBytes += (int) chunk.length(); + if (maxRemainingBodyLength == 0) { + res->end("counted"); + } + } + }); + }).post("/no-data-handler", [](auto *res, auto * /*req*/) { + res->end("ok"); + }).post("/old-replaces-new", [&](auto *res, auto * /*req*/) { + res->onData([&](auto *, std::string_view, uint64_t) { + replacedCalls++; + }); + res->onDataV2([&](std::string_view, uint64_t) { + oldChunks++; + }); + res->onAborted([&]() { + oldAborts++; + }); + }).post("/new-replaces-old", [&](auto *res, auto * /*req*/) { + res->onDataV2([&](std::string_view, uint64_t) { + replacedOldCalls++; + }); + res->onData([&](auto *res, std::string_view, uint64_t) { + if (!res) { + newAborts++; + } else { + newChunks++; + } + }); + }).post("/removed", [&](auto *res, auto * /*req*/) { + /* Removed with nullptr, of either kind: they get nothing, onAborted gets the abort */ + res->onData([&](auto *, std::string_view, uint64_t) { + removedCalls++; + }); + res->onWritable([&](auto *, uintmax_t) { + removedCalls++; + return true; + }); + res->onData(nullptr); + res->onWritable(nullptr); + res->onAborted([&]() { + removedAborts++; + }); + }).get("/writable-stream", [&](auto *res, auto * /*req*/) { + /* Streamed with tryEnd, no onAborted: onWritable gets the response to go on */ + if (!res->tryEnd(big, big.size()).first) { + res->onWritable([&](auto *res, uintmax_t offset) { + if (!res) { + writableAborts++; + return true; + } + writableCalls++; + return res->tryEnd(std::string_view(big).substr((size_t) offset), big.size()).first; + }); + } + writableAttached.store(true); }).listen(0, [&](us_listen_socket_t *listenSocket) { if (!listenSocket) { std::cerr << "Failed to listen" << std::endl; @@ -368,7 +529,7 @@ int main() { return 1; } - std::thread client([loop, &app, port, &backpressureWritten]() { + std::thread client([&, loop, port]() { expectChunked(port, "/write-end", "3\r\nfoo\r\n0\r\n\r\n"); expectChunked(port, "/begin-write-end", "3\r\nfoo\r\n0\r\n\r\n"); expectChunked(port, "/begin-end", "0\r\n\r\n"); @@ -398,6 +559,47 @@ int main() { } std::cout << "OK /backpressure (" << decoded.size() << " decoded bytes)" << std::endl; + sendThenClose(port, "GET /abort-wait HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n", 100, false); + expectCount("/abort-wait empty last chunk", waitEmptyChunks, 1); + expectCount("/abort-wait abort", waitAborts, 1); + + sendThenClose(port, "POST /abort-after-body HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 5\r\n\r\nhello", 100, false); + expectCount("/abort-after-body abort", bodyAborts, 1); + expectCount("/abort-after-body bytes", bodyBytes, 5); + + sendThenClose(port, "GET /abort-all HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n", 100, false); + expectCount("/abort-all calls", allCalls, 3); + expectCount("/abort-all order", allOrder, 123); + + sendThenClose(port, "POST /no-abort-after-end HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 2\r\n\r\nhi", 50, true); + expectCount("/no-abort-after-end", endedAborts, 0); + + sendThenClose(port, "POST /count-body HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 5\r\n\r\nhello" + "POST /no-data-handler HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 5\r\n\r\nworld", 50, true); + expectCount("/count-body then /no-data-handler on the same socket", countedBytes, 5); + + sendThenClose(port, "POST /old-replaces-new HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 5\r\n\r\nhello", 100, false); + expectCount("/old-replaces-new old chunks", oldChunks, 1); + expectCount("/old-replaces-new onAborted", oldAborts, 1); + expectCount("/old-replaces-new replaced", replacedCalls, 0); + + sendThenClose(port, "POST /new-replaces-old HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 5\r\n\r\nhello", 100, false); + expectCount("/new-replaces-old chunks", newChunks, 1); + expectCount("/new-replaces-old abort", newAborts, 1); + expectCount("/new-replaces-old replaced", replacedOldCalls, 0); + + sendThenClose(port, "POST /removed HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 5\r\n\r\nhello", 100, false); + expectCount("/removed onAborted", removedAborts, 1); + expectCount("/removed handlers", removedCalls, 0); + + response = requestAfterBackpressure(port, "/writable-stream", writableAttached); + if (bodyOf(response).size() != big.size() || writableCalls.load() == 0 || writableAborts.load() != 0) { + std::cerr << "/writable-stream got " << bodyOf(response).size() << " bytes, " << writableCalls.load() + << " writable calls, " << writableAborts.load() << " aborts" << std::endl; + std::abort(); + } + std::cout << "OK /writable-stream (" << writableCalls.load() << " writable calls)" << std::endl; + loop->defer([&app]() { app.close(); });