*_raw function (#125)

* publish_raw function

* add rmw_take_raw

* extract raw message

* add rmw_serialize functions

* linters

* address comments

* rename to rmw_raw_message_init

* documentation for new functions

* address comments

* raw->serialized

* documentation fixups (restored after being force pushed)

3e133b2
afb72ef
0a53658
8d0acff
35ee9e3

* fix rmw_take_serialized() -> rmw_take_serialized_message()

* use size_t (#139)

rmw/CMakeLists.txt

@@ -29,6 +29,7 @@ set(rmw_sources
   "src/convert_rcutils_ret_to_rmw_ret.c"
   "src/error_handling.c"
   "src/names_and_types.c"
+  "src/serialized_message.c"
   "src/sanity_checks.c"
   "src/node_security_options.c"
   "src/validate_full_topic_name.c"

rmw/include/rmw/rmw.h

@@ -171,6 +171,70 @@ RMW_WARN_UNUSED
 rmw_ret_t
 rmw_publish(const rmw_publisher_t * publisher, const void * ros_message);
 
+/// Publish an already serialized message.
+/**
+ * The publisher must already be registered with the correct message type
+ * support so that it can send serialized data corresponding to that type.
+ * This function sends the serialized byte stream directly over the wire without
+ * having to serialize the message first.
+ * A ROS message can be serialized manually using the rmw_serialize() function.
+ *
+ * \param publisher the publisher object registered to send the message
+ * \param serialized_message the serialized message holding the byte stream
+ * \return `RMW_RET_OK` if successful, or
+ * \return `RMW_RET_ERROR` if an unexpected error occurs.
+ */
+RMW_PUBLIC
+RMW_WARN_UNUSED
+rmw_ret_t
+rmw_publish_serialized_message(
+  const rmw_publisher_t * publisher, const rmw_serialized_message_t * serialized_message);
+
+/// Serialize a ROS message into a rmw_serialized_message_t.
+/**
+ * The ROS message is serialized into a byte stream contained within the
+ * rmw_serialized_message_t structure.
+ * The serialization format depends on the underlying middleware.
+ *
+ * \param ros_message the typed ROS message
+ * \param type_support the typesupport for the ROS message
+ * \param serialized_message the destination for the serialize ROS message
+ * \return `RMW_RET_OK` if successful, or
+ * \return `RMW_RET_BAD_ALLOC` if memory allocation failed, or
+ * \return `RMW_RET_ERROR` if an unexpected error occurs.
+ */
+RMW_PUBLIC
+RMW_WARN_UNUSED
+rmw_ret_t
+rmw_serialize(
+  const void * ros_message,
+  const rosidl_message_type_support_t * type_support,
+  rmw_serialized_message_t * serialized_message);
+
+/// Deserialize a ROS message.
+/**
+ * The given rmw_serialized_message_t's internal byte stream buffer is deserialized
+ * into the given ROS message.
+ * The ROS message must already be allocated and initialized, and must match
+ * the given typesupport structure.
+ * The serialization format expected in the rmw_serialized_message_t depends on the
+ * underlying middleware.
+ *
+ * \param serialized_message the serialized message holding the byte stream
+ * \param type_support the typesupport for the typed ros message
+ * \param ros_message destination for the deserialized ROS message
+ * \return `RMW_RET_OK` if successful, or
+ * \return `RMW_RET_BAD_ALLOC` if memory allocation failed, or
+ * \return `RMW_RET_ERROR` if an unexpected error occurs.
+ */
+RMW_PUBLIC
+RMW_WARN_UNUSED
+rmw_ret_t
+rmw_deserialize(
+  const rmw_serialized_message_t * serialized_message,
+  const rosidl_message_type_support_t * type_support,
+  void * ros_message);
+
 RMW_PUBLIC
 RMW_WARN_UNUSED
 rmw_subscription_t *
@@ -200,6 +264,51 @@ rmw_take_with_info(
   bool * taken,
   rmw_message_info_t * message_info);
 
+/// Take a message without deserializing it.
+/**
+ * The message is taken in its serialized form. In contrast to rmw_take, the message
+ * is not deserialized in its ROS type but rather returned as a byte stream.
+ * The subscriber has to be registered for a specific type. But instead of receiving
+ * the message as its corresponding message type, it is taken as a byte stream.
+ * If needed, this byte stream can then be deserialized in a ROS message with a call to
+ * rmw_deserialize.
+ *
+ * \param subscription subscription object to take from
+ * \param serialized_message the destination in which to store the serialized message
+ * \param taken boolean flag indicating if a message was taken or not
+ * \return `RMW_RET_OK` if successful, or
+ * \return `RMW_RET_BAD_ALLOC` if memory allocation failed, or
+ * \return `RMW_RET_ERROR` if an unexpected error occurs.
+ */
+RMW_PUBLIC
+RMW_WARN_UNUSED
+rmw_ret_t
+rmw_take_serialized_message(
+  const rmw_subscription_t * subscription,
+  rmw_serialized_message_t * serialized_message,
+  bool * taken);
+
+/// Take a message without deserializing it and with its additional message information.
+/**
+ * The same as rmw_take_serialized_message(), except it also includes the
+ * rmw_message_info_t.
+ *
+ * \param subscription subscription object to take from
+ * \param serialized_message the destination in which to store the serialized message
+ * \param taken boolean flag indicating if a message was taken or not
+ * \return `RMW_RET_OK` if successful, or
+ * \return `RMW_RET_BAD_ALLOC` if memory allocation failed, or
+ * \return `RMW_RET_ERROR` if an unexpected error occurs.
+ */
+RMW_PUBLIC
+RMW_WARN_UNUSED
+rmw_ret_t
+rmw_take_serialized_message_with_info(
+  const rmw_subscription_t * subscription,
+  rmw_serialized_message_t * serialized_message,
+  bool * taken,
+  rmw_message_info_t * message_info);
+
 RMW_PUBLIC
 RMW_WARN_UNUSED
 rmw_client_t *

rmw/include/rmw/serialized_message.h

@@ -0,0 +1,101 @@
+// Copyright 2018 Open Source Robotics Foundation, Inc.
+//
+// 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 RMW__SERIALIZED_MESSAGE_H_
+#define RMW__SERIALIZED_MESSAGE_H_
+
+#if __cplusplus
+extern "C"
+{
+#endif
+
+#include "rcutils/allocator.h"
+
+#include "rmw/macros.h"
+#include "rmw/types.h"
+#include "rmw/visibility_control.h"
+
+/// Return a zero initialized serialized message struct.
+/**
+ * \return rmw_serialized_message_t a zero initialized serialized message struct
+ */
+RMW_PUBLIC
+RMW_WARN_UNUSED
+rmw_serialized_message_t
+rmw_get_zero_initialized_serialized_message(void);
+
+/// Initialize a zero initialized serialized message struct.
+/**
+ * This function may leak if the serialized message struct is already
+ * pre-initialized.
+ * If the capacity is set to 0, no memory is allocated and the internal buffer
+ * is still NULL.
+ *
+ * \param msg a pointer to the to be initialized serialized message struct
+ * \param buffer_capacity the size of the memory to allocate for the byte stream
+ * \param allocator the allocator to use for the memory allocation
+ * \return `RMW_RET_OK` if successful, or
+ * \return `RMW_RET_ERROR` if an unexpected error occurs
+ */
+RMW_PUBLIC
+RMW_WARN_UNUSED
+rmw_ret_t
+rmw_serialized_message_init(
+  rmw_serialized_message_t * msg,
+  size_t buffer_capacity,
+  const rcutils_allocator_t * allocator);
+
+/// Finalize a serialized message struct.
+/**
+ * Cleans up and deallocates any resources used in a rmw_message_serialized_t.
+ * Passing a rmw_serialized_message_t which has not been zero initialized using
+ * rmw_get_zero_initialized_serialized_message() to this function is undefined
+ * behavior.
+ *
+ * \param msg pointer to the serialized message to be cleaned up
+ * \return `RMW_RET_OK` if successful, or
+ * \return `RMW_RET_BAD_ALLOC` if memory allocation failed, or
+ * \return `RMW_RET_ERROR` if an unexpected error occurs
+ */
+RMW_PUBLIC
+RMW_WARN_UNUSED
+rmw_ret_t
+rmw_serialized_message_fini(rmw_serialized_message_t * msg);
+
+/// Resize the internal buffer for the message byte stream.
+/**
+ * The internal buffer of the serialized message can be resized dynamically if needed.
+ * If the new size is smaller than the current capacity, then the memory is
+ * truncated.
+ * Be aware, that this will deallocate the memory and therefore invalidates any
+ * pointers to this storage.
+ * If the new size is larger, new memory is getting allocated and the existing
+ * content is copied over.
+ *
+ * \param msg pointer to the instance of rmw_serialized_message_t which is being resized
+ * \param new_size the new size of the internal buffer
+ * \return `RMW_RET_OK` if successful, or
+ * \return `RMW_RET_BAD_ALLOC` if memory allocation failed, or
+ * \return `RMW_RET_ERROR` if an unexpected error occurs
+ */
+RMW_PUBLIC
+RMW_WARN_UNUSED
+rmw_ret_t
+rmw_serialized_message_resize(rmw_serialized_message_t * msg, size_t new_size);
+
+#if __cplusplus
+}
+#endif
+
+#endif  // RMW__SERIALIZED_MESSAGE_H_

rmw/include/rmw/types.h

@@ -26,6 +26,7 @@ extern "C"
 
 // map rcutils specific log levels to rmw speicfic type
 #include <rcutils/logging.h>
+
 #include "rmw/visibility_control.h"
 
 typedef int rmw_ret_t;
@@ -221,6 +222,15 @@ typedef struct RMW_PUBLIC_TYPE rmw_gid_t
   uint8_t data[RMW_GID_STORAGE_SIZE];
 } rmw_gid_t;
 
+typedef struct RMW_PUBLIC_TYPE rmw_serialized_message_t
+{
+  // serialized message data
+  char * buffer;
+  size_t buffer_length;
+  size_t buffer_capacity;
+  rcutils_allocator_t allocator;
+} rmw_serialized_message_t;
+
 typedef struct RMW_PUBLIC_TYPE rmw_message_info_t
 {
   // const rmw_time_t received_timestamp;

rmw/src/serialized_message.c

@@ -0,0 +1,115 @@
+// Copyright 2018 Open Source Robotics Foundation, Inc.
+//
+// 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.
+
+#include "rcutils/error_handling.h"
+
+#include "rmw/error_handling.h"
+#include "rmw/serialized_message.h"
+
+rmw_serialized_message_t
+rmw_get_zero_initialized_serialized_message(void)
+{
+  static rmw_serialized_message_t serialized_message = {
+    .buffer = NULL,
+    .buffer_length = 0u,
+    .buffer_capacity = 0u
+  };
+  serialized_message.allocator = rcutils_get_zero_initialized_allocator();
+  return serialized_message;
+}
+
+rmw_ret_t
+rmw_serialized_message_init(
+  rmw_serialized_message_t * msg,
+  size_t buffer_capacity,
+  const rcutils_allocator_t * allocator)
+{
+  rcutils_allocator_t error_msg_allocator = rcutils_get_default_allocator();
+  RCUTILS_CHECK_FOR_NULL_WITH_MSG(
+    msg, "serialized message pointer is null", return RMW_RET_ERROR, error_msg_allocator);
+
+  if (!rcutils_allocator_is_valid(allocator)) {
+    RMW_SET_ERROR_MSG("serialized message has no valid allocator");
+    return RMW_RET_ERROR;
+  }
+
+  msg->buffer_length = 0;
+  msg->buffer_capacity = buffer_capacity;
+  msg->allocator = *allocator;
+
+  if (buffer_capacity > 0u) {
+    msg->buffer = (char *)allocator->allocate(buffer_capacity * sizeof(char), allocator->state);
+    RCUTILS_CHECK_FOR_NULL_WITH_MSG(
+      msg->buffer,
+      "failed to allocate memory for serialized message",
+      return RMW_RET_BAD_ALLOC,
+      *allocator);
+  }
+
+  return RMW_RET_OK;
+}
+
+rmw_ret_t
+rmw_serialized_message_fini(rmw_serialized_message_t * msg)
+{
+  rcutils_allocator_t error_msg_allocator = rcutils_get_default_allocator();
+  RCUTILS_CHECK_FOR_NULL_WITH_MSG(
+    msg, "serialized message pointer is null", return RMW_RET_ERROR, error_msg_allocator);
+
+  rcutils_allocator_t * allocator = &msg->allocator;
+  if (!rcutils_allocator_is_valid(allocator)) {
+    RMW_SET_ERROR_MSG("serialized message has no valid allocator");
+    return RMW_RET_ERROR;
+  }
+
+  allocator->deallocate(msg->buffer, allocator->state);
+  msg->buffer = NULL;
+  msg->buffer_length = 0u;
+  msg->buffer_capacity = 0u;
+
+  return RMW_RET_OK;
+}
+
+rmw_ret_t
+rmw_serialized_message_resize(rmw_serialized_message_t * msg, size_t new_size)
+{
+  rcutils_allocator_t error_msg_allocator = rcutils_get_default_allocator();
+  RCUTILS_CHECK_FOR_NULL_WITH_MSG(
+    msg, "serialized message pointer is null", return RMW_RET_ERROR, error_msg_allocator);
+
+  rcutils_allocator_t * allocator = &msg->allocator;
+  if (!rcutils_allocator_is_valid(allocator)) {
+    RMW_SET_ERROR_MSG("serialized message has no valid allocator");
+    return RMW_RET_ERROR;
+  }
+
+  if (new_size == msg->buffer_capacity) {
+    // nothing to do here
+    return RMW_RET_OK;
+  }
+
+  msg->buffer = rcutils_reallocf(msg->buffer, new_size * sizeof(char), allocator);
+  RCUTILS_CHECK_FOR_NULL_WITH_MSG(
+    msg->buffer,
+    "failed to reallocate memory for serialized message",
+    return RMW_RET_BAD_ALLOC,
+    *allocator);
+
+  msg->buffer_capacity = new_size;
+  if (new_size < msg->buffer_length) {
+    msg->buffer_length = new_size;
+  }
+
+  return RMW_RET_OK;
+}

rmw/test/CMakeLists.txt

@@ -1,5 +1,14 @@
 find_package(ament_cmake_gmock REQUIRED)
 
+ament_add_gmock(test_serialized_message
+  test_serialized_message.cpp
+  # Append the directory of librmw so it is found at test time.
+  APPEND_LIBRARY_DIRS "$<TARGET_FILE_DIR:${PROJECT_NAME}>"
+)
+if(TARGET test_serialized_message)
+  target_link_libraries(test_serialized_message ${PROJECT_NAME})
+endif()
+
 ament_add_gmock(test_validate_full_topic_name
   test_validate_full_topic_name.cpp
   # Append the directory of librmw so it is found at test time.