453 rewrite safe stream on new async model

aether/CMakeLists.txt

@@ -204,15 +204,6 @@ list(APPEND transport_srcs
 list(APPEND stream_api_src
             "stream_api/sized_packet_gate.cpp"
             "stream_api/stream_api.cpp"
-
-            "stream_api/safe_stream.cpp"
-            "stream_api/safe_stream/safe_stream_api.cpp"
-            "stream_api/safe_stream/sending_data_action.cpp"
-            "stream_api/safe_stream/send_data_buffer.cpp"
-            "stream_api/safe_stream/sending_chunk_list.cpp"
-            "stream_api/safe_stream/receiving_chunk_list.cpp"
-            "stream_api/safe_stream/safe_stream_send_action.cpp"
-            "stream_api/safe_stream/safe_stream_recv_action.cpp"
 )
 
 list(APPEND write_action_srcs

aether/actions/timer.h

@@ -0,0 +1,64 @@
+/*
+ * 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_ACTIONS_TIMER_H_
+#define AETHER_ACTIONS_TIMER_H_
+
+#include <utility>
+#include <optional>
+#include <type_traits>
+
+#include "aether/clock.h"
+#include "aether/ae_context.h"
+#include "aether/actions/action2_.h"
+#include "aether/types/small_function.h"
+
+namespace ae {
+class Timer final : public a2::Action {
+ public:
+  using TimerFunc = SmallFunction<void()>;
+
+  template <typename F, typename Time = TimePoint>
+    requires(std::is_same_v<Time, TimePoint> || std::is_same_v<Time, Duration>)
+  Timer(AeContext const& ae_context, F&& func, Time timeout)
+      : ae_context_{ae_context},
+        alive_ctx_{std::in_place, ae_context_, this},
+        func_{std::forward<F>(func)} {
+    // schedule timer
+    ae_context_.scheduler().DelayedTask(
+        [imalive{alive_ctx_->View()}]() {
+          if (imalive) {
+            imalive->Invoke();
+          }
+        },
+        timeout);
+  }
+
+  void Reset() { alive_ctx_.reset(); }
+
+ private:
+  void Invoke() {
+    func_();
+    Finish();
+  }
+
+  AeContext ae_context_;
+  std::optional<IndexCtx<Timer>> alive_ctx_;
+  TimerFunc func_;
+};
+}  // namespace ae
+
+#endif  // AETHER_ACTIONS_TIMER_H_

aether/all.h

@@ -61,8 +61,8 @@
 #include "aether/types/static_map.h"
 #include "aether/types/server_id.h"
 #include "aether/types/client_id.h"
+#include "aether/types/ring_index.h"
 #include "aether/types/data_buffer.h"
-#include "aether/types/ring_buffer.h"
 #include "aether/types/client_config.h"
 #include "aether/types/server_config.h"
 #include "aether/types/state_machine.h"
@@ -71,7 +71,6 @@
 #include "aether/types/address_parser.h"
 
 #include "aether/stream_api/istream.h"
-#include "aether/stream_api/safe_stream.h"
 #include "aether/stream_api/api_call_adapter.h"
 #include "aether/write_action/write_action.h"
 #include "aether/write_action/buffer_write.h"

aether/client_messages/p2p_safe_message_stream.cpp

@@ -19,28 +19,31 @@
 #include <utility>
 
 #include "aether/stream_api/tied_gates.h"
+#include "aether/safe_stream/safe_stream.h"
 
 namespace ae {
 P2pSafeStream::P2pSafeStream(AeContext const& ae_context,
                              SafeStreamConfig const& config,
                              RcPtr<P2pStream> p2p_stream)
     : sized_packet_gate_{},
-      safe_stream_{ae_context, config},
+      safe_stream_{std::make_unique<SafeStream<12000>>(ae_context, config)},
       p2p_stream_{std::move(p2p_stream)},
       out_data_sub_{TiedEventOutData(
           [this](auto const& data) { out_data_event_.Emit(data); },
-          sized_packet_gate_, safe_stream_)} {
-  Tie(safe_stream_, *p2p_stream_);
+          sized_packet_gate_, *safe_stream_)} {
+  Tie(*safe_stream_, *p2p_stream_);
 }
 
+P2pSafeStream::~P2pSafeStream() = default;
+
 WriteAction& P2pSafeStream::Write(DataBuffer&& data) {
   auto sized_data = sized_packet_gate_.WriteIn(std::move(data));
-  return safe_stream_.Write(std::move(sized_data));
+  return safe_stream_->Write(std::move(sized_data));
 }
 
 StreamInfo P2pSafeStream::stream_info() const {
   auto overhead = sized_packet_gate_.Overhead();
-  auto info = safe_stream_.stream_info();
+  auto info = safe_stream_->stream_info();
   if (info.max_element_size < overhead) {
     info.max_element_size = 0;
   } else {
@@ -51,13 +54,13 @@ StreamInfo P2pSafeStream::stream_info() const {
 
 P2pSafeStream::StreamUpdateEvent::Subscriber
 P2pSafeStream::stream_update_event() {
-  return safe_stream_.stream_update_event();
+  return safe_stream_->stream_update_event();
 }
 
 P2pSafeStream::OutDataEvent::Subscriber P2pSafeStream::out_data_event() {
   return EventSubscriber{out_data_event_};
 }
 
-void P2pSafeStream::Restream() { safe_stream_.Restream(); }
+void P2pSafeStream::Restream() { safe_stream_->Restream(); }
 
 }  // namespace ae

aether/client_messages/p2p_safe_message_stream.h

@@ -22,18 +22,20 @@
 #include "aether/actions/action_context.h"
 
 #include "aether/stream_api/istream.h"
-#include "aether/stream_api/safe_stream.h"
 #include "aether/stream_api/sized_packet_gate.h"
-#include "aether/stream_api/safe_stream/safe_stream_config.h"
+#include "aether/safe_stream/safe_stream_config.h"
 
 #include "aether/client_messages/p2p_message_stream.h"
 
 namespace ae {
+template <std::size_t Capacity>
+class SafeStream;
 
 class P2pSafeStream final : public ByteIStream {
  public:
   P2pSafeStream(AeContext const& ae_context, SafeStreamConfig const& config,
                 RcPtr<P2pStream> p2p_stream);
+  ~P2pSafeStream() override;
 
   AE_CLASS_NO_COPY_MOVE(P2pSafeStream)
 
@@ -45,7 +47,8 @@ class P2pSafeStream final : public ByteIStream {
 
  private:
   SizedPacketGate sized_packet_gate_;
-  SafeStream safe_stream_;
+  // TODO: add config
+  std::unique_ptr<SafeStream<12000>> safe_stream_;
   RcPtr<P2pStream> p2p_stream_;
   OutDataEvent out_data_event_;
   std::array<Subscription, 2> out_data_sub_;

aether/safe_stream/details/circular_buffer.h

@@ -0,0 +1,168 @@
+/*
+ * 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_SAFE_STREAM_DETAILS_CIRCULAR_BUFFER_H_
+#define AETHER_SAFE_STREAM_DETAILS_CIRCULAR_BUFFER_H_
+
+#include <span>
+#include <array>
+#include <cstddef>
+#include <algorithm>
+
+#include "aether/types/result.h"
+#include "aether/types/ring_index.h"
+
+namespace ae {
+enum class CircularBufferError : unsigned char {
+  kDataOverflow,
+  kEmptyBuffer,
+  kIndexOutOfRange,
+};
+
+/**
+ * \brief Circular buffer to work with sending and receiving buffers
+ */
+template <std::size_t Capacity>
+class CircularBuffer {
+ public:
+  using value_type = std::uint8_t;
+  static constexpr std::size_t kCapacity = Capacity;
+  using index_type = RingIndex<kCapacity>;
+  using index_range_type = RingIndexRange<index_type>;
+
+  struct DSpan {
+    constexpr std::size_t size() const noexcept {
+      return first.size() + second.size();
+    }
+
+    std::span<value_type const> first;
+    std::span<value_type const> second;
+  };
+
+  constexpr CircularBuffer() noexcept = default;
+  explicit constexpr CircularBuffer(std::size_t start_offset) noexcept
+      : begin_{index_type{start_offset}}, end_{begin_} {}
+
+  /**
+   * \brief Push new data and return its range
+   */
+  constexpr Result<index_range_type, CircularBufferError> Push(
+      std::span<value_type const> data) noexcept {
+    return Insert(end_, data);
+  }
+
+  /**
+   * \brief Insert data at the given position
+   * \param pos - the position to insert at must be after begin_
+   * \param data - the data to insert
+   */
+  constexpr Result<index_range_type, CircularBufferError> Insert(
+      index_type pos, std::span<value_type const> data) noexcept {
+    auto available = (begin_ == pos) ? kCapacity : Distance(pos, begin_);
+    if (data.size() > available) {
+      return Error{CircularBufferError::kDataOverflow};
+    }
+    auto start = pos;
+    // copy the data into array
+    // expect for 2 copies depends on does data overlaps buffer or not
+    while (!data.empty()) {
+      auto remaining = buffer_.size() - static_cast<std::size_t>(pos);
+      auto copy_size = data.size() > remaining ? remaining : data.size();
+      std::copy_n(std::begin(data), copy_size,
+                  buffer_.data() + static_cast<std::size_t>(pos));
+      data = data.subspan(copy_size);
+      pos += copy_size;
+    }
+    // also move end_ to the right
+    if (IndexComparable{pos, begin_} > end_) {
+      end_ = pos;
+    }
+    // return the range of the data that was pushed
+    return Ok{index_range_type{.left = start, .right = pos - 1}};
+  }
+
+  /**
+   * \brief Read the data with size from start
+   * \param start - the position to read from must be in range begin_..end_
+   */
+  constexpr Result<DSpan, CircularBufferError> Read(
+      index_type start, std::size_t size) const noexcept {
+    if ((begin_ == end_) || Distance(start, end_) == 0) {
+      return Error{CircularBufferError::kEmptyBuffer};
+    }
+    if (Distance(start, end_) > Distance(begin_, end_)) {
+      return Error{CircularBufferError::kIndexOutOfRange};
+    }
+
+    DSpan res;
+
+    // size may be bigger than the actual available data, so we clamp it
+    auto max_distance = Distance(start, end_);
+    size = size > max_distance ? max_distance : size;
+
+    // handle wrapping around the buffer
+    std::size_t first_max_size =
+        buffer_.size() - static_cast<std::size_t>(start);
+    std::size_t first_size = first_max_size > size ? size : first_max_size;
+    res.first = std::span<value_type const>(
+        buffer_.data() + static_cast<std::size_t>(start), first_size);
+    if (first_size < size) {
+      res.second =
+          std::span<value_type const>(buffer_.data(), size - first_size);
+    }
+    return Ok{res};
+  }
+  /**
+   * \brief Erase data range
+   * \param to - the position to erase to must be in range begin_..end_
+   */
+  constexpr void Erase(index_type to) noexcept {
+    if (Distance(to, end_) > Distance(begin_, end_)) {
+      assert(false && "Erase out of bounds");
+      return;
+    }
+    begin_ = to;
+  }
+  /**
+   * \brief Erase data range from the specified position
+   * \param from - the position to erase from must be in range begin_..end_
+   */
+  constexpr void EraseFrom(index_type from) noexcept {
+    if (Distance(from, end_) > Distance(begin_, end_)) {
+      assert(false && "Erase out of bounds");
+      return;
+    }
+    end_ = from;
+  }
+
+  constexpr void Reset(std::size_t offset = 0) noexcept {
+    begin_ = end_ = index_type{offset};
+  }
+
+  constexpr std::size_t size() const noexcept { return Distance(begin_, end_); }
+  constexpr index_type begin() const noexcept { return begin_; }
+  constexpr index_type end() const noexcept { return end_; }
+
+ private:
+  index_type begin_{0};
+  index_type end_{0};
+
+  std::array<value_type, kCapacity> buffer_;
+};
+
+}  // namespace ae
+
+#endif  // AETHER_SAFE_STREAM_DETAILS_CIRCULAR_BUFFER_H_