Connection.hh
Go to the documentation of this file.
1 /*
2  * Copyright (C) 2012-2015 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 _CONNECTION_HH_
18 #define _CONNECTION_HH_
19 
20 #include <tbb/task.h>
21 #include <google/protobuf/message.h>
22 
23 #include <boost/asio.hpp>
24 #include <boost/bind.hpp>
25 #include <boost/function.hpp>
26 #include <boost/thread.hpp>
27 #include <boost/tuple/tuple.hpp>
28 
29 #include <string>
30 #include <vector>
31 #include <iostream>
32 #include <iomanip>
33 #include <deque>
34 #include <utility>
35 
36 #include "gazebo/common/Event.hh"
37 #include "gazebo/common/Console.hh"
39 #include "gazebo/util/system.hh"
40 
41 #define HEADER_LENGTH 8
42 
43 namespace gazebo
44 {
45  namespace transport
46  {
47  extern bool is_stopped();
48 
49  class IOManager;
50  class Connection;
51  typedef boost::shared_ptr<Connection> ConnectionPtr;
52 
56  class GAZEBO_VISIBLE ConnectionReadTask : public tbb::task
57  {
62  public: ConnectionReadTask(
63  boost::function<void (const std::string &)> _func,
64  const std::string &_data)
65  {
66  this->func = _func;
67  this->data = _data;
68  }
69 
72  public: tbb::task *execute()
73  {
74  this->func(this->data);
75  return NULL;
76  }
77 
79  private: boost::function<void (const std::string &)> func;
80 
82  private: std::string data;
83  };
85 
101  public boost::enable_shared_from_this<Connection>
102  {
104  public: Connection();
105 
107  public: virtual ~Connection();
108 
113  public: bool Connect(const std::string &_host, unsigned int _port);
114 
116  typedef boost::function<void(const ConnectionPtr&)> AcceptCallback;
117 
122  public: void Listen(unsigned int _port, const AcceptCallback &_acceptCB);
123 
125  typedef boost::function<void(const std::string &_data)> ReadCallback;
126 
130  public: void StartRead(const ReadCallback &_cb);
131 
133  public: void StopRead();
134 
136  public: void Shutdown();
137 
140  public: bool IsOpen() const;
141 
143  private: void Close();
144 
146  public: void Cancel();
147 
151  public: bool Read(std::string &_data);
152 
160  public: void EnqueueMsg(const std::string &_buffer,
161  boost::function<void(uint32_t)> _cb, uint32_t _id,
162  bool _force = false);
163 
168  public: void EnqueueMsg(const std::string &_buffer, bool _force = false);
169 
172  public: std::string GetLocalURI() const;
173 
176  public: std::string GetRemoteURI() const;
177 
180  public: std::string GetLocalAddress() const;
181 
184  public: unsigned int GetLocalPort() const;
185 
188  public: std::string GetRemoteAddress() const;
189 
192  public: unsigned int GetRemotePort() const;
193 
196  public: std::string GetRemoteHostname() const;
197 
200  public: static std::string GetLocalHostname();
201 
204  public: template<typename Handler>
205  void AsyncRead(Handler _handler)
206  {
207  if (!this->IsOpen())
208  {
209  gzerr << "AsyncRead on a closed socket\n";
210  return;
211  }
212 
213  void (Connection::*f)(const boost::system::error_code &,
214  boost::tuple<Handler>) = &Connection::OnReadHeader<Handler>;
215 
216  this->inboundHeader.resize(HEADER_LENGTH);
217  boost::asio::async_read(*this->socket,
218  boost::asio::buffer(this->inboundHeader),
219  boost::bind(f, this,
220  boost::asio::placeholders::error,
221  boost::make_tuple(_handler)));
222  }
223 
231  private: template<typename Handler>
232  void OnReadHeader(const boost::system::error_code &_e,
233  boost::tuple<Handler> _handler)
234  {
235  if (_e)
236  {
237  if (_e.message() == "End of file")
238  this->isOpen = false;
239  }
240  else
241  {
242  std::size_t inboundData_size = 0;
243  std::string header(&this->inboundHeader[0],
244  this->inboundHeader.size());
245  this->inboundHeader.clear();
246 
247  inboundData_size = this->ParseHeader(header);
248 
249  if (inboundData_size > 0)
250  {
251  // Start the asynchronous call to receive data
252  this->inboundData.resize(inboundData_size);
253 
254  void (Connection::*f)(const boost::system::error_code &e,
255  boost::tuple<Handler>) =
256  &Connection::OnReadData<Handler>;
257 
258  boost::asio::async_read(*this->socket,
259  boost::asio::buffer(this->inboundData),
260  boost::bind(f, this,
261  boost::asio::placeholders::error,
262  _handler));
263  }
264  else
265  {
266  gzerr << "Header is empty\n";
267  boost::get<0>(_handler)("");
268  // This code tries to read the header again. We should
269  // never get here.
270  // this->inboundHeader.resize(HEADER_LENGTH);
271 
272  // void (Connection::*f)(const boost::system::error_code &,
273  // boost::tuple<Handler>) =
274  // &Connection::OnReadHeader<Handler>;
275 
276  // boost::asio::async_read(*this->socket,
277  // boost::asio::buffer(this->inboundHeader),
278  // boost::bind(f, this,
279  // boost::asio::placeholders::error, _handler));
280  }
281  }
282  }
283 
291  private: template<typename Handler>
292  void OnReadData(const boost::system::error_code &_e,
293  boost::tuple<Handler> _handler)
294  {
295  if (_e)
296  {
297  if (_e.message() == "End of file")
298  this->isOpen = false;
299  }
300 
301  // Inform caller that data has been received
302  std::string data(&this->inboundData[0],
303  this->inboundData.size());
304  this->inboundData.clear();
305 
306  if (data.empty())
307  gzerr << "OnReadData got empty data!!!\n";
308 
309  if (!_e && !transport::is_stopped())
310  {
311  ConnectionReadTask *task = new(tbb::task::allocate_root())
312  ConnectionReadTask(boost::get<0>(_handler), data);
313  tbb::task::enqueue(*task);
314 
315  // Non-tbb version:
316  // boost::get<0>(_handler)(data);
317  }
318  }
319 
323  public: event::ConnectionPtr ConnectToShutdown(boost::function<void()>
324  _subscriber)
325  { return this->shutdown.Connect(_subscriber); }
326 
330  public: void DisconnectShutdown(event::ConnectionPtr _subscriber)
331  {this->shutdown.Disconnect(_subscriber);}
332 
334  public: void ProcessWriteQueue(bool _blocking = false);
335 
338  public: unsigned int GetId() const;
339 
343  public: static bool ValidateIP(const std::string &_ip);
344 
348  public: std::string GetIPWhiteList() const;
349 
353  private: void OnWrite(const boost::system::error_code &_e);
354 
357  private: void OnAccept(const boost::system::error_code &_e);
358 
361  private: std::size_t ParseHeader(const std::string &_header);
362 
364  private: void ReadLoop(const ReadCallback &_cb);
365 
368  private: static boost::asio::ip::tcp::endpoint GetLocalEndpoint();
369 
372  private: boost::asio::ip::tcp::endpoint GetRemoteEndpoint() const;
373 
376  private: static std::string GetHostname(
377  boost::asio::ip::tcp::endpoint _ep);
378 
382  private: void OnConnect(const boost::system::error_code &_error,
383  boost::asio::ip::tcp::resolver::iterator _endPointIter);
384 
386  private: boost::asio::ip::tcp::socket *socket;
387 
389  private: boost::asio::ip::tcp::acceptor *acceptor;
390 
392  private: std::deque<std::string> writeQueue;
393 
396  private: std::deque<
397  std::pair<boost::function<void(uint32_t)>, uint32_t> > callbacks;
398 
400  private: boost::mutex connectMutex;
401 
403  private: boost::recursive_mutex writeMutex;
404 
406  private: boost::recursive_mutex readMutex;
407 
409  private: mutable boost::mutex socketMutex;
410 
412  private: boost::condition_variable connectCondition;
413 
415  private: AcceptCallback acceptCB;
416 
418  private: std::vector<char> inboundHeader;
419 
421  private: std::vector<char> inboundData;
422 
424  private: bool readQuit;
425 
427  private: unsigned int id;
428 
430  private: static unsigned int idCounter;
431 
433  private: ConnectionPtr acceptConn;
434 
437 
439  private: static IOManager *iomanager;
440 
442  private: unsigned int writeCount;
443 
445  private: std::string localURI;
446 
448  private: std::string localAddress;
449 
451  private: std::string remoteURI;
452 
454  private: std::string remoteAddress;
455 
457  private: bool connectError;
458 
460  private: std::string ipWhiteList;
461 
463  private: char *headerBuffer;
464 
466  private: bool dropMsgLogged;
467 
470  private: unsigned int callbackIndex;
471 
473  private: bool isOpen;
474  };
476  }
477 }
478 #endif
void DisconnectShutdown(event::ConnectionPtr _subscriber)
Unregister a function to be called when the connection is shut down.
Definition: Connection.hh:330
boost::shared_ptr< Connection > ConnectionPtr
Definition: CommonTypes.hh:144
bool is_stopped()
Is the transport system stopped?
boost::shared_ptr< Connection > ConnectionPtr
Definition: Connection.hh:50
#define HEADER_LENGTH
Definition: Connection.hh:41
#define gzerr
Output an error message.
Definition: Console.hh:49
Manages boost::asio IO.
Definition: IOManager.hh:33
void AsyncRead(Handler _handler)
Peform an asyncronous read param[in] _handler Callback to invoke on received data.
Definition: Connection.hh:205
boost::function< void(const ConnectionPtr &)> AcceptCallback
The signature of a connection accept callback.
Definition: Connection.hh:116
boost::function< void(const std::string &_data)> ReadCallback
The signature of a connection read callback.
Definition: Connection.hh:125
GAZEBO_VISIBLE bool shutdown()
Stop and cleanup simulation.
#define NULL
Definition: CommonTypes.hh:30
Single TCP/IP connection manager.
Definition: Connection.hh:100
#define GAZEBO_VISIBLE
Use to represent "symbol visible" if supported.
Definition: system.hh:48
event::ConnectionPtr ConnectToShutdown(boost::function< void()> _subscriber)
Register a function to be called when the connection is shut down.
Definition: Connection.hh:323