rclcpp  master
C++ ROS Client Library API
subscription.hpp
Go to the documentation of this file.
1 // Copyright 2014 Open Source Robotics Foundation, Inc.
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License");
4 // you may not use this file except in compliance with the License.
5 // You may obtain a copy of the License at
6 //
7 // http://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS,
11 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 // See the License for the specific language governing permissions and
13 // limitations under the License.
14 
15 #ifndef RCLCPP__SUBSCRIPTION_HPP_
16 #define RCLCPP__SUBSCRIPTION_HPP_
17 
18 #include <rmw/error_handling.h>
19 #include <rmw/rmw.h>
20 
21 #include <chrono>
22 #include <functional>
23 #include <iostream>
24 #include <memory>
25 #include <sstream>
26 #include <string>
27 #include <utility>
28 
29 #include "rcl/error_handling.h"
30 #include "rcl/subscription.h"
31 
35 #include "rclcpp/exceptions.hpp"
39 #include "rclcpp/logging.hpp"
40 #include "rclcpp/macros.hpp"
41 #include "rclcpp/message_info.hpp"
49 #include "rclcpp/waitable.hpp"
51 #include "tracetools/tracetools.h"
52 
53 namespace rclcpp
54 {
55 
56 namespace node_interfaces
57 {
58 class NodeTopicsInterface;
59 } // namespace node_interfaces
60 
62 template<
63  typename CallbackMessageT,
64  typename AllocatorT = std::allocator<void>,
65  typename MessageMemoryStrategyT = rclcpp::message_memory_strategy::MessageMemoryStrategy<
66  CallbackMessageT,
67  AllocatorT
68  >>
70 {
72 
73 public:
75  using MessageAllocator = typename MessageAllocatorTraits::allocator_type;
81 
83 
84 
85 
103  rclcpp::node_interfaces::NodeBaseInterface * node_base,
104  const rosidl_message_type_support_t & type_support_handle,
105  const std::string & topic_name,
106  const rclcpp::QoS & qos,
107  AnySubscriptionCallback<CallbackMessageT, AllocatorT> callback,
108  const rclcpp::SubscriptionOptionsWithAllocator<AllocatorT> & options,
109  typename MessageMemoryStrategyT::SharedPtr message_memory_strategy,
110  SubscriptionTopicStatisticsSharedPtr subscription_topic_statistics = nullptr)
112  node_base,
113  type_support_handle,
114  topic_name,
115  options.template to_rcl_subscription_options<CallbackMessageT>(qos),
116  rclcpp::subscription_traits::is_serialized_subscription_argument<CallbackMessageT>::value),
117  any_callback_(callback),
118  options_(options),
119  message_memory_strategy_(message_memory_strategy)
120  {
121  if (options.event_callbacks.deadline_callback) {
122  this->add_event_handler(
123  options.event_callbacks.deadline_callback,
124  RCL_SUBSCRIPTION_REQUESTED_DEADLINE_MISSED);
125  }
126  if (options.event_callbacks.liveliness_callback) {
127  this->add_event_handler(
128  options.event_callbacks.liveliness_callback,
129  RCL_SUBSCRIPTION_LIVELINESS_CHANGED);
130  }
131  if (options.event_callbacks.incompatible_qos_callback) {
132  this->add_event_handler(
133  options.event_callbacks.incompatible_qos_callback,
134  RCL_SUBSCRIPTION_REQUESTED_INCOMPATIBLE_QOS);
135  } else if (options_.use_default_callbacks) {
136  // Register default callback when not specified
137  try {
138  this->add_event_handler(
139  [this](QOSRequestedIncompatibleQoSInfo & info) {
141  },
142  RCL_SUBSCRIPTION_REQUESTED_INCOMPATIBLE_QOS);
143  } catch (UnsupportedEventTypeException & /*exc*/) {
144  // pass
145  }
146  }
147 
148  // Setup intra process publishing if requested.
149  if (rclcpp::detail::resolve_use_intra_process(options, *node_base)) {
151 
152  // Check if the QoS is compatible with intra-process.
154  if (qos_profile.history == RMW_QOS_POLICY_HISTORY_KEEP_ALL) {
155  throw std::invalid_argument(
156  "intraprocess communication is not allowed with keep all history qos policy");
157  }
158  if (qos_profile.depth == 0) {
159  throw std::invalid_argument(
160  "intraprocess communication is not allowed with 0 depth qos policy");
161  }
162  if (qos_profile.durability != RMW_QOS_POLICY_DURABILITY_VOLATILE) {
163  throw std::invalid_argument(
164  "intraprocess communication allowed only with volatile durability");
165  }
166 
167  // First create a SubscriptionIntraProcess which will be given to the intra-process manager.
168  auto context = node_base->get_context();
169  using SubscriptionIntraProcessT = rclcpp::experimental::SubscriptionIntraProcess<
170  CallbackMessageT,
171  AllocatorT,
172  typename MessageUniquePtr::deleter_type>;
173  auto subscription_intra_process = std::make_shared<SubscriptionIntraProcessT>(
174  callback,
175  options.get_allocator(),
176  context,
177  this->get_topic_name(), // important to get like this, as it has the fully-qualified name
178  qos_profile,
179  resolve_intra_process_buffer_type(options.intra_process_buffer_type, callback));
180  TRACEPOINT(
181  rclcpp_subscription_init,
182  (const void *)get_subscription_handle().get(),
183  (const void *)subscription_intra_process.get());
184 
185  // Add it to the intra process manager.
187  auto ipm = context->get_sub_context<IntraProcessManager>();
188  uint64_t intra_process_subscription_id = ipm->add_subscription(subscription_intra_process);
189  this->setup_intra_process(intra_process_subscription_id, ipm);
190  }
191 
192  if (subscription_topic_statistics != nullptr) {
193  this->subscription_topic_statistics_ = std::move(subscription_topic_statistics);
194  }
195 
196  TRACEPOINT(
197  rclcpp_subscription_init,
198  (const void *)get_subscription_handle().get(),
199  (const void *)this);
200  TRACEPOINT(
201  rclcpp_subscription_callback_added,
202  (const void *)this,
203  (const void *)&any_callback_);
204  // The callback object gets copied, so if registration is done too early/before this point
205  // (e.g. in `AnySubscriptionCallback::set()`), its address won't match any address used later
206  // in subsequent tracepoints.
207 #ifndef TRACETOOLS_DISABLED
208  any_callback_.register_callback_for_tracing();
209 #endif
210  }
211 
213  void
216  const rclcpp::QoS & qos,
218  {
219  (void)node_base;
220  (void)qos;
221  (void)options;
222  }
223 
225 
242  bool
243  take(CallbackMessageT & message_out, rclcpp::MessageInfo & message_info_out)
244  {
245  return this->take_type_erased(static_cast<void *>(&message_out), message_info_out);
246  }
247 
249  create_message() override
250  {
251  /* The default message memory strategy provides a dynamically allocated message on each call to
252  * create_message, though alternative memory strategies that re-use a preallocated message may be
253  * used (see rclcpp/strategies/message_pool_memory_strategy.hpp).
254  */
255  return message_memory_strategy_->borrow_message();
256  }
257 
260  {
261  return message_memory_strategy_->borrow_serialized_message();
262  }
263 
264  void
266  std::shared_ptr<void> & message,
267  const rclcpp::MessageInfo & message_info) override
268  {
270  // In this case, the message will be delivered via intra process and
271  // we should ignore this copy of the message.
272  return;
273  }
274  auto typed_message = std::static_pointer_cast<CallbackMessageT>(message);
275  any_callback_.dispatch(typed_message, message_info);
276 
277  if (subscription_topic_statistics_) {
278  const auto nanos = std::chrono::time_point_cast<std::chrono::nanoseconds>(
280  const auto time = rclcpp::Time(nanos.time_since_epoch().count());
281  subscription_topic_statistics_->handle_message(*typed_message, time);
282  }
283  }
284 
285  void
287  void * loaned_message,
288  const rclcpp::MessageInfo & message_info) override
289  {
290  auto typed_message = static_cast<CallbackMessageT *>(loaned_message);
291  // message is loaned, so we have to make sure that the deleter does not deallocate the message
293  typed_message, [](CallbackMessageT * msg) {(void) msg;});
294  any_callback_.dispatch(sptr, message_info);
295  }
296 
298 
301  void
303  {
304  auto typed_message = std::static_pointer_cast<CallbackMessageT>(message);
305  message_memory_strategy_->return_message(typed_message);
306  }
307 
309 
312  void
314  {
315  message_memory_strategy_->return_serialized_message(message);
316  }
317 
318  bool
320  {
321  return any_callback_.use_take_shared_method();
322  }
323 
324 private:
326 
329 
335  message_memory_strategy_;
337  SubscriptionTopicStatisticsSharedPtr subscription_topic_statistics_{nullptr};
338 };
339 
340 } // namespace rclcpp
341 
342 #endif // RCLCPP__SUBSCRIPTION_HPP_
rclcpp::Subscription::create_serialized_message
std::shared_ptr< rclcpp::SerializedMessage > create_serialized_message() override
Borrow a new serialized message.
Definition: subscription.hpp:259
resolve_intra_process_buffer_type.hpp
RMW_QOS_POLICY_HISTORY_KEEP_ALL
RMW_QOS_POLICY_HISTORY_KEEP_ALL
rclcpp::SubscriptionBase::setup_intra_process
void setup_intra_process(uint64_t intra_process_subscription_id, IntraProcessManagerWeakPtr weak_ipm)
Implemenation detail.
exceptions.hpp
std::shared_ptr
rclcpp::Subscription::handle_message
void handle_message(std::shared_ptr< void > &message, const rclcpp::MessageInfo &message_info) override
Check if we need to handle the message, and execute the callback if we do.
Definition: subscription.hpp:265
message_info.hpp
std::move
T move(T... args)
RCLCPP_DISABLE_COPY
#define RCLCPP_DISABLE_COPY(...)
Definition: macros.hpp:26
rmw.h
rmw_message_info_t::publisher_gid
rmw_gid_t publisher_gid
rclcpp::QoS::get_rmw_qos_profile
rmw_qos_profile_t & get_rmw_qos_profile()
Return the rmw qos profile.
rclcpp::Subscription::SubscriptionTopicStatisticsSharedPtr
std::shared_ptr< rclcpp::topic_statistics::SubscriptionTopicStatistics< CallbackMessageT > > SubscriptionTopicStatisticsSharedPtr
Definition: subscription.hpp:80
error_handling.h
rclcpp::Time
Definition: time.hpp:31
rclcpp::SubscriptionBase::get_subscription_handle
std::shared_ptr< rcl_subscription_t > get_subscription_handle()
std::shared_ptr::get
T get(T... args)
rclcpp::SubscriptionOptionsWithAllocator
Structure containing optional configuration for Subscriptions.
Definition: subscription_options.hpp:79
rclcpp::SubscriptionBase
Definition: subscription_base.hpp:59
rmw_qos_profile_t::durability
enum rmw_qos_durability_policy_t durability
rclcpp::Subscription::return_serialized_message
void return_serialized_message(std::shared_ptr< rclcpp::SerializedMessage > &message) override
Return the borrowed serialized message.
Definition: subscription.hpp:313
subscription_base.hpp
rclcpp::SubscriptionBase::default_incompatible_qos_callback
void default_incompatible_qos_callback(QOSRequestedIncompatibleQoSInfo &info) const
rclcpp
This header provides the get_node_base_interface() template function.
Definition: allocator_common.hpp:24
rclcpp::Subscription
Subscription implementation, templated on the type of message this subscription receives.
Definition: subscription.hpp:69
rclcpp::experimental::IntraProcessManager::add_subscription
uint64_t add_subscription(rclcpp::experimental::SubscriptionIntraProcessBase::SharedPtr subscription)
Register a subscription with the manager, returns subscriptions unique id.
rclcpp::node_interfaces::NodeBaseInterface
Pure virtual interface class for the NodeBase part of the Node API.
Definition: node_base_interface.hpp:36
rclcpp::QoS
Encapsulation of Quality of Service settings.
Definition: qos.hpp:59
RCLCPP_SMART_PTR_DEFINITIONS
#define RCLCPP_SMART_PTR_DEFINITIONS(...)
Definition: macros.hpp:36
subscription_traits.hpp
macros.hpp
rclcpp::Subscription::use_take_shared_method
bool use_take_shared_method() const
Definition: subscription.hpp:319
resolve_use_intra_process.hpp
logging.hpp
node_base_interface.hpp
rclcpp::allocator::AllocRebind
typename std::allocator_traits< Alloc >::template rebind_traits< T > AllocRebind
Definition: allocator_common.hpp:30
rclcpp::Subscription::MessageDeleter
allocator::Deleter< MessageAllocator, CallbackMessageT > MessageDeleter
Definition: subscription.hpp:76
rclcpp::MessageInfo::get_rmw_message_info
const rmw_message_info_t & get_rmw_message_info() const
Return the message info as the underlying rmw message info type.
expand_topic_or_service_name.hpp
rclcpp::Subscription::MessageAllocatorTraits
allocator::AllocRebind< CallbackMessageT, AllocatorT > MessageAllocatorTraits
Definition: subscription.hpp:74
rclcpp::experimental::SubscriptionIntraProcess
Definition: subscription_intra_process.hpp:45
rclcpp::AnySubscriptionCallback
Definition: any_subscription_callback.hpp:37
rclcpp::UnsupportedEventTypeException
Definition: qos_event.hpp:64
rmw_qos_profile_t::history
enum rmw_qos_history_policy_t history
std::invalid_argument
rclcpp::Subscription::take
bool take(CallbackMessageT &message_out, rclcpp::MessageInfo &message_info_out)
Take the next message from the inter-process subscription.
Definition: subscription.hpp:243
intra_process_manager.hpp
any_subscription_callback.hpp
rclcpp::node_interfaces::NodeTopicsInterface
Pure virtual interface class for the NodeTopics part of the Node API.
Definition: node_topics_interface.hpp:41
rclcpp::SubscriptionBase::take_type_erased
bool take_type_erased(void *message_out, rclcpp::MessageInfo &message_info_out)
Take the next inter-process message from the subscription as a type erased pointer.
subscription_options.hpp
rclcpp::MessageInfo
Additional meta data about messages taken from subscriptions.
Definition: message_info.hpp:26
rclcpp::detail::resolve_use_intra_process
bool resolve_use_intra_process(const OptionsT &options, const NodeBaseT &node_base)
Return whether or not intra process is enabled, resolving "NodeDefault" if needed.
Definition: resolve_use_intra_process.hpp:31
type_support_decl.hpp
visibility_control.hpp
rclcpp::allocator::Deleter
typename std::conditional< std::is_same< typename std::allocator_traits< Alloc >::template rebind_alloc< T >, typename std::allocator< void >::template rebind< T >::other >::value, std::default_delete< T >, AllocatorDeleter< Alloc > >::type Deleter
Definition: allocator_deleter.hpp:101
std
rclcpp::message_memory_strategy::MessageMemoryStrategy
Default allocation strategy for messages received by subscriptions.
Definition: message_memory_strategy.hpp:41
rmw_qos_incompatible_event_status_t
rclcpp::Subscription::handle_loaned_message
void handle_loaned_message(void *loaned_message, const rclcpp::MessageInfo &message_info) override
Definition: subscription.hpp:286
rmw_qos_profile_t
std::allocator
rclcpp::SubscriptionBase::matches_any_intra_process_publishers
bool matches_any_intra_process_publishers(const rmw_gid_t *sender_gid) const
message_memory_strategy.hpp
subscription_intra_process.hpp
rclcpp::experimental::IntraProcessManager
This class performs intra process communication between nodes.
Definition: intra_process_manager.hpp:91
rclcpp::Subscription::post_init_setup
void post_init_setup(rclcpp::node_interfaces::NodeBaseInterface *node_base, const rclcpp::QoS &qos, const rclcpp::SubscriptionOptionsWithAllocator< AllocatorT > &options)
Called after construction to continue setup that requires shared_from_this().
Definition: subscription.hpp:214
rclcpp::Subscription::return_message
void return_message(std::shared_ptr< void > &message) override
Return the borrowed message.
Definition: subscription.hpp:302
std::unique_ptr
rclcpp::SubscriptionBase::add_event_handler
void add_event_handler(const EventCallbackT &callback, const rcl_subscription_event_type_t event_type)
Definition: subscription_base.hpp:269
waitable.hpp
RMW_QOS_POLICY_DURABILITY_VOLATILE
RMW_QOS_POLICY_DURABILITY_VOLATILE
subscription_topic_statistics.hpp
rclcpp::SubscriptionBase::get_actual_qos
rclcpp::QoS get_actual_qos() const
Get the actual QoS settings, after the defaults have been determined.
rmw_qos_profile_t::depth
size_t depth
rclcpp::Subscription::create_message
std::shared_ptr< void > create_message() override
Borrow a new message.
Definition: subscription.hpp:249
rclcpp::detail::resolve_intra_process_buffer_type
rclcpp::IntraProcessBufferType resolve_intra_process_buffer_type(const rclcpp::IntraProcessBufferType buffer_type, const rclcpp::AnySubscriptionCallback< CallbackMessageT, AllocatorT > &any_subscription_callback)
Return the buffer type, resolving the "CallbackDefault" type to an actual type if needed.
Definition: resolve_intra_process_buffer_type.hpp:32
rclcpp::Subscription::MessageAllocator
typename MessageAllocatorTraits::allocator_type MessageAllocator
Definition: subscription.hpp:75
std::chrono::system_clock::now
T now(T... args)