Program Listing for File fastdds_transport.hpp

            

            
/*
* SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
* SPDX-License-Identifier: Apache-2.0
*
* 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 HOLOSCAN_PUBSUB_FASTDDS_PUBSUB_FASTDDS_TRANSPORT_HPP
#define HOLOSCAN_PUBSUB_FASTDDS_PUBSUB_FASTDDS_TRANSPORT_HPP

#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>
#include <vector>

#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fastdds/dds/publisher/DataWriterListener.hpp>
#include <fastdds/dds/publisher/Publisher.hpp>
#include <fastdds/dds/subscriber/DataReader.hpp>
#include <fastdds/dds/subscriber/DataReaderListener.hpp>
#include <fastdds/dds/subscriber/Subscriber.hpp>
#include <fastdds/dds/topic/Topic.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>

#include <gxf/pubsub/pubsub_native_buffer.hpp>
#include <gxf/pubsub/pubsub_transport.hpp>

#include <holoscan/pubsub/common/sidecar_dispatch_queue.hpp>
#include <holoscan/pubsub/fastdds/pubsub/fastdds_holoscan_entity_type_support.hpp>
#include <holoscan/pubsub/fastdds/pubsub/fastdds_qos_profiles.hpp>

namespace holoscan {

// Forward declarations
class FastDdsPubSubContext;

class FastDdsTransport : public nvidia::gxf::PubSubTransport {
 public:
  explicit FastDdsTransport(FastDdsPubSubContext* context);

  ~FastDdsTransport() override;

  // Delete copy operations
  FastDdsTransport(const FastDdsTransport&) = delete;
  FastDdsTransport& operator=(const FastDdsTransport&) = delete;

  //----------------------------------------------------------------------------
  // PubSubTransport Lifecycle
  //----------------------------------------------------------------------------

  nvidia::gxf::Expected<void> initialize() override;
  nvidia::gxf::Expected<void> shutdown() override;
  bool is_initialized() const override;

  //----------------------------------------------------------------------------
  // Transport Model
  //----------------------------------------------------------------------------

  nvidia::gxf::TransportModel transport_model() const override {
    return nvidia::gxf::TransportModel::kTopicBased;
  }

  //----------------------------------------------------------------------------
  // Backend Capabilities
  //----------------------------------------------------------------------------

  bool native_topic_matching() const override { return true; }

  bool native_qos_enforcement() const override { return true; }

  bool supports_multicast() const override { return true; }

  bool requires_explicit_connections() const override { return false; }

  //----------------------------------------------------------------------------
  // Native Buffer (CUDA IPC) Capabilities
  //----------------------------------------------------------------------------

  bool supports_native_buffers() const override { return native_buffers_enabled_; }
  bool supports_mixed_local_remote_fanout() const override { return native_buffers_enabled_; }
  bool supports_native_profile(const std::string& profile) const override {
    return profile == "cuda_ipc_same_gpu_v1";
  }

  nvidia::gxf::Expected<void> send_native_descriptor(
      const std::string& topic_name, const nvidia::gxf::NativeDescriptorPayload& descriptor,
      const nvidia::gxf::MessageMetadata& metadata) override;

  void set_native_buffers_enabled(bool enabled);

  //----------------------------------------------------------------------------
  // PubSubTransport Connection Management (mostly no-ops for DDS)
  //----------------------------------------------------------------------------

  nvidia::gxf::Expected<void> connect_to(const nvidia::gxf::EndpointInfo& remote_endpoint) override;

  nvidia::gxf::Expected<void> disconnect_from(const nvidia::gxf::Gid& remote_gid) override;

  bool is_connected_to(const nvidia::gxf::Gid& remote_gid) const override;

  //----------------------------------------------------------------------------
  // Topic-Based Endpoint Lifecycle (called by PubSubContext)
  //----------------------------------------------------------------------------

  nvidia::gxf::Expected<void> create_publisher_endpoint(
      const std::string& topic_name, const nvidia::gxf::Gid& publisher_gid,
      const nvidia::gxf::QoSProfile& qos = nvidia::gxf::QoSProfile{}) override;

  nvidia::gxf::Expected<void> create_subscriber_endpoint(
      const std::string& topic_name, const nvidia::gxf::Gid& subscriber_gid,
      const nvidia::gxf::QoSProfile& qos = nvidia::gxf::QoSProfile{}) override;

  nvidia::gxf::Expected<void> remove_publisher_endpoint(
      const nvidia::gxf::Gid& publisher_gid) override;

  nvidia::gxf::Expected<void> remove_subscriber_endpoint(
      const nvidia::gxf::Gid& subscriber_gid) override;

  //----------------------------------------------------------------------------
  // PubSubTransport Data Plane
  //----------------------------------------------------------------------------

  // Bring base class overloads into scope (topic-based send, move-semantic variants)
  using PubSubTransport::send;

  nvidia::gxf::Expected<void> send(const nvidia::gxf::Gid& destination_gid,
                                   const std::vector<uint8_t>& payload,
                                   const nvidia::gxf::MessageMetadata& metadata) override;

  nvidia::gxf::Expected<void> send(const std::string& topic_name,
                                   const std::vector<uint8_t>& payload,
                                   const nvidia::gxf::MessageMetadata& metadata) override;

  void set_on_receive(ReceiveCallback callback) override;
  void set_on_connection_established(ConnectionEstablishedCallback callback) override;
  void set_on_connection_lost(ConnectionLostCallback callback) override;

  //----------------------------------------------------------------------------
  // PubSubTransport Metrics
  //----------------------------------------------------------------------------

  size_t get_send_queue_size() const override;
  size_t get_receive_queue_size() const override;
  size_t get_connection_count() const override;

 private:
  class ReaderListener;

  class WriterListener;

  class SidecarReaderListener;

  eprosima::fastdds::dds::Topic* get_or_create_topic(const std::string& topic_name);

  eprosima::fastdds::dds::DataWriter* get_or_create_sidecar_writer(const std::string& topic_name);

  nvidia::gxf::Expected<void> create_sidecar_reader(const std::string& topic_name,
                                                    const nvidia::gxf::Gid& subscriber_gid);

  nvidia::gxf::Expected<void> send_impl(const std::string& topic_name,
                                        const std::vector<uint8_t>& payload,
                                        const nvidia::gxf::MessageMetadata& metadata);
  void enqueue_sidecar_receive(nvidia::gxf::Gid publisher_gid, std::vector<uint8_t>&& payload,
                               nvidia::gxf::MessageMetadata metadata);

  FastDdsPubSubContext* context_;
  bool initialized_ = false;

  // Type support (registered once per participant, managed by DDS via TypeSupport wrapper)
  // Note: We store the type name to use for creating topics after registration
  std::string registered_type_name_;
  bool type_registered_ = false;

  // DDS Publisher and Subscriber (one each, shared by all writers/readers)
  eprosima::fastdds::dds::Publisher* publisher_ = nullptr;
  eprosima::fastdds::dds::Subscriber* subscriber_ = nullptr;

  // Writers by publisher GID
  struct WriterInfo {
    eprosima::fastdds::dds::DataWriter* writer = nullptr;
    eprosima::fastdds::dds::Topic* topic = nullptr;
    std::string topic_name;
  };
  std::unordered_map<nvidia::gxf::Gid, WriterInfo> writers_;

  // Readers by subscriber GID
  struct ReaderInfo {
    eprosima::fastdds::dds::DataReader* reader = nullptr;
    eprosima::fastdds::dds::Topic* topic = nullptr;
    std::string topic_name;
    nvidia::gxf::Gid subscriber_gid;
  };
  std::unordered_map<nvidia::gxf::Gid, ReaderInfo> readers_;

  // Topics by name (shared between writers and readers)
  std::unordered_map<std::string, eprosima::fastdds::dds::Topic*> topics_;

  mutable std::mutex endpoints_mutex_;

  // Callbacks (receive_callback_ is guarded by callback_mutex_)
  mutable std::mutex callback_mutex_;
  ReceiveCallback receive_callback_;
  ConnectionEstablishedCallback connection_established_callback_;
  ConnectionLostCallback connection_lost_callback_;

  // Listeners
  std::unique_ptr<ReaderListener> reader_listener_;
  std::unique_ptr<WriterListener> writer_listener_;

  // Type support (stored to ensure proper lifetime)
  eprosima::fastdds::dds::TypeSupport type_support_;

  // Native buffer (CUDA IPC) state
  bool native_buffers_enabled_ = false;

  // Sidecar writers for native descriptor messages (keyed by base topic name)
  struct SidecarWriterInfo {
    eprosima::fastdds::dds::DataWriter* writer = nullptr;
    eprosima::fastdds::dds::Topic* topic = nullptr;
  };
  std::unordered_map<std::string, SidecarWriterInfo> sidecar_writers_;

  // Sidecar readers for native descriptor messages
  struct SidecarReaderInfo {
    eprosima::fastdds::dds::DataReader* reader = nullptr;
    eprosima::fastdds::dds::Topic* topic = nullptr;
  };
  std::unordered_map<std::string, SidecarReaderInfo> sidecar_readers_;
  std::unique_ptr<SidecarReaderListener> sidecar_reader_listener_;

  std::unique_ptr<SidecarDispatchQueue> sidecar_dispatch_queue_;
};

}  // namespace holoscan

#endif// HOLOSCAN_PUBSUB_FASTDDS_PUBSUB_FASTDDS_TRANSPORT_HPP