Introduced ScatteredBuffer::IFragmentsObserver interface

Author: serges147

include/libcyphal/transport/can/delegate.hpp

@@ -101,6 +101,14 @@ class CanardMemory final : public ScatteredBuffer::IStorage
         return bytes_to_copy;
     }
 
+    void observeFragments(ScatteredBuffer::IFragmentsObserver& observer) const override
+    {
+        if ((buffer_ != nullptr) && (payload_size_ > 0))
+        {
+            observer.onNext({buffer_, payload_size_});
+        }
+    }
+
 private:
     // MARK: Data members:
 

include/libcyphal/transport/msg_sessions.hpp

@@ -7,6 +7,7 @@
 #define LIBCYPHAL_TRANSPORT_MSG_SESSIONS_HPP_INCLUDED
 
 #include "errors.hpp"
+#include "scattered_buffer.hpp"
 #include "session.hpp"
 #include "types.hpp"
 
@@ -34,6 +35,18 @@ struct MessageTxParams final
     PortId subject_id{};
 };
 
+struct MessageRxMetadata final
+{
+    TransferRxMetadata     rx_meta{};
+    cetl::optional<NodeId> publisher_node_id;
+};
+
+struct MessageRxTransfer final
+{
+    MessageRxMetadata metadata{};
+    ScatteredBuffer   payload;
+};
+
 /// @brief Defines an abstract interface of a transport layer receive session for message subscription.
 ///
 /// Use transport's `makeMessageRxSession` factory function to create an instance of this interface.

include/libcyphal/transport/scattered_buffer.hpp

@@ -7,6 +7,7 @@
 #define LIBCYPHAL_TRANSPORT_SCATTERED_BUFFER_HPP_INCLUDED
 
 #include "libcyphal/config.hpp"
+#include "types.hpp"
 
 #include <cetl/pf17/cetlpf.hpp>
 #include <cetl/rtti.hpp>
@@ -33,6 +34,26 @@ class ScatteredBuffer final
     ///
     static constexpr std::size_t StorageVariantFootprint = config::Transport::ScatteredBuffer_StorageVariantFootprint();
 
+    /// @brief Defines interface for observing internal fragments of the scattered buffer.
+    ///
+    class IFragmentsObserver
+    {
+    public:
+        IFragmentsObserver(const IFragmentsObserver&)                = delete;
+        IFragmentsObserver& operator=(const IFragmentsObserver&)     = delete;
+        IFragmentsObserver& operator=(IFragmentsObserver&&) noexcept = delete;
+        IFragmentsObserver(IFragmentsObserver&&) noexcept            = delete;
+
+        /// @brief Notifies the observer about the next fragment of the scattered buffer.
+        ///
+        virtual void onNext(const PayloadFragment fragment) = 0;
+
+    protected:
+        IFragmentsObserver()  = default;
+        ~IFragmentsObserver() = default;
+
+    };  // IFragmentsObserver
+
     /// @brief Defines storage interface for the scattered buffer.
     ///
     /// @see ScatteredBuffer::ScatteredBuffer(AnyStorage&& any_storage)
@@ -72,6 +93,12 @@ class ScatteredBuffer final
                                  cetl::byte* const destination,
                                  const std::size_t length_bytes) const = 0;
 
+        /// @brief Reports the internal fragments of the storage to the specified observer.
+        ///
+        /// @param observer The observer will be called (by `onNext` method) for each fragment of the storage.
+        ///
+        virtual void observeFragments(IFragmentsObserver& observer) const = 0;
+
         // MARK: RTTI
 
         static constexpr cetl::type_id _get_type_id_() noexcept
@@ -201,6 +228,18 @@ class ScatteredBuffer final
         return storage_->copy(offset_bytes, static_cast<cetl::byte*>(destination), length_bytes);
     }
 
+    /// @brief Reports the internal fragments of the buffer to the specified observer.
+    ///
+    /// @param observer The observer will be called (by `onNext` method) for each fragment of the buffer.
+    ///
+    void observeFragments(IFragmentsObserver& observer) const
+    {
+        if (const auto* const storage = storage_)
+        {
+            storage->observeFragments(observer);
+        }
+    }
+
 private:
     cetl::unbounded_variant<StorageVariantFootprint, false, true> storage_variant_;
     const IStorage*                                               storage_;

include/libcyphal/transport/svc_sessions.hpp

@@ -7,6 +7,7 @@
 #define LIBCYPHAL_TRANSPORT_SVC_SESSION_HPP_INCLUDED
 
 #include "errors.hpp"
+#include "scattered_buffer.hpp"
 #include "session.hpp"
 #include "types.hpp"
 
