serializer: add custom request serializer with explicit body eof

example/nlohmann_json.cpp

@@ -51,7 +51,7 @@ tag_invoke(burl::body_from_tag<nlohmann::json>, const nlohmann::json& value)
         capy::io_task<>
         write(capy::any_buffer_sink& sink) const
         {
-            auto [ec, n] = co_await sink.write(capy::make_buffer(text_));
+            auto [ec, n] = co_await sink.write_eof(capy::make_buffer(text_));
             co_return { ec };
         }
     };

example/usage.cpp

@@ -491,7 +491,7 @@ use_proxy(corosio::tls_context tls_ctx)
     // tag::use_proxy[]
     burl::client::config cfg;
     // SOCKS5 and HTTP proxies are supported
-    cfg.proxy = urls::url("socks5h://user:pass@localhost:8080");
+    cfg.proxy = urls::url("socks5h://user:pass@localhost:1080");
 
     burl::client client(co_await capy::this_coro::executor, tls_ctx, cfg);
 

include/boost/burl/error.hpp

@@ -64,6 +64,14 @@ enum class error
     /** The proxy replied with an unsupported protocol version.
     */
     proxy_unsupported_version,
+
+    /** The request body size did not match its content length.
+
+        The number of bytes produced by the request body
+        differed from the `Content-Length` declared for the
+        request.
+    */
+    body_size_mismatch,
 };
 
 /** Error conditions corresponding to sets of error codes.

src/client.cpp

@@ -15,6 +15,7 @@
 #include "detail/connection_pool.hpp"
 #include "detail/drain_body.hpp"
 #include "detail/redirect.hpp"
+#include "detail/serializer.hpp"
 
 #include <boost/capy/buffers/make_buffer.hpp>
 #include <boost/capy/ex/execution_context.hpp>
@@ -27,7 +28,6 @@
 #include <boost/http/request.hpp>
 #include <boost/http/response_base.hpp>
 #include <boost/http/response_parser.hpp>
-#include <boost/http/serializer.hpp>
 #include <boost/http/status.hpp>
 #include <boost/http/zlib/inflate.hpp>
 
@@ -231,11 +231,8 @@ client::execute_impl(
     if(!headers.exists(field::accept_encoding))
         set_accept_encoding(parser_cfg, headers, config_);
 
-    http::serializer serializer(http::make_serializer_config({}));
-    serializer.reset();
-    serializer.set_message(headers);
-
-    http::response_parser parser(http::make_parser_config(parser_cfg));
+    http::response_parser parser(
+        http::make_parser_config(parser_cfg));
 
     auto url             = request.url;
     auto trusted         = true;
@@ -269,13 +266,16 @@ client::execute_impl(
 
         if(request.body.has_value())
         {
-            serializer.start_buffers();
-            capy::any_buffer_sink sink(serializer.sink_for(conn));
+            capy::any_write_stream ws(&conn);
+            detail::serializer sr(ws, headers);
+            capy::any_buffer_sink sink(&sr);
             if(auto [wec] = co_await request.body.write(sink); wec)
                 co_return { wec, {} };
-            // The body only writes its bytes; finalize the sink here.
-            if(auto [wec] = co_await sink.write_eof(); wec)
-                co_return { wec, {} };
+            if(!sr.is_done())
+            {
+                if(auto [wec] = co_await sr.write_eof(); wec)
+                    co_return { wec, {} };
+            }
         }
         else
         {

src/detail/connection_pool.cpp

@@ -54,11 +54,11 @@ connect_tcp(
     corosio::tcp_socket& socket,
     capy::executor_ref exec,
     const client::config& cfg,
-    std::string_view host,
-    std::string_view port)
+    urls::url_view url)
 {
     corosio::resolver resolver(exec);
-    auto [rec, eps] = co_await resolver.resolve(host, port);
+    auto [rec, eps] = co_await resolver.resolve(
+        url.encoded_host_address(), effective_port(url));
     if(rec)
         co_return rec;
 
@@ -264,6 +264,11 @@ connection_pool::release(pooled_connection pc)
 capy::io_task<std::unique_ptr<connection>>
 connection_pool::connect(urls::url_view url) const
 {
+    using urls::scheme;
+
+    if(url.scheme_id() != scheme::http && url.scheme_id() != scheme::https)
+        co_return { error::unsupported_url_scheme, {} };
+
     if(config_.connect_handler)
     {
         auto [ec, stream] = co_await config_.connect_handler(url);
@@ -273,41 +278,52 @@ connection_pool::connect(urls::url_view url) const
             {}, std::make_unique<stream_connection>(std::move(stream)) };
     }
 
-    auto target_port = effective_port(url);
-    if(target_port.empty())
-        co_return { error::unsupported_url_scheme, {} };
-
     corosio::tcp_socket socket(exec_);
 
     if(config_.proxy)
     {
         auto const& proxy = *config_.proxy;
-        auto proxy_port   = effective_port(proxy);
-        if(proxy_port.empty())
+        if(effective_port(proxy).empty())
             co_return { error::unsupported_proxy_scheme, {} };
 
-        auto [ec] = co_await connect_tcp(
-            socket, exec_, config_, proxy.encoded_host(), proxy_port);
-        if(ec)
+        if(auto [ec] = co_await connect_tcp(socket, exec_, config_, proxy); ec)
             co_return { ec, {} };
 
         if(proxy.scheme() == "http")
         {
             auto [ec] = co_await open_http_tunnel(
-                capy::any_stream(&socket),
-                url.encoded_host(),
-                target_port,
-                proxy);
+                capy::any_stream(&socket), url, proxy);
             if(ec)
                 co_return { ec, {} };
         }
-        else if(proxy.scheme() == "socks5" || proxy.scheme() == "socks5h")
+        else if(proxy.scheme() == "socks5")
         {
+            urls::url resolved;
+
+            corosio::resolver resolver(exec_);
+            auto [rec, eps] = co_await resolver.resolve(
+                url.encoded_host_address(), effective_port(url));
+            if(rec)
+                co_return { rec, {} };
+
+            auto const& ep = eps.front().get_endpoint();
+            resolved.set_port_number(ep.port());
+            if(ep.is_v4())
+                resolved.set_host_ipv4(
+                    urls::ipv4_address(ep.v4_address().to_bytes()));
+            else
+                resolved.set_host_ipv6(
+                    urls::ipv6_address(ep.v6_address().to_bytes()));
+
             auto [ec] = co_await open_socks5_tunnel(
-                capy::any_stream(&socket),
-                url.encoded_host(),
-                target_port,
-                proxy);
+                capy::any_stream(&socket), resolved, proxy);
+            if(ec)
+                co_return { ec, {} };
+        }
+        else if(proxy.scheme() == "socks5h")
+        {
+            auto [ec] = co_await open_socks5_tunnel(
+                capy::any_stream(&socket), url, proxy);
             if(ec)
                 co_return { ec, {} };
         }
@@ -318,13 +334,11 @@ connection_pool::connect(urls::url_view url) const
     }
     else
     {
-        auto [ec] = co_await connect_tcp(
-            socket, exec_, config_, url.encoded_host(), target_port);
-        if(ec)
+        if(auto [ec] = co_await connect_tcp(socket, exec_, config_, url); ec)
             co_return { ec, {} };
     }
 
-    if(url.scheme_id() == urls::scheme::https)
+    if(url.scheme_id() == scheme::https)
     {
         auto tls_ctx = tls_ctx_;
         tls_ctx.set_hostname(url.encoded_host());

src/detail/http_tunnel.cpp

@@ -12,6 +12,7 @@
 #include <boost/burl/error.hpp>
 
 #include "base64.hpp"
+#include "effective_port.hpp"
 
 #include <boost/capy/buffers/make_buffer.hpp>
 #include <boost/capy/write.hpp>
@@ -33,13 +34,12 @@ namespace detail
 capy::io_task<>
 open_http_tunnel(
     capy::any_stream stream,
-    std::string_view target_host,
-    std::string_view target_port,
+    urls::url_view target,
     urls::url_view proxy)
 {
-    std::string host_port(target_host);
+    std::string host_port(target.encoded_host());
     host_port += ':';
-    host_port += target_port;
+    host_port += effective_port(target);
 
     http::request req(http::method::connect, host_port);
     req.set(http::field::host, host_port);

src/detail/http_tunnel.hpp

@@ -14,8 +14,6 @@
 #include <boost/capy/io_task.hpp>
 #include <boost/url/url_view.hpp>
 
-#include <string_view>
-
 namespace boost
 {
 namespace burl
@@ -26,8 +24,7 @@ namespace detail
 capy::io_task<>
 open_http_tunnel(
     capy::any_stream stream,
-    std::string_view target_host,
-    std::string_view target_port,
+    urls::url_view target,
     urls::url_view proxy);
 
 } // namespace detail

src/detail/send_file.cpp

@@ -31,7 +31,8 @@ capy::io_task<>
 send_file(
     capy::any_buffer_sink& sink,
     std::filesystem::path const& path,
-    std::uint64_t size)
+    std::uint64_t size,
+    bool call_eof)
 {
     corosio::stream_file f(co_await capy::this_coro::executor);
     // TODO: switch to a non-throwing open() overload once available.
@@ -61,6 +62,11 @@ send_file(
         auto take = clamp(remaining, n);
         if(take)
         {
+            if(call_eof && take == remaining)
+            {
+                if(auto [ec] = co_await sink.commit_eof(take); ec)
+                    co_return { ec };
+            }
             if(auto [ec] = co_await sink.commit(take); ec)
                 co_return { ec };
             remaining -= take;

src/detail/send_file.hpp

@@ -34,7 +34,8 @@ capy::io_task<>
 send_file(
     capy::any_buffer_sink& sink,
     std::filesystem::path const& path,
-    std::uint64_t size);
+    std::uint64_t size,
+    bool call_eof = false);
 
 } // namespace detail
 } // namespace burl