use uint8_t array (ros2#309)

rmw_connext_cpp/src/rmw_publish.cpp

@@ -21,13 +21,11 @@
 #include "rmw_connext_cpp/connext_static_publisher_info.hpp"
 #include "rmw_connext_cpp/identifier.hpp"
 
-#include "rosidl_typesupport_connext_cpp/connext_static_cdr_stream.hpp"
-
 // include patched generated code from the build folder
 #include "connext_static_serialized_dataSupport.h"
 
 bool
-publish(DDSDataWriter * dds_data_writer, ConnextStaticCDRStream * cdr_stream)
+publish(DDSDataWriter * dds_data_writer, const rcutils_uint8_array_t * cdr_stream)
 {
   ConnextStaticSerializedDataDataWriter * data_writer =
     ConnextStaticSerializedDataDataWriter::narrow(dds_data_writer);
@@ -108,8 +106,9 @@ rmw_publish(const rmw_publisher_t * publisher, const void * ros_message)
   }
 
   auto ret = RMW_RET_OK;
-  ConnextStaticCDRStream cdr_stream;
+  rcutils_uint8_array_t cdr_stream = rcutils_get_zero_initialized_uint8_array();
   cdr_stream.allocator = rcutils_get_default_allocator();
+
   if (!callbacks->to_cdr_stream(ros_message, &cdr_stream)) {
     RMW_SET_ERROR_MSG("failed to convert ros_message to cdr stream");
     ret = RMW_RET_ERROR;
@@ -170,11 +169,7 @@ rmw_publish_serialized_message(
     return RMW_RET_ERROR;
   }
 
-  ConnextStaticCDRStream cdr_stream;
-  cdr_stream.buffer = serialized_message->buffer;
-  cdr_stream.buffer_length = serialized_message->buffer_length;
-  cdr_stream.buffer_capacity = serialized_message->buffer_capacity;
-  bool published = publish(topic_writer, &cdr_stream);
+  bool published = publish(topic_writer, serialized_message);
   if (!published) {
     RMW_SET_ERROR_MSG("failed to publish message");
     return RMW_RET_ERROR;

rmw_connext_cpp/src/rmw_serialize.cpp

@@ -37,20 +37,11 @@ rmw_serialize(
     return RMW_RET_ERROR;
   }
 
-  ConnextStaticCDRStream cdr_stream;
-  cdr_stream.buffer = serialized_message->buffer;
-  cdr_stream.buffer_length = serialized_message->buffer_length;
-  cdr_stream.buffer_capacity = serialized_message->buffer_capacity;
-  cdr_stream.allocator = serialized_message->allocator;
-
-  if (!callbacks->to_cdr_stream(ros_message, &cdr_stream)) {
+  if (!callbacks->to_cdr_stream(ros_message, serialized_message)) {
     RMW_SET_ERROR_MSG("failed to convert ros_message to cdr stream");
     return RMW_RET_ERROR;
   }
-  // reassgin buffer because it might have been resized
-  serialized_message->buffer = cdr_stream.buffer;
-  serialized_message->buffer_length = cdr_stream.buffer_length;
-  serialized_message->buffer_capacity = cdr_stream.buffer_capacity;
+
   return RMW_RET_OK;
 }
 
@@ -69,13 +60,8 @@ rmw_deserialize(
     return RMW_RET_ERROR;
   }
 
-  ConnextStaticCDRStream cdr_stream;
-  cdr_stream.buffer = serialized_message->buffer;
-  cdr_stream.buffer_length = serialized_message->buffer_length;
-  cdr_stream.buffer_capacity = serialized_message->buffer_capacity;
-  cdr_stream.allocator = serialized_message->allocator;
   // convert the cdr stream to the message
-  if (!callbacks->to_message(&cdr_stream, ros_message)) {
+  if (!callbacks->to_message(serialized_message, ros_message)) {
     RMW_SET_ERROR_MSG("can't convert cdr stream to ros message");
     return RMW_RET_ERROR;
   }

rmw_connext_cpp/src/rmw_take.cpp

@@ -23,8 +23,6 @@
 #include "rmw_connext_cpp/connext_static_subscriber_info.hpp"
 #include "rmw_connext_cpp/identifier.hpp"
 
-#include "rosidl_typesupport_connext_cpp/connext_static_cdr_stream.hpp"
-
 // include patched generated code from the build folder
 #include "./connext_static_serialized_dataSupport.h"
 #include "./connext_static_serialized_data.h"
@@ -33,7 +31,7 @@ static bool
 take(
   DDSDataReader * dds_data_reader,
   bool ignore_local_publications,
-  ConnextStaticCDRStream * cdr_stream,
+  rcutils_uint8_array_t * cdr_stream,
   bool * taken,
   void * sending_publication_handle)
 {
@@ -107,7 +105,7 @@ take(
     cdr_stream->buffer_length = dds_messages[0].serialized_data.length();
     // TODO(karsten1987): This malloc has to go!
     cdr_stream->buffer =
-      reinterpret_cast<char *>(malloc(cdr_stream->buffer_length * sizeof(char)));
+      reinterpret_cast<uint8_t *>(malloc(cdr_stream->buffer_length * sizeof(uint8_t)));
 
     if (cdr_stream->buffer_length > (std::numeric_limits<unsigned int>::max)()) {
       RMW_SET_ERROR_MSG("cdr_stream->buffer_length unexpectedly larger than max unsiged int value");
@@ -173,7 +171,7 @@ _take(
   }
 
   // fetch the incoming message as cdr stream
-  ConnextStaticCDRStream cdr_stream;
+  rcutils_uint8_array_t cdr_stream = rcutils_get_zero_initialized_uint8_array();
   if (!take(
       topic_reader, subscriber_info->ignore_local_publications, &cdr_stream, taken,
       sending_publication_handle))
@@ -270,20 +268,14 @@ _take_serialized_message(
   }
 
   // fetch the incoming message as cdr stream
-  ConnextStaticCDRStream cdr_stream;
   if (!take(
-      topic_reader, subscriber_info->ignore_local_publications, &cdr_stream, taken,
+      topic_reader, subscriber_info->ignore_local_publications, serialized_message, taken,
       sending_publication_handle))
   {
     RMW_SET_ERROR_MSG("error occured while taking message");
     return RMW_RET_ERROR;
   }
 
-  serialized_message->buffer_length = cdr_stream.buffer_length;
-  // we don't free the allocated memory
-  // and thus can directly set the pointer
-  serialized_message->buffer = cdr_stream.buffer;
-
   return RMW_RET_OK;
 }