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 <functional>
22 #include <iostream>
23 #include <memory>
24 #include <sstream>
25 #include <string>
26 
27 #include "rcl/error_handling.h"
28 #include "rcl/subscription.h"
29 
30 #include "rcl_interfaces/msg/intra_process_message.hpp"
31 
33 #include "rclcpp/exceptions.hpp"
35 #include "rclcpp/macros.hpp"
40 
41 namespace rclcpp
42 {
43 
44 namespace node_interfaces
45 {
46 class NodeTopicsInterface;
47 } // namespace node_interfaces
48 
52 {
53 public:
55 
56 
57 
66  std::shared_ptr<rcl_node_t> node_handle,
67  const rosidl_message_type_support_t & type_support_handle,
68  const std::string & topic_name,
69  const rcl_subscription_options_t & subscription_options,
70  bool is_serialized = false);
71 
74  virtual ~SubscriptionBase();
75 
78  const char *
79  get_topic_name() const;
80 
83  get_subscription_handle();
84 
87  get_subscription_handle() const;
88 
91  get_intra_process_subscription_handle() const;
92 
94 
95  virtual std::shared_ptr<void>
96  create_message() = 0;
97 
99 
101  create_serialized_message() = 0;
102 
104 
108  virtual void
109  handle_message(std::shared_ptr<void> & message, const rmw_message_info_t & message_info) = 0;
110 
112 
113  virtual void
114  return_message(std::shared_ptr<void> & message) = 0;
115 
117 
118  virtual void
119  return_serialized_message(std::shared_ptr<rcl_serialized_message_t> & message) = 0;
120 
121  virtual void
122  handle_intra_process_message(
123  rcl_interfaces::msg::IntraProcessMessage & ipm,
124  const rmw_message_info_t & message_info) = 0;
125 
126  const rosidl_message_type_support_t &
127  get_message_type_support_handle() const;
128 
129  bool
130  is_serialized() const;
131 
132 protected:
136 
137 private:
138  RCLCPP_DISABLE_COPY(SubscriptionBase)
139 
140  rosidl_message_type_support_t type_support_;
141  bool is_serialized_;
142 };
143 
145 template<
146  typename CallbackMessageT,
147  typename Alloc = std::allocator<void>>
149 {
151 
152 public:
154  using MessageAlloc = typename MessageAllocTraits::allocator_type;
157 
159 
160 
161 
172  std::shared_ptr<rcl_node_t> node_handle,
173  const rosidl_message_type_support_t & type_support_handle,
174  const std::string & topic_name,
175  const rcl_subscription_options_t & subscription_options,
176  AnySubscriptionCallback<CallbackMessageT, Alloc> callback,
177  typename message_memory_strategy::MessageMemoryStrategy<CallbackMessageT, Alloc>::SharedPtr
178  memory_strategy = message_memory_strategy::MessageMemoryStrategy<CallbackMessageT,
179  Alloc>::create_default())
181  node_handle,
182  type_support_handle,
183  topic_name,
184  subscription_options,
185  rclcpp::subscription_traits::is_serialized_subscription_argument<CallbackMessageT>::value),
186  any_callback_(callback),
187  message_memory_strategy_(memory_strategy),
188  get_intra_process_message_callback_(nullptr),
189  matches_any_intra_process_publishers_(nullptr)
190  {}
191 
193 
198  typename message_memory_strategy::MessageMemoryStrategy<CallbackMessageT,
199  Alloc>::SharedPtr message_memory_strategy)
200  {
201  message_memory_strategy_ = message_memory_strategy;
202  }
203 
205  {
206  /* The default message memory strategy provides a dynamically allocated message on each call to
207  * create_message, though alternative memory strategies that re-use a preallocated message may be
208  * used (see rclcpp/strategies/message_pool_memory_strategy.hpp).
209  */
210  return message_memory_strategy_->borrow_message();
211  }
212 
214  {
215  return message_memory_strategy_->borrow_serialized_message();
216  }
217 
218  void handle_message(std::shared_ptr<void> & message, const rmw_message_info_t & message_info)
219  {
220  if (matches_any_intra_process_publishers_) {
221  if (matches_any_intra_process_publishers_(&message_info.publisher_gid)) {
222  // In this case, the message will be delivered via intra process and
223  // we should ignore this copy of the message.
224  return;
225  }
226  }
227  auto typed_message = std::static_pointer_cast<CallbackMessageT>(message);
228  any_callback_.dispatch(typed_message, message_info);
229  }
230 
232 
234  {
235  auto typed_message = std::static_pointer_cast<CallbackMessageT>(message);
236  message_memory_strategy_->return_message(typed_message);
237  }
238 
240  {
241  message_memory_strategy_->return_serialized_message(message);
242  }
243 
245  rcl_interfaces::msg::IntraProcessMessage & ipm,
246  const rmw_message_info_t & message_info)
247  {
248  if (!get_intra_process_message_callback_) {
249  // throw std::runtime_error(
250  // "handle_intra_process_message called before setup_intra_process");
251  // TODO(wjwwood): for now, this could mean that intra process was just not enabled.
252  // However, this can only really happen if this node has it disabled, but the other doesn't.
253  return;
254  }
255  MessageUniquePtr msg;
256  get_intra_process_message_callback_(
257  ipm.publisher_id,
258  ipm.message_sequence,
259  intra_process_subscription_id_,
260  msg);
261  if (!msg) {
262  // This either occurred because the publisher no longer exists or the
263  // message requested is no longer being stored.
264  // TODO(wjwwood): should we notify someone of this? log error, log warning?
265  return;
266  }
267  any_callback_.dispatch_intra_process(msg, message_info);
268  }
269 
270  using GetMessageCallbackType =
273 
276  uint64_t intra_process_subscription_id,
277  GetMessageCallbackType get_message_callback,
278  MatchesAnyPublishersCallbackType matches_any_publisher_callback,
279  const rcl_subscription_options_t & intra_process_options)
280  {
281  std::string intra_process_topic_name = std::string(get_topic_name()) + "/_intra";
283  intra_process_subscription_handle_.get(),
284  node_handle_.get(),
286  intra_process_topic_name.c_str(),
287  &intra_process_options);
288  if (ret != RCL_RET_OK) {
289  if (ret == RCL_RET_TOPIC_NAME_INVALID) {
290  auto rcl_node_handle = node_handle_.get();
291  // this will throw on any validation problem
292  rcl_reset_error();
294  intra_process_topic_name,
295  rcl_node_get_name(rcl_node_handle),
296  rcl_node_get_namespace(rcl_node_handle));
297  }
298 
299  rclcpp::exceptions::throw_from_rcl_error(ret, "could not create intra process subscription");
300  }
301 
302  intra_process_subscription_id_ = intra_process_subscription_id;
303  get_intra_process_message_callback_ = get_message_callback;
304  matches_any_intra_process_publishers_ = matches_any_publisher_callback;
305  }
306 
310  {
311  if (!get_intra_process_message_callback_) {
312  return nullptr;
313  }
314  return intra_process_subscription_handle_;
315  }
316 
317 private:
319 
322  message_memory_strategy_;
323 
324  GetMessageCallbackType get_intra_process_message_callback_;
325  MatchesAnyPublishersCallbackType matches_any_intra_process_publishers_;
326  uint64_t intra_process_subscription_id_;
327 };
328 
329 } // namespace rclcpp
330 
331 #endif // RCLCPP__SUBSCRIPTION_HPP_
Default allocation strategy for messages received by subscriptions.
Definition: message_memory_strategy.hpp:40
std::shared_ptr< rcl_subscription_t > subscription_handle_
Definition: subscription.hpp:134
#define RCLCPP_DISABLE_COPY(...)
Definition: macros.hpp:26
#define rcl_reset_error
rmw_ret_t rcl_ret_t
std::shared_ptr< rcl_serialized_message_t > create_serialized_message()
Borrow a new serialized message.
Definition: subscription.hpp:213
void return_message(std::shared_ptr< void > &message)
Return the loaned message.
Definition: subscription.hpp:233
Definition: allocator_common.hpp:24
rcl_ret_t rcl_subscription_init(rcl_subscription_t *subscription, const rcl_node_t *node, const rosidl_message_type_support_t *type_support, const char *topic_name, const rcl_subscription_options_t *options)
void handle_message(std::shared_ptr< void > &message, const rmw_message_info_t &message_info)
Check if we need to handle the message, and execute the callback if we do.
Definition: subscription.hpp:218
std::string expand_topic_or_service_name(const std::string &name, const std::string &node_name, const std::string &namespace_, bool is_service=false)
Expand a topic or service name and throw if it is not valid.
allocator::Deleter< MessageAlloc, CallbackMessageT > MessageDeleter
Definition: subscription.hpp:155
#define RCL_RET_OK
Subscription implementation, templated on the type of message this subscription receives.
Definition: subscription.hpp:148
void handle_intra_process_message(rcl_interfaces::msg::IntraProcessMessage &ipm, const rmw_message_info_t &message_info)
Definition: subscription.hpp:244
typename MessageAllocTraits::allocator_type MessageAlloc
Definition: subscription.hpp:154
void throw_from_rcl_error(rcl_ret_t ret, const std::string &prefix="", const rcl_error_state_t *error_state=nullptr, void(*reset_error)()=rcl_reset_error)
Throw a C++ std::exception which was created based on an rcl error.
std::shared_ptr< rcl_subscription_t > intra_process_subscription_handle_
Definition: subscription.hpp:133
typename std::allocator_traits< Alloc >::template rebind_traits< T > AllocRebind
Definition: allocator_common.hpp:30
const rosidl_message_type_support_t * get_intra_process_message_msg_type_support()
Definition: any_subscription_callback.hpp:34
#define RCLCPP_SMART_PTR_DEFINITIONS(...)
Definition: macros.hpp:36
std::shared_ptr< void > create_message()
Borrow a new message.
Definition: subscription.hpp:204
allocator::AllocRebind< CallbackMessageT, Alloc > MessageAllocTraits
Definition: subscription.hpp:153
void return_serialized_message(std::shared_ptr< rcl_serialized_message_t > &message)
Return the message borrowed in create_serialized_message.
Definition: subscription.hpp:239
T static_pointer_cast(T... args)
#define RCLCPP_SMART_PTR_DEFINITIONS_NOT_COPYABLE(...)
Definition: macros.hpp:51
std::shared_ptr< rcl_node_t > node_handle_
Definition: subscription.hpp:135
const std::shared_ptr< rcl_subscription_t > get_intra_process_subscription_handle() const
Implemenation detail.
Definition: subscription.hpp:309
#define RCLCPP_PUBLIC
Definition: visibility_control.hpp:50
rmw_gid_t publisher_gid
Pure virtual interface class for the NodeTopics part of the Node API.
Definition: node_topics_interface.hpp:38
Definition: subscription.hpp:51
T c_str(T... args)
#define RCL_RET_TOPIC_NAME_INVALID
const char * rcl_node_get_name(const rcl_node_t *node)
const char * rcl_node_get_namespace(const rcl_node_t *node)
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
void set_message_memory_strategy(typename message_memory_strategy::MessageMemoryStrategy< CallbackMessageT, Alloc >::SharedPtr message_memory_strategy)
Support dynamically setting the message memory strategy.
Definition: subscription.hpp:197
void setup_intra_process(uint64_t intra_process_subscription_id, GetMessageCallbackType get_message_callback, MatchesAnyPublishersCallbackType matches_any_publisher_callback, const rcl_subscription_options_t &intra_process_options)
Implemenation detail.
Definition: subscription.hpp:275