diff --git a/CMakeLists.txt b/CMakeLists.txt index a12bef95..150d9446 100644 --- a/CMakeLists.txt +++ b/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() diff --git a/aether/CMakeLists.txt b/aether/CMakeLists.txt index 2348b6f5..65743ba4 100644 --- a/aether/CMakeLists.txt +++ b/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") diff --git a/aether/ae_actions/ping.cpp b/aether/ae_actions/ping.cpp index d358b256..a3a2b860 100644 --- a/aether/ae_actions/ping.cpp +++ b/aether/ae_actions/ping.cpp @@ -19,74 +19,127 @@ # include -# 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 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& auth_api) { - auto ping_interval_u64 = static_cast( - std::chrono::duration_cast( - 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& auth_api) { + auto ping_interval_u64 = static_cast( + std::chrono::duration_cast( + ping_interval_) + .count()); + auto rx_window_u64 = static_cast( + std::chrono::duration_cast( + 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& 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(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 diff --git a/aether/ae_actions/ping.h b/aether/ae_actions/ping.h index f0a29de6..633a84ba 100644 --- a/aether/ae_actions/ping.h +++ b/aether/ae_actions/ping.h @@ -31,16 +31,17 @@ IGNORE_IMPLICIT_CONVERSION() # include 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; @@ -48,20 +49,26 @@ class Ping { struct PingRequest { TimePoint start; RequestId request_id; + Subscription wait_result_sub; + TaskSubscription timeout_sub; + Subscription write_sub; }; public: - using PingFailed = Event; + using ResultEvent = Event)>; - Ping(AeContext const& ae_context, Ptr 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(); @@ -69,18 +76,18 @@ class Ping { void PingResponseTimeout(RequestId request_id); AeContext ae_context_; - PtrView channel_; - ClientServerConnection* client_server_connection_; + CloudServerConnection* cloud_server_connection_; Duration ping_interval_; + Duration rx_window_; + Duration timeout_; + ServerId server_id_; etl::circular_buffer, 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 diff --git a/aether/client.cpp b/aether/client.cpp index acae8640..67a297e6 100644 --- a/aether/client.cpp +++ b/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( + *aether_.Load().as(), *cloud_connection_); +#endif + #if AE_TELE_ENABLED // also create telemetry - telemetry_ = std::make_unique( - AeContext{*aether_.Load().as()}, *cloud_connection_); + telemetry_ = std::make_unique(*aether_.Load().as(), + *cloud_connection_); #endif } diff --git a/aether/client.h b/aether/client.h index 12f9f113..9090a556 100644 --- a/aether/client.h +++ b/aether/client.h @@ -27,11 +27,14 @@ #include "aether/types/uid.h" #include "aether/server_keys.h" +#include "aether/cloud_connections/ping_cloud_servers.h" #include "aether/cloud_connections/cloud_server_connections.h" + #include "aether/connection_manager/client_cloud_manager.h" -#include "aether/client_messages/p2p_message_stream_manager.h" #include "aether/connection_manager/server_connection_manager.h" +#include "aether/client_messages/p2p_message_stream_manager.h" + namespace ae { class Aether; class Telemetry; @@ -87,6 +90,9 @@ class Client : public Obj { std::unique_ptr cloud_connection_; std::unique_ptr message_stream_manager_; +#if AE_ENABLE_PING + std::unique_ptr ping_cloud_servers_; +#endif #if AE_TELE_ENABLED std::unique_ptr telemetry_; #endif diff --git a/aether/cloud_connections/ping_cloud_servers.cpp b/aether/cloud_connections/ping_cloud_servers.cpp new file mode 100644 index 00000000..a04aa9ab --- /dev/null +++ b/aether/cloud_connections/ping_cloud_servers.cpp @@ -0,0 +1,147 @@ +/* + * Copyright 2026 Aethernet Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "aether/cloud_connections/ping_cloud_servers.h" + +#if AE_ENABLE_PING + +# include "aether/server.h" +# include "aether/channels/channel.h" + +# include "aether/cloud_connections/cloud_connections_tele.h" + +namespace ae { +PingCloudServers::PingCloudServers( + AeContext const& ae_context, + CloudServerConnections& cloud_server_connections) + : ae_context_{ae_context}, + cloud_server_connections_{&cloud_server_connections}, + servers_update_{ + cloud_server_connections_->servers_update_event().Subscribe( + MethodPtr<&PingCloudServers::ServersUpdate>{this})} { + AE_TELED_INFO("PingCloudServers created"); + ServersUpdate(); +} + +void PingCloudServers::ServersUpdate() { + AE_TELED_DEBUG("Servers update"); + // enqueue redispatch for servers + if (!task_sub_) { + AE_TELED_DEBUG("Enqueue dispatch servers"); + task_sub_ = ae_context_.scheduler().Task([this]() { DispatchToServers(); }); + } +} + +void PingCloudServers::DispatchToServers() { + cloud_server_connections_->ForServers( + [this](CloudServerConnection* cloud_sc) { + if (cloud_sc == nullptr) { + AE_TELED_ERROR("Visit empty cloud server connection!"); + return; + } + MakePingToServer(*cloud_sc); + }, + // TODO: config request policy + RequestPolicy::All{}); +} + +void PingCloudServers::MakePingToServer(CloudServerConnection& cloud_sc) { + auto [sid, server_ping, is_new] = GetOrCreatePing(cloud_sc); + AE_TELED_DEBUG("Make ping to server {}, is_new {}", sid, is_new); + if (!is_new) { + return; + } + + auto c = cloud_sc.client_connection()->server_connection().current_channel(); + if (!c) { + AE_TELED_ERROR("Current channel value invalid"); + return; + } + auto timeout = c->ResponseTimeout(); + constexpr auto kDefaultInterval = + std::chrono::milliseconds{AE_PING_INTERVAL_MS}; + constexpr auto kDefaultRxWindow = kDefaultInterval; + + // TODO: different ping interval depends on priority + server_ping.ping = std::make_unique( + ae_context_, cloud_sc, kDefaultInterval, kDefaultRxWindow, timeout); + + // subscribe to ping result + // if success get ping time from result and save for statistics + // if failed request Restream for what server connection + server_ping.ping->result_event().Subscribe([this, sid_ = sid, + csc_ = &cloud_sc](auto res) { + auto it = server_pings_.find(sid_); + if (it == server_pings_.end()) { + return; + } + auto& sp = it->second; + if (res) { + // successful ping save timeout to the current channel + auto c = csc_->client_connection()->server_connection().current_channel(); + if (!c) { + AE_TELED_ERROR("Ping's channel value invalid"); + return; + } + c->channel_statistics().AddResponseTime(res.value()); + // and update timeout value + sp.ping->SetTimeout(c->ResponseTimeout()); + } else { + // ping error + AE_TELED_ERROR("Ping error!"); + // after restream if server is changed + csc_->Restream(); + } + }); + // reset ping on server error + server_ping.server_state_sub = cloud_sc.client_connection() + ->server_connection() + .server_error_event() + .Subscribe([this, sid_ = sid]() { + auto it = server_pings_.find(sid_); + if (it == server_pings_.end()) { + return; + } + it->second.ping.reset(); + }); +} + +auto PingCloudServers::GetOrCreatePing(CloudServerConnection& cloud_sc) + -> std::tuple { + auto server = cloud_sc.server(); + assert(server && "Server must exists"); + + auto server_id = server->server_id; + auto it = server_pings_.find(server_id); + // Create new ping or replace ping if server's priority changed + if ((it == server_pings_.end()) || + (it->second.priority != cloud_sc.priority())) { + auto [new_it, _] = server_pings_.insert_or_assign( + server_id, ServerPing{ + .ping{}, + .priority = cloud_sc.priority(), + .server_state_sub = {}, + }); + return {server_id, new_it->second, true}; + } + + // if ping not has value return is_new + return {server_id, it->second, !it->second.ping}; +} + +} // namespace ae + +#endif diff --git a/aether/cloud_connections/ping_cloud_servers.h b/aether/cloud_connections/ping_cloud_servers.h new file mode 100644 index 00000000..76738b75 --- /dev/null +++ b/aether/cloud_connections/ping_cloud_servers.h @@ -0,0 +1,65 @@ +/* + * Copyright 2026 Aethernet Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef AETHER_CLOUD_CONNECTIONS_PING_CLOUD_SERVERS_H_ +#define AETHER_CLOUD_CONNECTIONS_PING_CLOUD_SERVERS_H_ + +#include "aether/config.h" +#if AE_ENABLE_PING + +# include +# include +# include + +# include "aether/ae_context.h" +# include "aether/types/server_id.h" +# include "aether/events/event_subscription.h" +# include "aether/tasks/manual_task_scheduler.h" +# include "aether/cloud_connections/cloud_server_connections.h" + +# include "aether/ae_actions/ping.h" + +namespace ae { +class PingCloudServers { + struct ServerPing { + std::unique_ptr ping; + std::size_t priority; + Subscription server_state_sub; + }; + + public: + PingCloudServers(AeContext const& ae_context, + CloudServerConnections& cloud_server_connections); + + private: + void ServersUpdate(); + void DispatchToServers(); + void MakePingToServer(CloudServerConnection& cloud_sc); + std::tuple GetOrCreatePing( + CloudServerConnection& cloud_sc); + + AeContext ae_context_; + CloudServerConnections* cloud_server_connections_; + + Subscription servers_update_; + TaskSubscription task_sub_; + + std::map server_pings_; +}; +} // namespace ae + +#endif +#endif // AETHER_CLOUD_CONNECTIONS_PING_CLOUD_SERVERS_H_ diff --git a/aether/registration/api/registration_root_api.cpp b/aether/registration/api/registration_root_api.cpp index b11cd545..872dd0ec 100644 --- a/aether/registration/api/registration_root_api.cpp +++ b/aether/registration/api/registration_root_api.cpp @@ -24,6 +24,7 @@ RegistrationRootApi::RegistrationRootApi(ProtocolContext& protocol_context, : ApiClass{protocol_context}, get_asymmetric_public_key{protocol_context}, enter{protocol_context, EnterProc{*this}}, + get_my_ip{protocol_context}, enc_provider_{&root_encrypt}, server_registration_api_{protocol_context, global_encrypt} {} diff --git a/aether/registration/api/registration_root_api.h b/aether/registration/api/registration_root_api.h index b541aae4..979e19ac 100644 --- a/aether/registration/api/registration_root_api.h +++ b/aether/registration/api/registration_root_api.h @@ -27,6 +27,8 @@ # include "aether/crypto/icrypto_provider.h" # include "aether/api_protocol/api_protocol.h" +# include "aether/work_cloud_api/info_ip.h" + # include "aether/registration/api/server_registration_api.h" namespace ae { @@ -58,6 +60,8 @@ class RegistrationRootApi : public ApiClass { EnterProc> enter; + Method<6, ApiPromise()> get_my_ip; + private: DataBuffer Encrypt(DataBuffer const& data) const; diff --git a/aether/registration/registration.cpp b/aether/registration/registration.cpp index 28853e4f..a6d06194 100644 --- a/aether/registration/registration.cpp +++ b/aether/registration/registration.cpp @@ -117,6 +117,20 @@ auto Registration::GetKeys() { AE_TELED_INFO("Key received"); ex::set_value(std::move(r), Ignore{}); }); + +# if DEBUG + // For debug, call also for get my ip method to print our public ip + // visible to registration server + api_call->get_my_ip().Subscribe([](auto&& res) { + if (res) { + auto& iip = res.value(); + AE_TELED_INFO("Registration our public ip: {}:{}, coords: {},{}", + iip.ip, iip.port, iip.latitude, iip.longitude); + } else { + AE_TELED_ERROR("Get my ip failed"); + } + }); +# endif }); } @@ -327,13 +341,12 @@ void Registration::Run() { return ex::just_error(-1); }}); - waiter_.emplace( - ae_context_, std::move(s), - [&](std::optional>&& res) noexcept { - assert(res); - registration_event_.Emit(std::move(res).value()); - Finish(); - }); + waiter_.emplace(ae_context_, std::move(s), + [&](std::optional>&& res) noexcept { + assert(res); + registration_event_.Emit(std::move(res).value()); + Finish(); + }); } } // namespace ae #endif diff --git a/aether/server_connections/client_server_connection.cpp b/aether/server_connections/client_server_connection.cpp index 04ee2901..66d29b46 100644 --- a/aether/server_connections/client_server_connection.cpp +++ b/aether/server_connections/client_server_connection.cpp @@ -20,8 +20,8 @@ #include "aether/server.h" #include "aether/client.h" #include "aether/crypto/ikey_provider.h" -#include "aether/stream_api/api_call_adapter.h" #include "aether/api_protocol/api_protocol.h" +#include "aether/stream_api/api_call_adapter.h" #include "aether/crypto/sync_crypto_provider.h" #include "aether/tele/tele.h" @@ -78,8 +78,8 @@ class ClientDecryptKeyProvider : public ClientKeyProvider { class ClientCryptoProvider final : public ICryptoProvider { public: ClientCryptoProvider(Ptr const& client, ServerId server_id) - : encryptor_{ - std::make_unique(client, server_id)}, + : encryptor_{std::make_unique(client, + server_id)}, decryptor_{ std::make_unique(client, server_id)} {} @@ -155,10 +155,6 @@ ClientServerConnection::ClientServerConnection(AeContext const& ae_context, server_connection_.out_data_event().Subscribe( MethodPtr<&ClientServerConnection::OutData>{this}); - - server_connection_.server_connection.channel_changed_event().Subscribe( - MethodPtr<&ClientServerConnection::ChannelChanged>{this}); - ChannelChanged(); } ClientServerConnection::~ClientServerConnection() { @@ -206,37 +202,4 @@ void ClientServerConnection::OutData(DataBuffer const& data) { parser.Parse(client_api_unsafe_); } -void ClientServerConnection::ChannelChanged() { - AE_TELED_DEBUG("Channel is updated, make new ping"); - - auto make_ping = [&] { -#if AE_ENABLE_PING - auto& server_conn = server_connection_.server_connection; - auto channel = server_conn.current_channel(); - // Create new ping if channel is updated - static constexpr Duration kPingDefaultInterval = - std::chrono::milliseconds{AE_PING_INTERVAL_MS}; - // TODO: make ping interval depend on server priority - ping_.emplace(ae_context_, channel, *this, kPingDefaultInterval); - ping_sub_ = ping_->ping_failed().Subscribe([this]() { - AE_TELED_ERROR("Ping failed"); - server_connection_.Restream(); - }); -#endif - }; - - if (server_connection_.stream_info().link_state == LinkState::kLinked) { - make_ping(); - return; - } - wait_connection_sub_ = - server_connection_.stream_update_event().Subscribe([&, make_ping]() { - if (server_connection_.stream_info().link_state != LinkState::kLinked) { - return; - } - wait_connection_sub_.Reset(); - make_ping(); - }); -} - } // namespace ae diff --git a/aether/server_connections/client_server_connection.h b/aether/server_connections/client_server_connection.h index 2c5abb42..83a0aac2 100644 --- a/aether/server_connections/client_server_connection.h +++ b/aether/server_connections/client_server_connection.h @@ -19,7 +19,6 @@ #include "aether/common.h" #include "aether/ae_context.h" -#include "aether/ae_actions/ping.h" #include "aether/crypto/icrypto_provider.h" #include "aether/write_action/buffer_write.h" @@ -77,7 +76,6 @@ class ClientServerConnection { private: void OutData(DataBuffer const& data); - void ChannelChanged(); AeContext ae_context_; PtrView server_; @@ -90,13 +88,6 @@ class ClientServerConnection { client_server_connection_internal ::BufferedServerConnection server_connection_; - -#if AE_ENABLE_PING - std::optional ping_; -#endif - - Subscription ping_sub_; - Subscription wait_connection_sub_; }; } // namespace ae diff --git a/aether/work_cloud_api/info_ip.h b/aether/work_cloud_api/info_ip.h new file mode 100644 index 00000000..506b9ff0 --- /dev/null +++ b/aether/work_cloud_api/info_ip.h @@ -0,0 +1,37 @@ +/* + * Copyright 2026 Aethernet Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef AETHER_WORK_CLOUD_IP_H_ +#define AETHER_WORK_CLOUD_IP_H_ + +#include + +#include "aether-miscpp/reflect/reflect.h" + +#include "aether/types/address.h" + +namespace ae { +struct InfoIp { + AE_REFLECT_MEMBERS(ip, port, latitude, longitude) + + Address ip; + std::uint16_t port; + double latitude; + double longitude; +}; +} // namespace ae + +#endif // AETHER_WORK_CLOUD_IP_H_ diff --git a/aether/work_cloud_api/work_server_api/login_api.cpp b/aether/work_cloud_api/work_server_api/login_api.cpp index ea60f705..d76b28aa 100644 --- a/aether/work_cloud_api/work_server_api/login_api.cpp +++ b/aether/work_cloud_api/work_server_api/login_api.cpp @@ -25,6 +25,7 @@ LoginApi::LoginApi(ProtocolContext& protocol_context, get_time_utc{protocol_context}, login_by_uid{protocol_context, LoginProc{*this}}, login_by_alias{protocol_context, LoginProc{*this}}, + get_my_ip{protocol_context}, encrypt_provider_{&encrypt_provider}, auth_api_{protocol_context} {} diff --git a/aether/work_cloud_api/work_server_api/login_api.h b/aether/work_cloud_api/work_server_api/login_api.h index 2554705d..a71e4899 100644 --- a/aether/work_cloud_api/work_server_api/login_api.h +++ b/aether/work_cloud_api/work_server_api/login_api.h @@ -22,6 +22,7 @@ #include "aether/crypto/icrypto_provider.h" #include "aether/api_protocol/api_protocol.h" +#include "aether/work_cloud_api/info_ip.h" #include "aether/work_cloud_api/work_server_api/authorized_api.h" namespace ae { @@ -49,6 +50,8 @@ class LoginApi : public ApiClass { Method<5, void(Uid alias, SubApi sub_api), LoginProc> login_by_alias; + Method<6, ApiPromise()> get_my_ip; + AuthorizedApi& authorized_api() { return auth_api_; } private: