##

Node.hh
Go to the documentation of this file.
/*
* Copyright (C) 2012 Open Source Robotics Foundation
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
#ifndef GAZEBO_TRANSPORT_NODE_HH_
#define GAZEBO_TRANSPORT_NODE_HH_
#include <tbb/task.h>
#include <boost/bind.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <map>
#include <list>
#include <string>
#include <vector>
#include "gazebo/transport/TransportTypes.hh"
#include "gazebo/transport/TopicManager.hh"
#include "gazebo/util/system.hh"
namespace gazebo
{
namespace transport
{
class GZ_TRANSPORT_VISIBLE PublishTask : public tbb::task
{
public: PublishTask(transport::PublisherPtr _pub,
const google::protobuf::Message &_message)
: pub(_pub)
{
this->msg = _message.New();
this->msg->CopyFrom(_message);
}
public: tbb::task *execute()
{
this->pub->WaitForConnection();
this->pub->Publish(*this->msg, true);
this->pub->SendMessage();
delete this->msg;
this->pub.reset();
return NULL;
}
private: transport::PublisherPtr pub;
private: google::protobuf::Message *msg;
};
class GZ_TRANSPORT_VISIBLE Node :
public boost::enable_shared_from_this<Node>
{
public: Node();
public: virtual ~Node();
public: void Init(const std::string &_space ="");
public: bool TryInit(
const common::Time &_maxWait = common::Time(1, 0));
public: bool IsInitialized() const;
public: void Fini();
public: std::string GetTopicNamespace() const;
public: std::string DecodeTopicName(const std::string &_topic);
public: std::string EncodeTopicName(const std::string &_topic);
public: unsigned int GetId() const;
public: void ProcessPublishers();
public: void ProcessIncoming();
public: bool HasLatchedSubscriber(const std::string &_topic) const;
public: template<typename M>
void Publish(const std::string &_topic,
const google::protobuf::Message &_message)
{
transport::PublisherPtr pub = this->Advertise<M>(_topic);
PublishTask *task = new(tbb::task::allocate_root())
PublishTask(pub, _message);
tbb::task::enqueue(*task);
return;
}
public: template<typename M>
transport::PublisherPtr Advertise(const std::string &_topic,
unsigned int _queueLimit = 1000,
double _hzRate = 0)
{
std::string decodedTopic = this->DecodeTopicName(_topic);
PublisherPtr publisher =
transport::TopicManager::Instance()->Advertise<M>(
decodedTopic, _queueLimit, _hzRate);
boost::mutex::scoped_lock lock(this->publisherMutex);
publisher->SetNode(shared_from_this());
this->publishers.push_back(publisher);
return publisher;
}
public: template<typename M, typename T>
SubscriberPtr Subscribe(const std::string &_topic,
void(T::*_fp)(const boost::shared_ptr<M const> &), T *_obj,
bool _latching = false)
{
SubscribeOptions ops;
std::string decodedTopic = this->DecodeTopicName(_topic);
ops.template Init<M>(decodedTopic, shared_from_this(), _latching);
{
boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
this->callbacks[decodedTopic].push_back(CallbackHelperPtr(
new CallbackHelperT<M>(boost::bind(_fp, _obj, _1), _latching)));
}
SubscriberPtr result =
transport::TopicManager::Instance()->Subscribe(ops);
result->SetCallbackId(this->callbacks[decodedTopic].back()->GetId());
return result;
}
public: template<typename M>
SubscriberPtr Subscribe(const std::string &_topic,
void(*_fp)(const boost::shared_ptr<M const> &),
bool _latching = false)
{
SubscribeOptions ops;
std::string decodedTopic = this->DecodeTopicName(_topic);
ops.template Init<M>(decodedTopic, shared_from_this(), _latching);
{
boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
this->callbacks[decodedTopic].push_back(
CallbackHelperPtr(new CallbackHelperT<M>(_fp, _latching)));
}
SubscriberPtr result =
transport::TopicManager::Instance()->Subscribe(ops);
result->SetCallbackId(this->callbacks[decodedTopic].back()->GetId());
return result;
}
template<typename T>
SubscriberPtr Subscribe(const std::string &_topic,
void(T::*_fp)(const std::string &), T *_obj,
bool _latching = false)
{
SubscribeOptions ops;
std::string decodedTopic = this->DecodeTopicName(_topic);
ops.Init(decodedTopic, shared_from_this(), _latching);
{
boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
this->callbacks[decodedTopic].push_back(CallbackHelperPtr(
new RawCallbackHelper(boost::bind(_fp, _obj, _1))));
}
SubscriberPtr result =
transport::TopicManager::Instance()->Subscribe(ops);
result->SetCallbackId(this->callbacks[decodedTopic].back()->GetId());
return result;
}
SubscriberPtr Subscribe(const std::string &_topic,
void(*_fp)(const std::string &), bool _latching = false)
{
SubscribeOptions ops;
std::string decodedTopic = this->DecodeTopicName(_topic);
ops.Init(decodedTopic, shared_from_this(), _latching);
{
boost::recursive_mutex::scoped_lock lock(this->incomingMutex);
this->callbacks[decodedTopic].push_back(
CallbackHelperPtr(new RawCallbackHelper(_fp)));
}
SubscriberPtr result =
transport::TopicManager::Instance()->Subscribe(ops);
result->SetCallbackId(this->callbacks[decodedTopic].back()->GetId());
return result;
}
public: bool HandleData(const std::string &_topic,
const std::string &_msg);
public: bool HandleMessage(const std::string &_topic, MessagePtr _msg);
public: void InsertLatchedMsg(const std::string &_topic,
const std::string &_msg);
public: void InsertLatchedMsg(const std::string &_topic,
MessagePtr _msg);
public: std::string GetMsgType(const std::string &_topic) const;
public: void RemoveCallback(const std::string &_topic, unsigned int _id);
private: bool PrivateInit(const std::string &_space,
const common::Time &_maxWait,
const bool _fallbackToDefault);
private: std::string topicNamespace;
private: std::vector<PublisherPtr> publishers;
private: std::vector<PublisherPtr>::iterator publishersIter;
private: static unsigned int idCounter;
private: unsigned int id;
private: typedef std::list<CallbackHelperPtr> Callback_L;
private: typedef std::map<std::string, Callback_L> Callback_M;
private: Callback_M callbacks;
private: std::map<std::string, std::list<std::string> > incomingMsgs;
private: std::map<std::string, std::list<MessagePtr> > incomingMsgsLocal;
private: boost::mutex publisherMutex;
private: boost::mutex publisherDeleteMutex;
private: boost::recursive_mutex incomingMutex;
private: boost::recursive_mutex processIncomingMutex;
private: bool initialized;
};
}
}
#endif
gazebo::transport::Node::ProcessIncoming
void ProcessIncoming()
Process incoming messages.
gazebo::transport::Node::Subscribe
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:230
gazebo::transport::Node::Node
Node()
Constructor.
gazebo::transport::SubscribeOptions
Options for a subscription.
Definition: SubscribeOptions.hh:35
gazebo::transport::Node::~Node
virtual ~Node()
Destructor.
gazebo
Forward declarations for the common classes.
Definition: Animation.hh:26
gazebo::transport::MessagePtr
boost::shared_ptr< google::protobuf::Message > MessagePtr
Definition: TransportTypes.hh:45
TopicManager.hh
gazebo::common::Time
A Time class, can be used to hold wall- or sim-time. stored as sec and nano-sec.
Definition: Time.hh:47
NULL
#define NULL
Definition: CommonTypes.hh:31
gazebo::transport::PublisherPtr
boost::shared_ptr< Publisher > PublisherPtr
Definition: TransportTypes.hh:49
gazebo::transport::Node::HandleMessage
bool HandleMessage(const std::string &_topic, MessagePtr _msg)
Handle incoming msg.
gazebo::transport::Node::GetMsgType
std::string GetMsgType(const std::string &_topic) const
Get the message type for a topic.
TransportTypes.hh
Forward declarations for transport.
gazebo::transport::Node::HandleData
bool HandleData(const std::string &_topic, const std::string &_msg)
Handle incoming data.
system.hh
gazebo::transport::Node::EncodeTopicName
std::string EncodeTopicName(const std::string &_topic)
Encode a topic name.
gazebo::transport::Node::InsertLatchedMsg
void InsertLatchedMsg(const std::string &_topic, const std::string &_msg)
Add a latched message to the node for publication.
gazebo::transport::Node::DecodeTopicName
std::string DecodeTopicName(const std::string &_topic)
Decode a topic name.
gazebo::transport::SubscriberPtr
boost::shared_ptr< Subscriber > SubscriberPtr
Definition: TransportTypes.hh:53
gazebo::transport::Node::Publish
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:158
gazebo::transport::Node::Subscribe
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:289
gazebo::transport::Node
A node can advertise and subscribe topics, publish on advertised topics and listen to subscribed topi...
Definition: Node.hh:78
gazebo::transport::Node::GetId
unsigned int GetId() const
Get the unique ID of the node.
gazebo::msgs::Init
GAZEBO_VISIBLE void Init(google::protobuf::Message &_message, const std::string &_id="")
Initialize a message.
gazebo::transport::Node::TryInit
bool TryInit(const common::Time &_maxWait=common::Time(1, 0))
Try to initialize the node to use the global namespace, and specify the maximum wait time.
gazebo::transport::CallbackHelperT
Callback helper Template.
Definition: CallbackHelper.hh:111
SingletonT< TopicManager >::Instance
static TopicManager * Instance()
Get an instance of the singleton.
Definition: SingletonT.hh:36
gazebo::transport::Node::RemoveCallback
void RemoveCallback(const std::string &_topic, unsigned int _id)
gazebo::transport::Node::Init
void Init(const std::string &_space="")
Init the node.
gazebo::transport::SubscribeOptions::Init
void Init(const std::string &_topic, NodePtr _node, bool _latching)
Initialize the options.
Definition: SubscribeOptions.hh:48
transport
transport
Definition: ConnectionManager.hh:35
gazebo::transport::Node::ProcessPublishers
void ProcessPublishers()
Process all publishers, which has each publisher send it's most recent message over the wire.
gazebo::transport::Node::Advertise
transport::PublisherPtr Advertise(const std::string &_topic, unsigned int _queueLimit=1000, double _hzRate=0)
Adverise a topic.
Definition: Node.hh:177
gazebo::transport::Node::Subscribe
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:260
gazebo::transport::CallbackHelperPtr
boost::shared_ptr< CallbackHelper > CallbackHelperPtr
boost shared pointer to transport::CallbackHelper
Definition: CallbackHelper.hh:105
gazebo::transport::Node::GetTopicNamespace
std::string GetTopicNamespace() const
Get the topic namespace for this node.
gazebo::transport::Node::HasLatchedSubscriber
bool HasLatchedSubscriber(const std::string &_topic) const
Return true if a subscriber on a specific topic is latched.
gazebo::transport::Node::Subscribe
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:201
gazebo::transport::Node::Fini
void Fini()
Finalize the node.
gazebo::transport::Node::IsInitialized
bool IsInitialized() const
Check if this Node has been initialized.
gazebo::transport::RawCallbackHelper
Used to connect publishers to subscribers, where the subscriber wants the raw data from the publisher...
Definition: CallbackHelper.hh:177