18 #ifndef GAZEBO_TRANSPORT_NODE_HH_ 19 #define GAZEBO_TRANSPORT_NODE_HH_ 24 #include <boost/bind.hpp> 25 #include <boost/enable_shared_from_this.hpp> 41 class GZ_TRANSPORT_VISIBLE PublishTask :
public tbb::task
47 const google::protobuf::Message &_message)
50 this->msg = _message.New();
51 this->msg->CopyFrom(_message);
56 public: tbb::task *execute()
58 this->pub->WaitForConnection();
59 this->pub->Publish(*this->msg,
true);
60 this->pub->SendMessage();
70 private: google::protobuf::Message *msg;
80 class GZ_TRANSPORT_VISIBLE
Node :
81 public boost::enable_shared_from_this<Node>
87 public:
virtual ~
Node();
97 public:
void Init(
const std::string &_space =
"");
110 public:
bool TryInit(
117 public:
bool IsInitialized()
const;
124 public: std::string GetTopicNamespace()
const;
129 public: std::string DecodeTopicName(
const std::string &_topic);
134 public: std::string EncodeTopicName(
const std::string &_topic);
138 public:
unsigned int GetId()
const;
142 public:
void ProcessPublishers();
145 public:
void ProcessIncoming();
150 public:
bool HasLatchedSubscriber(
const std::string &_topic)
const;
159 public:
template<
typename M>
161 const google::protobuf::Message &_message)
164 PublishTask *task =
new(tbb::task::allocate_root())
165 PublishTask(pub, _message);
167 tbb::task::enqueue(*task);
178 public:
template<
typename M>
180 unsigned int _queueLimit = 1000,
183 std::string decodedTopic = this->DecodeTopicName(_topic);
186 decodedTopic, _queueLimit, _hzRate);
188 boost::mutex::scoped_lock lock(this->publisherMutex);
189 publisher->SetNode(shared_from_this());
190 this->publishers.push_back(publisher);
202 public:
template<
typename M,
typename T>
204 void(T::*_fp)(
const boost::shared_ptr<M const> &), T *_obj,
205 bool _latching =
false)
208 std::string decodedTopic = this->DecodeTopicName(_topic);
209 ops.template Init<M>(decodedTopic, shared_from_this(), _latching);
212 boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
220 result->SetCallbackId(this->callbacks[decodedTopic].back()->GetId());
231 public:
template<
typename M>
233 void(*_fp)(
const boost::shared_ptr<M const> &),
234 bool _latching =
false)
237 std::string decodedTopic = this->DecodeTopicName(_topic);
238 ops.template Init<M>(decodedTopic, shared_from_this(), _latching);
241 boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
242 this->callbacks[decodedTopic].push_back(
249 result->SetCallbackId(this->callbacks[decodedTopic].back()->GetId());
263 void(T::*_fp)(
const std::string &), T *_obj,
264 bool _latching =
false)
267 std::string decodedTopic = this->DecodeTopicName(_topic);
268 ops.
Init(decodedTopic, shared_from_this(), _latching);
271 boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
279 result->SetCallbackId(this->callbacks[decodedTopic].back()->GetId());
292 void(*_fp)(
const std::string &),
bool _latching =
false)
295 std::string decodedTopic = this->DecodeTopicName(_topic);
296 ops.
Init(decodedTopic, shared_from_this(), _latching);
299 boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
300 this->callbacks[decodedTopic].push_back(
307 result->SetCallbackId(this->callbacks[decodedTopic].back()->GetId());
316 public:
bool HandleData(
const std::string &_topic,
317 const std::string &_msg);
323 public:
bool HandleMessage(
const std::string &_topic,
MessagePtr _msg);
331 public:
void InsertLatchedMsg(
const std::string &_topic,
332 const std::string &_msg);
340 public:
void InsertLatchedMsg(
const std::string &_topic,
346 public: std::string GetMsgType(
const std::string &_topic)
const;
353 public:
void RemoveCallback(
const std::string &_topic,
unsigned int _id);
366 private:
bool PrivateInit(
const std::string &_space,
368 const bool _fallbackToDefault);
370 private: std::string topicNamespace;
371 private: std::vector<PublisherPtr> publishers;
372 private: std::vector<PublisherPtr>::iterator publishersIter;
373 private:
static unsigned int idCounter;
374 private:
unsigned int id;
376 private:
typedef std::list<CallbackHelperPtr> Callback_L;
377 private:
typedef std::map<std::string, Callback_L> Callback_M;
378 private: Callback_M callbacks;
379 private: std::map<std::string, std::list<std::string> > incomingMsgs;
382 private: std::map<std::string, std::list<MessagePtr> > incomingMsgsLocal;
384 private: boost::mutex publisherMutex;
385 private: boost::mutex publisherDeleteMutex;
386 private: boost::recursive_mutex incomingMutex;
390 private: boost::recursive_mutex processIncomingMutex;
392 private:
bool initialized;
Options for a subscription.
Definition: SubscribeOptions.hh:35
transport::PublisherPtr Advertise(const std::string &_topic, unsigned int _queueLimit=1000, double _hzRate=0)
Adverise a topic.
Definition: Node.hh:179
Forward declarations for the common classes.
Definition: Animation.hh:26
SubscriberPtr Subscribe(const std::string &_topic, void(T::*_fp)(const std::string &), T *_obj, bool _latching=false)
Subscribe to a topic using a class method as the callback.
Definition: Node.hh:262
boost::shared_ptr< google::protobuf::Message > MessagePtr
Definition: TransportTypes.hh:45
Forward declarations for transport.
boost::shared_ptr< Publisher > PublisherPtr
Definition: TransportTypes.hh:49
boost::shared_ptr< Subscriber > SubscriberPtr
Definition: TransportTypes.hh:53
boost::shared_ptr< CallbackHelper > CallbackHelperPtr
boost shared pointer to transport::CallbackHelper
Definition: CallbackHelper.hh:105
SubscriberPtr Subscribe(const std::string &_topic, void(T::*_fp)(const boost::shared_ptr< M const > &), T *_obj, bool _latching=false)
Subscribe to a topic using a class method as the callback.
Definition: Node.hh:203
Callback helper Template.
Definition: CallbackHelper.hh:111
SubscriberPtr Subscribe(const std::string &_topic, void(*_fp)(const boost::shared_ptr< M const > &), bool _latching=false)
Subscribe to a topic using a bare function as the callback.
Definition: Node.hh:232
A node can advertise and subscribe topics, publish on advertised topics and listen to subscribed topi...
Definition: Node.hh:80
#define NULL
Definition: CommonTypes.hh:31
SubscriberPtr Subscribe(const std::string &_topic, void(*_fp)(const std::string &), bool _latching=false)
Subscribe to a topic using a bare function as the callback.
Definition: Node.hh:291
GAZEBO_VISIBLE void Init(google::protobuf::Message &_message, const std::string &_id="")
Initialize a message.
void Publish(const std::string &_topic, const google::protobuf::Message &_message)
A convenience function for a one-time publication of a message.
Definition: Node.hh:160
Used to connect publishers to subscribers, where the subscriber wants the raw data from the publisher...
Definition: CallbackHelper.hh:177
void Init(const std::string &_topic, NodePtr _node, bool _latching)
Initialize the options.
Definition: SubscribeOptions.hh:48
A Time class, can be used to hold wall- or sim-time.
Definition: Time.hh:44
static TopicManager * Instance()
Get an instance of the singleton.
Definition: SingletonT.hh:36