Add API for taking a sequence of messages (#212)

* Add API for taking a sequence of messages

This will allow users to take multiple messages in one take call if they
are available in the underlying middleware, verus having to make
multiple rmw_take calls in a row.

Signed-off-by: Michael Carroll <michael@openrobotics.org>

* Apply suggestions from code review

Co-Authored-By: Jacob Perron <jacob@openrobotics.org>
Signed-off-by: Michael Carroll <michael@openrobotics.org>

* Clarify rmw_take_sequence arguments

Signed-off-by: Michael Carroll <michael@openrobotics.org>

* Add support for using rcutils allocator

Signed-off-by: Michael Carroll <michael@openrobotics.org>

* Store allocator inside sequence structure

Signed-off-by: Michael Carroll <michael@openrobotics.org>

* Apply suggestions from code review

Co-Authored-By: Jacob Perron <jacob@openrobotics.org>
Co-Authored-By: William Woodall <william+github@osrfoundation.org>
Signed-off-by: Michael Carroll <michael@openrobotics.org>

* Update docs per reviewers

Signed-off-by: Michael Carroll <michael@openrobotics.org>

* Address reviewer feedback

Signed-off-by: Michael Carroll <michael@openrobotics.org>

Co-authored-by: Jacob Perron <jacob@openrobotics.org>
Co-authored-by: William Woodall <william+github@osrfoundation.org>

rmw/CMakeLists.txt

@@ -29,7 +29,7 @@ set(rmw_sources
   "src/event.c"
   "src/init.c"
   "src/init_options.c"
-  "src/loaned_message_sequence.c"
+  "src/message_sequence.c"
   "src/names_and_types.c"
   "src/publisher_options.c"
   "src/sanity_checks.c"

rmw/include/rmw/message_sequence.h

@@ -0,0 +1,123 @@
+// Copyright 2020 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__MESSAGE_SEQUENCE_H_
+#define RMW__MESSAGE_SEQUENCE_H_
+
+#include <stddef.h>
+
+#include "rmw/macros.h"
+#include "rmw/visibility_control.h"
+#include "rmw/types.h"
+
+#if __cplusplus
+extern "C"
+{
+#endif
+
+/// Structure to hold a sequence of ROS messages.
+typedef struct RMW_PUBLIC_TYPE rmw_message_sequence_t
+{
+  /// Array of pointers to ROS messages.
+  void ** data;
+  /// The number of valid entries in `data`.
+  size_t size;
+  /// The total allocated capacity of the data array.
+  size_t capacity;
+  /// The allocator used to allocate the data array.
+  rcutils_allocator_t * allocator;
+} rmw_message_sequence_t;
+
+/// Structure to hold a sequence of message infos.
+typedef struct RMW_PUBLIC_TYPE rmw_message_info_sequence_t
+{
+  /// Array of message info.
+  rmw_message_info_t * data;
+  /// The number of valid entries in data.
+  size_t size;
+  /// The total allocated capacity of the data array.
+  size_t capacity;
+  /// The allocator used to allocate the data array.
+  rcutils_allocator_t * allocator;
+} rmw_message_info_sequence_t;
+
+/// Return an rmw_message_sequence_t struct with members initialized to `NULL`
+RMW_PUBLIC
+rmw_message_sequence_t
+rmw_get_zero_initialized_message_sequence(void);
+
+/// Initialize an rmw_message_sequence_t object.
+/**
+ * \param[inout] sequence sequence object to be initialized.
+ * \param[in] size capacity of the sequence to be allocated.
+ * \param[in] allocator the allcator used to allocate memory.
+ */
+RMW_PUBLIC
+rmw_ret_t
+rmw_message_sequence_init(
+  rmw_message_sequence_t * sequence,
+  size_t size,
+  rcutils_allocator_t * allocator);
+
+/// Finalize an rmw_message_sequence_t object.
+/**
+ * The rmw_message_sequence_t struct has members which require memory to be allocated to them
+ * before setting values.
+ * This function reclaims any allocated resources within the object and zeroes out all other
+ * members.
+ *
+ * Note: This will not call `fini` or deallocate the underlying message structures.
+ *
+ * \param[inout] sequence sequence object to be finalized.
+ */
+RMW_PUBLIC
+rmw_ret_t
+rmw_message_sequence_fini(rmw_message_sequence_t * sequence);
+
+/// Return an rmw_message_info_sequence_t struct with members initialized to `NULL`
+RMW_PUBLIC
+rmw_message_info_sequence_t
+rmw_get_zero_initialized_message_info_sequence(void);
+
+/// Initialize an rmw_message_info_sequence_t object.
+/**
+ * \param[inout] sequence sequence object to be initialized.
+ * \param[in] size capacity of the sequence to be allocated.
+ * \param[in] allocator the allcator used to allocate memory.
+ */
+RMW_PUBLIC
+rmw_ret_t
+rmw_message_info_sequence_init(
+  rmw_message_info_sequence_t * sequence,
+  size_t size,
+  rcutils_allocator_t * allocator);
+
+/// Finalize an rmw_message_sequence_t object.
+/**
+ * The rmw_message_sequence_t struct has members which require memory to be allocated to them
+ * before setting values.
+ * This function reclaims any allocated resources within the object and zeroes out all other
+ * members.
+ *
+ * \param[inout] sequence sequence object to be finalized.
+ */
+RMW_PUBLIC
+rmw_ret_t
+rmw_message_info_sequence_fini(rmw_message_info_sequence_t * sequence);
+
+#if __cplusplus
+}
+#endif
+
+#endif  // RMW__MESSAGE_SEQUENCE_H_

rmw/include/rmw/rmw.h

@@ -98,6 +98,7 @@ extern "C"
 #include "rmw/macros.h"
 #include "rmw/qos_profiles.h"
 #include "rmw/subscription_options.h"
+#include "rmw/message_sequence.h"
 #include "rmw/types.h"
 #include "rmw/visibility_control.h"
 
@@ -716,6 +717,44 @@ rmw_take_with_info(
   rmw_message_info_t * message_info,
   rmw_subscription_allocation_t * allocation);
 
+/// Take multiple incoming messages from a subscription with additional metadata.
+/**
+ * Take a sequence of ROS messgages from a given subscription.
+ *
+ * While `count` messages may be requested, fewer messages may be available on the subscription.
+ * In this case, only the currently available messages will be returned.
+ * The `taken` flag indicates the number of messages actually taken.
+ * The method will return `RMW_RET_OK` even in the case that fewer (or zero) messages were retrieved.
+ * from the subscription, and will `RMW_RET_ERROR` in the case of unexpected errors.
+ * In the case that `count` is zero, the function will return `RMW_RET_INVALID_ARGUMENT`.
+ *
+ * `message_sequence` and `message_info_sequence` should be initialized and have sufficient capacity.
+ * It is not critical that the sequence sizes match, and they may be reused from previous calls.
+ * Both must be valid (not NULL) for the method to run successfully.
+ *
+ * \param[in] subscription The subscription object to take from.
+ * \param[in] count Number of messages to attempt to take.
+ * \param[out] message_sequence The sequence of ROS message data on success.
+ * \param[out] message_info_sequence The sequence of additional message metadata on success.
+ * \param[out] taken Number of messages actually taken from subscription.
+ * \param[in] allocation Preallocated buffer to use (may be NULL).
+ * \return `RMW_RET_OK` if successful, or
+ * \return `RMW_RET_INVALID_ARGUMENT` if an argument is invalid, or
+ * \return `RMW_RET_BAD_ALLOC` if memory allocation failed, or
+ * \return `RMW_RET_INCORRECT_RMW_IMPLEMENTATION` if the rmw implementation does not match, or
+ * \return `RMW_RET_ERROR` if an unexpected error occurs.
+ */
+RMW_PUBLIC
+RMW_WARN_UNUSED
+rmw_ret_t
+rmw_take_sequence(
+  const rmw_subscription_t * subscription,
+  size_t count,
+  rmw_message_sequence_t * message_sequence,
+  rmw_message_info_sequence_t * message_info_sequence,
+  size_t * taken,
+  rmw_subscription_allocation_t * allocation);
+
 /// Take a message without deserializing it.
 /**
  * The message is taken in its serialized form. In contrast to rmw_take, the message

rmw/include/rmw/types.h

@@ -29,7 +29,6 @@ extern "C"
 
 #include "rmw/init.h"
 #include "rmw/init_options.h"
-#include "rmw/loaned_message_sequence.h"
 #include "rmw/ret_types.h"
 #include "rmw/security_options.h"
 #include "rmw/serialized_message.h"
@@ -346,13 +345,6 @@ RMW_WARN_UNUSED
 rmw_message_info_t
 rmw_get_zero_initialized_message_info(void);
 
-typedef struct RMW_PUBLIC_TYPE rmw_message_info_sequence_t
-{
-  rmw_message_info_t * message_info_sequence;
-  size_t size;
-  size_t capacity;
-} rmw_message_info_sequence_t;
-
 enum {RMW_QOS_POLICY_DEPTH_SYSTEM_DEFAULT = 0};
 
 /// Type mapping of rcutils log severity types to rmw specific types.