496 make message stream more poliformic

.github/workflows/ci-cppcheck.yml

@@ -27,23 +27,23 @@ jobs:
         uses: actions/cache@v4
         with:
           path: _cppcheck
-          key: ${{ runner.os }}-cppcheck-2.19.0
-          restore-keys: ${{ runner.os }}-cppcheck-2.19.0
+          key: ${{ runner.os }}-cppcheck-2.21.0
+          restore-keys: ${{ runner.os }}-cppcheck-2.21.0
 
       - if: ${{ steps.cache-cppcheck.outputs.cache-hit != 'true' }}
         name: download and build cppcheck
         run: |
           mkdir -p _cppcheck
-          wget -O _cppcheck/cppcheck.tar.gz https://github.com/danmar/cppcheck/archive/refs/tags/2.19.0.tar.gz
+          wget -O _cppcheck/cppcheck.tar.gz https://github.com/danmar/cppcheck/archive/refs/tags/2.21.0.tar.gz
           cd _cppcheck
           tar -xzf cppcheck.tar.gz
-          cd cppcheck-2.19.0
+          cd cppcheck-2.21.0
           make -j
           pwd
           ls ./cppcheck
           ./cppcheck --version
           cd ../..
-          _cppcheck/cppcheck-2.19.0/cppcheck --version
+          _cppcheck/cppcheck-2.21.0/cppcheck --version
 
       # we need configure cmake to get the compile_commands.json file
       - name: cmake configure
@@ -71,7 +71,7 @@ jobs:
       - name: cppcheck
         run: >
           cd build &&
-          ../_cppcheck/cppcheck-2.19.0/cppcheck --project=compile_commands.json
+          ../_cppcheck/cppcheck-2.21.0/cppcheck --project=compile_commands.json
           -D__GNUC__=4
           --safety
           --error-exitcode=-1

AGENTS.md

@@ -93,9 +93,13 @@
  To run tests, go into `<build-dir>` and run `ctest . --progress -j -E "((sodium)|(hydro)|(bcrypt)).*" --output-on-failure`.
  Or run specific test by name from `<build-dir>/tests/run/<test-name>`.
 
- To run smoke test, run `<build-dir>/aether-client-cpp-cloud`. 
+### Smoke test
+
+ The first smoke test is a `<build-dir>/aether-client-cpp-cloud`.
+ To run it simply `cd <build-dir>` and ran the `./aether-client-cpp-cloud` binary. 
  Notice! Run `aether-client-cpp-cloud` generates `state` dir there object state is saved. 
  Remove this `state` dir before run to make clean run. Keep it to run with previous state.
 
 ## Operational Rules
+
 - Do not analyze logs until everything is working fine.

aether/CMakeLists.txt

