Updates

Author: pablogs9

rmw_microxrcedds_c/src/rmw_init.c

@@ -268,9 +268,9 @@ rmw_init(const rmw_init_options_t * options, rmw_context_t * context)
     context_impl->connection_params.agent_port);
 #elif defined(MICRO_XRCEDDS_UDP) && defined(UCLIENT_BROKERLESS_ENABLE)
 
-  context_impl->udp_platform.comm = brokerless_comm_stub;
+  context_impl->transport.comm = brokerless_comm_stub;
 
-  printf("UDP mode Brokerless P2P\n");
+  printf("UDP mode => Brokerless P2P\n");
 #elif defined(MICRO_XRCEDDS_CUSTOM_SERIAL)
   int pseudo_fd = 0;
   if (strlen(options->impl->connection_params.serial_device) > 0) {

rmw_microxrcedds_c/src/rmw_microxrcedds_topic.c

@@ -109,7 +109,7 @@ rmw_ret_t destroy_topic(rmw_uxrce_topic_t * topic)
 
   uint16_t delete_topic = uxr_buffer_delete_entity(
     &topic->owner_node->context->session,
-    *custom_node->context->entity_creation_output,
+    *topic->owner_node->context->entity_creation_output,
     topic->topic_id);
 
   uint16_t requests[] = {delete_topic};

rmw_microxrcedds_c/src/rmw_node.c

@@ -85,7 +85,7 @@ rmw_node_t * create_node(
   participant_req =
     uxr_buffer_create_participant_xml(
     &node_info->context->session,
-    *custom_node->context->entity_creation_output,
+    *node_info->context->entity_creation_output,
     node_info->participant_id, 
     (uint16_t)domain_id, 
     rmw_uxrce_xml_buffer, 
@@ -104,10 +104,16 @@ rmw_node_t * create_node(
     rmw_uxrce_profile_name, 
     UXR_REPLACE);
 #endif
+
   uint8_t status[1];
   uint16_t requests[] = {participant_req};
 
-  if (!uxr_run_session_until_all_status(&node_info->context->session, 1000, requests, status, 1)) {
+  if (UXR_BEST_EFFORT_STREAM == node_info->context->entity_creation_output->type)
+  {
+    uxr_flash_output_streams(&node_info->context->session);
+  }
+  else if(!uxr_run_session_until_all_status(&node_info->context->session, 1000, requests, status, 1))
+  {
     rmw_uxrce_fini_node_memory(node_handle);
     RMW_SET_ERROR_MSG("Issues creating micro XRCE-DDS entities");
     return NULL;
@@ -212,9 +218,11 @@ rmw_ret_t rmw_destroy_node(rmw_node_t * node)
   uint8_t status[1];
   uint16_t requests[] = {participant_req};
 
-  if (!uxr_run_session_until_all_status(
-      &custom_node->context->session, 1000, requests, status,
-      1))
+  if (UXR_BEST_EFFORT_STREAM == custom_node->context->entity_creation_output->type)
+  {
+    uxr_flash_output_streams(&custom_node->context->session);
+  }
+  else if(!uxr_run_session_until_all_status(&custom_node->context->session, 1000, requests, status,1))
   {
     ret = RMW_RET_ERROR;
   }

rmw_microxrcedds_c/src/rmw_publisher.c

@@ -209,9 +209,14 @@ rmw_create_publisher(
 
     uint16_t requests[] = {publisher_req, datawriter_req};
     uint8_t status[sizeof(requests) / 2];
-    if (!uxr_run_session_until_all_status(
-        &custom_publisher->owner_node->context->session, 1000, requests,
-        status, sizeof(status)))
+
+    if (UXR_BEST_EFFORT_STREAM == custom_node->context->entity_creation_output->type)
+    {
+      uxr_flash_output_streams(&custom_node->context->session);
+    }
+    else if(!uxr_run_session_until_all_status(
+          &custom_publisher->owner_node->context->session, 1000, requests,
+          status, sizeof(status)))
     {
       RMW_SET_ERROR_MSG("Issues creating micro XRCE-DDS entities");
       put_memory(&publisher_memory, &custom_publisher->mem);
@@ -319,25 +324,37 @@ rmw_destroy_publisher(
 
     uint16_t delete_writer = uxr_buffer_delete_entity(
       &custom_publisher->owner_node->context->session,
-      *custom_node->context->entity_creation_output,
+      *custom_publisher->owner_node->context->entity_creation_output,
       custom_publisher->datawriter_id);
     uint16_t delete_publisher = uxr_buffer_delete_entity(
       &custom_publisher->owner_node->context->session,
-      *custom_node->context->entity_creation_output,
+      *custom_publisher->owner_node->context->entity_creation_output,
       custom_publisher->publisher_id);
 
     uint16_t requests[] = {delete_writer, delete_publisher};
     uint8_t status[sizeof(requests) / 2];
-    if (!uxr_run_session_until_all_status(
-        &custom_publisher->owner_node->context->session, 1000, requests, status,
-        sizeof(status)))
+
+    if (UXR_BEST_EFFORT_STREAM == custom_publisher->owner_node->context->entity_creation_output->type)
     {
-      RMW_SET_ERROR_MSG("unable to remove publisher from the server");
-      result_ret = RMW_RET_ERROR;
-    } else {
+      uxr_flash_output_streams(&custom_publisher->owner_node->context->session);
       rmw_uxrce_fini_publisher_memory(publisher);
       result_ret = RMW_RET_OK;
     }
+    else
+    {
+      if (!uxr_run_session_until_all_status(
+          &custom_publisher->owner_node->context->session, 1000, requests, status,
+          sizeof(status)))
+      {
+        RMW_SET_ERROR_MSG("unable to remove publisher from the server");
+        result_ret = RMW_RET_ERROR;
+      } 
+      else 
+      {
+        rmw_uxrce_fini_publisher_memory(publisher);
+        result_ret = RMW_RET_OK;
+      }
+    }
   }
 
   return result_ret;

rmw_microxrcedds_c/src/rmw_subscription.c

@@ -304,12 +304,12 @@ rmw_destroy_subscription(rmw_node_t * node, rmw_subscription_t * subscription)
     uint16_t delete_datareader =
       uxr_buffer_delete_entity(
       &custom_subscription->owner_node->context->session,
-      *custom_node->context->entity_creation_output,
+      *custom_subscription->owner_node->context->entity_creation_output,
       custom_subscription->datareader_id);
     uint16_t delete_subscriber =
       uxr_buffer_delete_entity(
       &custom_subscription->owner_node->context->session,
-      *custom_node->context->entity_creation_output,
+      *custom_subscription->owner_node->context->entity_creation_output,
       custom_subscription->subscriber_id);
 
     uint16_t requests[] = {delete_datareader, delete_subscriber};

rmw_microxrcedds_c/src/utils.c

@@ -373,7 +373,7 @@ bool build_datareader_profile(const char * topic_name, char profile_name[], size
 bool build_requester_profile(const char * service_name, char profile_name[], size_t buffer_size)
 {
   const char * const format = "%s___requester";
-  topic_name++;
+  service_name++;
   bool ret = false;
   int written = snprintf(profile_name, buffer_size, format, service_name);
   ret = (written > 0) && (written < (int)buffer_size);
@@ -383,7 +383,7 @@ bool build_requester_profile(const char * service_name, char profile_name[], siz
 bool build_replier_profile(const char * service_name, char profile_name[], size_t buffer_size)
 {
   const char * const format = "%s___replier";
-  topic_name++;
+  service_name++;
   bool ret = false;
   int written = snprintf(profile_name, buffer_size, format, service_name);
   ret = (written > 0) && (written < (int)buffer_size);