499 make ping on a clients cloud level

CMakeLists.txt

@@ -19,6 +19,27 @@ set(CMAKE_CXX_STANDARD_REQUIRED ON)
 
 set(AE_PROJECT_VERSION "0.1.0")
 
+find_program(GIT_COMMAND git)
+if ( GIT_COMMAND )
+  # get current git version
+  execute_process(
+        COMMAND ${GIT_COMMAND} -C ${CMAKE_CURRENT_SOURCE_DIR} rev-parse --verify HEAD
+        OUTPUT_VARIABLE GIT_VERSION
+    )
+  if ( NOT GIT_VERSION )
+    message(WARNING "Not a git repo")
+  else()
+    string(STRIP ${GIT_VERSION} GIT_VERSION)
+    message(STATUS "Get aether git version ${GIT_VERSION}")
+    # also show current commit message
+    execute_process(
+          COMMAND ${GIT_COMMAND} -C ${CMAKE_CURRENT_SOURCE_DIR} show -s --format=%h:%D:%s
+          OUTPUT_VARIABLE GIT_DESCRIBE
+      )
+    message(STATUS "Aether head on\n\t${GIT_DESCRIBE}")
+  endif()
+endif()
+
 project(aether VERSION ${AE_PROJECT_VERSION} LANGUAGES CXX C)
 
 set(TARGET_NAME "${PROJECT_NAME}")
@@ -210,21 +231,6 @@ if (CMAKE_SYSTEM_NAME STREQUAL "Windows")
   target_link_libraries(${TARGET_NAME} PRIVATE ws2_32)
 endif()
 
-find_program(GIT_COMMAND git)
-if ( GIT_COMMAND )
-  # get current git version
-  execute_process(
-        COMMAND ${GIT_COMMAND} -C ${CMAKE_CURRENT_SOURCE_DIR} rev-parse --verify HEAD
-        OUTPUT_VARIABLE GIT_VERSION
-    )
-  if ( NOT GIT_VERSION )
-    message(WARNING "Not a git repo")
-  else()
-    string(STRIP ${GIT_VERSION} GIT_VERSION)
-    message(STATUS "get aether git version ${GIT_VERSION}")
-  endif()
-endif()
-
 if (GIT_VERSION)
   target_compile_definitions(${TARGET_NAME} PUBLIC "AE_GIT_VERSION=\"${GIT_VERSION}\"")
 endif()

aether/CMakeLists.txt