@@ -46,6 +47,24 @@ struct ResponseTxParams final
     PortId service_id{};
 };
 
+struct ServiceTxMetadata final
+{
+    TransferTxMetadata tx_meta{};
+    NodeId             remote_node_id{};
+};
+
+struct ServiceRxMetadata final
+{
+    TransferRxMetadata rx_meta{};
+    NodeId             remote_node_id{};
+};
+
+struct ServiceRxTransfer final
+{
+    ServiceRxMetadata metadata{};
+    ScatteredBuffer   payload;
+};
+
 /// @brief Defines an abstract interface of a transport layer receive session for service.
 ///
 /// @see IRxSession, ISession

include/libcyphal/transport/transport.hpp

@@ -21,6 +21,13 @@ namespace libcyphal
 namespace transport
 {
 
+struct ProtocolParams final
+{
+    TransferId  transfer_id_modulo{};
+    std::size_t mtu_bytes{};
+    NodeId      max_nodes{};
+};
+
 /// @brief Interface for a transport layer.
 ///
 class ITransport

include/libcyphal/transport/types.hpp

@@ -6,8 +6,6 @@
 #ifndef LIBCYPHAL_TRANSPORT_TYPES_HPP_INCLUDED
 #define LIBCYPHAL_TRANSPORT_TYPES_HPP_INCLUDED
 
-#include "scattered_buffer.hpp"
-
 #include "libcyphal/types.hpp"
 
 #include <cetl/pf17/cetlpf.hpp>
@@ -48,59 +46,32 @@ enum class Priority : std::uint8_t
     Optional    = 7,
 };
 
-struct ProtocolParams final
-{
-    TransferId  transfer_id_modulo{};
-    std::size_t mtu_bytes{};
-    NodeId      max_nodes{};
-};
+/// @brief Defines immutable fragment of raw data (as span of const bytes).
+///
+using PayloadFragment = cetl::span<const cetl::byte>;
+
+/// @brief Defines a span of immutable raw data fragments.
+///
+using PayloadFragments = cetl::span<const PayloadFragment>;
 
 struct TransferMetadata final
 {
     TransferId transfer_id{};
     Priority   priority{};
 };
+
 struct TransferTxMetadata final
 {
     TransferMetadata base{};
     TimePoint        deadline;
 };
+
 struct TransferRxMetadata final
 {
     TransferMetadata base{};
     TimePoint        timestamp;
 };
 
-/// @brief Defines a span of immutable fragments of payload.
-using PayloadFragments = cetl::span<const cetl::span<const cetl::byte>>;
-
-struct MessageRxMetadata final
-{
-    TransferRxMetadata     rx_meta{};
-    cetl::optional<NodeId> publisher_node_id;
-};
-struct MessageRxTransfer final
-{
-    MessageRxMetadata metadata{};
-    ScatteredBuffer   payload;
-};
-
-struct ServiceTxMetadata final
-{
-    TransferTxMetadata tx_meta{};
-    NodeId             remote_node_id{};
-};
-struct ServiceRxMetadata final
-{
-    TransferRxMetadata rx_meta{};
-    NodeId             remote_node_id{};
-};
-struct ServiceRxTransfer final
-{
-    ServiceRxMetadata metadata{};
-    ScatteredBuffer   payload;
-};
-
 }  // namespace transport
 }  // namespace libcyphal
 

include/libcyphal/transport/udp/delegate.hpp

@@ -152,8 +152,8 @@ class UdpardMemory final : public ScatteredBuffer::IStorage
 
         // Find first fragment to start from (according to source `offset_bytes`).
         //
-        std::size_t                  src_offset = 0;
-        const struct UdpardFragment* frag       = &payload_;
+        std::size_t           src_offset = 0;
+        const UdpardFragment* frag       = &payload_;
         while ((nullptr != frag) && (offset_bytes >= (src_offset + frag->view.size)))
         {
             src_offset += frag->view.size;
@@ -190,6 +190,21 @@ class UdpardMemory final : public ScatteredBuffer::IStorage
         return total_bytes_copied;
     }
 
+    void observeFragments(ScatteredBuffer::IFragmentsObserver& observer) const override
+    {
+        const UdpardFragment* fragment = &payload_;
+        while (nullptr != fragment)
+        {
+            const auto& frag_view = fragment->view;
+            if ((nullptr != frag_view.data) && (frag_view.size > 0))
+            {
+                observer.onNext({static_cast<const cetl::byte*>(frag_view.data), frag_view.size});
+            }
+
+            fragment = fragment->next;
+        }
+    }
+
 private:
     // MARK: Data members: