TopicManager.hh
Go to the documentation of this file.
1 /*
2  * Copyright (C) 2012 Open Source Robotics Foundation
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  *
16 */
17 #ifndef GAZEBO_TRANSPORT_TOPICMANAGER_HH_
18 #define GAZEBO_TRANSPORT_TOPICMANAGER_HH_
19 
20 #include <boost/bind.hpp>
21 #include <boost/function.hpp>
22 #include <map>
23 #include <list>
24 #include <string>
25 #include <vector>
26 #include <boost/unordered/unordered_set.hpp>
27 
28 #include "gazebo/common/Assert.hh"
30 #include "gazebo/msgs/msgs.hh"
32 
41 #include "gazebo/util/system.hh"
42 
43 namespace gazebo
44 {
45  namespace transport
46  {
49 
52  class GZ_TRANSPORT_VISIBLE TopicManager : public SingletonT<TopicManager>
53  {
54  private: TopicManager();
55  private: virtual ~TopicManager();
56 
58  public: void Init();
59 
61  public: void Fini();
62 
66  public: PublicationPtr FindPublication(const std::string &_topic);
67 
70  public: void AddNode(NodePtr _node);
71 
74  public: void RemoveNode(unsigned int _id);
75 
79  public: void ProcessNodes(bool _onlyOut = false);
80 
84  public: SubscriberPtr Subscribe(const SubscribeOptions &_options);
85 
90  public: void Unsubscribe(const std::string &_topic, const NodePtr &_sub);
91 
99  public: PublisherPtr Advertise(const std::string &_topic,
100  const std::string &_msgTypeName,
101  unsigned int _queueLimit,
102  double _hzRate)
103  {
104  this->UpdatePublications(_topic, _msgTypeName);
105 
106  PublisherPtr pub = PublisherPtr(new Publisher(_topic,
107  _msgTypeName, _queueLimit, _hzRate));
108 
109  // Connect all local subscription to the publisher
110  PublicationPtr publication = this->FindPublication(_topic);
111  GZ_ASSERT(publication != nullptr,
112  "FindPublication returned nullptr");
113 
114  publication->AddPublisher(pub);
115  if (!publication->GetLocallyAdvertised())
116  {
117  ConnectionManager::Instance()->Advertise(_topic,
118  _msgTypeName);
119  }
120 
121  publication->SetLocallyAdvertised(true);
122  pub->SetPublication(publication);
123 
124  for (auto &iter2 : this->subscribedNodes)
125  {
126  if (iter2.first == _topic)
127  {
128  for (const auto liter : iter2.second)
129  {
130  publication->AddSubscription(liter);
131  }
132  }
133  }
134 
135  return pub;
136  }
137 
145  public: template<typename M>
146  PublisherPtr Advertise(const std::string &_topic,
147  unsigned int _queueLimit,
148  double _hzRate)
149  {
150  google::protobuf::Message *msg = nullptr;
151  M msgtype;
152  msg = dynamic_cast<google::protobuf::Message *>(&msgtype);
153  if (!msg)
154  gzthrow("Advertise requires a google protobuf type");
155 
156  return this->Advertise(_topic, msg->GetTypeName(), _queueLimit,
157  _hzRate);
158  }
159 
162  public: void Unadvertise(const std::string &_topic);
163 
166  public: void Unadvertise(PublisherPtr _pub);
167 
172  public: void Unadvertise(const std::string &_topic, const uint32_t _id);
173 
180  public: void Publish(const std::string &_topic, MessagePtr _message,
181  boost::function<void(uint32_t)> _cb, uint32_t _id);
182 
186  public: void ConnectPubToSub(const std::string &_topic,
187  const SubscriptionTransportPtr _sublink);
188 
191  public: void ConnectSubToPub(const msgs::Publish &_pub);
192 
197  public: void DisconnectPubFromSub(const std::string &_topic,
198  const std::string &_host,
199  unsigned int _port);
200 
205  public: void DisconnectSubFromPub(const std::string &_topic,
206  const std::string &_host,
207  unsigned int _port);
208 
211  public: void ConnectSubscribers(const std::string &_topic);
212 
218  public: PublicationPtr UpdatePublications(const std::string &_topic,
219  const std::string &_msgType);
220 
223  public: void RegisterTopicNamespace(const std::string &_name);
224 
227  public: void GetTopicNamespaces(std::list<std::string> &_namespaces);
228 
230  public: void ClearBuffers();
231 
234  public: void PauseIncoming(bool _pause);
235 
238  public: void AddNodeToProcess(NodePtr _ptr);
239 
241  typedef std::map<std::string, std::list<NodePtr> > SubNodeMap;
242 
243  private: typedef std::map<std::string, PublicationPtr> PublicationPtr_M;
244  private: PublicationPtr_M advertisedTopics;
245  private: PublicationPtr_M::iterator advertisedTopicsEnd;
246  private: SubNodeMap subscribedNodes;
247  private: std::vector<NodePtr> nodes;
248 
250  private: boost::unordered_set<NodePtr> nodesToProcess;
251 
252  private: boost::recursive_mutex nodeMutex;
253 
255  private: boost::mutex subscriberMutex;
256 
258  private: boost::mutex processNodesMutex;
259 
260  private: bool pauseIncoming;
261 
262  // Singleton implementation
263  private: friend class SingletonT<TopicManager>;
264  };
266  }
267 }
268 #endif
Options for a subscription.
Definition: SubscribeOptions.hh:35
#define GZ_ASSERT(_expr, _msg)
This macro define the standard way of launching an exception inside gazebo.
Definition: Assert.hh:24
Forward declarations for the common classes.
Definition: Animation.hh:26
boost::shared_ptr< google::protobuf::Message > MessagePtr
Definition: TransportTypes.hh:45
#define gzthrow(msg)
This macro logs an error to the throw stream and throws an exception that contains the file name and ...
Definition: Exception.hh:39
Forward declarations for transport.
boost::shared_ptr< Publisher > PublisherPtr
Definition: TransportTypes.hh:49
boost::shared_ptr< Subscriber > SubscriberPtr
Definition: TransportTypes.hh:53
Singleton template class.
Definition: SingletonT.hh:33
boost::shared_ptr< Node > NodePtr
Definition: TransportTypes.hh:57
boost::shared_ptr< Publication > PublicationPtr
Definition: TransportTypes.hh:61
boost::shared_ptr< SubscriptionTransport > SubscriptionTransportPtr
Definition: TransportTypes.hh:69
Manages topics and their subscriptions.
Definition: TopicManager.hh:52
A publisher of messages on a topic.
Definition: Publisher.hh:44
PublisherPtr Advertise(const std::string &_topic, const std::string &_msgTypeName, unsigned int _queueLimit, double _hzRate)
Advertise on a topic.
Definition: TopicManager.hh:99
GAZEBO_VISIBLE void Init(google::protobuf::Message &_message, const std::string &_id="")
Initialize a message.
PublisherPtr Advertise(const std::string &_topic, unsigned int _queueLimit, double _hzRate)
Advertise on a topic.
Definition: TopicManager.hh:146
std::map< std::string, std::list< NodePtr > > SubNodeMap
A map of string->list of Node pointers.
Definition: TopicManager.hh:241
static ConnectionManager * Instance()
Get an instance of the singleton.
Definition: SingletonT.hh:36