Program Listing for File fastdds_discovery.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_DISCOVERY_HPP
#define HOLOSCAN_PUBSUB_FASTDDS_PUBSUB_FASTDDS_DISCOVERY_HPP

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

#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/domain/DomainParticipantListener.hpp>
#include <fastdds/rtps/builtin/data/ParticipantBuiltinTopicData.hpp>
#include <fastdds/rtps/builtin/data/PublicationBuiltinTopicData.hpp>
#include <fastdds/rtps/builtin/data/SubscriptionBuiltinTopicData.hpp>
#include <fastdds/rtps/participant/ParticipantDiscoveryInfo.hpp>
#include <fastdds/rtps/reader/ReaderDiscoveryStatus.hpp>
#include <fastdds/rtps/writer/WriterDiscoveryStatus.hpp>

#include <gxf/pubsub/endpoint_info.hpp>
#include <gxf/pubsub/pubsub_discovery.hpp>

namespace holoscan {

// Forward declarations
class FastDdsPubSubContext;

class FastDdsDiscovery : public nvidia::gxf::PubSubDiscovery {
 public:
  explicit FastDdsDiscovery(FastDdsPubSubContext* context);

  ~FastDdsDiscovery() override;

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

  //----------------------------------------------------------------------------
  // PubSubDiscovery Lifecycle
  //----------------------------------------------------------------------------

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

  //----------------------------------------------------------------------------
  // Discovery Model
  //----------------------------------------------------------------------------

  nvidia::gxf::DiscoveryModel discovery_model() const override {
    return nvidia::gxf::DiscoveryModel::kDecentralizedPassive;
  }

  //----------------------------------------------------------------------------
  // PubSubDiscovery Registration
  //----------------------------------------------------------------------------

  nvidia::gxf::Expected<void> announce_publisher(const nvidia::gxf::PublisherInfo& info) override;

  nvidia::gxf::Expected<void> announce_subscriber(const nvidia::gxf::SubscriberInfo& info) override;

  //----------------------------------------------------------------------------
  // PubSubDiscovery Deregistration
  //----------------------------------------------------------------------------

  nvidia::gxf::Expected<void> remove_publisher(const nvidia::gxf::PublisherGid& gid) override;

  nvidia::gxf::Expected<void> remove_subscriber(const nvidia::gxf::SubscriberGid& gid) override;

  //----------------------------------------------------------------------------
  // PubSubDiscovery Query
  //----------------------------------------------------------------------------

  nvidia::gxf::Expected<std::vector<nvidia::gxf::PublisherInfo>> query_publishers(
      const std::string& topic_name) override;

  nvidia::gxf::Expected<std::vector<nvidia::gxf::SubscriberInfo>> query_subscribers(
      const std::string& topic_name) override;

  nvidia::gxf::Expected<std::vector<std::string>> get_all_topics() override;

  //----------------------------------------------------------------------------
  // PubSubDiscovery Callbacks
  //----------------------------------------------------------------------------

  void set_on_publisher_discovered(PublisherDiscoveredCallback callback) override;
  void set_on_subscriber_discovered(SubscriberDiscoveredCallback callback) override;
  void set_on_publisher_lost(PublisherLostCallback callback) override;
  void set_on_subscriber_lost(SubscriberLostCallback callback) override;

  //----------------------------------------------------------------------------
  // Native Buffer Capability Advertisement
  //----------------------------------------------------------------------------

  void set_local_native_capability(const nvidia::gxf::NativeBufferCapability& cap) {
    std::lock_guard<std::mutex> lock(mutex_);
    local_native_capability_ = cap;
  }

  nvidia::gxf::NativeBufferCapability local_native_capability(int device_id = -1) const override {
    std::lock_guard<std::mutex> lock(mutex_);
    if (device_id < 0 || !local_native_capability_.supports_native_buffers()) {
      return local_native_capability_;
    }
    // Build a capability for the requested device by looking up its UUID.
    return capability_for_device(device_id);
  }

  static std::vector<uint8_t> serialize_native_capability(
      const nvidia::gxf::NativeBufferCapability& cap);

  static nvidia::gxf::NativeBufferCapability parse_native_capability(
      const std::vector<uint8_t>& user_data);

 private:
  class DiscoveryListener;

  static nvidia::gxf::Gid guid_to_gid(const eprosima::fastdds::rtps::GUID_t& guid);

  FastDdsPubSubContext* context_;
  bool initialized_ = false;

  // Discovered remote endpoints (from DDS SEDP listener callbacks).
  // Local endpoints are tracked by the shared TopicRegistry (set by PubSubContext
  // via set_topic_registry()), so we don't need separate local_publishers_/
  // local_subscribers_ maps.
  std::unordered_map<nvidia::gxf::Gid, nvidia::gxf::NativeBufferCapability>
      discovered_participant_capabilities_;
  std::unordered_map<nvidia::gxf::Gid, nvidia::gxf::PublisherInfo> discovered_publishers_;
  std::unordered_map<nvidia::gxf::Gid, nvidia::gxf::SubscriberInfo> discovered_subscribers_;

  mutable std::mutex mutex_;

  // Callbacks
  PublisherDiscoveredCallback on_publisher_discovered_;
  SubscriberDiscoveredCallback on_subscriber_discovered_;
  PublisherLostCallback on_publisher_lost_;
  SubscriberLostCallback on_subscriber_lost_;

  // DDS listener
  std::unique_ptr<DiscoveryListener> listener_;

  nvidia::gxf::NativeBufferCapability capability_for_device(int device_id) const;

  // Local native buffer capability (set before initialize)
  nvidia::gxf::NativeBufferCapability local_native_capability_;
};

}  // namespace holoscan

#endif// HOLOSCAN_PUBSUB_FASTDDS_PUBSUB_FASTDDS_DISCOVERY_HPP