diff --git a/doc/modules/ROOT/pages/4.guide/4d.sockets.adoc b/doc/modules/ROOT/pages/4.guide/4d.sockets.adoc index 29f6d0cc..15046aec 100644 --- a/doc/modules/ROOT/pages/4.guide/4d.sockets.adoc +++ b/doc/modules/ROOT/pages/4.guide/4d.sockets.adoc @@ -136,6 +136,78 @@ For simpler code when errors are fatal: (co_await s.connect(endpoint)).value(); // Throws on error ---- +=== Range-Based Connect + +When connecting to a hostname, the resolver may return multiple endpoints +(IPv4, IPv6, multiple A records). The free function `corosio::connect()` tries +each in order, returning on the first success: + +[source,cpp] +---- +#include + +corosio::resolver r(ioc); +auto [rec, results] = co_await r.resolve("www.boost.org", "80"); +if (rec) + co_return; + +corosio::tcp_socket s(ioc); +auto [cec, ep] = co_await corosio::connect(s, results); +if (cec) + co_return; +// `ep` is the endpoint that accepted the connection. +---- + +Between attempts the socket is closed so the next `connect()` auto-opens with +the correct address family. This lets a single call try IPv4 and IPv6 +candidates transparently. + +The signature is generic over any range whose elements convert to the +socket's endpoint type: + +[source,cpp] +---- +template + requires std::convertible_to< + std::ranges::range_reference_t, + typename Socket::endpoint_type> +capy::task> +connect(Socket& s, Range endpoints); +---- + +On success, returns the connected endpoint. On all-fail, returns the error +from the last attempt. On empty range (or when a connect condition rejects +every candidate), returns `std::errc::no_such_device_or_address`. + +==== Filtering Candidates + +A second overload accepts a predicate invoked as `cond(last_ec, ep)` before +each attempt. Returning `false` skips the candidate: + +[source,cpp] +---- +auto [ec, ep] = co_await corosio::connect( + s, + results, + [](std::error_code const&, corosio::endpoint const& e) { + return e.is_v4(); // IPv4 only. + }); +---- + +==== Iterator Overload + +An iterator-pair overload returns the iterator to the successful endpoint on +success, or `end` on failure: + +[source,cpp] +---- +auto [ec, it] = co_await corosio::connect(s, v.begin(), v.end()); +if (!ec) + std::cout << "connected to index " << (it - v.begin()) << "\n"; +---- + +Both overloads accept an optional connect condition as a trailing argument. + == Reading Data === read_some() diff --git a/include/boost/corosio/connect.hpp b/include/boost/corosio/connect.hpp new file mode 100644 index 00000000..d106fa57 --- /dev/null +++ b/include/boost/corosio/connect.hpp @@ -0,0 +1,304 @@ +// +// Copyright (c) 2026 Michael Vandeberg +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// Official repository: https://github.com/cppalliance/corosio +// + +#ifndef BOOST_COROSIO_CONNECT_HPP +#define BOOST_COROSIO_CONNECT_HPP + +#include + +#include +#include +#include + +#include +#include +#include +#include +#include + +/* + Range-based composed connect operation. + + These free functions try each endpoint in a range (or iterator pair) + in order, returning on the first successful connect. Between attempts + the socket is closed so that the next attempt can auto-open with the + correct address family (e.g. going from IPv4 to IPv6 candidates). + + The iteration semantics follow Boost.Asio's range/iterator async_connect: + on success, the successful endpoint (or its iterator) is returned; on + all-fail, the last attempt's error code is returned; on an empty range + (or when a connect_condition rejects every candidate), + std::errc::no_such_device_or_address is returned, matching the error + the resolver uses for "no results" in posix_resolver_service. + + The operation is a plain coroutine; cancellation is propagated to the + inner per-endpoint connect via the affine awaitable protocol on io_env. +*/ + +namespace boost::corosio { + +namespace detail { + +/* Always-true connect condition used by the overloads that take no + user-supplied predicate. Kept at namespace-detail scope so it has a + stable linkage name across translation units. */ +struct default_connect_condition +{ + template + bool operator()(std::error_code const&, Endpoint const&) const noexcept + { + return true; + } +}; + +} // namespace detail + +/* Forward declarations so the non-condition overloads can delegate + to the condition overloads via qualified lookup (qualified calls + bind to the overload set visible at definition, not instantiation). */ + +template + requires std::convertible_to< + std::ranges::range_reference_t, + typename Socket::endpoint_type> && + std::predicate< + ConnectCondition&, + std::error_code const&, + typename Socket::endpoint_type const&> +capy::task> +connect(Socket& s, Range endpoints, ConnectCondition cond); + +template + requires std::convertible_to< + std::iter_reference_t, + typename Socket::endpoint_type> && + std::predicate< + ConnectCondition&, + std::error_code const&, + typename Socket::endpoint_type const&> +capy::task> +connect(Socket& s, Iter begin, Iter end, ConnectCondition cond); + +/** Asynchronously connect a socket by trying each endpoint in a range. + + Each candidate is tried in order. Before each attempt the socket is + closed (so the next `connect` auto-opens with the candidate's + address family). On first successful connect, the operation + completes with the connected endpoint. + + @par Cancellation + Supports cancellation via the affine awaitable protocol. If a + per-endpoint connect completes with `capy::cond::canceled` the + operation completes immediately with that error and does not try + further endpoints. + + @param s The socket to connect. Must have a `connect(endpoint)` + member returning an awaitable, plus `close()` and `is_open()`. + If the socket is already open, it will be closed before the + first attempt. + @param endpoints A range of candidate endpoints. Taken by value + so temporaries (e.g. `resolver_results` returned from + `resolver::resolve`) remain alive for the coroutine's lifetime. + + @return An awaitable completing with + `capy::io_result`: + - on success: default error_code and the connected endpoint; + - on failure of all attempts: the error from the last attempt + and a default-constructed endpoint; + - on empty range: `std::errc::no_such_device_or_address` and a + default-constructed endpoint. + + @note The socket is closed and re-opened before each attempt, so + any socket options set by the caller (e.g. `no_delay`, + `reuse_address`) are lost. Apply options after this operation + completes. + + @throws std::system_error if auto-opening the socket fails during + an attempt (inherits the contract of `Socket::connect`). + + @par Example + @code + resolver r(ioc); + auto [rec, results] = co_await r.resolve("www.boost.org", "80"); + if (rec) co_return; + tcp_socket s(ioc); + auto [cec, ep] = co_await corosio::connect(s, results); + @endcode +*/ +template + requires std::convertible_to< + std::ranges::range_reference_t, + typename Socket::endpoint_type> +capy::task> +connect(Socket& s, Range endpoints) +{ + return corosio::connect( + s, std::move(endpoints), detail::default_connect_condition{}); +} + +/** Asynchronously connect a socket by trying each endpoint in a range, + filtered by a user-supplied condition. + + For each candidate the condition is invoked as + `cond(last_ec, ep)` where `last_ec` is the error from the most + recent attempt (default-constructed before the first attempt). If + the condition returns `false` the candidate is skipped; otherwise a + connect is attempted. + + @param s The socket to connect. See the non-condition overload for + requirements. + @param endpoints A range of candidate endpoints. + @param cond A predicate invocable with + `(std::error_code const&, typename Socket::endpoint_type const&)` + returning a value contextually convertible to `bool`. + + @return Same as the non-condition overload. If every candidate is + rejected, completes with `std::errc::no_such_device_or_address`. + + @throws std::system_error if auto-opening the socket fails. +*/ +template + requires std::convertible_to< + std::ranges::range_reference_t, + typename Socket::endpoint_type> && + std::predicate< + ConnectCondition&, + std::error_code const&, + typename Socket::endpoint_type const&> +capy::task> +connect(Socket& s, Range endpoints, ConnectCondition cond) +{ + using endpoint_type = typename Socket::endpoint_type; + + std::error_code last_ec; + + for (auto&& e : endpoints) + { + endpoint_type ep = e; + + if (!cond(static_cast(last_ec), + static_cast(ep))) + continue; + + if (s.is_open()) + s.close(); + + auto [ec] = co_await s.connect(ep); + + if (!ec) + co_return {std::error_code{}, std::move(ep)}; + + if (ec == capy::cond::canceled) + co_return {ec, endpoint_type{}}; + + last_ec = ec; + } + + if (!last_ec) + last_ec = std::make_error_code(std::errc::no_such_device_or_address); + + co_return {last_ec, endpoint_type{}}; +} + +/** Asynchronously connect a socket by trying each endpoint in an + iterator range. + + Behaves like the range overload, except the return value carries + the iterator to the successfully connected endpoint on success, or + `end` on failure. This mirrors Boost.Asio's iterator-based + `async_connect`. + + @param s The socket to connect. + @param begin The first candidate. + @param end One past the last candidate. + + @return An awaitable completing with `capy::io_result`: + - on success: default error_code and the iterator of the + successful endpoint; + - on failure of all attempts: the error from the last attempt + and `end`; + - on empty range: `std::errc::no_such_device_or_address` and + `end`. + + @throws std::system_error if auto-opening the socket fails. +*/ +template + requires std::convertible_to< + std::iter_reference_t, + typename Socket::endpoint_type> +capy::task> +connect(Socket& s, Iter begin, Iter end) +{ + return corosio::connect( + s, + std::move(begin), + std::move(end), + detail::default_connect_condition{}); +} + +/** Asynchronously connect a socket by trying each endpoint in an + iterator range, filtered by a user-supplied condition. + + @param s The socket to connect. + @param begin The first candidate. + @param end One past the last candidate. + @param cond A predicate invocable with + `(std::error_code const&, typename Socket::endpoint_type const&)`. + + @return Same as the plain iterator overload. If every candidate is + rejected, completes with `std::errc::no_such_device_or_address`. + + @throws std::system_error if auto-opening the socket fails. +*/ +template + requires std::convertible_to< + std::iter_reference_t, + typename Socket::endpoint_type> && + std::predicate< + ConnectCondition&, + std::error_code const&, + typename Socket::endpoint_type const&> +capy::task> +connect(Socket& s, Iter begin, Iter end, ConnectCondition cond) +{ + using endpoint_type = typename Socket::endpoint_type; + + std::error_code last_ec; + + for (Iter it = begin; it != end; ++it) + { + endpoint_type ep = *it; + + if (!cond(static_cast(last_ec), + static_cast(ep))) + continue; + + if (s.is_open()) + s.close(); + + auto [ec] = co_await s.connect(ep); + + if (!ec) + co_return {std::error_code{}, std::move(it)}; + + if (ec == capy::cond::canceled) + co_return {ec, std::move(end)}; + + last_ec = ec; + } + + if (!last_ec) + last_ec = std::make_error_code(std::errc::no_such_device_or_address); + + co_return {last_ec, std::move(end)}; +} + +} // namespace boost::corosio + +#endif diff --git a/include/boost/corosio/local_stream_socket.hpp b/include/boost/corosio/local_stream_socket.hpp index b2d52632..864714fc 100644 --- a/include/boost/corosio/local_stream_socket.hpp +++ b/include/boost/corosio/local_stream_socket.hpp @@ -77,6 +77,9 @@ namespace boost::corosio { class BOOST_COROSIO_DECL local_stream_socket : public io_stream { public: + /// The endpoint type used by this socket. + using endpoint_type = corosio::local_endpoint; + using shutdown_type = corosio::shutdown_type; using enum corosio::shutdown_type; diff --git a/include/boost/corosio/tcp_socket.hpp b/include/boost/corosio/tcp_socket.hpp index 605f5ac3..8611d26d 100644 --- a/include/boost/corosio/tcp_socket.hpp +++ b/include/boost/corosio/tcp_socket.hpp @@ -78,6 +78,9 @@ namespace boost::corosio { class BOOST_COROSIO_DECL tcp_socket : public io_stream { public: + /// The endpoint type used by this socket. + using endpoint_type = corosio::endpoint; + using shutdown_type = corosio::shutdown_type; using enum corosio::shutdown_type; diff --git a/test/unit/connect.cpp b/test/unit/connect.cpp new file mode 100644 index 00000000..6de45b45 --- /dev/null +++ b/test/unit/connect.cpp @@ -0,0 +1,450 @@ +// +// Copyright (c) 2026 Michael Vandeberg +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// Official repository: https://github.com/cppalliance/corosio +// + +// Test that header file is self-contained. +#include + +#include +#include +#include +#include +#include + +#include +#include +#include + +#include +#include +#include + +#include "context.hpp" +#include "test_suite.hpp" + +namespace boost::corosio { + +// Range-based and iterator-based connect composed operation tests. +// Templated on backend so every available reactor exercises the same paths. + +template +struct connect_test +{ + /* Bind+listen on loopback ephemeral port; return (acceptor, port). + Caller keeps the acceptor alive. */ + static std::uint16_t open_listener(tcp_acceptor& acc, tcp proto = tcp::v4()) + { + acc.open(proto); + acc.set_option(socket_option::reuse_address(true)); + std::error_code ec; + if (proto == tcp::v6()) + ec = acc.bind(endpoint(ipv6_address::loopback(), 0)); + else + ec = acc.bind(endpoint(ipv4_address::loopback(), 0)); + BOOST_TEST(!ec); + ec = acc.listen(); + BOOST_TEST(!ec); + return acc.local_endpoint().port(); + } + + /* Get a port that is known to be unused right now: bind to + port 0, read the assigned port, close. A subsequent connect to + that port will fail with connection_refused (subject to the + usual TOCTOU caveat, which is unavoidable for a "closed port" + test and rarely flakes in practice). */ + static std::uint16_t pick_closed_port(io_context& ioc) + { + tcp_acceptor tmp(ioc); + tmp.open(); + tmp.set_option(socket_option::reuse_address(true)); + auto ec = tmp.bind(endpoint(ipv4_address::loopback(), 0)); + BOOST_TEST(!ec); + auto port = tmp.local_endpoint().port(); + tmp.close(); + return port; + } + + void testEmptyRange() + { + io_context ioc(Backend); + tcp_socket sock(ioc); + + std::error_code result_ec{}; + endpoint result_ep; + bool done = false; + + auto task = [&]() -> capy::task<> { + std::vector endpoints; + auto [ec, ep] = co_await corosio::connect(sock, endpoints); + result_ec = ec; + result_ep = ep; + done = true; + }; + capy::run_async(ioc.get_executor())(task()); + ioc.run(); + + BOOST_TEST(done); + BOOST_TEST(result_ec == + std::make_error_code(std::errc::no_such_device_or_address)); + BOOST_TEST(result_ep == endpoint()); + } + + void testSingleGood() + { + io_context ioc(Backend); + tcp_acceptor acc(ioc); + auto port = open_listener(acc); + + tcp_socket client(ioc); + tcp_socket peer(ioc); + + std::error_code connect_ec{}; + endpoint connected_ep; + bool connect_done = false; + + auto connect_task = [&]() -> capy::task<> { + std::vector endpoints{ + endpoint(ipv4_address::loopback(), port)}; + auto [ec, ep] = co_await corosio::connect(client, endpoints); + connect_ec = ec; + connected_ep = ep; + connect_done = true; + }; + + auto accept_task = [&]() -> capy::task<> { + (void)co_await acc.accept(peer); + }; + + capy::run_async(ioc.get_executor())(accept_task()); + capy::run_async(ioc.get_executor())(connect_task()); + ioc.run(); + + BOOST_TEST(connect_done); + BOOST_TEST(!connect_ec); + BOOST_TEST_EQ(connected_ep.port(), port); + BOOST_TEST(connected_ep.is_v4()); + BOOST_TEST_EQ(client.remote_endpoint().port(), port); + } + + void testBadThenGood() + { + io_context ioc(Backend); + tcp_acceptor acc(ioc); + auto good_port = open_listener(acc); + auto bad_port = pick_closed_port(ioc); + + tcp_socket client(ioc); + tcp_socket peer(ioc); + + std::error_code connect_ec{}; + endpoint connected_ep; + bool connect_done = false; + + auto connect_task = [&]() -> capy::task<> { + std::vector endpoints{ + endpoint(ipv4_address::loopback(), bad_port), + endpoint(ipv4_address::loopback(), good_port)}; + auto [ec, ep] = co_await corosio::connect(client, endpoints); + connect_ec = ec; + connected_ep = ep; + connect_done = true; + }; + + auto accept_task = [&]() -> capy::task<> { + (void)co_await acc.accept(peer); + }; + + capy::run_async(ioc.get_executor())(accept_task()); + capy::run_async(ioc.get_executor())(connect_task()); + ioc.run(); + + BOOST_TEST(connect_done); + BOOST_TEST(!connect_ec); + BOOST_TEST_EQ(connected_ep.port(), good_port); + } + + void testAllBad() + { + io_context ioc(Backend); + auto bad1 = pick_closed_port(ioc); + auto bad2 = pick_closed_port(ioc); + + tcp_socket client(ioc); + + std::error_code connect_ec{}; + endpoint connected_ep; + bool done = false; + + auto task = [&]() -> capy::task<> { + std::vector endpoints{ + endpoint(ipv4_address::loopback(), bad1), + endpoint(ipv4_address::loopback(), bad2)}; + auto [ec, ep] = co_await corosio::connect(client, endpoints); + connect_ec = ec; + connected_ep = ep; + done = true; + }; + capy::run_async(ioc.get_executor())(task()); + ioc.run(); + + BOOST_TEST(done); + BOOST_TEST(connect_ec); + // Distinguish from the empty-range case. + BOOST_TEST(connect_ec != + std::make_error_code(std::errc::no_such_device_or_address)); + BOOST_TEST(connected_ep == endpoint()); + } + + void testV4ThenV6() + { + io_context ioc(Backend); + tcp_acceptor acc(ioc); + auto port = open_listener(acc, tcp::v6()); + auto bad_v4 = pick_closed_port(ioc); + + tcp_socket client(ioc); + tcp_socket peer(ioc); + + std::error_code connect_ec{}; + endpoint connected_ep; + bool connect_done = false; + + auto connect_task = [&]() -> capy::task<> { + std::vector endpoints{ + endpoint(ipv4_address::loopback(), bad_v4), + endpoint(ipv6_address::loopback(), port)}; + auto [ec, ep] = co_await corosio::connect(client, endpoints); + connect_ec = ec; + connected_ep = ep; + connect_done = true; + }; + + auto accept_task = [&]() -> capy::task<> { + (void)co_await acc.accept(peer); + }; + + capy::run_async(ioc.get_executor())(accept_task()); + capy::run_async(ioc.get_executor())(connect_task()); + ioc.run(); + + BOOST_TEST(connect_done); + BOOST_TEST(!connect_ec); + BOOST_TEST(connected_ep.is_v6()); + BOOST_TEST_EQ(connected_ep.port(), port); + } + + void testCondSkipsAll() + { + io_context ioc(Backend); + tcp_acceptor acc(ioc); + auto port = open_listener(acc); + + tcp_socket client(ioc); + + std::error_code connect_ec{}; + endpoint connected_ep; + bool done = false; + + auto task = [&]() -> capy::task<> { + std::vector endpoints{ + endpoint(ipv4_address::loopback(), port)}; + auto [ec, ep] = co_await corosio::connect( + client, + endpoints, + [](std::error_code const&, endpoint const&) { return false; }); + connect_ec = ec; + connected_ep = ep; + done = true; + }; + + // Must also cancel the acceptor since nothing will ever connect. + auto cancel_task = [&]() -> capy::task<> { + timer t(ioc); + t.expires_after(std::chrono::milliseconds(50)); + (void)co_await t.wait(); + acc.cancel(); + }; + + capy::run_async(ioc.get_executor())(task()); + capy::run_async(ioc.get_executor())(cancel_task()); + ioc.run(); + + BOOST_TEST(done); + BOOST_TEST(connect_ec == + std::make_error_code(std::errc::no_such_device_or_address)); + BOOST_TEST(connected_ep == endpoint()); + } + + void testCondSelectiveSkip() + { + io_context ioc(Backend); + tcp_acceptor acc(ioc); + auto good_port = open_listener(acc); + auto bad_port = pick_closed_port(ioc); + + tcp_socket client(ioc); + tcp_socket peer(ioc); + + std::error_code connect_ec{}; + endpoint connected_ep; + int cond_calls = 0; + bool connect_done = false; + + auto connect_task = [&]() -> capy::task<> { + std::vector endpoints{ + endpoint(ipv4_address::loopback(), bad_port), + endpoint(ipv4_address::loopback(), good_port)}; + // Skip the first (bad) endpoint, allow the second. + auto cond = [&, good_port]( + std::error_code const&, endpoint const& e) { + ++cond_calls; + return e.port() == good_port; + }; + auto [ec, ep] = co_await corosio::connect( + client, endpoints, cond); + connect_ec = ec; + connected_ep = ep; + connect_done = true; + }; + + auto accept_task = [&]() -> capy::task<> { + (void)co_await acc.accept(peer); + }; + + capy::run_async(ioc.get_executor())(accept_task()); + capy::run_async(ioc.get_executor())(connect_task()); + ioc.run(); + + BOOST_TEST(connect_done); + BOOST_TEST(!connect_ec); + BOOST_TEST_EQ(connected_ep.port(), good_port); + BOOST_TEST_EQ(cond_calls, 2); + } + + void testIteratorOverload() + { + io_context ioc(Backend); + tcp_acceptor acc(ioc); + auto port = open_listener(acc); + + tcp_socket client(ioc); + tcp_socket peer(ioc); + + std::vector endpoints{ + endpoint(ipv4_address::loopback(), port)}; + + std::error_code connect_ec{}; + bool connect_done = false; + bool iter_matches_good = false; + + auto connect_task = [&]() -> capy::task<> { + auto [ec, it] = co_await corosio::connect( + client, endpoints.begin(), endpoints.end()); + connect_ec = ec; + iter_matches_good = (it == endpoints.begin()); + connect_done = true; + }; + + auto accept_task = [&]() -> capy::task<> { + (void)co_await acc.accept(peer); + }; + + capy::run_async(ioc.get_executor())(accept_task()); + capy::run_async(ioc.get_executor())(connect_task()); + ioc.run(); + + BOOST_TEST(connect_done); + BOOST_TEST(!connect_ec); + BOOST_TEST(iter_matches_good); + } + + void testIteratorAllFailReturnsEnd() + { + io_context ioc(Backend); + auto bad = pick_closed_port(ioc); + + tcp_socket client(ioc); + + std::vector endpoints{ + endpoint(ipv4_address::loopback(), bad)}; + + std::error_code connect_ec{}; + bool connect_done = false; + bool iter_is_end = false; + + auto task = [&]() -> capy::task<> { + auto [ec, it] = co_await corosio::connect( + client, endpoints.begin(), endpoints.end()); + connect_ec = ec; + iter_is_end = (it == endpoints.end()); + connect_done = true; + }; + capy::run_async(ioc.get_executor())(task()); + ioc.run(); + + BOOST_TEST(connect_done); + BOOST_TEST(connect_ec); + BOOST_TEST(iter_is_end); + } + + void testCancellation() + { + io_context ioc(Backend); + tcp_socket client(ioc); + + // RFC 5737 TEST-NET-1 addresses. These are reserved for + // documentation and will not route; connect attempts hang + // until timeout, giving us a window to cancel. + std::vector endpoints{ + endpoint(ipv4_address("192.0.2.1"), 80), + endpoint(ipv4_address("192.0.2.2"), 80), + endpoint(ipv4_address("192.0.2.3"), 80)}; + + std::error_code connect_ec{}; + bool connect_done = false; + + auto connect_task = [&]() -> capy::task<> { + auto [ec, ep] = co_await corosio::connect(client, endpoints); + connect_ec = ec; + connect_done = true; + (void)ep; + }; + + auto cancel_task = [&]() -> capy::task<> { + timer t(ioc); + t.expires_after(std::chrono::milliseconds(50)); + (void)co_await t.wait(); + client.cancel(); + }; + + capy::run_async(ioc.get_executor())(connect_task()); + capy::run_async(ioc.get_executor())(cancel_task()); + ioc.run(); + + BOOST_TEST(connect_done); + BOOST_TEST(connect_ec == capy::cond::canceled); + } + + void run() + { + testEmptyRange(); + testSingleGood(); + testBadThenGood(); + testAllBad(); + testV4ThenV6(); + testCondSkipsAll(); + testCondSelectiveSkip(); + testIteratorOverload(); + testIteratorAllFailReturnsEnd(); + testCancellation(); + } +}; + +COROSIO_BACKEND_TESTS(connect_test, "boost.corosio.connect") + +} // namespace boost::corosio