From 1bfc8138424e764baf3558ba6c923f7f490633cd Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 27 Aug 2026 04:58:43 -0400 Subject: [PATCH 1/5] THRIFT-6060: Reconnect after HTTP Connection: close Signed-off-by: 1fanwang <1fannnw@gmail.com> --- lib/cpp/src/thrift/transport/THttpClient.cpp | 54 +++++++--- lib/cpp/src/thrift/transport/THttpClient.h | 1 + lib/cpp/test/OneWayHTTPTest.cpp | 107 +++++++++++++++++-- 3 files changed, 141 insertions(+), 21 deletions(-) diff --git a/lib/cpp/src/thrift/transport/THttpClient.cpp b/lib/cpp/src/thrift/transport/THttpClient.cpp index 3cc83345018..5e1bd3c95e4 100644 --- a/lib/cpp/src/thrift/transport/THttpClient.cpp +++ b/lib/cpp/src/thrift/transport/THttpClient.cpp @@ -18,10 +18,11 @@ */ #include -#include +#include #include +#include #include -#include +#include #include #include @@ -38,23 +39,32 @@ THttpClient::THttpClient(std::shared_ptr transport, std::string path, std::shared_ptr config) : THttpTransport(transport, config), - host_(host), + host_(host), path_(path), - onewayResponsePending_(false) { -} + onewayResponsePending_(false), + closeAfterResponse_(false) {} -THttpClient::THttpClient(string host, int port, string path, - std::shared_ptr config) +THttpClient::THttpClient(string host, int port, string path, std::shared_ptr config) : THttpTransport(std::shared_ptr(new TSocket(host, port)), config), host_(host), path_(path), - onewayResponsePending_(false) { -} + onewayResponsePending_(false), + closeAfterResponse_(false) {} THttpClient::~THttpClient() = default; void THttpClient::close() { onewayResponsePending_ = false; + closeAfterResponse_ = false; + readBuffer_.resetBuffer(); + readHeaders_ = true; + chunked_ = false; + chunkedDone_ = false; + chunkSize_ = 0; + contentLength_ = 0; + httpPos_ = 0; + httpBufLen_ = 0; + httpBuf_[0] = '\0'; THttpTransport::close(); } @@ -72,10 +82,20 @@ void THttpClient::parseHeader(char* header) { } else if (boost::istarts_with(header, "Content-Length")) { chunked_ = false; contentLength_ = atoi(value); + } else if (boost::istarts_with(header, "Connection")) { + std::vector options; + boost::split(options, value, boost::is_any_of(",")); + for (const string& option : options) { + if (boost::iequals(boost::trim_copy(option), "close")) { + closeAfterResponse_ = true; + break; + } + } } } bool THttpClient::parseStatusLine(char* status) { + closeAfterResponse_ = false; char* http = status; char* code = strchr(http, ' '); @@ -107,6 +127,13 @@ bool THttpClient::parseStatusLine(char* status) { void THttpClient::flush() { resetConsumedMessageSize(); + uint8_t* buf; + uint32_t len; + writeBuffer_.getBuffer(&buf, &len); + if (len == 0) { + return; + } + if (onewayResponsePending_) { if (transport_->isOpen()) { drainPendingOnewayResponse(); @@ -115,15 +142,14 @@ void THttpClient::flush() { } } + if (closeAfterResponse_) { + close(); + } + if (!transport_->isOpen()) { transport_->open(); } - // Fetch the contents of the write buffer - uint8_t* buf; - uint32_t len; - writeBuffer_.getBuffer(&buf, &len); - // Construct the HTTP header std::ostringstream h; h << "POST " << path_ << " HTTP/1.1" << CRLF << "Host: " << host_ << CRLF diff --git a/lib/cpp/src/thrift/transport/THttpClient.h b/lib/cpp/src/thrift/transport/THttpClient.h index b8c14c4a6e2..96e6e1fa10f 100644 --- a/lib/cpp/src/thrift/transport/THttpClient.h +++ b/lib/cpp/src/thrift/transport/THttpClient.h @@ -65,6 +65,7 @@ class THttpClient : public THttpTransport { std::string host_; std::string path_; bool onewayResponsePending_; + bool closeAfterResponse_; void parseHeader(char* header) override; bool parseStatusLine(char* status) override; diff --git a/lib/cpp/test/OneWayHTTPTest.cpp b/lib/cpp/test/OneWayHTTPTest.cpp index 2c545678d59..cdfb00ea811 100644 --- a/lib/cpp/test/OneWayHTTPTest.cpp +++ b/lib/cpp/test/OneWayHTTPTest.cpp @@ -17,22 +17,23 @@ * under the License. */ +#include "gen-cpp/OneWayService.h" #include #include -#include #include -#include +#include +#include +#include #include #include #include #include -#include +#include #include +#include #include #include -#include -#include -#include "gen-cpp/OneWayService.h" +#include BOOST_AUTO_TEST_SUITE(OneWayHTTPTest) @@ -126,13 +127,61 @@ class TServerReadyEventHandler : public TServerEventHandler, public Monitor { return nullptr; } bool isListening() const { return isListening_; } - uint64_t acceptedCount() const { return accepted_; } + uint64_t acceptedCount() { + Synchronized sync(*this); + return accepted_; + } private: bool isListening_; uint64_t accepted_; }; +class TClosingHttpServer : public THttpServer { +public: + explicit TClosingHttpServer(std::shared_ptr transport) + : THttpServer(transport, transport->getConfiguration()) {} + + void flush() override { + resetConsumedMessageSize(); + + uint8_t* buf; + uint32_t len; + writeBuffer_.getBuffer(&buf, &len); + + std::ostringstream header; + header << "HTTP/1.1 200 OK" << CRLF << "Content-Type: application/x-thrift" << CRLF + << "Transfer-Encoding: chunked" << CRLF << "Connection: keep-alive, close" << CRLF + << "Connection: keep-alive" << CRLF << CRLF; + const string headerText = header.str(); + transport_->write(reinterpret_cast(headerText.data()), + static_cast(headerText.size())); + + if (len > 0) { + std::ostringstream chunkSize; + chunkSize << std::hex << len << CRLF; + const string chunkPrefix = chunkSize.str(); + transport_->write(reinterpret_cast(chunkPrefix.data()), + static_cast(chunkPrefix.size())); + transport_->write(buf, len); + transport_->write(reinterpret_cast(CRLF), CRLF_LEN); + } + transport_->write(reinterpret_cast("0\r\n\r\n"), 5); + transport_->flush(); + + writeBuffer_.resetBuffer(); + readHeaders_ = true; + close(); + } +}; + +class TClosingHttpServerTransportFactory : public apache::thrift::transport::TTransportFactory { +public: + std::shared_ptr getTransport(std::shared_ptr transport) override { + return std::make_shared(transport); + } +}; + class TBlockableBufferedTransport : public TBufferedTransport { public: TBlockableBufferedTransport(std::shared_ptr transport) @@ -286,4 +335,48 @@ BOOST_AUTO_TEST_CASE( JSON_HTTP_OneWayWrapperDoesNotPoisonNextCall ) thread.join(); } +BOOST_AUTO_TEST_CASE(HTTP_ClientReconnectsAfterConnectionClose) { + std::shared_ptr ss = std::make_shared(0); + TThreadedServer server(std::make_shared( + std::make_shared()), + ss, std::make_shared(), + std::make_shared()); + + std::shared_ptr pEventHandler(new TServerReadyEventHandler); + server.setServerEventHandler(pEventHandler); + + RPC0ThreadClass t(server); + boost::thread thread(&RPC0ThreadClass::Run, &t); + + { + Synchronized sync(*(pEventHandler.get())); + while (!pEventHandler->isListening()) { + pEventHandler->wait(); + } + } + + { + std::shared_ptr socket(new TSocket("localhost", ss->getPort())); + socket->setRecvTimeout(10000); + std::shared_ptr httpTransport(new THttpClient(socket, "localhost", "/service")); + std::shared_ptr transport(new TBufferedTransport(httpTransport)); + std::shared_ptr protocol(new TBinaryProtocol(transport)); + onewaytest::OneWayServiceClient client(protocol); + + transport->open(); + client.roundTripRPC(); + BOOST_CHECK_EQUAL(pEventHandler->acceptedCount(), 1U); + BOOST_CHECK_NO_THROW(client.roundTripRPC()); + BOOST_CHECK_EQUAL(pEventHandler->acceptedCount(), 2U); + client.oneWayRPC(); + BOOST_CHECK_NO_THROW(client.roundTripRPC()); + BOOST_CHECK_EQUAL(pEventHandler->acceptedCount(), 4U); + transport->close(); + BOOST_CHECK_EQUAL(pEventHandler->acceptedCount(), 4U); + } + + server.stop(); + thread.join(); +} + BOOST_AUTO_TEST_SUITE_END() From c356a06311030c2c6e5ee0853773e04874247e3a Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 27 Aug 2026 07:41:41 -0400 Subject: [PATCH 2/5] THRIFT-6060: Fix the Windows test build Signed-off-by: 1fanwang <1fannnw@gmail.com> --- lib/cpp/test/OneWayHTTPTest.cpp | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/lib/cpp/test/OneWayHTTPTest.cpp b/lib/cpp/test/OneWayHTTPTest.cpp index cdfb00ea811..4867b2339c1 100644 --- a/lib/cpp/test/OneWayHTTPTest.cpp +++ b/lib/cpp/test/OneWayHTTPTest.cpp @@ -150,21 +150,23 @@ class TClosingHttpServer : public THttpServer { writeBuffer_.getBuffer(&buf, &len); std::ostringstream header; - header << "HTTP/1.1 200 OK" << CRLF << "Content-Type: application/x-thrift" << CRLF - << "Transfer-Encoding: chunked" << CRLF << "Connection: keep-alive, close" << CRLF - << "Connection: keep-alive" << CRLF << CRLF; + header << "HTTP/1.1 200 OK\r\n" + << "Content-Type: application/x-thrift\r\n" + << "Transfer-Encoding: chunked\r\n" + << "Connection: keep-alive, close\r\n" + << "Connection: keep-alive\r\n\r\n"; const string headerText = header.str(); transport_->write(reinterpret_cast(headerText.data()), static_cast(headerText.size())); if (len > 0) { std::ostringstream chunkSize; - chunkSize << std::hex << len << CRLF; + chunkSize << std::hex << len << "\r\n"; const string chunkPrefix = chunkSize.str(); transport_->write(reinterpret_cast(chunkPrefix.data()), static_cast(chunkPrefix.size())); transport_->write(buf, len); - transport_->write(reinterpret_cast(CRLF), CRLF_LEN); + transport_->write(reinterpret_cast("\r\n"), 2); } transport_->write(reinterpret_cast("0\r\n\r\n"), 5); transport_->flush(); From a34449a9113ca1590c19cdabec8008f199ae4ad1 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 27 Aug 2026 16:32:10 -0400 Subject: [PATCH 3/5] THRIFT-6060: Match Connection response header exactly Signed-off-by: 1fanwang <1fannnw@gmail.com> --- lib/cpp/src/thrift/transport/THttpClient.cpp | 2 +- lib/cpp/test/OneWayHTTPTest.cpp | 19 +++++++++++++++++++ 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/lib/cpp/src/thrift/transport/THttpClient.cpp b/lib/cpp/src/thrift/transport/THttpClient.cpp index 5e1bd3c95e4..de6092b5342 100644 --- a/lib/cpp/src/thrift/transport/THttpClient.cpp +++ b/lib/cpp/src/thrift/transport/THttpClient.cpp @@ -82,7 +82,7 @@ void THttpClient::parseHeader(char* header) { } else if (boost::istarts_with(header, "Content-Length")) { chunked_ = false; contentLength_ = atoi(value); - } else if (boost::istarts_with(header, "Connection")) { + } else if (boost::iequals(string(header, colon), "Connection")) { std::vector options; boost::split(options, value, boost::is_any_of(",")); for (const string& option : options) { diff --git a/lib/cpp/test/OneWayHTTPTest.cpp b/lib/cpp/test/OneWayHTTPTest.cpp index 4867b2339c1..92382545667 100644 --- a/lib/cpp/test/OneWayHTTPTest.cpp +++ b/lib/cpp/test/OneWayHTTPTest.cpp @@ -62,6 +62,18 @@ namespace utf = boost::unit_test; // Define this env var to enable some logging (in case you need to debug) #undef ENABLE_STDERR_LOGGING +class TInspectableHttpClient : public THttpClient { +public: + explicit TInspectableHttpClient(std::shared_ptr transport) : THttpClient(transport) {} + + bool closesAfterHeader(const string& header) { + std::vector buffer(header.begin(), header.end()); + buffer.push_back('\0'); + parseHeader(buffer.data()); + return closeAfterResponse_; + } +}; + class OneWayServiceHandler : public onewaytest::OneWayServiceIf { public: OneWayServiceHandler() = default; @@ -381,4 +393,11 @@ BOOST_AUTO_TEST_CASE(HTTP_ClientReconnectsAfterConnectionClose) { thread.join(); } +BOOST_AUTO_TEST_CASE(HTTP_ClientRequiresExactConnectionHeaderName) { + TInspectableHttpClient client(std::make_shared()); + + BOOST_CHECK(!client.closesAfterHeader("Connection-Timeout: close")); + BOOST_CHECK(client.closesAfterHeader("Connection: keep-alive, close")); +} + BOOST_AUTO_TEST_SUITE_END() From ddbd0aaf141bf6d45911b922a1410190be630bc3 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 27 Aug 2026 16:42:18 -0400 Subject: [PATCH 4/5] THRIFT-6060: Reset test client header state Signed-off-by: 1fanwang <1fannnw@gmail.com> --- lib/cpp/test/OneWayHTTPTest.cpp | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/lib/cpp/test/OneWayHTTPTest.cpp b/lib/cpp/test/OneWayHTTPTest.cpp index 92382545667..6b79a670847 100644 --- a/lib/cpp/test/OneWayHTTPTest.cpp +++ b/lib/cpp/test/OneWayHTTPTest.cpp @@ -67,6 +67,7 @@ class TInspectableHttpClient : public THttpClient { explicit TInspectableHttpClient(std::shared_ptr transport) : THttpClient(transport) {} bool closesAfterHeader(const string& header) { + closeAfterResponse_ = false; std::vector buffer(header.begin(), header.end()); buffer.push_back('\0'); parseHeader(buffer.data()); @@ -396,8 +397,8 @@ BOOST_AUTO_TEST_CASE(HTTP_ClientReconnectsAfterConnectionClose) { BOOST_AUTO_TEST_CASE(HTTP_ClientRequiresExactConnectionHeaderName) { TInspectableHttpClient client(std::make_shared()); - BOOST_CHECK(!client.closesAfterHeader("Connection-Timeout: close")); BOOST_CHECK(client.closesAfterHeader("Connection: keep-alive, close")); + BOOST_CHECK(!client.closesAfterHeader("Connection-Timeout: close")); } BOOST_AUTO_TEST_SUITE_END() From 334671a3f1f15d85e242cb934743be418bec0715 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Fri, 28 Aug 2026 00:34:21 -0400 Subject: [PATCH 5/5] THRIFT-6060: Match every response header exactly Signed-off-by: 1fanwang <1fannnw@gmail.com> --- lib/cpp/src/thrift/transport/THttpClient.cpp | 7 ++++--- lib/cpp/test/OneWayHTTPTest.cpp | 20 ++++++++++++++++++++ 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/lib/cpp/src/thrift/transport/THttpClient.cpp b/lib/cpp/src/thrift/transport/THttpClient.cpp index de6092b5342..dc85a0e82c8 100644 --- a/lib/cpp/src/thrift/transport/THttpClient.cpp +++ b/lib/cpp/src/thrift/transport/THttpClient.cpp @@ -75,14 +75,15 @@ void THttpClient::parseHeader(char* header) { } char* value = colon + 1; - if (boost::istarts_with(header, "Transfer-Encoding")) { + const string name(header, colon); + if (boost::iequals(name, "Transfer-Encoding")) { if (boost::iends_with(value, "chunked")) { chunked_ = true; } - } else if (boost::istarts_with(header, "Content-Length")) { + } else if (boost::iequals(name, "Content-Length")) { chunked_ = false; contentLength_ = atoi(value); - } else if (boost::iequals(string(header, colon), "Connection")) { + } else if (boost::iequals(name, "Connection")) { std::vector options; boost::split(options, value, boost::is_any_of(",")); for (const string& option : options) { diff --git a/lib/cpp/test/OneWayHTTPTest.cpp b/lib/cpp/test/OneWayHTTPTest.cpp index 6b79a670847..643cdb333e0 100644 --- a/lib/cpp/test/OneWayHTTPTest.cpp +++ b/lib/cpp/test/OneWayHTTPTest.cpp @@ -73,6 +73,22 @@ class TInspectableHttpClient : public THttpClient { parseHeader(buffer.data()); return closeAfterResponse_; } + + bool chunksAfterHeader(const string& header) { + chunked_ = false; + std::vector buffer(header.begin(), header.end()); + buffer.push_back('\0'); + parseHeader(buffer.data()); + return chunked_; + } + + uint32_t contentLengthAfterHeader(const string& header) { + contentLength_ = 0; + std::vector buffer(header.begin(), header.end()); + buffer.push_back('\0'); + parseHeader(buffer.data()); + return contentLength_; + } }; class OneWayServiceHandler : public onewaytest::OneWayServiceIf { @@ -399,6 +415,10 @@ BOOST_AUTO_TEST_CASE(HTTP_ClientRequiresExactConnectionHeaderName) { BOOST_CHECK(client.closesAfterHeader("Connection: keep-alive, close")); BOOST_CHECK(!client.closesAfterHeader("Connection-Timeout: close")); + BOOST_CHECK(client.chunksAfterHeader("Transfer-Encoding: chunked")); + BOOST_CHECK(!client.chunksAfterHeader("Transfer-Encoding-Other: chunked")); + BOOST_CHECK_EQUAL(client.contentLengthAfterHeader("Content-Length: 42"), 42U); + BOOST_CHECK_EQUAL(client.contentLengthAfterHeader("Content-Length-Mismatch: 42"), 0U); } BOOST_AUTO_TEST_SUITE_END()