Support 100-continue of server and remove expect header of request

src/brpc/details/http_message.cpp

@@ -559,6 +559,11 @@ void MakeRawHttpRequest(butil::IOBuf* request,
         os << "Content-Length: " << (content ? content->length() : 0)
            << BRPC_CRLF;
     }
+    // `Expect: 100-continue' is not supported, remove it.
+    const std::string* expect = h->GetHeader("Expect");
+    if (expect && *expect == "100-continue") {
+        h->RemoveHeader("Expect");
+    }
     //rfc 7230#section-5.4:
     //A client MUST send a Host header field in all HTTP/1.1 request
     //messages. If the authority component is missing or undefined for

src/brpc/policy/http_rpc_protocol.cpp

@@ -127,6 +127,9 @@ CommonStrings::CommonStrings()
     , AUTHORIZATION("authorization")
     , ACCEPT_ENCODING("accept-encoding")
     , CONTENT_ENCODING("content-encoding")
+    , CONTENT_LENGTH("content_length")
+    , EXPECT("expect")
+    , CONTINUE_100("100-continue")
     , GZIP("gzip")
     , CONNECTION("connection")
     , KEEP_ALIVE("keep-alive")
@@ -1168,6 +1171,29 @@ ParseResult ParseHttpMessage(butil::IOBuf *source, Socket *socket,
             }
             return result;
         } else if (http_imsg->stage() >= HTTP_ON_HEADERS_COMPLETE) {
+            // https://datatracker.ietf.org/doc/html/rfc7231#section-5.1.1
+            // A server that receives a 100-continue expectation in an HTTP/1.0
+            // request MUST ignore that expectation.
+            //
+            // A server MAY omit sending a 100 (Continue) response if it has
+            // already received some or all of the message body for the
+            // corresponding request, or if the framing indicates that there is
+            // no message body.
+            if (http_imsg->parser().type == HTTP_REQUEST &&
+                !http_imsg->header().before_http_1_1()) {
+                const std::string* expect = http_imsg->header().GetHeader(common->EXPECT);
+                if (expect && *expect ==  common->CONTINUE_100) {
+                    // Send 100-continue response back.
+                    butil::IOBuf resp;
+                    HttpHeader header;
+                    header.set_status_code(HTTP_STATUS_CONTINUE);
+                    MakeRawHttpResponse(&resp, &header, NULL);
+                    Socket::WriteOptions wopt;
+                    wopt.ignore_eovercrowded = true;
+                    socket->Write(&resp, &wopt);
+                }
+            }
+
             http_imsg->CheckProgressiveRead(arg, socket);
             if (socket->is_read_progressive()) {
                 // header part of a progressively-read http message is complete,

src/brpc/policy/http_rpc_protocol.h

@@ -43,6 +43,8 @@ struct CommonStrings {
     std::string ACCEPT_ENCODING;
     std::string CONTENT_ENCODING;
     std::string CONTENT_LENGTH;
+    std::string EXPECT;
+    std::string CONTINUE_100;
     std::string GZIP;
     std::string CONNECTION;
     std::string KEEP_ALIVE;

test/brpc_http_rpc_protocol_unittest.cpp

@@ -24,11 +24,17 @@
 #include <string>
 #include <sys/ioctl.h>
 #include <sys/socket.h>
+#include <butil/build_config.h> // OS_MACOSX
+#if defined(OS_MACOSX)
+#include <sys/event.h>
+#endif
 #include <gtest/gtest.h>
 #include <gflags/gflags.h>
 #include <google/protobuf/text_format.h>
 #include <unistd.h>
 #include <butil/strings/string_number_conversions.h>
+#include <brpc/policy/http_rpc_protocol.h>
+#include <butil/base64.h>
 #include "brpc/http_method.h"
 #include "butil/iobuf.h"
 #include "butil/logging.h"
@@ -47,6 +53,7 @@
 #include "json2pb/json_to_pb.h"
 #include "brpc/details/method_status.h"
 #include "brpc/rpc_dump.h"
+#include "bthread/unstable.h"
 
 namespace brpc {
 DECLARE_bool(rpc_dump);
@@ -1785,8 +1792,7 @@ class HttpServiceImpl : public ::test::HttpService {
         ::test::HttpResponse*,
         ::google::protobuf::Closure* done) override {
         brpc::ClosureGuard done_guard(done);
-        brpc::Controller* cntl =
-            static_cast<brpc::Controller*>(cntl_base);
+        brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
         ASSERT_EQ(cntl->http_request().method(), brpc::HTTP_METHOD_HEAD);
         const std::string* index = cntl->http_request().GetHeader("x-db-index");
         ASSERT_NE(nullptr, index);
@@ -1801,6 +1807,19 @@ class HttpServiceImpl : public ::test::HttpService {
                 EXP_RESPONSE_TRANSFER_ENCODING);
         }
     }
+
+    void Expect(::google::protobuf::RpcController* cntl_base,
+        const ::test::HttpRequest*,
+        ::test::HttpResponse*,
+        ::google::protobuf::Closure* done) override {
+        brpc::ClosureGuard done_guard(done);
+        brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
+        const std::string* expect = cntl->http_request().GetHeader("Expect");
+        ASSERT_TRUE(expect != NULL);
+        ASSERT_EQ("100-continue", *expect);
+        ASSERT_EQ(cntl->http_request().method(), brpc::HTTP_METHOD_POST);
+        cntl->response_attachment().append("world");
+    }
 };
 
 TEST_F(HttpTest, http_head) {
@@ -1836,4 +1855,148 @@ TEST_F(HttpTest, http_head) {
     }
 }
 
+#define BRPC_CRLF "\r\n"
+
+void MakeHttpRequestHeaders(butil::IOBuf* out,
+                            brpc::HttpHeader* h,
+                            const butil::EndPoint& remote_side) {
+    butil::IOBufBuilder os;
+    os << HttpMethod2Str(h->method()) << ' ';
+    const brpc::URI& uri = h->uri();
+    uri.PrintWithoutHost(os); // host is sent by "Host" header.
+    os << " HTTP/" << h->major_version() << '.'
+       << h->minor_version() << BRPC_CRLF;
+    //rfc 7230#section-5.4:
+    //A client MUST send a Host header field in all HTTP/1.1 request
+    //messages. If the authority component is missing or undefined for
+    //the target URI, then a client MUST send a Host header field with an
+    //empty field-value.
+    //rfc 7231#sec4.3:
+    //the request-target consists of only the host name and port number of
+    //the tunnel destination, seperated by a colon. For example,
+    //Host: server.example.com:80
+    if (h->GetHeader("host") == NULL) {
+        os << "Host: ";
+        if (!uri.host().empty()) {
+            os << uri.host();
+            if (uri.port() >= 0) {
+                os << ':' << uri.port();
+            }
+        } else if (remote_side.port != 0) {
+            os << remote_side;
+        }
+        os << BRPC_CRLF;
+    }
+    if (!h->content_type().empty()) {
+        os << "Content-Type: " << h->content_type()
+           << BRPC_CRLF;
+    }
+    for (brpc::HttpHeader::HeaderIterator it = h->HeaderBegin();
+         it != h->HeaderEnd(); ++it) {
+        os << it->first << ": " << it->second << BRPC_CRLF;
+    }
+    if (h->GetHeader("Accept") == NULL) {
+        os << "Accept: */*" BRPC_CRLF;
+    }
+    // The fake "curl" user-agent may let servers return plain-text results.
+    if (h->GetHeader("User-Agent") == NULL) {
+        os << "User-Agent: brpc/1.0 curl/7.0" BRPC_CRLF;
+    }
+    const std::string& user_info = h->uri().user_info();
+    if (!user_info.empty() && h->GetHeader("Authorization") == NULL) {
+        // NOTE: just assume user_info is well formatted, namely
+        // "<user_name>:<password>". Users are very unlikely to add extra
+        // characters in this part and even if users did, most of them are
+        // invalid and rejected by http_parser_parse_url().
+        std::string encoded_user_info;
+        butil::Base64Encode(user_info, &encoded_user_info);
+        os << "Authorization: Basic " << encoded_user_info << BRPC_CRLF;
+    }
+    os << BRPC_CRLF;  // CRLF before content
+    os.move_to(*out);
+}
+
+#undef BRPC_CRLF
+
+void ReadOneResponse(brpc::SocketUniquePtr& sock,
+    brpc::DestroyingPtr<brpc::policy::HttpContext>& imsg_guard) {
+#if defined(OS_LINUX)
+    ASSERT_EQ(0, bthread_fd_wait(sock->fd(), EPOLLIN));
+#elif defined(OS_MACOSX)
+    ASSERT_EQ(0, bthread_fd_wait(sock->fd(), EVFILT_READ));
+#endif
+
+    butil::IOPortal read_buf;
+    int64_t start_time = butil::gettimeofday_us();
+    while (true) {
+        const ssize_t nr = read_buf.append_from_file_descriptor(sock->fd(), 4096);
+        LOG(INFO) << "nr=" << nr;
+        LOG(INFO) << butil::ToPrintableString(read_buf);
+        ASSERT_TRUE(nr > 0 || (nr < 0 && errno == EAGAIN));
+        if (errno == EAGAIN) {
+            ASSERT_LT(butil::gettimeofday_us(), start_time + 1000000L) << "Too long!";
+            bthread_usleep(1000);
+            continue;
+        }
+        brpc::ParseResult pr = brpc::policy::ParseHttpMessage(&read_buf, sock.get(), false, NULL);
+        ASSERT_TRUE(pr.error() == brpc::PARSE_ERROR_NOT_ENOUGH_DATA || pr.is_ok());
+        if (pr.is_ok()) {
+            imsg_guard.reset(static_cast<brpc::policy::HttpContext*>(pr.message()));
+            break;
+        }
+    }
+    ASSERT_TRUE(read_buf.empty());
+}
+
+TEST_F(HttpTest, http_expect) {
+    const int port = 8923;
+    brpc::Server server;
+    HttpServiceImpl svc;
+    EXPECT_EQ(0, server.AddService(&svc, brpc::SERVER_DOESNT_OWN_SERVICE));
+    EXPECT_EQ(0, server.Start(port, NULL));
+
+    butil::EndPoint ep;
+    ASSERT_EQ(0, butil::str2endpoint("127.0.0.1:8923", &ep));
+    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");
+    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);
+    LOG(INFO) << butil::ToPrintableString(header_buf);
+    butil::IOBuf request_buf(header_buf);
+    request_buf.append(content);
+
+    ASSERT_EQ(0, sock->Write(&header_buf));
+    int64_t start_time = butil::gettimeofday_us();
+    while (sock->fd() < 0) {
+        bthread_usleep(1000);
+        ASSERT_LT(butil::gettimeofday_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);
+
+    ASSERT_EQ(0, sock->Write(&content));
+    // 200 Ok
+    ReadOneResponse(sock, imsg_guard);
+    ASSERT_EQ(imsg_guard->header().status_code(), brpc::HTTP_STATUS_OK);
+
+    ASSERT_EQ(0, sock->Write(&request_buf));
+    // 200 Ok
+    ReadOneResponse(sock, imsg_guard);
+    ASSERT_EQ(imsg_guard->header().status_code(), brpc::HTTP_STATUS_OK);
+}
+
 } //namespace

test/echo.proto

@@ -90,6 +90,7 @@ service NacosNamingService {
 
 service HttpService {
     rpc Head(HttpRequest) returns (HttpResponse);
+    rpc Expect(HttpRequest) returns (HttpResponse);
 }
 
 enum State0 {