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_
 
#undef emit
#include <tbb/task.h>
#define emit
 
// If TBB_VERSION_MAJOR is not defined, this means that
// tbb >= 2021 and we can include the tbb/version.h header
#ifndef TBB_VERSION_MAJOR
#include <tbb/version.h>
#endif
 
#include <google/protobuf/message.h>
 
// This fixes compiler warnings, see #3147 and #3160
#ifndef BOOST_BIND_GLOBAL_PLACEHOLDERS
#define BOOST_BIND_GLOBAL_PLACEHOLDERS
#endif
#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/common/WeakBind.hh"
#if TBB_VERSION_MAJOR >= 2021
#include "gazebo/transport/TaskGroup.hh"
#endif
#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;
 
#if TBB_VERSION_MAJOR < 2021
    class GZ_TRANSPORT_VISIBLE ConnectionReadTask : public tbb::task
#else
    class GZ_TRANSPORT_VISIBLE ConnectionReadTask
#endif
    {
      public: ConnectionReadTask(
                  boost::function<void (const std::string &)> _func,
                  const std::string &_data) :
                func(_func),
                data(_data)
              {
              }
 
#if TBB_VERSION_MAJOR < 2021
      public: tbb::task *execute()
              {
                this->func(this->data);
                return NULL;
              }
#else
      public: void operator()() const
              { this->func(this->data); }
#endif
      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),
                    common::weakBind(f, this->shared_from_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),
                        common::weakBind(f, this->shared_from_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),
                    //    common::weakBind(f, this->shared_from_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())
                {
#if TBB_VERSION_MAJOR < 2021
                  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);
#else
                  this->taskGroup.run<ConnectionReadTask>(boost::get<0>(_handler), data);
#endif
                }
              }
 
      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;
 
#if TBB_VERSION_MAJOR >= 2021
      private: TaskGroup taskGroup;
#endif
    };
  }
}
#endif