ros2
diff --git a/‎rclcpp/CMakeLists.txt‎
Lines changed: 19 additions & 8 deletions b/‎rclcpp/CMakeLists.txt‎
Lines changed: 19 additions & 8 deletions
diff --git a/‎rclcpp/include/rclcpp/any_subscription_callback.hpp‎
Lines changed: 1 addition & 1 deletion b/‎rclcpp/include/rclcpp/any_subscription_callback.hpp‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎rclcpp/src/rclcpp/intra_process_manager_impl.cpp‎ renamed to ‎rclcpp/include/rclcpp/buffers/buffer_implementation_base.hpp‎
Lines changed: 21 additions & 7 deletions b/‎rclcpp/src/rclcpp/intra_process_manager_impl.cpp‎ renamed to ‎rclcpp/include/rclcpp/buffers/buffer_implementation_base.hpp‎
Lines changed: 21 additions & 7 deletions
diff --git a/‎rclcpp/include/rclcpp/buffers/intra_process_buffer.hpp‎
Lines changed: 230 additions & 0 deletions b/‎rclcpp/include/rclcpp/buffers/intra_process_buffer.hpp‎
Lines changed: 230 additions & 0 deletions
@@ -46,7 +46,6 @@ set(${PROJECT_NAME}_SRCS
   src/rclcpp/graph_listener.cpp
   src/rclcpp/init_options.cpp
   src/rclcpp/intra_process_manager.cpp
-  src/rclcpp/intra_process_manager_impl.cpp
   src/rclcpp/logger.cpp
   src/rclcpp/memory_strategies.cpp
   src/rclcpp/memory_strategy.cpp
@@ -74,6 +73,7 @@ set(${PROJECT_NAME}_SRCS
   src/rclcpp/service.cpp
   src/rclcpp/signal_handler.cpp
   src/rclcpp/subscription_base.cpp
+  src/rclcpp/subscription_intra_process_base.cpp
   src/rclcpp/time.cpp
   src/rclcpp/time_source.cpp
   src/rclcpp/timer.cpp
@@ -196,25 +196,36 @@ if(BUILD_TESTING)
       "rosidl_typesupport_cpp"
     )
   endif()
-  ament_add_gtest(test_mapped_ring_buffer test/test_mapped_ring_buffer.cpp)
-  if(TARGET test_mapped_ring_buffer)
-    ament_target_dependencies(test_mapped_ring_buffer
+  ament_add_gmock(test_intra_process_manager test/test_intra_process_manager.cpp)
+  if(TARGET test_intra_process_manager)
+    ament_target_dependencies(test_intra_process_manager
       "rcl"
       "rcl_interfaces"
       "rmw"
       "rosidl_generator_cpp"
       "rosidl_typesupport_cpp"
     )
+    target_link_libraries(test_intra_process_manager ${PROJECT_NAME})
   endif()
-  ament_add_gtest(test_intra_process_manager test/test_intra_process_manager.cpp)
-  if(TARGET test_intra_process_manager)
-    ament_target_dependencies(test_intra_process_manager
-      "rcl"
+  ament_add_gtest(test_ring_buffer_implementation test/test_ring_buffer_implementation.cpp)
+  if(TARGET test_ring_buffer_implementation)
+    ament_target_dependencies(test_ring_buffer_implementation
+      "rcl_interfaces"
+      "rmw"
+      "rosidl_generator_cpp"
+      "rosidl_typesupport_cpp"
+    )
+    target_link_libraries(test_ring_buffer_implementation ${PROJECT_NAME})
+  endif()
+  ament_add_gtest(test_intra_process_buffer test/test_intra_process_buffer.cpp)
+  if(TARGET test_intra_process_buffer)
+    ament_target_dependencies(test_intra_process_buffer
       "rcl_interfaces"
       "rmw"
       "rosidl_generator_cpp"
       "rosidl_typesupport_cpp"
     )
+    target_link_libraries(test_intra_process_buffer ${PROJECT_NAME})
   endif()
   ament_add_gtest(test_node test/test_node.cpp)
   if(TARGET test_node)
 
@@ -219,7 +219,7 @@ class AnySubscriptionCallback
     }
   }
 
-  bool use_take_shared_method()
+  bool use_take_shared_method() const
   {
     return const_shared_ptr_callback_ || const_shared_ptr_with_info_callback_;
   }
 
@@ -1,4 +1,4 @@
-// Copyright 2015 Open Source Robotics Foundation, Inc.
+// Copyright 2019 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.
@@ -12,12 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#include "rclcpp/intra_process_manager_impl.hpp"
+#ifndef RCLCPP__BUFFERS__BUFFER_IMPLEMENTATION_BASE_HPP_
+#define RCLCPP__BUFFERS__BUFFER_IMPLEMENTATION_BASE_HPP_
 
-#include <memory>
+namespace rclcpp
+{
+namespace intra_process_buffer
+{
 
-rclcpp::intra_process_manager::IntraProcessManagerImplBase::SharedPtr
-rclcpp::intra_process_manager::create_default_impl()
+template<typename BufferT>
+class BufferImplementationBase
 {
-  return std::make_shared<IntraProcessManagerImpl<>>();
-}
+public:
+  virtual BufferT dequeue() = 0;
+  virtual void enqueue(BufferT request) = 0;
+
+  virtual void clear() = 0;
+  virtual bool has_data() const = 0;
+};
+
+}  // namespace intra_process_buffer
+}  // namespace rclcpp
+
+#endif  // RCLCPP__BUFFERS__BUFFER_IMPLEMENTATION_BASE_HPP_
@@ -0,0 +1,230 @@
+// Copyright 2019 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 RCLCPP__BUFFERS__INTRA_PROCESS_BUFFER_HPP_
+#define RCLCPP__BUFFERS__INTRA_PROCESS_BUFFER_HPP_
+
+#include <memory>
+#include <type_traits>
+#include <utility>
+
+#include "rclcpp/allocator/allocator_common.hpp"
+#include "rclcpp/allocator/allocator_deleter.hpp"
+#include "rclcpp/buffers/buffer_implementation_base.hpp"
+
+namespace rclcpp
+{
+namespace intra_process_buffer
+{
+
+class IntraProcessBufferBase
+{
+public:
+  RCLCPP_SMART_PTR_ALIASES_ONLY(IntraProcessBufferBase)
+
+  virtual void clear() = 0;
+
+  virtual bool has_data() const = 0;
+  virtual bool use_take_shared_method() const = 0;
+};
+
+template<
+  typename MessageT,
+  typename Alloc = std::allocator<void>,
+  typename MessageDeleter = std::default_delete<MessageT>>
+class IntraProcessBuffer : public IntraProcessBufferBase
+{
+public:
+  RCLCPP_SMART_PTR_ALIASES_ONLY(IntraProcessBuffer)
+
+  using MessageUniquePtr = std::unique_ptr<MessageT, MessageDeleter>;
+  using MessageSharedPtr = std::shared_ptr<const MessageT>;
+
+  virtual void add_shared(MessageSharedPtr msg) = 0;
+  virtual void add_unique(MessageUniquePtr msg) = 0;
+
+  virtual MessageSharedPtr consume_shared() = 0;
+  virtual MessageUniquePtr consume_unique() = 0;
+};
+
+template<
+  typename MessageT,
+  typename Alloc = std::allocator<void>,
+  typename MessageDeleter = std::default_delete<MessageT>,
+  typename BufferT = std::unique_ptr<MessageT>>
+class TypedIntraProcessBuffer : public IntraProcessBuffer<MessageT, Alloc, MessageDeleter>
+{
+public:
+  RCLCPP_SMART_PTR_DEFINITIONS(TypedIntraProcessBuffer)
+
+  using MessageAllocTraits = allocator::AllocRebind<MessageT, Alloc>;
+  using MessageAlloc = typename MessageAllocTraits::allocator_type;
+  using MessageUniquePtr = std::unique_ptr<MessageT, MessageDeleter>;
+  using MessageSharedPtr = std::shared_ptr<const MessageT>;
+
+  TypedIntraProcessBuffer(
+    std::unique_ptr<BufferImplementationBase<BufferT>> buffer_impl,
+    std::shared_ptr<Alloc> allocator = nullptr)
+  {
+    bool valid_type = (std::is_same<BufferT, MessageSharedPtr>::value ||
+      std::is_same<BufferT, MessageUniquePtr>::value);
+    if (!valid_type) {
+      throw std::runtime_error("Creating TypedIntraProcessBuffer with not valid BufferT");
+    }
+
+    buffer_ = std::move(buffer_impl);
+
+    if (!allocator) {
+      message_allocator_ = std::make_shared<MessageAlloc>();
+    } else {
+      message_allocator_ = std::make_shared<MessageAlloc>(*allocator.get());
+    }
+  }
+
+  void add_shared(MessageSharedPtr msg)
+  {
+    add_shared_impl<BufferT>(std::move(msg));
+  }
+
+  void add_unique(MessageUniquePtr msg)
+  {
+    buffer_->enqueue(std::move(msg));
+  }
+
+  MessageSharedPtr consume_shared()
+  {
+    return consume_shared_impl<BufferT>();
+  }
+
+  MessageUniquePtr consume_unique()
+  {
+    return consume_unique_impl<BufferT>();
+  }
+
+  bool has_data() const
+  {
+    return buffer_->has_data();
+  }
+
+  void clear()
+  {
+    buffer_->clear();
+  }
+
+  bool use_take_shared_method() const
+  {
+    return std::is_same<BufferT, MessageSharedPtr>::value;
+  }
+
+private:
+  std::unique_ptr<BufferImplementationBase<BufferT>> buffer_;
+
+  std::shared_ptr<MessageAlloc> message_allocator_;
+
+  // MessageSharedPtr to MessageSharedPtr
+  template<typename DestinationT>
+  typename std::enable_if<
+    std::is_same<DestinationT, MessageSharedPtr>::value
+  >::type
+  add_shared_impl(MessageSharedPtr shared_msg)
+  {
+    buffer_->enqueue(std::move(shared_msg));
+  }
+
+  // MessageSharedPtr to MessageUniquePtr
+  template<typename DestinationT>
+  typename std::enable_if<
+    std::is_same<DestinationT, MessageUniquePtr>::value
+  >::type
+  add_shared_impl(MessageSharedPtr shared_msg)
+  {
+    // This should not happen: here a copy is unconditionally made, while the intra-process manager
+    // can decide whether a copy is needed depending on the number and the type of buffers
+
+    MessageUniquePtr unique_msg;
+    MessageDeleter * deleter = std::get_deleter<MessageDeleter, const MessageT>(shared_msg);
+    auto ptr = MessageAllocTraits::allocate(*message_allocator_.get(), 1);
+    MessageAllocTraits::construct(*message_allocator_.get(), ptr, *shared_msg);
+    if (deleter) {
+      unique_msg = MessageUniquePtr(ptr, *deleter);
+    } else {
+      unique_msg = MessageUniquePtr(ptr);
+    }
+
+    buffer_->enqueue(std::move(unique_msg));
+  }
+
+  // MessageSharedPtr to MessageSharedPtr
+  template<typename OriginT>
+  typename std::enable_if<
+    (std::is_same<OriginT, MessageSharedPtr>::value),
+    MessageSharedPtr
+  >::type
+  consume_shared_impl()
+  {
+    return buffer_->dequeue();
+  }
+
+  // MessageUniquePtr to MessageSharedPtr
+  template<typename OriginT>
+  typename std::enable_if<
+    (std::is_same<OriginT, MessageUniquePtr>::value),
+    MessageSharedPtr
+  >::type
+  consume_shared_impl()
+  {
+    // automatic cast from unique ptr to shared ptr
+    return buffer_->dequeue();
+  }
+
+  // MessageSharedPtr to MessageUniquePtr
+  template<typename OriginT>
+  typename std::enable_if<
+    (std::is_same<OriginT, MessageSharedPtr>::value),
+    MessageUniquePtr
+  >::type
+  consume_unique_impl()
+  {
+    MessageSharedPtr buffer_msg = buffer_->dequeue();
+
+    MessageUniquePtr unique_msg;
+    MessageDeleter * deleter = std::get_deleter<MessageDeleter, const MessageT>(buffer_msg);
+    auto ptr = MessageAllocTraits::allocate(*message_allocator_.get(), 1);
+    MessageAllocTraits::construct(*message_allocator_.get(), ptr, *buffer_msg);
+    if (deleter) {
+      unique_msg = MessageUniquePtr(ptr, *deleter);
+    } else {
+      unique_msg = MessageUniquePtr(ptr);
+    }
+
+    return unique_msg;
+  }
+
+  // MessageUniquePtr to MessageUniquePtr
+  template<typename OriginT>
+  typename std::enable_if<
+    (std::is_same<OriginT, MessageUniquePtr>::value),
+    MessageUniquePtr
+  >::type
+  consume_unique_impl()
+  {
+    return buffer_->dequeue();
+  }
+};
+
+}  // namespace intra_process_buffer
+}  // namespace rclcpp
+
+
+#endif  // RCLCPP__BUFFERS__INTRA_PROCESS_BUFFER_HPP_
Original file line number	Diff line number	Diff line change
`@@ -219,7 +219,7 @@ class AnySubscriptionCallback`
`219`	`219`	`}`
`220`	`220`	`}`
`221`	`221`
`222`		`- bool use_take_shared_method()`
	`222`	`+ bool use_take_shared_method() const`
`223`	`223`	`{`
`224`	`224`	`return const_shared_ptr_callback_ \|\| const_shared_ptr_with_info_callback_;`
`225`	`225`	`}`