Wrap zenoh session with a shared pointer

.github/workflows/style.yaml

@@ -19,6 +19,6 @@ jobs:
     steps:
     - uses: actions/checkout@v4
     - name: uncrustify
-      run: /ros_entrypoint.sh ament_uncrustify --exclude rmw_zenoh_cpp/src/detail/ordered_hash.hpp rmw_zenoh_cpp/src/detail/ordered_map.hpp rmw_zenoh_cpp/
+      run: /ros_entrypoint.sh ament_uncrustify rmw_zenoh_cpp/ --exclude rmw_zenoh_cpp/src/detail/ordered_hash.hpp rmw_zenoh_cpp/src/detail/ordered_map.hpp
     - name: cpplint
-      run: /ros_entrypoint.sh ament_cpplint --exclude rmw_zenoh_cpp/src/detail/ordered_hash.hpp rmw_zenoh_cpp/src/detail/ordered_map.hpp rmw_zenoh_cpp/
+      run: /ros_entrypoint.sh ament_cpplint rmw_zenoh_cpp/ --exclude rmw_zenoh_cpp/src/detail/ordered_hash.hpp rmw_zenoh_cpp/src/detail/ordered_map.hpp

rmw_zenoh_cpp/CMakeLists.txt

@@ -22,10 +22,6 @@ find_package(rosidl_typesupport_fastrtps_c REQUIRED)
 find_package(rosidl_typesupport_fastrtps_cpp REQUIRED)
 find_package(rmw REQUIRED)
 find_package(zenoh_c_vendor REQUIRED)
