Fix for issue #412 Only one single RPC client is possible per port (#415)

1. Response Rx Session now doesn't own lizard's rpc port subscription -
instead now session references the shared subscription (with reference
counting).
2. Transport delegates now manage subscriptions.
3. CAN Response Rx Sessions now stored in cavl tree (for UDP it was
already like this) - matching now by both port and node ids (previously
it was by the port only).
4. CAN filters now also are made by the delegate (b/c it knows about
subscriptions (see bullet # 2).
5. Extended unit tests to cover multiple response Rx sessions (both CAN
and UDP). Also covered unsolicited responses.

Github's Hide whitespaces is recommended.

Author: serges147

include/libcyphal/transport/can/can_transport_impl.hpp

@@ -50,15 +50,17 @@ class MessageRxSession final : private IRxSessionDelegate, public IMessageRxSess
     };
 
 public:
-    CETL_NODISCARD static Expected<UniquePtr<IMessageRxSession>, AnyFailure> make(TransportDelegate&     delegate,
-                                                                                  const MessageRxParams& params)
+    CETL_NODISCARD static Expected<UniquePtr<IMessageRxSession>, AnyFailure> make(  //
+        cetl::pmr::memory_resource& memory,
+        TransportDelegate&          delegate,
+        const MessageRxParams&      params)
     {
         if (params.subject_id > CANARD_SUBJECT_ID_MAX)
         {
             return ArgumentError{};
         }
 
-        auto session = libcyphal::detail::makeUniquePtr<Spec>(delegate.memory(), Spec{}, delegate, params);
+        auto session = libcyphal::detail::makeUniquePtr<Spec>(memory, Spec{}, delegate, params);
         if (session == nullptr)
         {
             return MemoryError{};
@@ -72,20 +74,12 @@ class MessageRxSession final : private IRxSessionDelegate, public IMessageRxSess
         , params_{params}
         , subscription_{}
     {
-        const std::int8_t result = ::canardRxSubscribe(&delegate.canardInstance(),
-                                                       CanardTransferKindMessage,
-                                                       params_.subject_id,
-                                                       params_.extent_bytes,
-                                                       CANARD_DEFAULT_TRANSFER_ID_TIMEOUT_USEC,
-                                                       &subscription_);
-        (void) result;
-        CETL_DEBUG_ASSERT(result >= 0, "There is no way currently to get an error here.");
-        CETL_DEBUG_ASSERT(result > 0, "New subscription supposed to be made.");
+        delegate.listenForRxSubscription(subscription_, params);
 
         // No Sonar `cpp:S5356` b/c we integrate here with C libcanard API.
         subscription_.user_reference = static_cast<IRxSessionDelegate*>(this);  // NOSONAR cpp:S5356
 
-        delegate_.onSessionEvent(TransportDelegate::SessionEvent::MsgRxLifetime{true /* is_added */});
+        delegate_.onSessionEvent(TransportDelegate::SessionEvent::MsgCreated{});
     }
 
     MessageRxSession(const MessageRxSession&)                = delete;
@@ -95,13 +89,8 @@ class MessageRxSession final : private IRxSessionDelegate, public IMessageRxSess
 
     ~MessageRxSession()
     {
-        const std::int8_t result =
-            ::canardRxUnsubscribe(&delegate_.canardInstance(), CanardTransferKindMessage, params_.subject_id);
-        (void) result;
-        CETL_DEBUG_ASSERT(result >= 0, "There is no way currently to get an error here.");
-        CETL_DEBUG_ASSERT(result > 0, "Subscription supposed to be made at constructor.");
-
-        delegate_.onSessionEvent(TransportDelegate::SessionEvent::MsgRxLifetime{false /* is_added */});
+        delegate_.cancelRxSubscriptionFor(subscription_, CanardTransferKindMessage);
+        delegate_.onSessionEvent(TransportDelegate::SessionEvent::MsgDestroyed{});
     }
 
 private:
@@ -141,26 +130,15 @@ class MessageRxSession final : private IRxSessionDelegate, public IMessageRxSess
 
     // MARK: IRxSessionDelegate
 
-    void acceptRxTransfer(const CanardRxTransfer& transfer) override
+    void acceptRxTransfer(CanardMemory&&            lizard_memory,
+                          const TransferRxMetadata& rx_metadata,
+                          const NodeId              source_node_id) override
     {
-        const auto priority    = static_cast<Priority>(transfer.metadata.priority);
-        const auto transfer_id = static_cast<TransferId>(transfer.metadata.transfer_id);
-        const auto timestamp   = TimePoint{std::chrono::microseconds{transfer.timestamp_usec}};
-
         const cetl::optional<NodeId> publisher_node_id =
-            transfer.metadata.remote_node_id > CANARD_NODE_ID_MAX
-                ? cetl::nullopt
-                : cetl::make_optional<NodeId>(transfer.metadata.remote_node_id);
-
-        // No Sonar `cpp:S5356` and `cpp:S5357` b/c we need to pass raw data from C libcanard api.
-        auto* const buffer = static_cast<cetl::byte*>(transfer.payload.data);  // NOSONAR cpp:S5356 cpp:S5357
-        TransportDelegate::CanardMemory canard_memory{delegate_,
-                                                      transfer.payload.allocated_size,
-                                                      buffer,
-                                                      transfer.payload.size};
-
-        const MessageRxMetadata meta{{{transfer_id, priority}, timestamp}, publisher_node_id};
-        MessageRxTransfer       msg_rx_transfer{meta, ScatteredBuffer{std::move(canard_memory)}};
+            source_node_id > CANARD_NODE_ID_MAX ? cetl::nullopt : cetl::make_optional(source_node_id);
+
+        const MessageRxMetadata meta{rx_metadata, publisher_node_id};
+        MessageRxTransfer       msg_rx_transfer{meta, ScatteredBuffer{std::move(lizard_memory)}};
         if (on_receive_cb_fn_)
         {
             on_receive_cb_fn_(OnReceiveCallback::Arg{msg_rx_transfer});

include/libcyphal/transport/can/delegate.hpp

@@ -42,15 +42,17 @@ class MessageTxSession final : public IMessageTxSession
     };
 
 public:
-    CETL_NODISCARD static Expected<UniquePtr<IMessageTxSession>, AnyFailure> make(TransportDelegate&     delegate,
-                                                                                  const MessageTxParams& params)
+    CETL_NODISCARD static Expected<UniquePtr<IMessageTxSession>, AnyFailure> make(  //
+        cetl::pmr::memory_resource& memory,
+        TransportDelegate&          delegate,
+        const MessageTxParams&      params)
     {
         if (params.subject_id > CANARD_SUBJECT_ID_MAX)
         {
             return ArgumentError{};
         }
 
-        auto session = libcyphal::detail::makeUniquePtr<Spec>(delegate.memory(), Spec{}, delegate, params);
+        auto session = libcyphal::detail::makeUniquePtr<Spec>(memory, Spec{}, delegate, params);
         if (session == nullptr)
         {
             return MemoryError{};

include/libcyphal/transport/can/msg_rx_session.hpp

@@ -0,0 +1,44 @@
+/// @copyright
+/// Copyright (C) OpenCyphal Development Team  <opencyphal.org>
+/// Copyright Amazon.com Inc. or its affiliates.
+/// SPDX-License-Identifier: MIT
+
+#ifndef LIBCYPHAL_TRANSPORT_CAN_RX_SESSION_TREE_NODE_HPP_INCLUDED
+#define LIBCYPHAL_TRANSPORT_CAN_RX_SESSION_TREE_NODE_HPP_INCLUDED
+
+#include "libcyphal/transport/session_tree.hpp"
+
+namespace libcyphal
+{
+namespace transport
+{
+namespace can
+{
+
+/// Internal implementation details of the UDP transport.
+/// Not supposed to be used directly by the users of the library.
+///
+namespace detail
+{
+
+class IRxSessionDelegate;
+
+/// Umbrella type for various RX session tree nodes in use at the CAN transport.
+///
+/// Currently, it contains only one `Response` subtype,
+/// but still kept nested to match the UDP transport approach (where there are several subtypes).
+///
+struct RxSessionTreeNode
+{
+    /// @brief Represents a service response RX session node.
+    ///
+    using Response = transport::detail::ResponseRxSessionNode<IRxSessionDelegate>;
+
+};  // RxSessionTreeNode
+
+}  // namespace detail
+}  // namespace can
+}  // namespace transport
+}  // namespace libcyphal
+
+#endif  // LIBCYPHAL_TRANSPORT_CAN_RX_SESSION_TREE_NODE_HPP_INCLUDED