##

Connection.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 _CONNECTION_HH_
#define _CONNECTION_HH_
#include <tbb/task.h>
#include <google/protobuf/message.h>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <boost/thread.hpp>
#include <boost/tuple/tuple.hpp>
#include <string>
#include <vector>
#include <iostream>
#include <iomanip>
#include <deque>
#include <utility>
#include "gazebo/common/Event.hh"
#include "gazebo/common/Console.hh"
#include "gazebo/common/Exception.hh"
#include "gazebo/util/system.hh"
#define HEADER_LENGTH 8
namespace gazebo
{
namespace transport
{
extern GZ_TRANSPORT_VISIBLE bool is_stopped();
class IOManager;
class Connection;
typedef boost::shared_ptr<Connection> ConnectionPtr;
class GZ_TRANSPORT_VISIBLE ConnectionReadTask : public tbb::task
{
public: ConnectionReadTask(
boost::function<void (const std::string &)> _func,
const std::string &_data) :
func(_func),
data(_data)
{
}
public: tbb::task *execute()
{
this->func(this->data);
return NULL;
}
private: boost::function<void (const std::string &)> func;
private: std::string data;
};
class GZ_TRANSPORT_VISIBLE Connection :
public boost::enable_shared_from_this<Connection>
{
public: Connection();
public: virtual ~Connection();
public: bool Connect(const std::string &_host, unsigned int _port);
typedef boost::function<void(const ConnectionPtr&)> AcceptCallback;
public: void Listen(unsigned int _port, const AcceptCallback &_acceptCB);
typedef boost::function<void(const std::string &_data)> ReadCallback;
public: void StartRead(const ReadCallback &_cb);
public: void StopRead();
public: void Shutdown();
public: bool IsOpen() const;
private: void Close();
public: void Cancel();
public: bool Read(std::string &_data);
public: void EnqueueMsg(const std::string &_buffer,
boost::function<void(uint32_t)> _cb, uint32_t _id,
bool _force = false);
public: void EnqueueMsg(const std::string &_buffer, bool _force = false);
public: std::string GetLocalURI() const;
public: std::string GetRemoteURI() const;
public: std::string GetLocalAddress() const;
public: unsigned int GetLocalPort() const;
public: std::string GetRemoteAddress() const;
public: unsigned int GetRemotePort() const;
public: std::string GetRemoteHostname() const;
public: static std::string GetLocalHostname();
public: template<typename Handler>
void AsyncRead(Handler _handler)
{
boost::mutex::scoped_lock lock(this->socketMutex);
if (!this->IsOpen())
{
gzerr << "AsyncRead on a closed socket\n";
return;
}
void (Connection::*f)(const boost::system::error_code &,
boost::tuple<Handler>) = &Connection::OnReadHeader<Handler>;
this->inboundHeader.resize(HEADER_LENGTH);
boost::asio::async_read(*this->socket,
boost::asio::buffer(this->inboundHeader),
boost::bind(f, this,
boost::asio::placeholders::error,
boost::make_tuple(_handler)));
}
private: template<typename Handler>
void OnReadHeader(const boost::system::error_code &_e,
boost::tuple<Handler> _handler)
{
if (_e)
{
if (_e.value() == boost::asio::error::eof)
this->isOpen = false;
}
else
{
std::size_t inboundData_size = 0;
std::string header(&this->inboundHeader[0],
this->inboundHeader.size());
this->inboundHeader.clear();
inboundData_size = this->ParseHeader(header);
if (inboundData_size > 0)
{
// Start the asynchronous call to receive data
this->inboundData.resize(inboundData_size);
void (Connection::*f)(const boost::system::error_code &e,
boost::tuple<Handler>) =
&Connection::OnReadData<Handler>;
boost::asio::async_read(*this->socket,
boost::asio::buffer(this->inboundData),
boost::bind(f, this,
boost::asio::placeholders::error,
_handler));
}
else
{
gzerr << "Header is empty\n";
boost::get<0>(_handler)("");
// This code tries to read the header again. We should
// never get here.
// this->inboundHeader.resize(HEADER_LENGTH);
// void (Connection::*f)(const boost::system::error_code &,
// boost::tuple<Handler>) =
// &Connection::OnReadHeader<Handler>;
// boost::asio::async_read(*this->socket,
// boost::asio::buffer(this->inboundHeader),
// boost::bind(f, this,
// boost::asio::placeholders::error, _handler));
}
}
}
private: template<typename Handler>
void OnReadData(const boost::system::error_code &_e,
boost::tuple<Handler> _handler)
{
if (_e)
{
if (_e.value() == boost::asio::error::eof)
this->isOpen = false;
}
// Inform caller that data has been received
std::string data(&this->inboundData[0],
this->inboundData.size());
this->inboundData.clear();
if (data.empty())
gzerr << "OnReadData got empty data!!!\n";
if (!_e && !transport::is_stopped())
{
ConnectionReadTask *task = new(tbb::task::allocate_root())
ConnectionReadTask(boost::get<0>(_handler), data);
tbb::task::enqueue(*task);
// Non-tbb version:
// boost::get<0>(_handler)(data);
}
}
public: event::ConnectionPtr ConnectToShutdown(boost::function<void()>
_subscriber)
{ return this->shutdown.Connect(_subscriber); }
public: void ProcessWriteQueue(bool _blocking = false);
public: unsigned int GetId() const;
public: static bool ValidateIP(const std::string &_ip);
public: std::string GetIPWhiteList() const;
private: void PostWrite();
private: void OnWrite(const boost::system::error_code &_e);
private: void OnAccept(const boost::system::error_code &_e);
private: std::size_t ParseHeader(const std::string &_header);
private: void ReadLoop(const ReadCallback &_cb);
private: static boost::asio::ip::tcp::endpoint GetLocalEndpoint();
private: boost::asio::ip::tcp::endpoint GetRemoteEndpoint() const;
private: static std::string GetHostname(
boost::asio::ip::tcp::endpoint _ep);
private: void OnConnect(const boost::system::error_code &_error,
boost::asio::ip::tcp::resolver::iterator _endPointIter);
private: boost::asio::ip::tcp::socket *socket;
private: boost::asio::ip::tcp::acceptor *acceptor;
private: std::deque<std::string> writeQueue;
private: std::deque< std::vector<
std::pair<boost::function<void(uint32_t)>, uint32_t> > >
callbacks;
private: boost::mutex connectMutex;
private: boost::recursive_mutex writeMutex;
private: boost::recursive_mutex readMutex;
private: mutable boost::mutex socketMutex;
private: boost::condition_variable connectCondition;
private: AcceptCallback acceptCB;
private: std::vector<char> inboundHeader;
private: std::vector<char> inboundData;
private: bool readQuit;
private: unsigned int id;
private: static unsigned int idCounter;
private: ConnectionPtr acceptConn;
private: event::EventT<void()> shutdown;
private: static IOManager *iomanager;
private: unsigned int writeCount;
private: std::string localURI;
private: std::string localAddress;
private: std::string remoteURI;
private: std::string remoteAddress;
private: bool connectError;
private: std::string ipWhiteList;
private: bool dropMsgLogged;
private: bool isOpen;
};
}
}
#endif
gazebo::transport::Connection::GetRemotePort
unsigned int GetRemotePort() const
Get the remote port number.
gazebo::transport::Connection::EnqueueMsg
void EnqueueMsg(const std::string &_buffer, boost::function< void(uint32_t)> _cb, uint32_t _id, bool _force=false)
Write data to the socket.
gazebo::transport::Connection::Listen
void Listen(unsigned int _port, const AcceptCallback &_acceptCB)
Start a server that listens on a port.
gazebo
Forward declarations for the common classes.
Definition: Animation.hh:26
Exception.hh
NULL
#define NULL
Definition: CommonTypes.hh:31
gazebo::transport::Connection::ReadCallback
boost::function< void(const std::string &_data)> ReadCallback
The signature of a connection read callback.
Definition: Connection.hh:125
gazebo::transport::Connection::StartRead
void StartRead(const ReadCallback &_cb)
Start a thread that reads from the connection and passes new message to the ReadCallback.
gazebo::transport::Connection::AsyncRead
void AsyncRead(Handler _handler)
Peform an asyncronous read param[in] _handler Callback to invoke on received data.
Definition: Connection.hh:205
gazebo::transport::Connection::StopRead
void StopRead()
Stop the read loop.
Console.hh
system.hh
gazebo::transport::Connection::GetRemoteAddress
std::string GetRemoteAddress() const
Get the remote address.
gazebo::transport::Connection::ValidateIP
static bool ValidateIP(const std::string &_ip)
Return true if the _ip is a valid.
gazebo::transport::Connection::Connect
bool Connect(const std::string &_host, unsigned int _port)
Connect to a remote host.
gazebo::transport::is_stopped
bool is_stopped()
Is the transport system stopped?
gazebo::transport::Connection::GetLocalPort
unsigned int GetLocalPort() const
Get the port of this connection.
HEADER_LENGTH
#define HEADER_LENGTH
Definition: Connection.hh:41
gzerr
#define gzerr
Output an error message.
Definition: Console.hh:50
gazebo::event::EventT
A class for event processing.
Definition: Event.hh:97
gazebo::transport::Connection::GetRemoteURI
std::string GetRemoteURI() const
Get the remote URI.
gazebo::transport::Connection::GetLocalAddress
std::string GetLocalAddress() const
Get the local address of this connection.
gazebo::transport::Connection::~Connection
virtual ~Connection()
Destructor.
gazebo::transport::Connection::ConnectToShutdown
event::ConnectionPtr ConnectToShutdown(boost::function< void()> _subscriber)
Register a function to be called when the connection is shut down.
Definition: Connection.hh:324
gazebo::event::EventT::Connect
ConnectionPtr Connect(const std::function< T > &_subscriber)
Connect a callback to this event.
Definition: Event.hh:558
gazebo::transport::Connection::Cancel
void Cancel()
Cancel all async operations on an open socket.
gazebo::transport::ConnectionPtr
boost::shared_ptr< Connection > ConnectionPtr
Definition: Connection.hh:50
gazebo::event::ConnectionPtr
boost::shared_ptr< Connection > ConnectionPtr
Definition: CommonTypes.hh:134
gazebo::transport::Connection::Read
bool Read(std::string &_data)
Read data from the socket.
gazebo::shutdown
GAZEBO_VISIBLE bool shutdown()
Stop and cleanup simulation.
gazebo::transport::IOManager
Manages boost::asio IO.
Definition: IOManager.hh:33
gazebo::transport::Connection::Shutdown
void Shutdown()
Shutdown the socket.
gazebo::transport::Connection::Connection
Connection()
Constructor.
transport
transport
Definition: ConnectionManager.hh:35
gazebo::transport::Connection
Single TCP/IP connection manager.
Definition: Connection.hh:100
gazebo::transport::Connection::GetLocalURI
std::string GetLocalURI() const
Get the local URI.
Event.hh
gazebo::transport::Connection::ProcessWriteQueue
void ProcessWriteQueue(bool _blocking=false)
Handle on-write callbacks.
gazebo::transport::Connection::GetId
unsigned int GetId() const
Get the ID of the connection.
gazebo::transport::Connection::GetRemoteHostname
std::string GetRemoteHostname() const
Get the remote hostname.
gazebo::transport::Connection::AcceptCallback
boost::function< void(const ConnectionPtr &)> AcceptCallback
The signature of a connection accept callback.
Definition: Connection.hh:116
gazebo::transport::Connection::GetLocalHostname
static std::string GetLocalHostname()
Get the local hostname.
gazebo::transport::Connection::IsOpen
bool IsOpen() const
Is the connection open?
gazebo::transport::Connection::GetIPWhiteList
std::string GetIPWhiteList() const
Get the IP white list, from GAZEBO_IP_WHITE_LIST environment variable.