diff --git a/CMakeLists.txt b/CMakeLists.txt index 76f2806c..afe913f6 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -364,6 +364,9 @@ if(AE_BUILD_EXAMPLES) set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}) add_subdirectory(examples/cloud) + if(CMAKE_SYSTEM_NAME STREQUAL "Windows") + add_subdirectory(examples/windows_message_receiver) + endif() add_subdirectory(examples/capi/oddity) add_subdirectory(examples/benches/send_message_delays) add_subdirectory(examples/benches/send_messages_bandwidth) diff --git a/aether/CMakeLists.txt b/aether/CMakeLists.txt index 13a303d1..0418d8f4 100644 --- a/aether/CMakeLists.txt +++ b/aether/CMakeLists.txt @@ -198,6 +198,9 @@ list(APPEND aether_srcs list(APPEND aether_srcs "server_connections/client_server_connection.cpp" + "prepared_packet/prepared_send_message.cpp" + "prepared_packet/prepare_send_message.cpp" + "prepared_packet/packet_encoder.cpp" "server_connections/channel_connection.cpp" "server_connections/server_connection.cpp") diff --git a/aether/client_messages/p2p_message_stream.cpp b/aether/client_messages/p2p_message_stream.cpp index c202645f..23ac7ecf 100644 --- a/aether/client_messages/p2p_message_stream.cpp +++ b/aether/client_messages/p2p_message_stream.cpp @@ -25,6 +25,7 @@ #include "aether/cloud_connections/cloud_request.h" #include "aether/cloud_connections/cloud_subscription.h" +#include "aether/cloud_connections/cloud_server_connection.h" #include "aether/client_messages/client_messages_tele.h" @@ -48,6 +49,30 @@ class MessageSendStream final : public IStream { }}, request_policy_); } + + std::optional + ExportPreparedSendMessageBlock(Uid target_uid, + std::uint32_t reserve_nonce_count) { + for (auto* sc : cloud_connection_->servers()) { + if (sc == nullptr) { + continue; + } + + auto* conn = sc->client_connection(); + if (conn == nullptr) { + continue; + } + + auto block = conn->ExportPreparedSendMessageBlock(target_uid, + reserve_nonce_count); + if (block) { + return block; + } + } + + return std::nullopt; + } + StreamInfo stream_info() const override { return stream_info_; } OutDataEvent::Subscriber out_data_event() override { return out_data_event_; } StreamUpdateEvent::Subscriber stream_update_event() override { @@ -203,6 +228,16 @@ void P2pStream::WriteOut(DataBuffer const& data) { Uid const& P2pStream::destination() const { return destination_; } +std::optional +P2pStream::ExportPreparedSendMessageBlock(std::uint32_t reserve_nonce_count) { + if (message_send_stream_ == nullptr) { + return std::nullopt; + } + + return message_send_stream_->ExportPreparedSendMessageBlock( + destination_, reserve_nonce_count); +} + void P2pStream::ConnectReceive() { out_data_sub_ = handle_.out_data_event().Subscribe(MethodPtr<&P2pStream::WriteOut>{this}); diff --git a/aether/client_messages/p2p_message_stream.h b/aether/client_messages/p2p_message_stream.h index 64da048b..e243174f 100644 --- a/aether/client_messages/p2p_message_stream.h +++ b/aether/client_messages/p2p_message_stream.h @@ -17,6 +17,9 @@ #ifndef AETHER_CLIENT_MESSAGES_P2P_MESSAGE_STREAM_H_ #define AETHER_CLIENT_MESSAGES_P2P_MESSAGE_STREAM_H_ +#include +#include + #include "aether/common.h" #include "aether/ae_context.h" @@ -28,6 +31,7 @@ #include "aether/client_messages/p2p_port_handle.h" #include "aether/cloud_connections/cloud_server_connections.h" #include "aether/connection_manager/client_cloud_manager.h" +#include "aether/prepared_packet/prepared_send_message.h" namespace ae { class Client; @@ -54,6 +58,9 @@ class P2pStream final : public ByteIStream { void WriteOut(DataBuffer const& data); Uid const& destination() const; + std::optional + ExportPreparedSendMessageBlock(std::uint32_t reserve_nonce_count); + private: void ConnectReceive(); void ConnectSend(); diff --git a/aether/prepared_packet/packet_encoder.cpp b/aether/prepared_packet/packet_encoder.cpp new file mode 100644 index 00000000..0c87e7bb --- /dev/null +++ b/aether/prepared_packet/packet_encoder.cpp @@ -0,0 +1,73 @@ +#include "aether/prepared_packet/packet_encoder.h" + +#include +#include + +#include "aether/crypto/ikey_provider.h" +#include "aether/crypto/sync_crypto_provider.h" + +#include "aether/api_protocol/api_context.h" +#include "aether/api_protocol/sub_api.h" + +#include "aether/work_cloud_api/ae_message.h" +#include "aether/work_cloud_api/work_server_api/authorized_api.h" +#include "aether/work_cloud_api/work_server_api/login_api.h" + +namespace ae::prepared_packet { +namespace { + +class PreparedSendMessageKeyProvider final : public ISyncKeyProvider { + public: + explicit PreparedSendMessageKeyProvider(PreparedSendMessageBlock& block) + : block_{&block} {} + + Key GetKey() const override { + return block_->client_to_server_key; + } + + CryptoNonce const& Nonce() const override { + return block_->next_nonce; + } + + private: + PreparedSendMessageBlock* block_; +}; + +} // namespace + +EncodePacketResult EncodePacket(PreparedSendMessageBlock& block, + DataBuffer const& payload, + DataBuffer& out) { + if (block.nonce_left == 0) { + out.clear(); + return EncodePacketResult{EncodePacketError::kNonceExhausted, 0}; + } + + // Match the existing ClientKeyProvider semantics: + // consume next nonce before encryption. + block.next_nonce.Next(); + --block.nonce_left; + + auto key_provider = + std::make_unique(block); + SyncEncryptProvider encrypt_provider{std::move(key_provider)}; + + ProtocolContext protocol_context; + LoginApi login_api{protocol_context, encrypt_provider}; + + auto api_context = ApiContext{login_api}; + + api_context->login_by_alias( + block.sender_ephemeral_uid, + SubApi{ + [&block, &payload](ApiContext& auth_api) { + auth_api->send_message( + AeMessage{block.target_uid, DataBuffer{payload}}); + }}); + + out = std::move(api_context).Pack(); + + return EncodePacketResult{EncodePacketError::kNone, out.size()}; +} + +} // namespace ae::prepared_packet diff --git a/aether/prepared_packet/packet_encoder.h b/aether/prepared_packet/packet_encoder.h new file mode 100644 index 00000000..6f9d7371 --- /dev/null +++ b/aether/prepared_packet/packet_encoder.h @@ -0,0 +1,21 @@ +/* + * Prepared packet encoder. + * + * EncodePacket only builds Aether packet bytes and advances the reserved nonce + * range. It does not send, open sockets, resolve DNS, or know platform + * transport. + */ +#ifndef AETHER_PREPARED_PACKET_PACKET_ENCODER_H_ +#define AETHER_PREPARED_PACKET_PACKET_ENCODER_H_ + +#include "aether/prepared_packet/prepared_send_message.h" + +namespace ae::prepared_packet { + +EncodePacketResult EncodePacket(PreparedSendMessageBlock& block, + DataBuffer const& payload, + DataBuffer& out); + +} // namespace ae::prepared_packet + +#endif // AETHER_PREPARED_PACKET_PACKET_ENCODER_H_ diff --git a/aether/prepared_packet/prepare_send_message.cpp b/aether/prepared_packet/prepare_send_message.cpp new file mode 100644 index 00000000..2a56800f --- /dev/null +++ b/aether/prepared_packet/prepare_send_message.cpp @@ -0,0 +1,28 @@ +/* + * Copyright 2025 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/prepared_packet/prepare_send_message.h" + +#include "aether/client_messages/p2p_message_stream.h" + +namespace ae::prepared_packet { + +std::optional PrepareSendMessage( + P2pStream& stream, std::uint32_t reserve_nonce_count) { + return stream.ExportPreparedSendMessageBlock(reserve_nonce_count); +} + +} // namespace ae::prepared_packet diff --git a/aether/prepared_packet/prepare_send_message.h b/aether/prepared_packet/prepare_send_message.h new file mode 100644 index 00000000..08d0e710 --- /dev/null +++ b/aether/prepared_packet/prepare_send_message.h @@ -0,0 +1,37 @@ +/* + * Copyright 2025 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_PREPARED_PACKET_PREPARE_SEND_MESSAGE_H_ +#define AETHER_PREPARED_PACKET_PREPARE_SEND_MESSAGE_H_ + +#include +#include + +#include "aether/prepared_packet/prepared_send_message.h" + +namespace ae { + +class P2pStream; + +namespace prepared_packet { + +std::optional PrepareSendMessage( + P2pStream& stream, std::uint32_t reserve_nonce_count); + +} // namespace prepared_packet +} // namespace ae + +#endif // AETHER_PREPARED_PACKET_PREPARE_SEND_MESSAGE_H_ diff --git a/aether/prepared_packet/prepared_send_message.cpp b/aether/prepared_packet/prepared_send_message.cpp new file mode 100644 index 00000000..db965ee6 --- /dev/null +++ b/aether/prepared_packet/prepared_send_message.cpp @@ -0,0 +1,41 @@ +#include "aether/prepared_packet/prepared_send_message.h" + +#include + +namespace ae::prepared_packet { + +std::optional MakePreparedEndpoint(Endpoint const& endpoint) { + PreparedEndpoint candidate; + candidate.protocol = endpoint.protocol; + candidate.port = endpoint.port; + + bool is_ip = false; + + std::visit( + [&](auto const& addr) { + using T = std::decay_t; + + if constexpr (std::is_same_v) { + candidate.version = PreparedIpVersion::kIpV4; + for (std::size_t i = 0; i < 4; ++i) { + candidate.ip[i] = addr.ipv4_value[i]; + } + is_ip = true; + } else if constexpr (std::is_same_v) { + candidate.version = PreparedIpVersion::kIpV6; + for (std::size_t i = 0; i < 16; ++i) { + candidate.ip[i] = addr.ipv6_value[i]; + } + is_ip = true; + } + }, + endpoint.address); + + if (!is_ip) { + return std::nullopt; + } + + return candidate; +} + +} // namespace ae::prepared_packet diff --git a/aether/prepared_packet/prepared_send_message.h b/aether/prepared_packet/prepared_send_message.h new file mode 100644 index 00000000..fff70874 --- /dev/null +++ b/aether/prepared_packet/prepared_send_message.h @@ -0,0 +1,94 @@ +/* + * Prepared send_message block. + * + * This is transport-neutral state for encoding a send_message packet. + * It may contain an endpoint selected by the full Aether client, but it does + * not own sockets, DNS, connections, channels, or timers. + */ +#ifndef AETHER_PREPARED_PACKET_PREPARED_SEND_MESSAGE_H_ +#define AETHER_PREPARED_PACKET_PREPARED_SEND_MESSAGE_H_ + +#include +#include +#include +#include + +#include "aether/types/address.h" +#include "aether/types/data_buffer.h" +#include "aether/types/uid.h" + +#include "aether/crypto/key.h" +#include "aether/crypto/crypto_nonce.h" + +namespace ae::prepared_packet { +static constexpr std::uint32_t kMagic = 0x50534456; // "PSDV" +static constexpr std::uint32_t kVersion = 1; +// Serialized PreparedSendMessageBlock is a few hundred bytes. Keep the RTC +// footprint small enough for ESP32 RTC slow memory (8 KiB on ESP32-C6). +static constexpr std::size_t kMaxPreparedBlockBytes = 512; + +struct RetainedPreparedBlock { + std::uint32_t magic; + std::uint32_t version; + std::uint32_t size; + std::uint32_t checksum; + std::array bytes; +}; + +enum class PreparedIpVersion : std::uint8_t { + kIpV4 = 4, + kIpV6 = 6, +}; + +struct PreparedEndpoint { + AE_REFLECT_MEMBERS(version, protocol, port, ip) + PreparedIpVersion version = PreparedIpVersion::kIpV4; + Protocol protocol = Protocol::kUdp; + std::uint16_t port = 0; + + // IPv4 uses first 4 bytes. + // IPv6 uses all 16 bytes. + std::array ip{}; +}; + +struct PreparedSendMessageBlock { + AE_REFLECT_MEMBERS(endpoint, sender_ephemeral_uid, target_uid, + client_to_server_key, next_nonce, nonce_left) + PreparedEndpoint endpoint; + + Uid sender_ephemeral_uid; + Uid target_uid; + + Key client_to_server_key; + + CryptoNonce next_nonce; + std::uint32_t nonce_left = 0; +}; + +enum class EncodePacketError { + kNone = 0, + kNonceExhausted, +}; + +inline char const* ToString(EncodePacketError error) { + switch (error) { + case EncodePacketError::kNone: + return "none"; + case EncodePacketError::kNonceExhausted: + return "nonce_exhausted"; + } + return "unknown"; +} + +struct EncodePacketResult { + EncodePacketError error = EncodePacketError::kNone; + std::size_t bytes_written = 0; + + explicit operator bool() const { return error == EncodePacketError::kNone; } +}; + +std::optional MakePreparedEndpoint(Endpoint const& endpoint); + +} // namespace ae::prepared_packet + +#endif // AETHER_PREPARED_PACKET_PREPARED_SEND_MESSAGE_H_ diff --git a/aether/server_connections/client_server_connection.cpp b/aether/server_connections/client_server_connection.cpp index 66d29b46..bf173e8e 100644 --- a/aether/server_connections/client_server_connection.cpp +++ b/aether/server_connections/client_server_connection.cpp @@ -142,6 +142,7 @@ ClientServerConnection::ClientServerConnection(AeContext const& ae_context, Ptr const& client, Ptr const& server) : ae_context_{ae_context}, + client_{client}, server_{server}, ephemeral_uid_{client->ephemeral_uid()}, crypto_provider_{std::make_unique< @@ -197,6 +198,60 @@ ServerConnection& ClientServerConnection::server_connection() { return server_connection_.server_connection; } +std::optional +ClientServerConnection::ExportPreparedSendMessageBlock( + Uid target_uid, std::uint32_t reserve_nonce_count) { + auto client = client_.Lock(); + auto server = server_.Lock(); + + if ((client == nullptr) || (server == nullptr) || + (reserve_nonce_count == 0)) { + return std::nullopt; + } + + std::optional prepared_endpoint; + + for (auto const& e : server->endpoints) { + if (e.protocol != Protocol::kUdp) { + continue; + } + + auto endpoint = prepared_packet::MakePreparedEndpoint(e); + if (endpoint) { + prepared_endpoint = *endpoint; + break; + } + } + + if (!prepared_endpoint) { + return std::nullopt; + } + + auto* server_key = client->server_state(server->server_id); + if (server_key == nullptr) { + return std::nullopt; + } + + prepared_packet::PreparedSendMessageBlock block; + block.endpoint = *prepared_endpoint; + block.sender_ephemeral_uid = ephemeral_uid_; + block.target_uid = target_uid; + block.client_to_server_key = server_key->client_to_server(); + + // Store current nonce. EncodePacket() will call Next() before encryption, + // matching existing ClientKeyProvider behavior. + block.next_nonce = server_key->nonce(); + block.nonce_left = reserve_nonce_count; + + // Burn/reserve the same nonce range in the full client, + // so the normal Aether path cannot reuse it later. + for (std::uint32_t i = 0; i < reserve_nonce_count; ++i) { + server_key->Next(); + } + + return block; +} + void ClientServerConnection::OutData(DataBuffer const& data) { auto parser = ApiParser{protocol_context_, data}; parser.Parse(client_api_unsafe_); diff --git a/aether/server_connections/client_server_connection.h b/aether/server_connections/client_server_connection.h index 83a0aac2..d1c54796 100644 --- a/aether/server_connections/client_server_connection.h +++ b/aether/server_connections/client_server_connection.h @@ -17,6 +17,9 @@ #ifndef AETHER_SERVER_CONNECTIONS_CLIENT_SERVER_CONNECTION_H_ #define AETHER_SERVER_CONNECTIONS_CLIENT_SERVER_CONNECTION_H_ +#include +#include + #include "aether/common.h" #include "aether/ae_context.h" #include "aether/crypto/icrypto_provider.h" @@ -27,6 +30,7 @@ #include "aether/work_cloud_api/client_api/client_api_unsafe.h" #include "aether/work_cloud_api/work_server_api/authorized_api.h" +#include "aether/prepared_packet/prepared_send_message.h" #include "aether/server_connections/server_connection.h" namespace ae { @@ -74,10 +78,15 @@ class ClientServerConnection { ServerConnection& server_connection(); + std::optional + ExportPreparedSendMessageBlock(Uid target_uid, + std::uint32_t reserve_nonce_count); + private: void OutData(DataBuffer const& data); AeContext ae_context_; + PtrView client_; PtrView server_; Uid ephemeral_uid_; diff --git a/examples/windows_message_receiver/CMakeLists.txt b/examples/windows_message_receiver/CMakeLists.txt new file mode 100644 index 00000000..ede10f15 --- /dev/null +++ b/examples/windows_message_receiver/CMakeLists.txt @@ -0,0 +1,50 @@ +# Copyright 2024 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. + +cmake_minimum_required(VERSION 3.16.0) + +set(CMAKE_CXX_STANDARD 20) +set(CMAKE_CXX_STANDARD_REQUIRED ON) + +list( APPEND src_list + windows_message_receiver.cpp +) + +if(NOT CM_PLATFORM) + project("aether-client-cpp-windows_message_receiver" VERSION "1.0.0" LANGUAGES C CXX) + + add_executable(${PROJECT_NAME} ${src_list}) + target_include_directories(${PROJECT_NAME} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}) + target_link_libraries(${PROJECT_NAME} PRIVATE aether) +else() + idf_build_get_property(CM_PLATFORM CM_PLATFORM) + if(CM_PLATFORM STREQUAL "ESP32") + #ESP32 CMake + idf_component_register(SRCS ${src_list} + INCLUDE_DIRS ${CMAKE_CURRENT_SOURCE_DIR} + REQUIRES + esp_wifi + esp_netif + nvs_flash + spiffs + esp_driver_uart + ) + + add_subdirectory("../../" aether) + target_link_libraries(${COMPONENT_LIB} PRIVATE aether) + else() + #Other platforms + message(FATAL_ERROR "Platform ${CM_PLATFORM} is not supported") + endif() +endif() diff --git a/examples/windows_message_receiver/config/cloud_config.ini b/examples/windows_message_receiver/config/cloud_config.ini new file mode 100644 index 00000000..0c275714 --- /dev/null +++ b/examples/windows_message_receiver/config/cloud_config.ini @@ -0,0 +1,38 @@ +; Static configuration for cloud test +; Register to save state as a header file by +; aether-registrator +; Use header file to build the client test cmake -DFS_INIT= . && cmake --build . + +; Required section +; Aether configuration +[Aether] +; Must have one of the following keys or both depending on the project configuration \see aether/config.h +ed25519_sign_key = 4F202A94AB729FE9B381613AE77A8A7D89EDAB9299C3320D1A0B994BA710CCEB +hydrogen_sign_key = 883B4D7E0FB04A38CA12B3A451B00942048858263EE6E6D61150F2EF15F40343 + +; Optional section +; If provided, wifi adapter would be configured and saved to the state +[AdapterWifi] +ssid = Test123 +pass = Test123 + +; Must have at least one registration server +; Registration server configuration +; Supported protocols: kTcp, kUdp +[RegServer_1] +address = registration.aethernet.io +port=9010 +protocol=kTcp + +[RegServer_2] +address = 34.60.244.148 +port=9010 +protocol=kTcp + +; List of clients +; Suffix after _ is used as user defined client id \see Aether::SelectClient +[Client_A] +parent_uid = 3ac93165-3d37-4970-87a6-fa4ee27744e4 + +[Client_B] +parent_uid = 3ac93165-3d37-4970-87a6-fa4ee27744e4 diff --git a/examples/windows_message_receiver/user_config.h b/examples/windows_message_receiver/user_config.h new file mode 100644 index 00000000..a2fd7142 --- /dev/null +++ b/examples/windows_message_receiver/user_config.h @@ -0,0 +1,39 @@ +/* + * Copyright 2024 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 USER_CONFIG_H_ +#define USER_CONFIG_H_ + +#include "aether/config_consts.h" +/** + * \brief For full config list and default values \see aether/config.h + */ + +// use hydrogen encryption +#define AE_CRYPTO_ASYNC AE_HYDRO_CRYPTO_PK +#define AE_CRYPTO_SYNC AE_HYDRO_CRYPTO_SK +#define AE_SIGNATURE AE_HYDRO_SIGNATURE +#define AE_KDF AE_HYDRO_KDF + +// disable debug telemetry on release +#define AE_TELE_ENABLED 1 +#define AE_TELE_LOG_CONSOLE 1 +#if defined NDEBUG +# define AE_TELE_DEBUG_MODULES 0 +#else +# define AE_TELE_DEBUG_MODULES AE_ALL +#endif +#endif // USER_CONFIG_H_ diff --git a/examples/windows_message_receiver/windows_message_receiver.cpp b/examples/windows_message_receiver/windows_message_receiver.cpp new file mode 100644 index 00000000..647760fe --- /dev/null +++ b/examples/windows_message_receiver/windows_message_receiver.cpp @@ -0,0 +1,144 @@ +/* + * Copyright 2024 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 +#include + +#include "aether/all.h" + +static constexpr auto kParentUid = + ae::Uid::FromString("B1AC52C8-8D94-BD39-4C01-A631AC594165"); + +class TimeSynchronizer { + public: + TimeSynchronizer() = default; + + void SetPingSentTime(ae::TimePoint ping_sent_time); + void SetPongSentTime(ae::TimePoint pong_sent_time); + + ae::Duration GetPingDuration() const; + ae::Duration GetPongDuration() const; + + private: + ae::TimePoint ping_sent_time_; + ae::TimePoint pong_sent_time_; +}; + +// Alice sends "ping"s to Bob +/* class Alice { + public: + explicit Alice(ae::AetherApp& aether_app, ae::Client::ptr client_alice, + TimeSynchronizer& time_synchronizer, ae::Uid bobs_uid); + + private: + void SendMessage(); + void ResponseReceived(ae::DataBuffer const& data_buffer); + + ae::AetherApp* aether_app_; + ae::Client::ptr client_alice_; + TimeSynchronizer* time_synchronizer_; + ae::P2pStream p2pstream_; + ae::RepeatableTask interval_sender_; + ae::Subscription receive_data_sub_; + ae::MultiSubscription send_subs_; +};*/ + +// Bob answers "pong" to each "ping" +class Bob { + public: + explicit Bob(ae::AetherApp& aether_app, ae::Client::ptr client_bob, + TimeSynchronizer& time_synchronizer); + + private: + void OnNewStream(ae::P2pPortHandle p2p_port); + void OnMessageReceived(ae::DataBuffer const& data_buffer); + + ae::AetherApp* aether_app_; + ae::Client::ptr client_bob_; + TimeSynchronizer* time_synchronizer_; + std::unique_ptr p2pstream_; + ae::Subscription new_stream_receive_sub_; + ae::Subscription message_receive_sub_; +}; + +int main() { + auto aether_app = ae::AetherApp::Construct(ae::AetherAppContext{}); + + //std::unique_ptr alice; + std::unique_ptr bob; + TimeSynchronizer time_synchronizer; + + // register or load clients + auto& bob_select = aether_app->aether()->SelectClient(kParentUid, "Bob"); + bob_select.result_event().Subscribe([&](auto const& bob_res) { + if (bob_res) { + bob = + ae::make_unique(*aether_app, bob_res.value(), time_synchronizer); + } else { + aether_app->Exit(1); + } + }); + + while (!aether_app->IsExited()) { + auto next_time = aether_app->Update(ae::Now()); + aether_app->WaitUntil(next_time); + } + return aether_app->ExitCode(); +} + +void TimeSynchronizer::SetPingSentTime(ae::TimePoint ping_sent_time) { + ping_sent_time_ = ping_sent_time; +} +void TimeSynchronizer::SetPongSentTime(ae::TimePoint pong_sent_time) { + pong_sent_time_ = pong_sent_time; +} +ae::Duration TimeSynchronizer::GetPingDuration() const { + return std::chrono::duration_cast(ae::Now() - ping_sent_time_); +} +ae::Duration TimeSynchronizer::GetPongDuration() const { + return std::chrono::duration_cast(ae::Now() - pong_sent_time_); +} + +Bob::Bob(ae::AetherApp& aether_app, ae::Client::ptr client_bob, + TimeSynchronizer& time_synchronizer) + : aether_app_{&aether_app}, + client_bob_{std::move(client_bob)}, + time_synchronizer_{&time_synchronizer}, + new_stream_receive_sub_{ + client_bob_->message_stream_manager().new_port_event().Subscribe( + ae::MethodPtr<&Bob::OnNewStream>{this})} {} + +void Bob::OnNewStream(ae::P2pPortHandle p2p_port) { + p2pstream_ = std::make_unique(*aether_app_, client_bob_.Load(), + p2p_port.destination(), + std::move(p2p_port)); + message_receive_sub_ = p2pstream_->out_data_event().Subscribe( + ae::MethodPtr<&Bob::OnMessageReceived>{this}); +} + +void Bob::OnMessageReceived(ae::DataBuffer const& data_buffer) { + auto ping_message = std::string_view{ + reinterpret_cast(data_buffer.data()), data_buffer.size()}; + std::cout << ae::Format( + ">>>\n>>>\n>>>\n[{:%H:%M:%S}] Bob received \"{}\" within time {} ms\n>>>\n>>>\n>>>\n", ae::Now(), + ping_message, + std::chrono::duration_cast( + time_synchronizer_->GetPingDuration()) + .count()); + std::cout << ae::Format("Hex string [{}]\n", data_buffer); + time_synchronizer_->SetPongSentTime(ae::Now()); + constexpr std::string_view pong_message = "pong"; +}