diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml index e88d23c..5e2f766 100644 --- a/.gitea/workflows/ci.yml +++ b/.gitea/workflows/ci.yml @@ -47,7 +47,7 @@ jobs: key: conan-${{ runner.os }}-${{ hashFiles('conanfile.py', 'modulus.json') }} restore-keys: conan-${{ runner.os }}- - name: Configure native release - run: python3 modulus.py configure --target x86_64_gcc12_release + run: python3 modulus.py configure --target x86_64_gcc12_release --update - name: Check code style run: | python3 -m pip install --no-cache-dir -r build/build_tools/docker/format-requirements.txt @@ -63,4 +63,4 @@ jobs: CONAN_TOKEN: ${{ secrets.CONAN_TOKEN }} LIGO_CONAN_UPLOAD: "1" MODULUS_ALLOW_PACKAGE_REPLACE: "1" - run: python3 modulus.py package --all --replace + run: python3 modulus.py package --all --replace --update diff --git a/conanfile.py b/conanfile.py index 9d67c0f..6126567 100644 --- a/conanfile.py +++ b/conanfile.py @@ -28,8 +28,11 @@ class ligo_conan(conan.ConanFile): "fPIC": True, "with_tests": False, } - exports = "modulus.json" + exports = "modulus.json", "!build", "!.git", "!test_package/build" exports_sources = ( + "!build", + "!.git", + "!test_package/build", "CMakeLists.txt", "cmake/*.in", "include/*.hpp", @@ -37,6 +40,7 @@ class ligo_conan(conan.ConanFile): "include/ligo/detail/*.hpp", "src/*.cpp", "src/detail/*.cpp", + "src/detail/*.hpp", "tests/CMakeLists.txt", "tests/*.cpp", "tests/*.hpp", diff --git a/include/ligo/detail/lifecycle.hpp b/include/ligo/detail/lifecycle.hpp index 00d3e18..21e72e7 100644 --- a/include/ligo/detail/lifecycle.hpp +++ b/include/ligo/detail/lifecycle.hpp @@ -202,6 +202,9 @@ auto ligo::detail::stop_accepting_server(std::shared_ptr state) -> void { return; } state->sessions->cancel_all(); + if constexpr (requires { state->tls_sessions; }) { + state->tls_sessions->cancel_all(); + } auto server_executor = state->strand; boost::asio::post(server_executor, [state = std::move(state)] { auto ignored = boost::system::error_code{}; @@ -216,6 +219,9 @@ auto ligo::detail::stop_accepting_server_now(std::shared_ptr const &state) -> return; } state->sessions->cancel_all(); + if constexpr (requires { state->tls_sessions; }) { + state->tls_sessions->cancel_all(); + } auto ignored = boost::system::error_code{}; state->acceptor->close(ignored); } @@ -232,6 +238,9 @@ auto ligo::detail::async_stop_accepting_server(std::shared_ptr state, std::ch should_stop = !state->stopped.exchange(true); if (should_stop) { state->sessions->cancel_all(); + if constexpr (requires { state->tls_sessions; }) { + state->tls_sessions->cancel_all(); + } } } if (should_stop) { @@ -239,5 +248,12 @@ auto ligo::detail::async_stop_accepting_server(std::shared_ptr state, std::ch auto ignored = boost::system::error_code{}; state->acceptor->close(ignored); } - co_return co_await state->sessions->wait_empty(timeout); + auto drained = co_await state->sessions->wait_empty(timeout); + if (!drained) { + co_return drained; + } + if constexpr (requires { state->tls_sessions; }) { + co_return co_await state->tls_sessions->wait_empty(timeout); + } + co_return drained; } diff --git a/modulus.py b/modulus.py index 1b69392..856cec7 100755 --- a/modulus.py +++ b/modulus.py @@ -1,4 +1,5 @@ #!/usr/bin/env python3 +# Canonical consumer bootstrap. Copy this file unchanged to a project's modulus.py. import json import os import pathlib @@ -17,6 +18,7 @@ COMMANDS = ( "lsp", "clangd", "format", + "run", ) COMMON_DIR = pathlib.Path("build") / "build_common" TOOLS_DIR = pathlib.Path("build") / "build_tools" @@ -66,13 +68,6 @@ def runtime_ready(root): and (root / RUNTIME_DIR / "modulus" / "format_layout.py").is_file() ) -def optional_credentials(): - user = os.environ.get("CONAN_USER") - token = os.environ.get("CONAN_TOKEN") - if bool(user) != bool(token): - raise SystemExit("CONAN_USER and CONAN_TOKEN must either both be set or both be unset") - return user, token - def env_flag(name): return os.environ.get(name, "").lower() in {"1", "true", "yes", "on"} @@ -128,7 +123,7 @@ RUN python -m pip install --no-cache-dir conan=={CONAN_VERSION} WORKDIR /workspace """ -def ensure_fetch_image(): +def ensure_fetch_image(offline=False): inspected = subprocess.run( [ "docker", @@ -140,12 +135,15 @@ def ensure_fetch_image(): stderr=subprocess.DEVNULL, check=False, ) - if inspected.returncode == 0: + if inspected.returncode == 0 and (offline or not docker_platform_args()): return + if offline: + raise SystemExit('offline bootstrap requires the existing Modulus fetch image or MODULUS_DIRECT=1 with Conan installed') run( [ "docker", "build", + *docker_platform_args(), "--tag", FETCH_IMAGE, "-", @@ -156,6 +154,10 @@ def ensure_fetch_image(): def in_container(): return pathlib.Path("/.dockerenv").exists() +def docker_platform_args(): + platform = os.environ.get("MODULUS_DOCKER_PLATFORM", "").strip() + return ["--platform", platform] if platform else [] + def docker_workspace_args(root): if in_container(): container = os.environ.get("HOSTNAME", "").strip() @@ -172,18 +174,24 @@ def docker_workspace_args(root): def fetch_env(root): env = os.environ.copy() + for name in list(env): + if name in {"CONAN_USER", "CONAN_TOKEN"} or name.startswith(("CONAN_LOGIN_USERNAME", "CONAN_PASSWORD")): + del env[name] env["CONAN_HOME"] = str(root / CONAN_HOME_DIR) + env["CONAN_NON_INTERACTIVE"] = "1" return env -def run_fetch(root, command, capture=False, display_command=None, check=True): +def run_fetch(root, command, capture=False, display_command=None, check=True, offline=False): if direct_mode(root): return run(command, capture=capture, display_command=display_command, env=fetch_env(root), check=check) - ensure_fetch_image() + ensure_fetch_image(offline=offline) docker_command = [ "docker", "run", "--rm", "--interactive", + *(["--pull=never", "--network=none"] if offline else []), + *docker_platform_args(), "--user", f"{os.getuid()}:{os.getgid()}", "--env", @@ -191,9 +199,7 @@ def run_fetch(root, command, capture=False, display_command=None, check=True): "--env", "CONAN_HOME", "--env", - "CONAN_USER", - "--env", - "CONAN_TOKEN", + "CONAN_NON_INTERACTIVE", *docker_workspace_args(root), "--workdir", str(root), @@ -203,7 +209,6 @@ def run_fetch(root, command, capture=False, display_command=None, check=True): return run(docker_command, capture=capture, display_command=display_command, env=fetch_env(root), check=check) def configure_remote(root, remote_name, remote_url): - user, token = optional_credentials() run_fetch( root, [ @@ -215,29 +220,7 @@ def configure_remote(root, remote_name, remote_url): remote_url, ], ) - if not user and not token: - return - run_fetch( - root, - [ - "conan", - "remote", - "login", - remote_name, - user, - "-p", - token, - ], - display_command=[ - "conan", - "remote", - "login", - remote_name, - user, - "-p", - "", - ], - ) + run_fetch(root, ["conan", "remote", "logout", remote_name]) def package_refs(data): refs = [] @@ -274,46 +257,26 @@ def package_refs(data): refs.append(f"{PACKAGE_REF}#{recipe_revision}:{package_id}") return sorted(refs) -def remove_cached_modulus(root): - run_fetch( - root, - [ - "conan", - "remove", - PACKAGE_REF + ":*", - "-c", - ], - check=False, - ) - -def download_modulus(root, remote_name): - remove_cached_modulus(root) - run_fetch( - root, - [ - "conan", - "download", - PACKAGE_REF + ":*", - "-r", - remote_name, - ], - ) +def find_modulus(root, remote_name=None): completed = run_fetch( root, [ "conan", "list", - PACKAGE_REF + ":*", - "-c", + PACKAGE_REF + "#latest:*#latest", + *(["-r", remote_name] if remote_name else ["-c"]), "-f", "json", ], capture=True, + offline=remote_name is None, ) refs = package_refs(json.loads(completed.stdout)) if not refs: - raise SystemExit(f"no downloaded {PACKAGE_REF} package found in the Conan cache") + raise SystemExit(f"no {PACKAGE_REF} package available in {remote_name or 'the local Conan cache'}") package_ref = refs[-1] + if remote_name: + run_fetch(root, ["conan", "download", package_ref, "-r", remote_name]) completed = run_fetch( root, [ @@ -323,24 +286,25 @@ def download_modulus(root, remote_name): package_ref, ], capture=True, + offline=remote_name is None, ) package_root = pathlib.Path(completed.stdout.strip()) if not (package_root / "modulus" / "export.py").is_file() or not (package_root / "tools").is_dir(): raise SystemExit(f"downloaded {PACKAGE_REF} package does not contain Modulus export files: {package_root}") return package_root -def materialize_runtime(root): +def materialize_runtime(root, no_remote=False): remote_name, remote_url = remote_config(root) - configure_remote(root, remote_name, remote_url) - package_root = download_modulus(root, remote_name) + if not no_remote: + configure_remote(root, remote_name, remote_url) + package_root = find_modulus(root, None if no_remote else remote_name) sys.path.insert(0, str(package_root)) import modulus.export modulus.export.sync_project(root, package_root) -def ensure_runtime(root): - if runtime_ready(root): - return - materialize_runtime(root) +def ensure_runtime(root, update=False, no_remote=False): + if update or not runtime_ready(root): + materialize_runtime(root, no_remote=no_remote) def delegate(command, argv): root = repo_root() @@ -349,7 +313,7 @@ def delegate(command, argv): print(f"Modulus {command} tooling is missing; run ./modulus.py configure first", file=sys.stderr) return 1 else: - ensure_runtime(root) + ensure_runtime(root, update=command == "configure" and "--update" in argv, no_remote="--no-remote" in argv) sys.path.insert(0, str(root / RUNTIME_DIR)) import modulus.cli previous_argv = sys.argv diff --git a/src/detail/stream_server.hpp b/src/detail/stream_server.hpp new file mode 100644 index 0000000..7b6eaf3 --- /dev/null +++ b/src/detail/stream_server.hpp @@ -0,0 +1,85 @@ +#pragma once + +#include +#include +#include + +#include +#include +#include + +#include +#include +#include + + +namespace ligo::detail { + +struct stream_listener { + stream_listener(boost::asio::any_io_executor executor, ligo::service_router router) + : executor{std::move(executor)}, strand{boost::asio::make_strand(this->executor)}, acceptor{std::make_shared(strand)}, router{std::make_shared(std::move(router))} {} + boost::asio::any_io_executor executor; + boost::asio::strand strand; + std::shared_ptr acceptor{}; + std::shared_ptr router{}; + std::shared_ptr> sessions{std::make_shared>()}; + std::shared_ptr>> tls_sessions{std::make_shared>>()}; + std::shared_ptr tls_context{}; + std::atomic_uint16_t selected_port{}; + std::atomic start_called{false}; + std::atomic stopped{false}; + mutable std::mutex lifecycle_mutex{}; +}; + +template +auto start_stream_server(std::shared_ptr const &state, std::string_view protocol, ACCEPT accept_loop, ACCEPT_TLS accept_tls_loop) -> std::expected; + +} + +template +auto ligo::detail::start_stream_server(std::shared_ptr const &state, std::string_view protocol, ACCEPT accept_loop, ACCEPT_TLS accept_tls_loop) -> std::expected { + auto lock = std::scoped_lock{state->lifecycle_mutex}; + auto valid_timeout = ligo::detail::validate_positive_timeout(state->options.timeout, std::string{protocol} + " server timeout"); + if (!valid_timeout) { + return valid_timeout; + } + if (state->options.security.mode == ligo::security_mode::DTLS) { + return ligo::detail::fail_void(ligo::unsupported_security(std::string{protocol} + " server", state->options.security.mode)); + } + if (state->start_called.exchange(true)) { + return ligo::detail::fail_void(ligo::detail::unavailable(std::string{protocol} + " server has already been started")); + } + if (state->options.security.mode == ligo::security_mode::TLS) { + auto valid_config = ligo::validate_tls_server_config(protocol, state->options.security); + if (!valid_config) { + state->start_called.store(false); + return valid_config; + } + auto context = ligo::make_tls_server_context(state->options.security); + if (!context) { + state->start_called.store(false); + return ligo::detail::fail_void(context.error()); + } + auto valid_context = ligo::validate_tls_server_context(*context); + if (!valid_context) { + state->start_called.store(false); + return valid_context; + } + state->tls_context = std::move(*context); + } + auto selected = ligo::detail::start_tcp_acceptor(*state->acceptor, state->endpoint.host, state->endpoint.port); + if (!selected) { + state->start_called.store(false); + return ligo::detail::fail_void(selected.error()); + } + state->selected_port.store(*selected); + assert(state->selected_port.load() == *selected); + state->stopped.store(false); + modulus::log_debug("ligo " + std::string{protocol} + " server started on port " + std::to_string(*selected)); + if (state->options.security.mode == ligo::security_mode::TLS) { + boost::asio::co_spawn(state->strand, accept_tls_loop(state), boost::asio::detached); + } else { + boost::asio::co_spawn(state->strand, accept_loop(state), boost::asio::detached); + } + return {}; +} diff --git a/src/http.cpp b/src/http.cpp index 59306b1..4a6b019 100755 --- a/src/http.cpp +++ b/src/http.cpp @@ -36,7 +36,7 @@ #include #include -#include +#include "detail/stream_server.hpp" #include #include #include @@ -46,21 +46,10 @@ namespace { -struct http_server_state { +struct http_server_state : ligo::detail::stream_listener { http_server_state(boost::asio::any_io_executor executor, ligo::http::endpoint endpoint, ligo::service_router router, ligo::http::server_options options); - boost::asio::any_io_executor executor; - boost::asio::strand strand; - std::shared_ptr acceptor{}; ligo::http::endpoint endpoint{}; ligo::http::server_options options{}; - std::shared_ptr router{}; - std::shared_ptr> sessions{std::make_shared>()}; - std::shared_ptr>> tls_sessions{std::make_shared>>()}; - std::shared_ptr tls_context{}; - std::atomic_uint16_t selected_port{}; - std::atomic start_called{false}; - std::atomic stopped{false}; - mutable std::mutex lifecycle_mutex{}; }; auto http_status(std::int32_t code) -> boost::beast::http::status { @@ -297,111 +286,26 @@ auto uninitialized_http_server() -> std::expected { } auto start_http_server(std::shared_ptr const &state) -> std::expected { - assert(!state || state->acceptor != nullptr); - assert(!state || state->router != nullptr); if (!state) { return uninitialized_http_server(); } - auto lock = std::scoped_lock{state->lifecycle_mutex}; - auto valid_timeout = ligo::detail::validate_positive_timeout(state->options.timeout, "HTTP server timeout"); - if (!valid_timeout) { - return valid_timeout; - } - if (state->options.security.mode == ligo::security_mode::DTLS) { - return ligo::detail::fail_void(ligo::unsupported_security("HTTP server", state->options.security.mode)); - } - if (state->start_called.exchange(true)) { - return ligo::detail::fail_void(ligo::detail::unavailable("HTTP server has already been started")); - } - if (state->options.security.mode == ligo::security_mode::TLS) { - auto valid_config = ligo::validate_tls_server_config("HTTP", state->options.security); - if (!valid_config) { - state->start_called.store(false); - return valid_config; - } - auto context = ligo::make_tls_server_context(state->options.security); - if (!context) { - state->start_called.store(false); - return ligo::detail::fail_void(context.error()); - } - auto valid_context = ligo::validate_tls_server_context(*context); - if (!valid_context) { - state->start_called.store(false); - return valid_context; - } - state->tls_context = std::move(*context); - } - auto selected = ligo::detail::start_tcp_acceptor(*state->acceptor, state->endpoint.host, state->endpoint.port); - if (!selected) { - state->start_called.store(false); - return ligo::detail::fail_void(selected.error()); - } - state->selected_port.store(*selected); - assert(state->selected_port.load() == *selected); - state->stopped.store(false); - modulus::log_debug("ligo HTTP server started on port " + std::to_string(*selected)); - if (state->options.security.mode == ligo::security_mode::TLS) { - boost::asio::co_spawn(state->strand, accept_tls_loop(state), boost::asio::detached); - } else { - boost::asio::co_spawn(state->strand, accept_loop(state), boost::asio::detached); - } - return {}; + return ligo::detail::start_stream_server(state, "HTTP", accept_loop, accept_tls_loop); } auto stop_http_server(std::shared_ptr state) -> void { - assert(state != nullptr); - auto lock = std::scoped_lock{state->lifecycle_mutex}; - if (state->stopped.exchange(true)) { - return; - } modulus::log_debug("ligo HTTP server stopping"); - state->sessions->cancel_all(); - state->tls_sessions->cancel_all(); - auto server_executor = state->strand; - boost::asio::post(server_executor, [state = std::move(state)] { - auto ignored = boost::system::error_code{}; - state->acceptor->close(ignored); - }); + ligo::detail::stop_accepting_server(std::move(state)); } auto stop_http_server_now(std::shared_ptr const &state) -> void { - auto lock = std::scoped_lock{state->lifecycle_mutex}; - if (state->stopped.exchange(true)) { - return; - } - state->sessions->cancel_all(); - state->tls_sessions->cancel_all(); - auto ignored = boost::system::error_code{}; - state->acceptor->close(ignored); + ligo::detail::stop_accepting_server_now(state); } auto async_stop_http_server(std::shared_ptr state, std::chrono::milliseconds timeout) -> boost::asio::awaitable> { if (!state) { co_return uninitialized_http_server(); } - auto valid_timeout = ligo::detail::validate_positive_timeout(timeout, "shutdown timeout"); - if (!valid_timeout) { - co_return valid_timeout; - } - auto should_stop = false; - { - auto lock = std::scoped_lock{state->lifecycle_mutex}; - should_stop = !state->stopped.exchange(true); - if (should_stop) { - state->sessions->cancel_all(); - state->tls_sessions->cancel_all(); - } - } - if (should_stop) { - co_await boost::asio::post(state->strand, boost::asio::use_awaitable); - auto ignored = boost::system::error_code{}; - state->acceptor->close(ignored); - } - auto drained = co_await state->sessions->wait_empty(timeout); - if (!drained) { - co_return drained; - } - co_return co_await state->tls_sessions->wait_empty(timeout); + co_return co_await ligo::detail::async_stop_accepting_server(std::move(state), timeout); } auto selected_http_port(std::shared_ptr const &state) -> std::uint16_t { @@ -546,7 +450,7 @@ auto call_http_endpoint(ligo::http::endpoint endpoint, std::shared_ptrexecutor)}, acceptor{std::make_shared(strand)}, endpoint{std::move(endpoint)}, options{options}, router{std::make_shared(std::move(router))} {} + : ligo::detail::stream_listener{std::move(executor), std::move(router)}, endpoint{std::move(endpoint)}, options{options} {} } diff --git a/src/tcp.cpp b/src/tcp.cpp index 125356a..946283e 100755 --- a/src/tcp.cpp +++ b/src/tcp.cpp @@ -39,7 +39,7 @@ #include #include -#include +#include "detail/stream_server.hpp" #include #include #include @@ -49,21 +49,10 @@ namespace { -struct tcp_server_state { +struct tcp_server_state : ligo::detail::stream_listener { tcp_server_state(boost::asio::any_io_executor executor, ligo::tcp::endpoint endpoint, ligo::service_router router, ligo::tcp::server_options options); - boost::asio::any_io_executor executor; - boost::asio::strand strand; - std::shared_ptr acceptor{}; ligo::tcp::endpoint endpoint{}; ligo::tcp::server_options options{}; - std::shared_ptr router{}; - std::shared_ptr> sessions{std::make_shared>()}; - std::shared_ptr>> tls_sessions{std::make_shared>>()}; - std::shared_ptr tls_context{}; - std::atomic_uint16_t selected_port{}; - std::atomic start_called{false}; - std::atomic stopped{false}; - mutable std::mutex lifecycle_mutex{}; }; constexpr auto max_frame_size = std::uint32_t{16U * 1024U * 1024U}; @@ -283,111 +272,26 @@ auto uninitialized_tcp_server() -> std::expected { } auto start_tcp_server(std::shared_ptr const &state) -> std::expected { - assert(!state || state->acceptor != nullptr); - assert(!state || state->router != nullptr); if (!state) { return uninitialized_tcp_server(); } - auto lock = std::scoped_lock{state->lifecycle_mutex}; - auto valid_timeout = ligo::detail::validate_positive_timeout(state->options.timeout, "TCP server timeout"); - if (!valid_timeout) { - return valid_timeout; - } - if (state->options.security.mode == ligo::security_mode::DTLS) { - return ligo::detail::fail_void(ligo::unsupported_security("TCP server", state->options.security.mode)); - } - if (state->start_called.exchange(true)) { - return ligo::detail::fail_void(ligo::detail::unavailable("TCP server has already been started")); - } - if (state->options.security.mode == ligo::security_mode::TLS) { - auto valid_config = ligo::validate_tls_server_config("TCP", state->options.security); - if (!valid_config) { - state->start_called.store(false); - return valid_config; - } - auto context = ligo::make_tls_server_context(state->options.security); - if (!context) { - state->start_called.store(false); - return ligo::detail::fail_void(context.error()); - } - auto valid_context = ligo::validate_tls_server_context(*context); - if (!valid_context) { - state->start_called.store(false); - return valid_context; - } - state->tls_context = std::move(*context); - } - auto selected = ligo::detail::start_tcp_acceptor(*state->acceptor, state->endpoint.host, state->endpoint.port); - if (!selected) { - state->start_called.store(false); - return ligo::detail::fail_void(selected.error()); - } - state->selected_port.store(*selected); - assert(state->selected_port.load() == *selected); - state->stopped.store(false); - modulus::log_debug("ligo TCP server started on port " + std::to_string(*selected)); - if (state->options.security.mode == ligo::security_mode::TLS) { - boost::asio::co_spawn(state->strand, accept_tls_loop(state), boost::asio::detached); - } else { - boost::asio::co_spawn(state->strand, accept_loop(state), boost::asio::detached); - } - return {}; + return ligo::detail::start_stream_server(state, "TCP", accept_loop, accept_tls_loop); } auto stop_tcp_server(std::shared_ptr state) -> void { - assert(state != nullptr); - auto lock = std::scoped_lock{state->lifecycle_mutex}; - if (state->stopped.exchange(true)) { - return; - } modulus::log_debug("ligo TCP server stopping"); - state->sessions->cancel_all(); - state->tls_sessions->cancel_all(); - auto server_executor = state->strand; - boost::asio::post(server_executor, [state = std::move(state)] { - auto ignored = boost::system::error_code{}; - state->acceptor->close(ignored); - }); + ligo::detail::stop_accepting_server(std::move(state)); } auto stop_tcp_server_now(std::shared_ptr const &state) -> void { - auto lock = std::scoped_lock{state->lifecycle_mutex}; - if (state->stopped.exchange(true)) { - return; - } - state->sessions->cancel_all(); - state->tls_sessions->cancel_all(); - auto ignored = boost::system::error_code{}; - state->acceptor->close(ignored); + ligo::detail::stop_accepting_server_now(state); } auto async_stop_tcp_server(std::shared_ptr state, std::chrono::milliseconds timeout) -> boost::asio::awaitable> { if (!state) { co_return uninitialized_tcp_server(); } - auto valid_timeout = ligo::detail::validate_positive_timeout(timeout, "shutdown timeout"); - if (!valid_timeout) { - co_return valid_timeout; - } - auto should_stop = false; - { - auto lock = std::scoped_lock{state->lifecycle_mutex}; - should_stop = !state->stopped.exchange(true); - if (should_stop) { - state->sessions->cancel_all(); - state->tls_sessions->cancel_all(); - } - } - if (should_stop) { - co_await boost::asio::post(state->strand, boost::asio::use_awaitable); - auto ignored = boost::system::error_code{}; - state->acceptor->close(ignored); - } - auto drained = co_await state->sessions->wait_empty(timeout); - if (!drained) { - co_return drained; - } - co_return co_await state->tls_sessions->wait_empty(timeout); + co_return co_await ligo::detail::async_stop_accepting_server(std::move(state), timeout); } auto selected_tcp_port(std::shared_ptr const &state) -> std::uint16_t { @@ -498,7 +402,7 @@ auto validate_tls_connect_security(ligo::security_config const &security) -> std } tcp_server_state::tcp_server_state(boost::asio::any_io_executor executor, ligo::tcp::endpoint endpoint, ligo::service_router router, ligo::tcp::server_options options) - : executor{std::move(executor)}, strand{boost::asio::make_strand(this->executor)}, acceptor{std::make_shared(strand)}, endpoint{std::move(endpoint)}, options{options}, router{std::make_shared(std::move(router))} {} + : ligo::detail::stream_listener{std::move(executor), std::move(router)}, endpoint{std::move(endpoint)}, options{options} {} } @@ -749,51 +653,17 @@ auto ligo::tcp::connect_tls_configured(boost::asio::any_io_executor executor, li } auto ligo::tcp::connect_tls_handshaken(boost::asio::any_io_executor executor, ligo::tcp::endpoint endpoint, ligo::tcp::client_options options, ligo::security_config security) -> boost::asio::awaitable> { - auto host = endpoint.host; - auto valid_timeout = ligo::detail::validate_positive_timeout(options.timeout, "TCP timeout"); - if (!valid_timeout) { - co_return ligo::detail::fail(valid_timeout.error()); + auto socket = co_await ligo::tcp::connect_tls_configured(std::move(executor), std::move(endpoint), options, std::move(security)); + if (!socket) { + co_return ligo::detail::fail(socket.error()); } - auto valid_security = validate_tls_connect_security(security); - if (!valid_security) { - co_return ligo::detail::fail(valid_security.error()); + auto stream = socket->stream(); + if (!stream) { + co_return ligo::detail::fail(stream.error()); } - if (endpoint.port == 0) { - co_return ligo::detail::fail(ligo::detail::error_invalid_argument, "TCP port must be non-zero"); - } - auto context = ligo::make_tls_client_context(security); - if (!context) { - co_return ligo::detail::fail(context.error()); - } - auto endpoints = co_await ligo::detail::resolve_tcp_endpoints(executor, endpoint.host, endpoint.port, options.timeout, "TCP connect timed out"); - if (!endpoints) { - co_return ligo::detail::fail(endpoints.error()); - } - auto stream = boost::beast::tcp_stream{executor}; - stream.expires_after(options.timeout); - auto [connect_error, _] = co_await stream.async_connect(*endpoints, boost::asio::as_tuple(boost::asio::use_awaitable)); - if (connect_error) { - auto failure_status = ligo::detail::operation_failure(connect_error, "TCP connect timed out"); - co_return ligo::detail::fail(failure_status); - } - auto socket = ligo::tcp::tls_socket{ - std::move(*context), - stream.release_socket(), - }; - auto socket_stream = socket.stream(); - if (!socket_stream) { - co_return ligo::detail::fail(socket_stream.error()); - } - auto verification = ligo::apply_tls_client_verification(socket_stream->get(), host, security); - if (!verification) { - co_return ligo::detail::fail(verification.error()); - } - auto handshake = co_await ligo::detail::async_tls_handshake(socket_stream->get(), boost::asio::ssl::stream_base::client, options.timeout, "TCP TLS handshake timed out"); + auto handshake = co_await ligo::detail::async_tls_handshake(stream->get(), boost::asio::ssl::stream_base::client, options.timeout, "TCP TLS handshake timed out"); if (!handshake) { co_return ligo::detail::fail(handshake.error()); } - co_return std::expected{ - std::in_place, - std::move(socket), - }; + co_return std::move(socket); }