Connection.hh

Go to the documentation of this file.

/*
 * Copyright (C) 2012-2014 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"

#define HEADER_LENGTH 8

namespace gazebo
{
  namespace transport
  {
    extern bool is_stopped();

    class IOManager;
    class Connection;
    typedef boost::shared_ptr<Connection> ConnectionPtr;

    class ConnectionReadTask : public tbb::task
    {
      public: ConnectionReadTask(
                  boost::function<void (const std::string &)> _func,
                  const std::string &_data)
              {
                this->func = _func;
                this->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 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)
              {
                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.message() == "End of file")
                    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.message() == "End of file")
                    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 DisconnectShutdown(event::ConnectionPtr _subscriber)
              {this->shutdown.Disconnect(_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 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::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: char *headerBuffer;

      private: bool dropMsgLogged;

      private: unsigned int callbackIndex;

      private: bool isOpen;
    };
  }
}
#endif