All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
TopicManager.hh
Go to the documentation of this file.
1 /*
2  * Copyright 2012 Nate Koenig
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 _TOPICMANAGER_HH_
18 #define _TOPICMANAGER_HH_
19 
20 #include <boost/bind.hpp>
21 #include <map>
22 #include <list>
23 #include <string>
24 #include <vector>
25 
26 #include "common/Exception.hh"
27 #include "msgs/msgs.hh"
28 #include "common/SingletonT.hh"
29 
35 #include "transport/Publisher.hh"
36 #include "transport/Publication.hh"
37 #include "transport/Subscriber.hh"
38 
39 namespace gazebo
40 {
41  namespace transport
42  {
45 
48  class TopicManager : public SingletonT<TopicManager>
49  {
50  private: TopicManager();
51  private: virtual ~TopicManager();
52 
54  public: void Init();
55 
57  public: void Fini();
58 
62  public: PublicationPtr FindPublication(const std::string &_topic);
63 
66  public: void AddNode(NodePtr _node);
67 
70  public: void RemoveNode(unsigned int _id);
71 
75  public: void ProcessNodes(bool _onlyOut = false);
76 
80  public: bool IsAdvertised(const std::string &_topic);
81 
85  public: SubscriberPtr Subscribe(const SubscribeOptions &_options);
86 
91  public: void Unsubscribe(const std::string &_topic, const NodePtr &_sub);
92 
100  public: template<typename M>
101  PublisherPtr Advertise(const std::string &_topic,
102  unsigned int _queueLimit,
103  bool _latch)
104  {
105  google::protobuf::Message *msg = NULL;
106  M msgtype;
107  msg = dynamic_cast<google::protobuf::Message *>(&msgtype);
108  if (!msg)
109  gzthrow("Advertise requires a google protobuf type");
110 
111  this->UpdatePublications(_topic, msg->GetTypeName());
112 
113  PublisherPtr pub = PublisherPtr(new Publisher(_topic,
114  msg->GetTypeName(), _queueLimit, _latch));
115 
116  std::string msgTypename;
117  PublicationPtr publication;
118 
119  // Connect all local subscription to the publisher
120  for (int i = 0; i < 2; i ++)
121  {
122  std::string t;
123  if (i == 0)
124  {
125  t = _topic;
126  msgTypename = msg->GetTypeName();
127  }
128  else
129  {
130  t = _topic + "/__dbg";
131  msgs::GzString tmp;
132  msgTypename = tmp.GetTypeName();
133  }
134 
135  publication = this->FindPublication(t);
136  publication->AddPublisher(pub);
137  if (!publication->GetLocallyAdvertised())
138  {
139  ConnectionManager::Instance()->Advertise(t, msgTypename);
140  }
141 
142  publication->SetLocallyAdvertised(true);
143  pub->SetPublication(publication, i);
144 
145  SubNodeMap::iterator iter2;
146  SubNodeMap::iterator st_end2 = this->subscribedNodes.end();
147  for (iter2 = this->subscribedNodes.begin();
148  iter2 != st_end2; iter2++)
149  {
150  if (iter2->first == t)
151  {
152  std::list<NodePtr>::iterator liter;
153  std::list<NodePtr>::iterator l_end = iter2->second.end();
154  for (liter = iter2->second.begin();
155  liter != l_end; liter++)
156  {
157  publication->AddSubscription(*liter);
158  }
159  }
160  }
161  }
162 
163  return pub;
164  }
165 
168  public: void Unadvertise(const std::string &_topic);
169 
175  public: void Publish(const std::string &_topic,
176  const google::protobuf::Message &_message,
177  const boost::function<void()> &_cb = NULL);
178 
182  public: void ConnectPubToSub(const std::string &_topic,
183  const SubscriptionTransportPtr &_sublink);
184 
187  public: void ConnectSubToPub(const msgs::Publish &_pub);
188 
193  public: void DisconnectPubFromSub(const std::string &_topic,
194  const std::string &_host,
195  unsigned int _port);
196 
201  public: void DisconnectSubFromPub(const std::string &_topic,
202  const std::string &_host,
203  unsigned int _port);
204 
207  public: void ConnectSubscribers(const std::string &_topic);
208 
214  public: PublicationPtr UpdatePublications(const std::string &_topic,
215  const std::string &_msgType);
216 
219  public: void RegisterTopicNamespace(const std::string &_name);
220 
223  public: void GetTopicNamespaces(std::list<std::string> &_namespaces);
224 
226  public: void ClearBuffers();
227 
230  public: void PauseIncoming(bool _pause);
231 
233  typedef std::map<std::string, std::list<NodePtr> > SubNodeMap;
234 
235  private: typedef std::map<std::string, PublicationPtr> PublicationPtr_M;
236  private: PublicationPtr_M advertisedTopics;
237  private: PublicationPtr_M::iterator advertisedTopicsEnd;
238  private: SubNodeMap subscribedNodes;
239  private: std::vector<NodePtr> nodes;
240 
241  private: boost::recursive_mutex *nodeMutex;
242 
243  private: bool pauseIncoming;
244 
245  // Singleton implementation
246  private: friend class SingletonT<TopicManager>;
247  };
249  }
250 }
251 #endif
252