@@ -213,7 +213,8 @@ list(APPEND aether_srcs
 list(APPEND aether_srcs
             "client_messages/p2p_message_stream.cpp"
             "client_messages/p2p_message_stream_manager.cpp"
-            "client_messages/p2p_safe_message_stream.cpp")
+            "client_messages/p2p_safe_message_stream.cpp"
+            "client_messages/p2p_port_handle.cpp")
 
 list(APPEND aether_srcs
             "domain_storage/domain_storage_factory.cpp"

aether/aether_c/aether_capi.cpp

@@ -17,6 +17,7 @@
 #include "aether/aether_c/aether_capi.h"
 
 #include <cassert>
+#include <memory>
 
 #include <string_view>
 
@@ -52,9 +53,9 @@
 static ae::RcPtr<ae::AetherApp> aether_app;
 
 struct AetherClient {
-  ClientConfig config;
+  ClientConfig config{};
   ae::Client::ptr client;
-  std::map<ae::Uid, ae::RcPtr<ae::P2pStream>> streams;
+  std::map<ae::Uid, std::shared_ptr<ae::ByteIStream>> message_streams;
   std::unique_ptr<ae::ActionsQueue> actions_queue_;
 };
 
@@ -63,8 +64,9 @@ std::unique_ptr<AetherClient, void (*)(AetherClient*)> default_client{
 
 void Listen(AetherClient* client, void* user_data);
 
-void ListenStream(AetherClient* client, void* user_data, CUid const& c_dest,
-                  ae::RcPtr<ae::P2pStream> const& stream);
+void ListenMessageStream(AetherClient* client, void* user_data,
+                         CUid const& c_dest,
+                         std::shared_ptr<ae::ByteIStream> const& stream);
 
 ae::SelectClientAction& SelectClientImpl(AetherClient* client,
                                          ClientConfig const* config) {
@@ -110,16 +112,20 @@ ae::WriteAction& WriteMessageImpl(AetherClient* client, ae::Uid destination,
                                   ae::DataBuffer&& data,
                                   ActionStatusCb status_cb, void* user_data) {
   assert(client->client);
-  auto it = client->streams.find(destination);
-  if (it == std::end(client->streams)) {
+  auto it = client->message_streams.find(destination);
+  if (it == std::end(client->message_streams)) {
     // create new stream
-    auto stream =
-        client->client->message_stream_manager().CreateStream(destination);
+    auto handle =
+        client->client->message_stream_manager().CreatePort(destination);
+    auto stream = std::make_shared<ae::P2pStream>(
+        ae::AeContext{*aether_app}, client->client.Load(), destination,
+        std::move(handle));
     auto c_dest =
         CUidFromBytes(destination.value.data(), destination.value.size());
-    ListenStream(client, user_data, c_dest, stream);
+    ListenMessageStream(client, user_data, c_dest, stream);
 
-    std::tie(it, std::ignore) = client->streams.emplace(destination, stream);
+    std::tie(it, std::ignore) =
+        client->message_streams.emplace(destination, stream);
   }
 
   auto& action = it->second->Write(std::move(data));
@@ -169,21 +175,22 @@ void Listen(AetherClient* client, void* user_data) {
     return;
   }
 
-  client->client->message_stream_manager().new_stream_event().Subscribe(
-      [client, user_data](ae::RcPtr<ae::P2pStream> stream) {
-        auto dest = stream->destination();
+  client->client->message_stream_manager().new_port_event().Subscribe(
+      [client, user_data](ae::P2pPortHandle handle) {
+        auto dest = handle.destination();
+        auto stream = std::make_shared<ae::P2pStream>(
+            ae::AeContext{*aether_app}, client->client.Load(), dest,
+            std::move(handle));
         // store the stream for future use
-        client->streams.emplace(dest, stream);
-
-        // create cuid for callback
+        client->message_streams.emplace(dest, stream);
         auto c_dest = CUidFromBytes(dest.value.data(), dest.value.size());
-        // listen for incoming messages
-        ListenStream(client, user_data, c_dest, stream);
+        ListenMessageStream(client, user_data, c_dest, stream);
       });
 }
 
-void ListenStream(AetherClient* client, void* user_data, CUid const& c_dest,
-                  ae::RcPtr<ae::P2pStream> const& stream) {
+void ListenMessageStream(AetherClient* client, void* user_data,
+                         CUid const& c_dest,
+                         std::shared_ptr<ae::ByteIStream> const& stream) {
   // special context saved in heap
   struct ListenContext {
     void* user_data;

aether/client_messages/p2p_message_stream.cpp

@@ -144,45 +144,16 @@ class MessageSendStream final : public IStream<AeMessage, AeMessage> {
   Subscription servers_update_sub_;
   MultiSubscription streams_update_sub_;
 };
-
-class ReadMessageGate {
- public:
-  using OutDataEvent = Event<void(DataBuffer const& data)>;
-
-  ReadMessageGate(Uid uid, CloudServerConnections& cloud_connection,
-                  RequestPolicy::Variant request_policy)
-      : uid_{uid},
-        message_event_sub_{
-            ApiEventSubscriber{[this](ClientApiSafe& client_api, auto*) {
-              return client_api.send_message_event().Subscribe(
-                  [this](auto const& message) {
-                    if (message.uid == uid_) {
-                      out_data_event_.Emit(message.data);
-                    }
-                  });
-            }},
-            cloud_connection,
-            request_policy,
-        } {}
-
-  OutDataEvent::Subscriber out_data_event() { return out_data_event_; }
-
- private:
-  Uid uid_;
-  CloudEventListener message_event_sub_;
-  OutDataEvent out_data_event_;
-};
-
 }  // namespace p2p_stream_internal
 
 P2pStream::P2pStream(AeContext const& ae_context, Ptr<Client> const& client,
-                     Uid destination)
+                     Uid destination, P2pPortHandle handle)
     : ae_context_{ae_context},
       client_{client},
-      destination_{destination},
+      destination_{std::move(destination)},
+      handle_{std::move(handle)},
       buffer_write_{ae_context_, MethodPtr<&P2pStream::OnWrite>{this}} {
   AE_TELE_DEBUG(kP2pMessageStreamNew, "P2pStream created for {}", destination_);
-  // destination uid must not be empty
   assert(!destination_.empty());
 
   ConnectReceive();
@@ -203,7 +174,6 @@ P2pStream::StreamUpdateEvent::Subscriber P2pStream::stream_update_event() {
 }
 
 StreamInfo P2pStream::stream_info() const {
-  // TODO: Combine info for receive and send streams
   if (message_send_stream_) {
     return message_send_stream_->stream_info();
   }
@@ -232,14 +202,7 @@ void P2pStream::WriteOut(DataBuffer const& data) {
 Uid const& P2pStream::destination() const { return destination_; }
 
 void P2pStream::ConnectReceive() {
-  auto client_ptr = client_.Lock();
-  assert(client_ptr);
-  // TODO: config request policy
-  read_message_gate_ = std::make_unique<p2p_stream_internal::ReadMessageGate>(
-      destination_, client_ptr->cloud_connection(),
-      RequestPolicy::Replica{client_ptr->cloud_connection().max_connections()});
-  // write out received data
-  read_message_gate_->out_data_event().Subscribe(
+  out_data_sub_ = handle_.out_data_event().Subscribe(
       MethodPtr<&P2pStream::WriteOut>{this});
 }
 
@@ -256,7 +219,6 @@ void P2pStream::ConnectSend() {
           dest_cloud_conn_ = MakeDestinationCloudConn(
               cloud.Load(), client_ptr->server_connection_manager()
                                 .GetServerConnectionFactory());
-          // TODO: add config for request policy
           message_send_stream_ =
               std::make_unique<p2p_stream_internal::MessageSendStream>(
                   *dest_cloud_conn_, RequestPolicy::MainServer{});

aether/client_messages/p2p_message_stream.h

@@ -25,38 +25,33 @@
 
 #include "aether/write_action/buffer_write.h"
 
-#include "aether/cloud_connections/cloud_server_connections.h"
+#include "aether/client_messages/p2p_port_handle.h"
 #include "aether/connection_manager/client_cloud_manager.h"
+#include "aether/cloud_connections/cloud_server_connections.h"
 
 namespace ae {
 class Client;
 class Cloud;
 namespace p2p_stream_internal {
 class MessageSendStream;
-class ReadMessageGate;
 }  // namespace p2p_stream_internal
 
 class P2pStream final : public ByteIStream {
  public:
   P2pStream(AeContext const& ae_context, Ptr<Client> const& client,
-            Uid destination);
+            Uid destination, P2pPortHandle handle);
 
   ~P2pStream() override;
 
   AE_CLASS_NO_COPY_MOVE(P2pStream);
 
   WriteAction& Write(DataBuffer&& data) override;
-
   StreamUpdateEvent::Subscriber stream_update_event() override;
-
   StreamInfo stream_info() const override;
-
   OutDataEvent::Subscriber out_data_event() override;
-
   void Restream() override;
 
   void WriteOut(DataBuffer const& data);
-
   Uid const& destination() const;
 
  private:
@@ -72,16 +67,18 @@ class P2pStream final : public ByteIStream {
   PtrView<Client> client_;
   Uid destination_{};
 
+  P2pPortHandle handle_;
+
   // connection to destination cloud
   std::unique_ptr<CloudServerConnections> dest_cloud_conn_;
   BufferWrite<AeMessage, 100> buffer_write_;
   std::unique_ptr<p2p_stream_internal::MessageSendStream> message_send_stream_;
-  std::unique_ptr<p2p_stream_internal::ReadMessageGate> read_message_gate_;
 
   OutDataEvent out_data_event_;
   StreamUpdateEvent stream_update_event_;
 
   Subscription get_client_cloud_sub_;
+  Subscription out_data_sub_;
 };
 
 }  // namespace ae