@@ -197,6 +197,7 @@ list(APPEND aether_srcs
 list(APPEND aether_srcs
             "cloud_connections/cloud_server_connection.cpp"
             "cloud_connections/cloud_server_connections.cpp"
+            "cloud_connections/ping_cloud_servers.cpp"
             "cloud_connections/cloud_subscription.cpp"
             "cloud_connections/cloud_request.cpp")
 

aether/ae_actions/ping.cpp

@@ -19,74 +19,127 @@
 
 #  include <optional>
 
-#  include "aether/channels/channel.h"
+#  include "aether/server.h"
+#  include "aether/cloud_connections/cloud_server_connection.h"
 #  include "aether/server_connections/client_server_connection.h"
 #  include "aether/work_cloud_api/work_server_api/authorized_api.h"
 
 #  include "aether/ae_actions/ae_actions_tele.h"
 
 namespace ae {
-Ping::Ping(AeContext const& ae_context, Ptr<Channel> const& channel,
-           ClientServerConnection& client_server_connection,
-           Duration ping_interval)
+Ping::Ping(AeContext const& ae_context,
+           CloudServerConnection& cloud_server_connection,
+           Duration ping_interval, Duration rx_window, Duration timeout)
     : ae_context_{ae_context},
-      channel_{channel},
-      client_server_connection_{&client_server_connection},
-      ping_interval_{ping_interval} {
-  AE_TELE_INFO(kPing, "Ping action created, interval {:%S}s", ping_interval);
-  // send ping on the next tick
-  schedule_sub_ = ae_context_.scheduler().Task([&]() { SendPing(); });
+      cloud_server_connection_{&cloud_server_connection},
+      ping_interval_{ping_interval},
+      rx_window_{rx_window},
+      timeout_{timeout},
+      server_id_{cloud_server_connection_->server()->server_id} {
+  AE_TELE_INFO(
+      kPing,
+      "Ping action created to server id: {}, interval: {:%S}s, rx_window: "
+      "{:%S}s, timeout: {:%S}s",
+      server_id_, ping_interval_, rx_window_, timeout_);
+
+  ScheduleFirstPing();
 }
 
-Ping::PingFailed::Subscriber Ping::ping_failed() {
-  return EventSubscriber{ping_failed_};
+Ping::ResultEvent::Subscriber Ping::result_event() { return result_event_; }
+
+void Ping::SetTimeout(Duration timeout) {
+  // Only next ping will use new timeout
+  timeout_ = timeout;
+}
+
+void Ping::ScheduleFirstPing() {
+  // TODO: calculate actual next ping time
+  auto* cc = cloud_server_connection_->client_connection();
+  assert(cc != nullptr && "Client connection is null");
+
+  // send first ping only after client connection is fully linked
+  if (cc->stream_info().link_state == LinkState::kLinked) {
+    // send ping on the next tick
+    schedule_sub_ = ae_context_.scheduler().Task([&]() { SendPing(); });
+  } else {
+    link_state_sub_ = cc->stream_update_event().Subscribe([this, cc]() {
+      if (cc->stream_info().link_state == LinkState::kLinked) {
+        link_state_sub_.Reset();
+        // send ping on the next tick
+        schedule_sub_ = ae_context_.scheduler().Task([&]() { SendPing(); });
+      }
+    });
+  }
 }
 
 void Ping::SendPing() {
   AE_TELE_DEBUG(kPingSend, "Send ping");
 
-  auto& write_action = client_server_connection_->AuthorizedApiCall(
-      SubApi{[this](ApiContext<AuthorizedApi>& auth_api) {
-        auto ping_interval_u64 = static_cast<std::uint64_t>(
-            std::chrono::duration_cast<std::chrono::milliseconds>(
-                ping_interval_)
-                .count());
-        // FIXME: for rx_window send interval value
-        auto& pong_promise =
-            auth_api->ping(ping_interval_u64, ping_interval_u64);
-        auto req_id = pong_promise.request_id();
-
-        // save the ping request
-        auto channel_ptr = channel_.Lock();
-        assert(channel_ptr);
-        auto expected_ping_time = channel_ptr->ResponseTimeout();
-        auto current_time = Now();
-        auto end_time = current_time + expected_ping_time;
-        AE_TELED_DEBUG("Ping request expected time {:%S}s", expected_ping_time);
-
-        ping_requests_.push(PingRequest{
-            current_time,
-            req_id,
-        });
-
-        // Wait for response
-        wait_responses_ += pong_promise.Subscribe(
-            [&, req_id](auto&&...) { PingResponse(req_id); });
-
-        // setup response timeout
-        // FIXME multi sub
-        timeout_sub_ = ae_context_.scheduler().DelayedTask(
-            [this, req_id]() { PingResponseTimeout(req_id); }, end_time);
-      }});
-
-  write_subs_ += write_action.status_event().Subscribe([&](auto status) {
+  auto& write_action =
+      cloud_server_connection_->client_connection()->AuthorizedApiCall(
+          SubApi{[this](ApiContext<AuthorizedApi>& auth_api) {
+            auto ping_interval_u64 = static_cast<std::uint64_t>(
+                std::chrono::duration_cast<std::chrono::milliseconds>(
+                    ping_interval_)
+                    .count());
+            auto rx_window_u64 = static_cast<std::uint64_t>(
+                std::chrono::duration_cast<std::chrono::milliseconds>(
+                    rx_window_)
+                    .count());
+
+            auto& pong_promise =
+                auth_api->ping(ping_interval_u64, rx_window_u64);
+            auto req_id = pong_promise.request_id();
+
+            // save the ping request
+            AE_TELED_DEBUG("Ping server id {}, request {} expected time {:%S}s",
+                           server_id_, req_id, timeout_);
+            auto current_time = Now();
+            auto end_time = current_time + timeout_;
+
+            ping_requests_.push(PingRequest{
+                .start = current_time,
+                .request_id = req_id,
+                // Wait for response
+                .wait_result_sub = pong_promise.Subscribe(
+                    [&, req_id](auto&&...) { PingResponse(req_id); }),
+                .timeout_sub = ae_context_.scheduler().DelayedTask(
+                    [this, req_id]() { PingResponseTimeout(req_id); },
+                    end_time),
+                .write_sub = {},
+            });
+          }});
+
+  auto& req = ping_requests_.back();
+  assert(req.has_value() &&
+         "After call AuthorizedApiCall ping request should be saved");
+
+  req->write_sub = write_action.status_event().Subscribe([&](auto status) {
     if (status == WriteAction::Status::kFail) {
       AE_TELE_ERROR(kPingWriteError, "Ping write error");
-      ping_failed_.Emit();
+      result_event_.Emit(Error{1});
     }
   });
 
-  // setup ping interval
+#  if DEBUG
+  // For debug, call also for get my ip method to print our public ip
+  // visible to work server
+  cloud_server_connection_->client_connection()->LoginApiCall(
+      SubApi{[&](ApiContext<LoginApi>& login_api) {
+        login_api->get_my_ip().Subscribe([sid_ =
+                                              server_id_](auto&& res) noexcept {
+          if (res) {
+            auto& iip = res.value();
+            AE_TELED_DEBUG("Server id: {}, our public ip: {}:{}, coords: {},{}",
+                           sid_, iip.ip, iip.port, iip.latitude, iip.longitude);
+          } else {
+            AE_TELED_ERROR("Get my ip failed!");
+          }
+        });
+      }});
+#  endif
+
+  // setup next ping interval
   schedule_sub_ = ae_context_.scheduler().DelayedTask([this]() { SendPing(); },
                                                       ping_interval_);
 }
@@ -97,7 +150,7 @@ void Ping::PingResponse(RequestId request_id) {
       [&](auto const& p) { return p && (p->request_id == request_id); });
 
   if (request_it == std::end(ping_requests_)) {
-    AE_TELED_DEBUG("Got lost, or not our pong response");
+    AE_TELED_WARNING("Got lost, or not our pong response");
     return;
   }
 
@@ -107,24 +160,28 @@ void Ping::PingResponse(RequestId request_id) {
   auto ping_duration =
       std::chrono::duration_cast<Duration>(current_time - request->start);
 
-  AE_TELED_DEBUG("Ping received by {:%S} s", ping_duration);
-  auto channel_ptr = channel_.Lock();
-  assert(channel_ptr);
-  channel_ptr->channel_statistics().AddResponseTime(ping_duration);
-
   // reset request as finished
   request.reset();
+
+  AE_TELED_DEBUG("Ping server id {} request {} received by {:%S} s", server_id_,
+                 request_id, ping_duration);
+  result_event_.Emit(Ok{ping_duration});
 }
 
 void Ping::PingResponseTimeout(RequestId request_id) {
-  for (auto& p : ping_requests_) {
-    if (p && (p->request_id == request_id)) {
-      p.reset();
-      // timeout
-      AE_TELE_ERROR(kPingTimeout, "Ping timeout");
-      ping_failed_.Emit();
-    }
+  auto request_it = std::find_if(
+      std::begin(ping_requests_), std::end(ping_requests_),
+      [&](auto const& p) { return p && (p->request_id == request_id); });
+
+  if (request_it == std::end(ping_requests_)) {
+    AE_TELED_WARNING("Timeout for lost, or not our pong response");
+    return;
   }
+
+  request_it->reset();
+  AE_TELE_ERROR(kPingTimeout, "Ping server id {} request {} timeout",
+                server_id_, request_id);
+  result_event_.Emit(Error{2});
 }
 
 }  // namespace ae

aether/ae_actions/ping.h

@@ -31,56 +31,63 @@ IGNORE_IMPLICIT_CONVERSION()
 #  include <etl/circular_buffer.h>
 DISABLE_WARNING_POP()
 
-#  include "aether/ptr/ptr.h"
+#  include "aether-miscpp/types/result.h"
+
 #  include "aether/ae_context.h"
-#  include "aether/ptr/ptr_view.h"
 #  include "aether/events/events.h"
+#  include "aether/types/server_id.h"
 #  include "aether/api_protocol/request_id.h"
-#  include "aether/events/multi_subscription.h"
+#  include "aether/events/event_subscription.h"
 
 namespace ae {
 class Channel;
-class ClientServerConnection;
+class CloudServerConnection;
 
 class Ping {
   static constexpr std::uint8_t kMaxStorePingTimes = 10;
 
   struct PingRequest {
     TimePoint start;
     RequestId request_id;
+    Subscription wait_result_sub;
+    TaskSubscription timeout_sub;
+    Subscription write_sub;
   };
 
  public:
-  using PingFailed = Event<void()>;
+  using ResultEvent = Event<void(Result<Duration, int>)>;
 
-  Ping(AeContext const& ae_context, Ptr<Channel> const& channel,
-       ClientServerConnection& client_server_connection,
-       Duration ping_interval);
+  Ping(AeContext const& ae_context,
+       CloudServerConnection& cloud_server_connection, Duration ping_interval,
+       Duration rx_window, Duration timeout);
 
   AE_CLASS_NO_COPY_MOVE(Ping);
 
-  PingFailed::Subscriber ping_failed();
+  ResultEvent::Subscriber result_event();
+
+  void SetTimeout(Duration timeout);
 
  private:
+  void ScheduleFirstPing();
   void SendPing();
   TimePoint WaitInterval();
   TimePoint WaitResponse();
   void PingResponse(RequestId request_id);
   void PingResponseTimeout(RequestId request_id);
 
   AeContext ae_context_;
-  PtrView<Channel> channel_;
-  ClientServerConnection* client_server_connection_;
+  CloudServerConnection* cloud_server_connection_;
   Duration ping_interval_;
+  Duration rx_window_;
+  Duration timeout_;
+  ServerId server_id_;
 
   etl::circular_buffer<std::optional<PingRequest>, kMaxStorePingTimes>
       ping_requests_;
 
-  PingFailed ping_failed_;
-  MultiSubscription write_subs_;
-  MultiSubscription wait_responses_;
+  ResultEvent result_event_;
+  Subscription link_state_sub_;
   TaskSubscription schedule_sub_;
-  TaskSubscription timeout_sub_;
 };
 }  // namespace ae
 #endif  // AE_ENABLE_PING

aether/client.cpp

@@ -70,10 +70,15 @@ CloudServerConnections& Client::cloud_connection() {
         server_connection_manager().GetServerConnectionFactory(),
         AE_CLOUD_MAX_SERVER_CONNECTIONS);
 
+#if AE_ENABLE_PING
+    ping_cloud_servers_ = std::make_unique<PingCloudServers>(
+        *aether_.Load().as<Aether>(), *cloud_connection_);
+#endif
+
 #if AE_TELE_ENABLED
     // also create telemetry
-    telemetry_ = std::make_unique<Telemetry>(
-        AeContext{*aether_.Load().as<Aether>()}, *cloud_connection_);
+    telemetry_ = std::make_unique<Telemetry>(*aether_.Load().as<Aether>(),
+                                             *cloud_connection_);
 #endif
   }