22 #include <boost/enable_shared_from_this.hpp>
43 const google::protobuf::Message &_message)
46 this->msg = _message.New();
47 this->msg->CopyFrom(_message);
55 this->pub->
Publish(*this->msg,
true);
65 private: google::protobuf::Message *msg;
75 class Node :
public boost::enable_shared_from_this<Node>
81 public:
virtual ~Node();
87 public:
void Init(
const std::string &_space =
"");
108 public:
unsigned int GetId()
const;
129 public:
template<
typename M>
131 const google::protobuf::Message &_message)
134 PublishTask *task =
new(tbb::task::allocate_root())
137 tbb::task::enqueue(*task);
148 public:
template<
typename M>
150 unsigned int _queueLimit = 1000,
156 decodedTopic, _queueLimit, _hzRate);
158 boost::recursive_mutex::scoped_lock lock(this->publisherMutex);
159 this->publishers.push_back(publisher);
160 this->publishersEnd = this->publishers.end();
172 public:
template<
typename M,
typename T>
174 void(T::*_fp)(
const boost::shared_ptr<M const> &), T *_obj,
175 bool _latching =
false)
179 ops.template Init<M>(decodedTopic, shared_from_this(), _latching);
182 boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
201 public:
template<
typename M>
203 void(*_fp)(
const boost::shared_ptr<M const> &),
204 bool _latching =
false)
208 ops.template Init<M>(decodedTopic, shared_from_this(), _latching);
211 boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
212 this->callbacks[decodedTopic].push_back(
233 void(T::*_fp)(
const std::string &), T *_obj,
234 bool _latching =
false)
238 ops.
Init(decodedTopic, shared_from_this(), _latching);
241 boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
262 void(*_fp)(
const std::string &),
bool _latching =
false)
266 ops.
Init(decodedTopic, shared_from_this(), _latching);
269 boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
270 this->callbacks[decodedTopic].push_back(
286 public:
bool HandleData(
const std::string &_topic,
287 const std::string &_msg);
302 const std::string &_msg);
316 public: std::string
GetMsgType(
const std::string &_topic)
const;
323 public:
void RemoveCallback(
const std::string &_topic,
unsigned int _id);
325 private: std::string topicNamespace;
326 private: std::vector<PublisherPtr> publishers;
327 private: std::vector<PublisherPtr>::iterator publishersIter;
328 private: std::vector<PublisherPtr>::iterator publishersEnd;
329 private:
static unsigned int idCounter;
330 private:
unsigned int id;
332 private:
typedef std::list<CallbackHelperPtr> Callback_L;
333 private:
typedef std::map<std::string, Callback_L> Callback_M;
334 private: Callback_M callbacks;
335 private: std::map<std::string, std::list<std::string> > incomingMsgs;
338 private: std::map<std::string, std::list<MessagePtr> > incomingMsgsLocal;
340 private: boost::recursive_mutex publisherMutex;
341 private: boost::recursive_mutex incomingMutex;
343 private:
bool initialized;