-find_package(zenohc_debug QUIET)
-if(NOT zenohc_debug_FOUND)
-  find_package(zenohc REQUIRED)
-endif()
 
 add_library(rmw_zenoh_cpp SHARED
   src/detail/attachment_helpers.cpp

rmw_zenoh_cpp/src/detail/attachment_helpers.cpp

@@ -16,83 +16,92 @@
 
 #include <cstdlib>
 #include <cstring>
-#include <string>
+#include <stdexcept>
+#include <string_view>
+#include <utility>
 
 #include "rmw/types.h"
 
 #include "attachment_helpers.hpp"
 
 namespace rmw_zenoh_cpp
 {
-//==============================================================================
-bool get_gid_from_attachment(
-  const z_attachment_t * const attachment, uint8_t gid[RMW_GID_STORAGE_SIZE])
-{
-  if (!z_check(*attachment)) {
-    return false;
-  }
 
-  z_bytes_t index = z_attachment_get(*attachment, z_bytes_new("source_gid"));
-  if (!z_check(index)) {
-    return false;
-  }
-
-  if (index.len != RMW_GID_STORAGE_SIZE) {
-    return false;
-  }
+attachment_data_t::attachment_data_t(
+  const int64_t _sequence_number,
+  const int64_t _source_timestamp,
+  const uint8_t _source_gid[RMW_GID_STORAGE_SIZE])
+{
+  sequence_number = _sequence_number;
+  source_timestamp = _source_timestamp;
+  memcpy(source_gid, _source_gid, RMW_GID_STORAGE_SIZE);
+}
 
-  memcpy(gid, index.start, index.len);
+attachment_data_t::attachment_data_t(attachment_data_t && data)
+{
+  sequence_number = std::move(data.sequence_number);
+  source_timestamp = std::move(data.source_timestamp);
+  memcpy(source_gid, data.source_gid, RMW_GID_STORAGE_SIZE);
+}
 
-  return true;
+void attachment_data_t::serialize_to_zbytes(z_owned_bytes_t * attachment)
+{
+  ze_owned_serializer_t serializer;
+  ze_serializer_empty(&serializer);
+  ze_serializer_serialize_str(z_loan_mut(serializer), "sequence_number");
+  ze_serializer_serialize_int64(z_loan_mut(serializer), this->sequence_number);
+  ze_serializer_serialize_str(z_loan_mut(serializer), "source_timestamp");
+  ze_serializer_serialize_int64(z_loan_mut(serializer), this->source_timestamp);
+  ze_serializer_serialize_str(z_loan_mut(serializer), "source_gid");
+  ze_serializer_serialize_buf(z_loan_mut(serializer), this->source_gid, RMW_GID_STORAGE_SIZE);
+  ze_serializer_finish(z_move(serializer), attachment);
 }
 
-//==============================================================================
-int64_t get_int64_from_attachment(
-  const z_attachment_t * const attachment, const std::string & name)
+attachment_data_t::attachment_data_t(const z_loaned_bytes_t * attachment)
 {
-  if (!z_check(*attachment)) {
-    // A valid request must have had an attachment
-    return -1;
+  ze_deserializer_t deserializer = ze_deserializer_from_bytes(attachment);
+  z_owned_string_t key;
+
+  // Deserialize the sequence_number
+  ze_deserializer_deserialize_string(&deserializer, &key);
+  if (std::string_view(
+      z_string_data(z_loan(key)),
+      z_string_len(z_loan(key))) != "sequence_number")
+  {
+    throw std::runtime_error("sequence_number is not found in the attachment.");
   }
-
-  z_bytes_t index = z_attachment_get(*attachment, z_bytes_new(name.c_str()));
-  if (!z_check(index)) {
-    return -1;
+  z_drop(z_move(key));
+  if (ze_deserializer_deserialize_int64(&deserializer, &this->sequence_number)) {
+    throw std::runtime_error("Failed to deserialize the sequence_number.");
   }
 
-  if (index.len < 1) {
-    return -1;
+  // Deserialize the source_timestamp
+  ze_deserializer_deserialize_string(&deserializer, &key);
+  if (std::string_view(
+      z_string_data(z_loan(key)),
+      z_string_len(z_loan(key))) != "source_timestamp")
+  {
+    throw std::runtime_error("source_timestamp is not found in the attachment");
   }
-
-  if (index.len > 19) {
-    // The number was larger than we expected
-    return -1;
+  z_drop(z_move(key));
+  if (ze_deserializer_deserialize_int64(&deserializer, &this->source_timestamp)) {
+    throw std::runtime_error("Failed to deserialize the source_timestamp.");
   }
 
-  // The largest possible int64_t number is INT64_MAX, i.e. 9223372036854775807.
-  // That is 19 characters long, plus one for the trailing \0, means we need 20 bytes.
-  char int64_str[20];
-
-  memcpy(int64_str, index.start, index.len);
-  int64_str[index.len] = '\0';
-
-  errno = 0;
-  char * endptr;
-  int64_t num = strtol(int64_str, &endptr, 10);
-  if (num == 0) {
-    // This is an error regardless; the client should never send this
-    return -1;
-  } else if (endptr == int64_str) {
-    // No values were converted, this is an error
-    return -1;
-  } else if (*endptr != '\0') {
-    // There was junk after the number
-    return -1;
-  } else if (errno != 0) {
-    // Some other error occurred, which may include overflow or underflow
-    return -1;
+  // Deserialize the source_gid
+  ze_deserializer_deserialize_string(&deserializer, &key);
+  if (std::string_view(z_string_data(z_loan(key)), z_string_len(z_loan(key))) != "source_gid") {
+    throw std::runtime_error("Invalid attachment: the key source_gid is not found");
   }
-
-  return num;
+  z_drop(z_move(key));
+  z_owned_slice_t slice;
+  if (ze_deserializer_deserialize_slice(&deserializer, &slice)) {
+    throw std::runtime_error("Failed to deserialize the source_gid.");
+  }
+  if (z_slice_len(z_loan(slice)) != RMW_GID_STORAGE_SIZE) {
+    throw std::runtime_error("The length of source_gid mismatched.");
+  }
+  memcpy(this->source_gid, z_slice_data(z_loan(slice)), z_slice_len(z_loan(slice)));
+  z_drop(z_move(slice));
 }
 }  // namespace rmw_zenoh_cpp

rmw_zenoh_cpp/src/detail/attachment_helpers.hpp

@@ -17,19 +17,27 @@
 
 #include <zenoh.h>
 
-#include <string>
-
 #include "rmw/types.h"
 
 namespace rmw_zenoh_cpp
 {
-//==============================================================================
-bool get_gid_from_attachment(
-  const z_attachment_t * const attachment, uint8_t gid[RMW_GID_STORAGE_SIZE]);
 
-//==============================================================================
-int64_t get_int64_from_attachment(
-  const z_attachment_t * const attachment, const std::string & name);
+class attachment_data_t final
+{
+public:
+  explicit attachment_data_t(
+    const int64_t _sequence_number,
+    const int64_t _source_timestamp,
+    const uint8_t _source_gid[RMW_GID_STORAGE_SIZE]);
+  explicit attachment_data_t(const z_loaned_bytes_t *);
+  explicit attachment_data_t(attachment_data_t && data);
+
+  int64_t sequence_number;
+  int64_t source_timestamp;
+  uint8_t source_gid[RMW_GID_STORAGE_SIZE];
+
+  void serialize_to_zbytes(z_owned_bytes_t *);
+};
 }  // namespace rmw_zenoh_cpp
 
 #endif  // DETAIL__ATTACHMENT_HELPERS_HPP_

rmw_zenoh_cpp/src/detail/graph_cache.cpp

@@ -612,9 +612,10 @@ void GraphCache::parse_del(
 
   if (entity->type() == EntityType::Node) {
     // Node
-    // The liveliness tokens to remove pub/subs should be received before the one to remove a node
-    // given the reliability QoS for liveliness subs. However, if we find any pubs/subs present in
-    // the node below, we should update the count in graph_topics_.
+    // When destroying a node, Zenoh does not guarantee that liveliness tokens to remove pub/subs
+    // arrive before the one to remove the node from the graph despite un-registering those entities
+    // earlier. In such scenarios, if we find any pub/subs present in the node, we reduce their
+    // counts in graph_topics_.
     const GraphNodePtr graph_node = node_it->second;
     if (!graph_node->pubs_.empty() ||
       !graph_node->subs_.empty() ||

rmw_zenoh_cpp/src/detail/liveliness_utils.cpp

@@ -374,14 +374,11 @@ std::optional<rmw_qos_profile_t> keyexpr_to_qos(const std::string & keyexpr)
 ///=============================================================================
 std::string zid_to_str(const z_id_t & id)
 {
-  std::stringstream ss;
-  ss << std::hex;
-  for (std::size_t i = 0; i < sizeof(id.id); i++) {
-    // By Zenoh convention a z_id_t is a little endian u128.
-    const std::size_t le_idx = sizeof(id.id) - 1 - i;
-    ss << static_cast<int>(id.id[le_idx]);
-  }
-  return ss.str();
+  z_owned_string_t z_str;
+  z_id_to_string(&id, &z_str);
+  std::string str(z_string_data(z_loan(z_str)), z_string_len(z_loan(z_str)));
+  z_drop(z_move(z_str));
+  return str;
 }
 
 ///=============================================================================