Unique network flows optional checks

README.md

@@ -63,6 +63,13 @@ It is important to note that this mode typically yields higher throughput rates
 
 If `RMW_FASTRTPS_PUBLICATION_MODE` is not set, then both `rmw_fastrtps_cpp` and `rmw_fastrtps_dynamic_cpp` behave as if it were set to `SYNCHRONOUS`.
 
+For the internal `ros_discovery_info` graph subscription, `rmw_fastrtps` requests unique network flow endpoints by default. The `RMW_FASTRTPS_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS` environment variable can be used to change the unique network flows request mode. The admissible values are:
+
+* `DISABLED`: Unique network flow endpoints not required.
+* `STRICT`: Unique network flow endpoins strictly required. This can result in an error if the `ros_discovery_info` graph subscription is configured with a conflicting option, such as static endpoint discovery or custom locators.
+* `OPTIONAL`: Unique network flow endpoints optionally required and will not be used if the `ros_discovery_info` graph subscription is configured with a conflicting option, such as static endpoint discovery or custom locators.
+* `DEFAULT`: Unique network flow endpoints requirement decided by system.
+
 ### Full QoS configuration
 
 Fast DDS QoS policies can be fully configured through a combination of the [rmw QoS profile] API, and the [Fast DDS XML] file's QoS elements. Configuration depends on the environment variable `RMW_FASTRTPS_USE_QOS_FROM_XML`.

rmw_fastrtps_cpp/src/init_rmw_context_impl.cpp

@@ -1,4 +1,5 @@
 // Copyright 2020 Open Source Robotics Foundation, Inc.
+// Copyright 2026 Torc Robotics, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -29,6 +30,7 @@
 #include "rmw_fastrtps_cpp/subscription.hpp"
 
 #include "rmw_fastrtps_shared_cpp/custom_participant_info.hpp"
+#include "rmw_fastrtps_shared_cpp/init_rmw_context_impl.hpp"
 #include "rmw_fastrtps_shared_cpp/namespace_prefix.hpp"
 #include "rmw_fastrtps_shared_cpp/participant.hpp"
 #include "rmw_fastrtps_shared_cpp/publisher.hpp"
