WIP add cancellation

Author: mathisloge

.clang-tidy

@@ -24,8 +24,6 @@ CheckOptions:
   - { key: readability-identifier-naming.ProtectedMemberSuffix,     value: _          }
   - { key: readability-identifier-naming.MacroDefinitionCase,       value: UPPER_CASE }
   - { key: readability-identifier-naming.EnumConstantCase,          value: CamelCase }
-  - { key: readability-identifier-naming.ConstexprVariableCase,     value: CamelCase }
-  - { key: readability-identifier-naming.ConstexprVariablePrefix,   value: k         }
   - { key: readability-identifier-naming.GlobalConstantCase,        value: CamelCase }
   - { key: readability-identifier-naming.GlobalConstantPrefix,      value: g         }
   - { key: readability-identifier-naming.MemberConstantCase,        value: CamelCase }

cppesphomeapi/include/cppesphomeapi/api_client.hpp

@@ -2,6 +2,7 @@
 #define CPPESPHOMEAPI_API_CLIENT_HPP
 #include <cstdint>
 #include <memory>
+#include <stop_token>
 #include <string>
 #include <cppesphomeapi/cppesphomeapi_export.hpp>
 #include "api_version.hpp"
@@ -22,6 +23,7 @@ class CPPESPHOMEAPI_EXPORT ApiClient
 {
   public:
     explicit ApiClient(const boost::asio::any_io_executor &executor,
+                       std::stop_source &stop_source,
                        std::string hostname,
                        std::uint16_t port = 6053,
                        std::string password = "");
@@ -36,6 +38,7 @@ class CPPESPHOMEAPI_EXPORT ApiClient
     AsyncResult<void> async_light_command(LightCommand light_command);
     AsyncResult<void> enable_logs(EspHomeLogLevel log_level, bool config_dump);
     AsyncResult<LogEntry> receive_log();
+    void close();
 
     ApiClient(const ApiClient &) = delete;
     ApiClient(ApiClient &&) = delete;

cppesphomeapi/src/CMakeLists.txt

@@ -1,8 +1,12 @@
-target_sources(cppesphomeapi PRIVATE
-    api_client.cpp
-    make_unexpected_result.cpp
-    plain_text_protocol.cpp
-    api_connection.cpp
-    entity_conversion.cpp
-    entity_conversion.hpp
+target_sources(cppesphomeapi
+    PRIVATE
+        api_client.cpp
+        make_unexpected_result.cpp
+        plain_text_protocol.cpp
+        api_connection.cpp
+        entity_conversion.cpp
+        entity_conversion.hpp
+        executor.hpp
+        net.hpp
+        net.cpp
 )

cppesphomeapi/src/api_client.cpp

@@ -3,10 +3,12 @@
 namespace cppesphomeapi
 {
 ApiClient::ApiClient(const boost::asio::any_io_executor &executor,
+                     std::stop_source &stop_source,
                      std::string hostname,
                      std::uint16_t port,
                      std::string password)
-    : connection_{std::make_unique<ApiConnection>(std::move(hostname), port, std::move(password), executor)}
+    : connection_{
+          std::make_unique<ApiConnection>(std::move(hostname), port, std::move(password), stop_source, executor)}
 {}
 
 ApiClient::~ApiClient() = default;
@@ -56,4 +58,9 @@ const std::string &ApiClient::device_name() const
     return connection_->device_name();
 }
 
+void ApiClient::close()
+{
+    connection_->cancel();
+}
+
 } // namespace cppesphomeapi

cppesphomeapi/src/api_connection.cpp

@@ -3,7 +3,9 @@
 #include <boost/asio.hpp>
 #include "api.pb.h"
 #include "entity_conversion.hpp"
+#include "executor.hpp"
 #include "make_unexpected_result.hpp"
+#include "net.hpp"
 #include "plain_text_protocol.hpp"
 
 namespace asio = boost::asio;
@@ -23,25 +25,51 @@ namespace cppesphomeapi
 ApiConnection::ApiConnection(std::string hostname,
                              std::uint16_t port,
                              std::string password,
+                             std::stop_source &stop_source,
                              const asio::any_io_executor &executor)
     : hostname_{std::move(hostname)}
     , port_{port}
     , password_{std::move(password)}
-    , strand_{asio::make_strand(executor)}
-    , socket_{strand_}
-{}
+    , socket_{executor}
+{
+    executor::addStopService(executor::getContext(executor), stop_source);
+}
+
+void ApiConnection::cancel()
+{
+    executor::stopAssetOf(executor::getContext(socket_.get_executor())).request_stop();
+}
 
 AsyncResult<void> ApiConnection::connect()
 {
     auto executor = co_await this_coro::executor;
-    tcp::resolver resolver{executor};
-    const auto resolved = co_await resolver.async_resolve(hostname_, std::to_string(port_));
+    net::Timer timer{executor};
+    timer.expires_after(std::chrono::milliseconds{500});
+
+    auto endpoints = co_await net::resolveHostEndpoints(hostname_, net::Port{port_}, timer);
+    if (not endpoints.has_value())
+    {
+        co_return make_unexpected_result(
+            ApiErrorCode::UnexpectedMessage,
+            std::format(
+                "Could not resolve host {}:{}. Failed with error: {}", hostname_, port_, endpoints.error().message()));
+    }
+
+    timer.expires_after(std::chrono::milliseconds{500});
+    auto connect_result = co_await net::connectTo(*endpoints, timer);
+    if (not connect_result.has_value())
+    {
+        co_return make_unexpected_result(ApiErrorCode::UnexpectedMessage,
+                                         std::format("Could not connect to host {}:{}. Failed with error: {}",
+                                                     hostname_,
+                                                     port_,
+                                                     endpoints.error().message()));
+    }
 
-    co_await socket_.async_connect(resolved->endpoint());
-    socket_.set_option(asio::socket_base::keep_alive{true});
+    socket_ = std::move(*connect_result);
 
-    asio::co_spawn(strand_, std::bind(&ApiConnection::async_receive, this), asio::detached);
-    spawn_heartbeat();
+    executor::commission(socket_.get_executor(), &ApiConnection::receive_loop, this);
+    executor::commission(socket_.get_executor(), &ApiConnection::heartbeat_loop, this);
 
     REQUIRE_SUCCESS(co_await send_message_hello());
     REQUIRE_SUCCESS(co_await send_message_connect());
@@ -165,14 +193,25 @@ AsyncResult<void> ApiConnection::send_message(const google::protobuf::Message &m
     if (packet.has_value())
     {
         std::println("Sending {}", message.GetTypeName());
-        const auto written = co_await socket_.async_write_some(asio::buffer(packet.value()));
-        if (written != packet->size())
+        auto executor = co_await this_coro::executor;
+        net::Timer timer{executor};
+        const auto watch_dog = executor::abort(socket_, timer);
+        timer.expires_after(std::chrono::milliseconds{500});
+
+        const auto written = co_await net::sendTo(socket_, timer, packet.value());
+        if (written.has_value() && written.value() != packet->size())
         {
             co_return make_unexpected_result(
                 ApiErrorCode::SendError,
                 std::format("Could not send message. Bytes written are different: expected={}, written={}",
                             packet->size(),
-                            written));
+                            *written));
+        }
+        else if (not written.has_value())
+        {
+            co_return make_unexpected_result(
+                ApiErrorCode::SendError,
+                std::format("Could not send message in time. Error: {}", written.error().message()));
         }
         co_return Result<void>{};
     }
@@ -207,15 +246,23 @@ AsyncResult<LogEntry> ApiConnection::receive_log()
     };
 }
 
