rclcpp  master
C++ ROS Client Library API
publisher.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__PUBLISHER_HPP_
16 #define RCLCPP__PUBLISHER_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/publisher.h"
29 
30 #include "rcl_interfaces/msg/intra_process_message.hpp"
31 
34 #include "rclcpp/macros.hpp"
38 
39 namespace rclcpp
40 {
41 
42 // Forward declaration is used for friend statement.
43 namespace node_interfaces
44 {
45 class NodeTopicsInterface;
46 }
47 
49 {
50  friend ::rclcpp::node_interfaces::NodeTopicsInterface;
51 
52 public:
54 
55 
56 
67  const std::string & topic,
68  const rosidl_message_type_support_t & type_support,
69  const rcl_publisher_options_t & publisher_options);
70 
72  virtual ~PublisherBase();
73 
75 
77  const char *
78  get_topic_name() const;
79 
81 
83  size_t
84  get_queue_size() const;
85 
87 
89  const rmw_gid_t &
90  get_gid() const;
91 
93 
95  const rmw_gid_t &
96  get_intra_process_gid() const;
97 
99 
103 
105 
107  const rcl_publisher_t *
108  get_publisher_handle() const;
109 
111 
117  bool
118  operator==(const rmw_gid_t & gid) const;
119 
121 
127  bool
128  operator==(const rmw_gid_t * gid) const;
129 
131 
134  void
136  uint64_t intra_process_publisher_id,
137  StoreMessageCallbackT callback,
138  const rcl_publisher_options_t & intra_process_options);
139 
140 protected:
142 
145 
148 
151 };
152 
154 template<typename MessageT, typename Alloc = std::allocator<void>>
155 class Publisher : public PublisherBase
156 {
157 public:
159  using MessageAlloc = typename MessageAllocTraits::allocator_type;
162 
164 
166  rclcpp::node_interfaces::NodeBaseInterface * node_base,
167  const std::string & topic,
168  const rcl_publisher_options_t & publisher_options,
169  const std::shared_ptr<MessageAlloc> & allocator)
170  : PublisherBase(
171  node_base,
172  topic,
173  *rosidl_typesupport_cpp::get_message_type_support_handle<MessageT>(),
174  publisher_options),
175  message_allocator_(allocator)
176  {
178  }
179 
180  virtual ~Publisher()
181  {}
182 
184 
188  virtual void
190  {
191  this->do_inter_process_publish(msg.get());
193  // Take the pointer from the unique_msg, release it and pass as a void *
194  // to the ipm. The ipm should then capture it again as a unique_ptr of
195  // the correct type.
196  // TODO(wjwwood):
197  // investigate how to transfer the custom deleter (if there is one)
198  // from the incoming unique_ptr through to the ipm's unique_ptr.
199  // See: http://stackoverflow.com/questions/11002641/dynamic-casting-for-unique-ptr
200  MessageT * msg_ptr = msg.get();
201  msg.release();
202  uint64_t message_seq =
203  store_intra_process_message_(intra_process_publisher_id_, msg_ptr, typeid(MessageT));
204  rcl_interfaces::msg::IntraProcessMessage ipm;
205  ipm.publisher_id = intra_process_publisher_id_;
206  ipm.message_sequence = message_seq;
207  auto status = rcl_publish(&intra_process_publisher_handle_, &ipm);
208  if (RCL_RET_OK != status) {
209  rclcpp::exceptions::throw_from_rcl_error(status, "failed to publish intra process message");
210  }
211  } else {
212  // Always destroy the message, even if we don't consume it, for consistency.
213  msg.reset();
214  }
215  }
216 
217  virtual void
219  {
220  // Avoid allocating when not using intra process.
222  // In this case we're not using intra process.
223  return this->do_inter_process_publish(msg.get());
224  }
225  // Otherwise we have to allocate memory in a unique_ptr and pass it along.
226  // TODO(wjwwood):
227  // The intra process manager should probably also be able to store
228  // shared_ptr's and do the "smart" thing based on other intra process
229  // subscriptions. For now call the other publish().
230  auto ptr = MessageAllocTraits::allocate(*message_allocator_.get(), 1);
231  MessageAllocTraits::construct(*message_allocator_.get(), ptr, *msg.get());
232  MessageUniquePtr unique_msg(ptr, message_deleter_);
233  return this->publish(unique_msg);
234  }
235 
236  virtual void
238  {
239  // Avoid allocating when not using intra process.
241  // In this case we're not using intra process.
242  return this->do_inter_process_publish(msg.get());
243  }
244  // Otherwise we have to allocate memory in a unique_ptr and pass it along.
245  // TODO(wjwwood):
246  // The intra process manager should probably also be able to store
247  // shared_ptr's and do the "smart" thing based on other intra process
248  // subscriptions. For now call the other publish().
249  auto ptr = MessageAllocTraits::allocate(*message_allocator_.get(), 1);
250  MessageAllocTraits::construct(*message_allocator_.get(), ptr, *msg.get());
251  MessageUniquePtr unique_msg(ptr, message_deleter_);
252  return this->publish(unique_msg);
253  }
254 
255  virtual void
256  publish(const MessageT & msg)
257  {
258  // Avoid allocating when not using intra process.
260  // In this case we're not using intra process.
261  return this->do_inter_process_publish(&msg);
262  }
263  // Otherwise we have to allocate memory in a unique_ptr and pass it along.
264  auto ptr = MessageAllocTraits::allocate(*message_allocator_.get(), 1);
265  MessageAllocTraits::construct(*message_allocator_.get(), ptr, msg);
266  MessageUniquePtr unique_msg(ptr, message_deleter_);
267  return this->publish(unique_msg);
268  }
269 
270  virtual void
271  publish(const MessageT * msg)
272  {
273  if (!msg) {
274  throw std::runtime_error("msg argument is nullptr");
275  }
276  return this->publish(*msg);
277  }
278 
279  void
280  publish(const rcl_serialized_message_t * serialized_msg)
281  {
283  // TODO(Karsten1987): support serialized message passed by intraprocess
284  throw std::runtime_error("storing serialized messages in intra process is not supported yet");
285  }
286  auto status = rcl_publish_serialized_message(&publisher_handle_, serialized_msg);
287  if (RCL_RET_OK != status) {
288  rclcpp::exceptions::throw_from_rcl_error(status, "failed to publish serialized message");
289  }
290  }
291 
293  {
294  return message_allocator_;
295  }
296 
297 protected:
298  void
299  do_inter_process_publish(const MessageT * msg)
300  {
301  auto status = rcl_publish(&publisher_handle_, msg);
302  if (RCL_RET_OK != status) {
303  rclcpp::exceptions::throw_from_rcl_error(status, "failed to publish message");
304  }
305  }
306 
308 
310 };
311 
312 } // namespace rclcpp
313 
314 #endif // RCLCPP__PUBLISHER_HPP_
void setup_intra_process(uint64_t intra_process_publisher_id, StoreMessageCallbackT callback, const rcl_publisher_options_t &intra_process_options)
Implementation utility function used to setup intra process publishing after creation.
virtual void publish(std::shared_ptr< const MessageT > msg)
Definition: publisher.hpp:237
virtual void publish(const std::shared_ptr< MessageT > &msg)
Definition: publisher.hpp:218
A publisher publishes messages of any type to a topic.
Definition: publisher.hpp:155
std::shared_ptr< MessageAlloc > message_allocator_
Definition: publisher.hpp:307
size_t get_queue_size() const
Get the queue size for this publisher.
typename MessageAllocTraits::allocator_type MessageAlloc
Definition: publisher.hpp:159
PublisherBase(rclcpp::node_interfaces::NodeBaseInterface *node_base, const std::string &topic, const rosidl_message_type_support_t &type_support, const rcl_publisher_options_t &publisher_options)
Default constructor.
std::shared_ptr< rcl_node_t > rcl_node_handle_
Definition: publisher.hpp:141
Definition: publisher.hpp:48
Definition: allocator_common.hpp:24
virtual ~Publisher()
Definition: publisher.hpp:180
std::shared_ptr< MessageAlloc > get_allocator() const
Definition: publisher.hpp:292
rmw_gid_t intra_process_rmw_gid_
Definition: publisher.hpp:150
StoreMessageCallbackT store_intra_process_message_
Definition: publisher.hpp:147
T release(T... args)
#define RCL_RET_OK
virtual void publish(const MessageT *msg)
Definition: publisher.hpp:271
void set_allocator_for_deleter(D *deleter, Alloc *alloc)
Definition: allocator_deleter.hpp:72
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.
allocator::Deleter< MessageAlloc, rcl_interfaces::msg::ParameterEvent > MessageDeleter
Definition: publisher.hpp:160
MessageDeleter message_deleter_
Definition: publisher.hpp:309
typename std::allocator_traits< Alloc >::template rebind_traits< T > AllocRebind
Definition: allocator_common.hpp:30
rcl_publisher_t rcl_get_zero_initialized_publisher(void)
#define RCLCPP_SMART_PTR_DEFINITIONS(...)
Definition: macros.hpp:36
virtual void publish(const MessageT &msg)
Definition: publisher.hpp:256
const char * get_topic_name() const
Get the topic that this publisher publishes on.
T reset(T... args)
uint64_t intra_process_publisher_id_
Definition: publisher.hpp:146
rcl_publisher_t intra_process_publisher_handle_
Definition: publisher.hpp:144
T get(T... args)
rcl_publisher_t * get_publisher_handle()
Get the rcl publisher handle.
rcl_ret_t rcl_publish_serialized_message(const rcl_publisher_t *publisher, const rcl_serialized_message_t *serialized_message)
#define RCLCPP_PUBLIC
Definition: visibility_control.hpp:50
rcl_publisher_t publisher_handle_
Definition: publisher.hpp:143
rmw_gid_t rmw_gid_
Definition: publisher.hpp:149
virtual void publish(std::unique_ptr< MessageT, MessageDeleter > &msg)
Send a message to the topic for this publisher.
Definition: publisher.hpp:189
bool operator==(const rmw_gid_t &gid) const
Compare this publisher to a gid.
Pure virtual interface class for the NodeBase part of the Node API.
Definition: node_base_interface.hpp:36
void do_inter_process_publish(const MessageT *msg)
Definition: publisher.hpp:299
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
allocator::AllocRebind< rcl_interfaces::msg::ParameterEvent, std::allocator< void > > MessageAllocTraits
Definition: publisher.hpp:158
const rmw_gid_t & get_intra_process_gid() const
Get the global identifier for this publisher used by intra-process communication. ...
rcl_ret_t rcl_publish(const rcl_publisher_t *publisher, const void *ros_message)
void publish(const rcl_serialized_message_t *serialized_msg)
Definition: publisher.hpp:280
const rmw_gid_t & get_gid() const
Get the global identifier for this publisher (used in rmw and by DDS).