This is an automated email from the ASF dual-hosted git repository.
chenBright pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new 7c7ad123 answer Expect: 100-continue once per request in
ParseHttpMessage (#3704)
7c7ad123 is described below
commit 7c7ad123d12baabb0dbcbdf35bcb1fb39fcb1fe7
Author: sahvx655-wq <[email protected]>
AuthorDate: Tue Nov 3 17:29:00 2026 +0530
answer Expect: 100-continue once per request in ParseHttpMessage (#3704)
---
src/brpc/policy/http_rpc_protocol.cpp | 7 ++++-
src/brpc/policy/http_rpc_protocol.h | 8 +++++
test/brpc_http_rpc_protocol_unittest.cpp | 50 ++++++++++++++++++++++++++++++++
3 files changed, 64 insertions(+), 1 deletion(-)
diff --git a/src/brpc/policy/http_rpc_protocol.cpp
b/src/brpc/policy/http_rpc_protocol.cpp
index dd9abf05..e4bac8ab 100644
--- a/src/brpc/policy/http_rpc_protocol.cpp
+++ b/src/brpc/policy/http_rpc_protocol.cpp
@@ -1309,8 +1309,13 @@ ParseResult ParseHttpMessage(butil::IOBuf *source,
Socket *socket,
// already received some or all of the message body for the
// corresponding request, or if the framing indicates that there is
// no message body.
+ //
+ // This branch is reached again for every read that leaves the
+ // body incomplete, so answer the expectation on the first one
only.
if (http_imsg->parser().type == HTTP_REQUEST &&
- !http_imsg->header().before_http_1_1()) {
+ !http_imsg->header().before_http_1_1() &&
+ !http_imsg->expect_handled()) {
+ http_imsg->set_expect_handled();
const std::string* expect =
http_imsg->header().GetHeader(common->EXPECT);
if (expect && *expect == common->CONTINUE_100) {
// Send 100-continue response back.
diff --git a/src/brpc/policy/http_rpc_protocol.h
b/src/brpc/policy/http_rpc_protocol.h
index 92d0cae7..862f0200 100644
--- a/src/brpc/policy/http_rpc_protocol.h
+++ b/src/brpc/policy/http_rpc_protocol.h
@@ -98,6 +98,7 @@ public:
: InputMessageBase()
, HttpMessage(read_body_progressively, request_method)
, _is_stage2(false)
+ , _expect_handled(false)
, _socket_id(INVALID_SOCKET_ID) {
// add one ref for Destroy
butil::intrusive_ptr<HttpContext>(this).detach();
@@ -115,6 +116,12 @@ public:
// True if AddOneRefForStage2() was ever called.
bool is_stage2() const { return _is_stage2; }
+ // True once ParseHttpMessage() has dealt with the `Expect' header of this
+ // request. The parser is re-entered on every read while the body is still
+ // incomplete; the expectation is answered on the first of them only.
+ bool expect_handled() const { return _expect_handled; }
+ void set_expect_handled() { _expect_handled = true; }
+
// @InputMessageBase
void DestroyImpl() override {
RemoveOneRefForStage2();
@@ -129,6 +136,7 @@ public:
private:
bool _is_stage2;
+ bool _expect_handled;
SocketId _socket_id;
};
diff --git a/test/brpc_http_rpc_protocol_unittest.cpp
b/test/brpc_http_rpc_protocol_unittest.cpp
index af2b43e5..f3e07f4a 100644
--- a/test/brpc_http_rpc_protocol_unittest.cpp
+++ b/test/brpc_http_rpc_protocol_unittest.cpp
@@ -3154,6 +3154,56 @@ TEST_F(HttpTest, http_expect) {
ASSERT_EQ(imsg_guard->header().status_code(), brpc::HTTP_STATUS_OK);
}
+TEST_F(HttpTest, http_expect_answered_once) {
+ brpc::Server server;
+ HttpServiceImpl svc;
+ ASSERT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+ ASSERT_EQ(0, server.Start(0, nullptr));
+
+ const butil::EndPoint ep = server.listen_address();
+ brpc::SocketOptions options;
+ options.remote_side = ep;
+ brpc::SocketId id;
+ ASSERT_EQ(0, brpc::Socket::Create(options, &id));
+ brpc::SocketUniquePtr sock;
+ ASSERT_EQ(0, brpc::Socket::Address(id, &sock));
+
+ butil::IOBuf content;
+ content.append("hello world");
+ brpc::HttpHeader header;
+ header.set_method(brpc::HTTP_METHOD_POST);
+ header.uri().set_path("/HttpService/Expect");
+ header.SetHeader("Expect", "100-continue");
+ header.SetHeader("Content-Length", std::to_string(content.size()));
+ butil::IOBuf header_buf;
+ MakeHttpRequestHeaders(&header_buf, &header, ep);
+
+ ASSERT_EQ(0, sock->Write(&header_buf));
+ int64_t start_time = butil::cpuwide_time_us();
+ while (sock->fd() < 0) {
+ bthread_usleep(1000);
+ ASSERT_LT(butil::cpuwide_time_us(), start_time + 1000000L) << "Too
long!";
+ }
+ // 100 Continue
+ brpc::DestroyingPtr<brpc::policy::HttpContext> imsg_guard;
+ ReadOneResponse(sock, imsg_guard);
+ ASSERT_EQ(imsg_guard->header().status_code(), brpc::HTTP_STATUS_CONTINUE);
+
+ // Send the body in pieces with a pause in between, so that the server
+ // gets each piece in a separate read. The expectation was already
+ // answered above, so no piece may be answered with another 100 Continue.
+ while (content.size() > 1) {
+ butil::IOBuf piece;
+ content.cutn(&piece, content.size() / 2);
+ ASSERT_EQ(0, sock->Write(&piece));
+ bthread_usleep(50 * 1000);
+ }
+ ASSERT_EQ(0, sock->Write(&content));
+ // 200 Ok
+ ReadOneResponse(sock, imsg_guard);
+ ASSERT_EQ(imsg_guard->header().status_code(), brpc::HTTP_STATUS_OK);
+}
+
TEST_F(HttpTest, pipelined_requests_get_one_response_each) {
brpc::Server server;
server.set_version("pipelined");
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
