Merge 'Add HTTP/1.1 100 Continue' from Wojciech

This series adds a missing feature of HTTP 1.1: the "Expect" header
with value "100-continue" and a corresponding reply with status code
"100 Continue"

Test: unit(dev)
Fixes: scylladb#768
Also available on https://github.com/wmitros/seastar/tree/100-cont

    * '100-cont' of github.com:wmitros/seastar:
      tests: httpd: add test for the 100 Continue status code
      http: add "Expect: 100-continue" handling

* '100-cont' of github.com:wmitros/seastar:
  tests: httpd: add test for the 100 Continue status code
  http: add "Expect: 100-continue" handling

include/seastar/http/reply.hh

@@ -53,6 +53,7 @@ struct reply {
      * The status of the reply.
      */
     enum class status_type {
+        continue_ = 100, //!< continue
         ok = 200, //!< ok
         created = 201, //!< created
         accepted = 202, //!< accepted

src/http/httpd.cc

@@ -236,11 +236,29 @@ future<> connection::read_one() {
             generate_error_reply_and_close(std::move(req), reply::status_type::payload_too_large, std::move(msg));
             return make_ready_future<>();
         }
-        return read_request_body(_read_buf, std::move(req)).then([this] (std::unique_ptr<httpd::request> req) {
-            return _replies.not_full().then([req = std::move(req), this] () mutable {
-                return generate_reply(std::move(req));
-            }).then([this](bool done) {
-                _done = done;
+
+        auto maybe_reply_continue = [this, req = std::move(req)] () mutable {
+            if (req->_version == "1.1" && req->get_header("Expect") == "100-continue") {
+                auto continue_reply = std::make_unique<reply>();
+                set_headers(*continue_reply);
+                continue_reply->set_version(req->_version);
+                continue_reply->set_status(reply::status_type::continue_).done();
+                return _replies.not_full().then([continue_reply = std::move(continue_reply), req = std::move(req), this] () mutable {
+                    this->_replies.push(std::move(continue_reply));
+                    return make_ready_future<std::unique_ptr<httpd::request>>(std::move(req));
+                });
+            } else {
+                return make_ready_future<std::unique_ptr<httpd::request>>(std::move(req));
+            }
+        };
+
+        return maybe_reply_continue().then([this] (std::unique_ptr<httpd::request> req) {
+            return read_request_body(_read_buf, std::move(req)).then([this] (std::unique_ptr<httpd::request> req) {
+                return _replies.not_full().then([req = std::move(req), this] () mutable {
+                    return generate_reply(std::move(req));
+                }).then([this](bool done) {
+                    _done = done;
+                });
             });
         });
     });

src/http/reply.cc

@@ -38,6 +38,7 @@ namespace httpd {
 
 namespace status_strings {
 
+const sstring continue_ = " 100 Continue\r\n";
 const sstring ok = " 200 OK\r\n";
 const sstring created = " 201 Created\r\n";
 const sstring accepted = " 202 Accepted\r\n";
@@ -59,6 +60,8 @@ const sstring service_unavailable = " 503 Service Unavailable\r\n";
 
 static const sstring& to_string(reply::status_type status) {
     switch (status) {
+    case reply::status_type::continue_:
+        return continue_;
     case reply::status_type::ok:
         return ok;
     case reply::status_type::created:

tests/unit/httpd_test.cc

@@ -775,6 +775,75 @@ SEASTAR_TEST_CASE(content_length_limit) {
     });
 }
 
+SEASTAR_TEST_CASE(test_100_continue) {
+    return seastar::async([] {
+        loopback_connection_factory lcf;
+        http_server server("test");
+        server.set_content_length_limit(11);
+        loopback_socket_impl lsi(lcf);
+        httpd::http_server_tester::listeners(server).emplace_back(lcf.get_server_socket());
+        future<> client = seastar::async([&lsi, &server] {
+            connected_socket c_socket = std::get<connected_socket>(lsi.connect(socket_address(ipv4_addr()), socket_address(ipv4_addr())).get());
+            input_stream<char> input(c_socket.input());
+            output_stream<char> output(c_socket.output());
+
+            for (auto version : {sstring("1.0"), sstring("1.1")}) {
+                for (auto content : {sstring(""), sstring("xxxxxxxxxxx")}) {
+                    for (auto expect : {sstring(""), sstring("Expect: 100-continue\r\n")}) {
+                        auto content_len = content.empty() ? sstring("") : (sstring("Content-Length: ") + to_sstring(content.length()) + sstring("\r\n"));
+                        sstring req = sstring("GET /test HTTP/") + version + sstring("\r\nHost: test\r\nConnection: Keep-Alive\r\n") + content_len + expect + sstring("\r\n");
+                        output.write(req).get();
+                        output.flush().get();
+                        bool already_ok = false;
+                        if (version == "1.1" && expect.length()) {
+                            auto resp = input.read().get0();
+                            BOOST_REQUIRE_NE(std::string(resp.get(), resp.size()).find("100 Continue"), std::string::npos);
+                            already_ok = content.empty() && std::string(resp.get(), resp.size()).find("200 OK") != std::string::npos;
+                        }
+                        if (!already_ok) {
+                            //If the body is empty, the final response might have already been read
+                            output.write(content).get();
+                            output.flush().get();
+                            auto resp = input.read().get0();
+                            BOOST_REQUIRE_EQUAL(std::string(resp.get(), resp.size()).find("100 Continue"), std::string::npos);
+                            BOOST_REQUIRE_NE(std::string(resp.get(), resp.size()).find("200 OK"), std::string::npos);
+                        }
+                    }
+                }
+            }
+            output.write(sstring("GET /test HTTP/1.1\r\nHost: test\r\nContent-Length: 17\r\nExpect: 100-continue\r\n\r\n")).get();
+            output.flush().get();
+            auto resp = input.read().get0();
+            BOOST_REQUIRE_EQUAL(std::string(resp.get(), resp.size()).find("100 Continue"), std::string::npos);
+            BOOST_REQUIRE_NE(std::string(resp.get(), resp.size()).find("413 Payload Too Large"), std::string::npos);
+
+            input.close().get();
+            output.close().get();
+        });
+
+        future<> writer = seastar::async([&server] {
+            class test_handler : public handler_base {
+                std::function<future<>(output_stream<char> &&)> _write_func;
+            public:
+                test_handler(http_server& server, std::function<future<>(output_stream<char> &&)>&& write_func) : _write_func(write_func) {
+                }
+                future<std::unique_ptr<reply>> handle(const sstring& path,
+                        std::unique_ptr<request> req, std::unique_ptr<reply> rep) override {
+                    rep->write_body("json", _write_func);
+                    return make_ready_future<std::unique_ptr<reply>>(std::move(rep));
+                }
+            };
+            auto handler = new test_handler(server, json::stream_object("hello"));
+            server._routes.put(GET, "/test", handler);
+            server.do_accepts(0).get();
+        });
+
+        client.get();
+        writer.get();
+        server.stop().get();
+    });
+}
+
 SEASTAR_TEST_CASE(case_insensitive_header) {
     std::unique_ptr<seastar::httpd::request> req = std::make_unique<seastar::httpd::request>();
     req->_headers["conTEnt-LengtH"] = "17";