@@ -53,9 +55,10 @@ init_context_impl(
 
   // Avoid receiving graph updates from our own publication
   subscription_options.ignore_local_publications = true;
-  // Improve graph discovery by using a unique listening port for its subscription
+  // Improve graph discovery by using a unique listening port for its subscription,
+  // unless disabled by the user
   subscription_options.require_unique_network_flow_endpoints =
-    RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;
+    rmw_fastrtps_shared_cpp::get_unique_network_flows_for_ros_discovery_info();
 
   std::unique_ptr<rmw_dds_common::Context> common_context(
     new(std::nothrow) rmw_dds_common::Context());

rmw_fastrtps_dynamic_cpp/src/init_rmw_context_impl.cpp

@@ -1,4 +1,5 @@
 // Copyright 2020 Open Source Robotics Foundation, Inc.
+// Copyright 2026 Torc Robotics, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -27,6 +28,7 @@
 #include "rmw_fastrtps_dynamic_cpp/identifier.hpp"
 
 #include "rmw_fastrtps_shared_cpp/custom_participant_info.hpp"
+#include "rmw_fastrtps_shared_cpp/init_rmw_context_impl.hpp"
 #include "rmw_fastrtps_shared_cpp/listener_thread.hpp"
 #include "rmw_fastrtps_shared_cpp/participant.hpp"
 #include "rmw_fastrtps_shared_cpp/publisher.hpp"
@@ -50,9 +52,10 @@ init_context_impl(
 
   // Avoid receiving graph updates from our own publication
   subscription_options.ignore_local_publications = true;
-  // Improve graph discovery by using a unique listening port for its subscription
+  // Improve graph discovery by using a unique listening port for its subscription,
+  // unless disabled by the user
   subscription_options.require_unique_network_flow_endpoints =
-    RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;
+    rmw_fastrtps_shared_cpp::get_unique_network_flows_for_ros_discovery_info();
 
   std::unique_ptr<rmw_dds_common::Context> common_context(
     new(std::nothrow) rmw_dds_common::Context());

rmw_fastrtps_shared_cpp/include/rmw_fastrtps_shared_cpp/init_rmw_context_impl.hpp

@@ -1,4 +1,5 @@
 // Copyright 2020 Open Source Robotics Foundation, Inc.
+// Copyright 2026 Torc Robotics, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -23,6 +24,24 @@
 namespace rmw_fastrtps_shared_cpp
 {
 
+/// Check if unique network flows should be used for ROS discovery info.
+/**
+* This function is thread-safe and by default caches the result of the environment
+* variable check to avoid repeated lookups during concurrent context initialization.
+*
+* If `use_cached` is true (the default) the cached value is returned; the cache
+* is populated on the first call and subsequent calls ignore changes to the
+* environment variable. If `use_cached` is false the environment variable is
+* read and mapped to the returned enumeration on each call without modifying the
+* cached value.
+*
+* \param[in] use_cached whether to return the cached value (default: true)
+* \return the requirement for unique network flow endpoints for ROS discovery info
+*/
+RMW_FASTRTPS_SHARED_CPP_PUBLIC
+rmw_unique_network_flow_endpoints_requirement_t
+get_unique_network_flows_for_ros_discovery_info(bool use_cached = true);
+
 /// Increment `rmw_context_impl_t` reference count, destroying it if the count reaches zero.
 /**
  * Function that should be called when destroying a node.

rmw_fastrtps_shared_cpp/src/init_rmw_context_impl.cpp

@@ -1,4 +1,5 @@
 // Copyright 2020 Open Source Robotics Foundation, Inc.
+// Copyright 2026 Torc Robotics, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -15,6 +16,9 @@
 #include "rmw_fastrtps_shared_cpp/init_rmw_context_impl.hpp"
 
 #include <cassert>
+#include <mutex>
+
+#include "rcutils/env.h"
 
 #include "rmw/error_handling.h"
 #include "rmw/init.h"
@@ -30,6 +34,73 @@
 
 #include "rmw_fastrtps_shared_cpp/listener_thread.hpp"
 
+namespace
+{
+
+rmw_unique_network_flow_endpoints_requirement_t g_unique_network_flows_for_ros_discovery_info =
+  RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;
+std::once_flag g_init_flag;
+
+static rmw_unique_network_flow_endpoints_requirement_t
+read_unique_network_flows_from_env()
+{
+  const char * env_value = nullptr;
+  const char * error_str = rcutils_get_env(
+    "RMW_FASTRTPS_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS", &env_value);
+
+  if (error_str != nullptr) {
+    RCUTILS_LOG_WARN_NAMED(
+      "rmw_fastrtps_shared_cpp",
+      "Error getting env var RMW_FASTRTPS_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS: %s. "
+      "Using default behavior.",
+      error_str);
+    return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;
+  }
+
+  if (env_value == nullptr) {
+    RCUTILS_LOG_WARN_NAMED(
+      "rmw_fastrtps_shared_cpp",
+      "Invalid value for env var RMW_FASTRTPS_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS: null. "
+      "Using default behavior.");
+    return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;
+  }
+
+  if (strcmp(env_value, "DISABLED") == 0) {
+    return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_NOT_REQUIRED;
+  } else if (strcmp(env_value, "OPTIONAL") == 0) {
+    return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;
+  } else if (strcmp(env_value, "STRICT") == 0) {
+    return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED;
+  } else if (strcmp(env_value, "DEFAULT") == 0) {
+    return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_SYSTEM_DEFAULT;
+  }
+
+  RCUTILS_LOG_WARN_NAMED(
+    "rmw_fastrtps_shared_cpp",
+    "Invalid value for env var RMW_FASTRTPS_ROS_DISCOVERY_INFO_UNIQUE_NETWORK_FLOWS: %s. "
+    "Using default behavior.",
+    env_value);
+  return RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED;
+}
+
+void initialize_unique_network_flows()
+{
+  g_unique_network_flows_for_ros_discovery_info = read_unique_network_flows_from_env();
+}
+
+}  // anonymous namespace
+
+rmw_unique_network_flow_endpoints_requirement_t
+rmw_fastrtps_shared_cpp::get_unique_network_flows_for_ros_discovery_info(bool use_cached)
+{
+  if (use_cached) {
+    std::call_once(g_init_flag, initialize_unique_network_flows);
+    return g_unique_network_flows_for_ros_discovery_info;
+  }
+  // When bypassing the cache, use the shared helper so logic is consistent.
+  return read_unique_network_flows_from_env();
+}
+
 rmw_ret_t
 rmw_fastrtps_shared_cpp::decrement_context_impl_ref_count(
   rmw_context_t * context)

rmw_fastrtps_shared_cpp/src/utils.cpp

@@ -1,4 +1,5 @@
 // Copyright 2021 Proyectos y Sistemas de Mantenimiento SL (eProsima).
+// Copyright 2026 Torc Robotics, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -140,19 +141,84 @@ create_datareader(
     default:
     case RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_SYSTEM_DEFAULT:
     case RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_NOT_REQUIRED:
-      // Unique network flow endpoints not required. We leave the decission to the XML profile.
+      // Unique network flow endpoints not required. We leave the decision to the XML profile.
       break;
 
     case RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED:
     case RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED:
-      // Ensure we request unique network flow endpoints
-      using PropertyPolicyHelper = eprosima::fastdds::rtps::PropertyPolicyHelper;
-      if (nullptr ==
-        PropertyPolicyHelper::find_property(
-          updated_qos.properties(),
-          "fastdds.unique_network_flows"))
       {
-        updated_qos.properties().properties().emplace_back("fastdds.unique_network_flows", "");
+        bool strict_unique_flows_required =
+          (RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED ==
+          subscription_options->require_unique_network_flow_endpoints);
+
+        eprosima::fastdds::dds::DomainParticipantQos participant_qos;
+        subscriber->get_participant()->get_qos(participant_qos);
+
+        // Check if locators are explicitly specified at endpoint level.
+        bool has_explicit_locators =
+          !datareader_qos.endpoint().unicast_locator_list.empty() ||
+          !datareader_qos.endpoint().multicast_locator_list.empty() ||
+          !datareader_qos.endpoint().remote_locator_list.empty();
+
+        // Also consider participant defaults as explicit user intent. This covers
+        // scenarios where only default multicast locators are configured and unique
+        // flow creation would fail due to missing unicast locators.
+        if (!has_explicit_locators) {
+          has_explicit_locators =
+            !participant_qos.wire_protocol().default_unicast_locator_list.empty() ||
+            !participant_qos.wire_protocol().default_multicast_locator_list.empty();
+        }
+
+        if (has_explicit_locators) {
+          if (strict_unique_flows_required) {
+            RMW_SET_ERROR_MSG(
+              "RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED cannot be used when "
+              "endpoint or participant default locators are explicitly configured");
+            return false;
+          } else {
+            RCUTILS_LOG_WARN_NAMED(
+              "rmw_fastrtps_shared_cpp",
+              "RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED is ignored when "
+              "endpoint or participant default locators are explicitly configured");
+          }
+        }
+
+        // Check if Endpoint Discovery Protocol (EDP) is configured as STATIC
+        bool is_edp_static =
+          participant_qos.wire_protocol().builtin.discovery_config.
+          use_STATIC_EndpointDiscoveryProtocol;
+
+        if (is_edp_static) {
+          if (strict_unique_flows_required) {
+            RMW_SET_ERROR_MSG(
+              "RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_STRICTLY_REQUIRED cannot be used with "
+              "static endpoint discovery");
+            return false;
+          } else {
+            RCUTILS_LOG_WARN_NAMED(
+              "rmw_fastrtps_shared_cpp",
+              "RMW_UNIQUE_NETWORK_FLOW_ENDPOINTS_OPTIONALLY_REQUIRED is ignored with "
+              "static endpoint discovery");
+          }
+        }
+
+        using PropertyPolicyHelper = eprosima::fastdds::rtps::PropertyPolicyHelper;
+        if (nullptr ==
+          PropertyPolicyHelper::find_property(
+            updated_qos.properties(),
+            "fastdds.unique_network_flows"))
+        {
+          // For STRICTLY_REQUIRED: always request unique flows (Fast-DDS will enforce it)
+          // For OPTIONALLY_REQUIRED: only request if no locators are explicitly configured and not
+          //   using static endpoint discovery
+          bool should_request_unique_flows =
+            strict_unique_flows_required ||
+            (!has_explicit_locators && !is_edp_static);
+
+          if (should_request_unique_flows) {
+            updated_qos.properties().properties().emplace_back("fastdds.unique_network_flows", "");
+          }
+        }
       }
       break;
   }

rmw_fastrtps_shared_cpp/test/CMakeLists.txt

@@ -54,3 +54,12 @@ ament_add_gtest(test_logging test_logging.cpp)
 if(TARGET test_logging)
   target_link_libraries(test_logging ${PROJECT_NAME} rmw::rmw)
 endif()
+
+ament_add_gtest(test_unique_network_flows test_unique_network_flows.cpp)
+if(TARGET test_unique_network_flows)
+  target_link_libraries(test_unique_network_flows
+    ${PROJECT_NAME}
+    rcutils::rcutils
+    rmw::rmw
+  )
+endif()