-boost::asio::awaitable<void> ApiConnection::async_receive()
+boost::asio::awaitable<void> ApiConnection::receive_loop()
 {
-    namespace asio = boost::asio;
-    std::array<std::uint8_t, 4096> buffer{};
-    bool do_receive{true};
+    std::array<std::byte, 4096> buffer{};
+    auto executor = co_await this_coro::executor;
+    net::Timer timer{executor};
+    const auto watch_dog = executor::abort(socket_, timer);
 
+    bool do_receive{true};
     while (do_receive)
     {
-        const auto received_bytes = co_await socket_.async_receive(asio::buffer(buffer));
+        timer.expires_after(std::chrono::seconds{100}); // at least every 90secs. a message should be received
+        const auto received_bytes = co_await net::receiveFrom(socket_, timer, buffer);
+        if (not received_bytes.has_value())
+        {
+            std::println("Could not receive bytes. Error {}", received_bytes.error().message());
+            break;
+        }
         auto result = PlainTextProtocol{}
                           .decode_multiple<proto::SubscribeLogsResponse,
                                            proto::DeviceInfoResponse,
@@ -247,7 +294,7 @@ boost::asio::awaitable<void> ApiConnection::async_receive()
                                            proto::ListEntitiesTimeResponse,
                                            proto::ListEntitiesUpdateResponse,
                                            proto::ListEntitiesValveResponse>(
-                              std::span{buffer.begin(), received_bytes}, [this](auto &&message) {
+                              std::span{buffer.cbegin(), received_bytes.value()}, [this](auto &&message) {
                                   std::vector<boost::asio::any_completion_handler<void(MessageWrapper)>> handlers;
                                   {
                                       std::unique_lock l{handler_mtx_};
@@ -276,11 +323,6 @@ boost::asio::awaitable<void> ApiConnection::async_receive()
     std::println("RECEIVE ENDED!");
 }
 
-void ApiConnection::spawn_heartbeat()
-{
-    asio::co_spawn(strand_, std::bind(&ApiConnection::heartbeat_loop, this), asio::detached);
-}
-
 boost::asio::awaitable<void> ApiConnection::heartbeat_loop()
 {
     auto executor = co_await this_coro::executor;
@@ -296,6 +338,7 @@ boost::asio::awaitable<void> ApiConnection::heartbeat_loop()
         // otherwise the connection is dead.
         co_await receive_message<proto::PingResponse>(asio::use_awaitable);
     }
+    executor::stopAssetOf(executor).request_stop();
 }
 
 } // namespace cppesphomeapi

cppesphomeapi/src/api_connection.hpp

@@ -1,8 +1,10 @@
 #pragma once
 #include <cstdint>
+#include <stop_token>
 #include <string>
 #include <boost/asio/any_completion_handler.hpp>
 #include <boost/asio/executor.hpp>
+#include <boost/asio/experimental/awaitable_operators.hpp>
 #include <boost/asio/strand.hpp>
 #include <boost/asio/use_awaitable.hpp>
 #include <google/protobuf/message.h>
@@ -14,8 +16,8 @@
 #include "cppesphomeapi/device_info.hpp"
 #include "make_unexpected_result.hpp"
 #include "message_wrapper.hpp"
+#include "net.hpp"
 #include "overloaded.hpp"
-#include "tcp.hpp"
 
 namespace cppesphomeapi
 {
@@ -25,6 +27,7 @@ class ApiConnection
     explicit ApiConnection(std::string hostname,
                            std::uint16_t port,
                            std::string password,
+                           std::stop_source &stop_source,
                            const boost::asio::any_io_executor &executor);
 
     AsyncResult<void> connect();
@@ -39,6 +42,8 @@ class ApiConnection
     AsyncResult<void> enable_logs(EspHomeLogLevel log_level, bool config_dump);
     AsyncResult<LogEntry> receive_log();
 
+    void cancel();
+
   private:
     AsyncResult<void> send_message(const google::protobuf::Message &message);
 
@@ -49,19 +54,29 @@ class ApiConnection
             std::unique_lock l{handler_mtx_};
             handlers_.emplace_back(std::move(handler));
         };
-        return boost::asio::async_initiate<CompletionToken, void(MessageWrapper)>(
-            init, // First, pass the function object that launches the operation,
-            token // then the completion token that will be transformed to a handler,
-        );        // and, finally, any additional arguments to the function object.
+        return boost::asio::async_initiate<CompletionToken, void(MessageWrapper)>(init, token);
     }
 
     template <typename TMsg>
     auto receive_message(auto &&completion_token) -> AsyncResult<std::shared_ptr<TMsg>>
     {
+        namespace aex = boost::asio::experimental;
+        using aex::awaitable_operators::operator||;
+        auto executor = co_await boost::asio::this_coro::executor;
+        net::Timer timer{executor};
+        timer.expires_after(std::chrono::milliseconds{250});
         while (true)
         {
-            const auto received_message =
-                co_await async_receive_message(std::forward<decltype(completion_token)>(completion_token));
+            //! TODO: the added handler might get aborted. How to remove it from the vector holding the completion
+            //! handlers?
+            const auto received_message_or_error =
+                co_await (async_receive_message(std::forward<decltype(completion_token)>(completion_token)) ||
+                          timer.async_wait());
+            if (std::holds_alternative<std::tuple<net::ErrorCode>>(received_message_or_error))
+            {
+                break;
+            }
+            auto &&received_message = std::get<cppesphomeapi::MessageWrapper>(received_message_or_error);
             if (received_message.template holds_message<TMsg>())
             {
                 co_return received_message.template as<TMsg>();
@@ -114,16 +129,14 @@ class ApiConnection
     }
 
   private:
-    boost::asio::awaitable<void> async_receive();
-    void spawn_heartbeat();
+    boost::asio::awaitable<void> receive_loop();
     boost::asio::awaitable<void> heartbeat_loop();
 
   private:
     std::string hostname_;
     std::uint16_t port_;
     std::string password_;
-    boost::asio::strand<boost::asio::any_io_executor> strand_;
-    tcp::socket socket_;
+    net::Socket socket_;
 
     std::string device_name_;
     std::optional<ApiVersion> api_version_;