Connection.hh

/*
 * Copyright 2011 Nate Koenig & Andrew Howard
 *
 * 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 <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/function.hpp>
#include <boost/thread.hpp>
#include <boost/tuple/tuple.hpp>
#include <iostream>
#include <iomanip>
#include <deque>

#include <google/protobuf/message.h>

#include "common/Event.hh"
#include "common/Console.hh"
#include "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 Connection : public boost::enable_shared_from_this<Connection>
    {
      public: Connection();

      public: virtual ~Connection();

      public: bool Connect(const std::string &host,  unsigned short port);

      typedef boost::function<void(const ConnectionPtr&)> AcceptCallback;

      public: void Listen(unsigned short port, const AcceptCallback &accept_cb);

      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, bool _force = false);

      public: std::string GetLocalURI() const;
              
      public: std::string GetRemoteURI() const;

      public: std::string GetLocalAddress() const;

      public: unsigned short GetLocalPort() const;

      public: std::string GetRemoteAddress() const;

      public: unsigned short GetRemotePort() const;

      public: std::string GetRemoteHostname() const;

      public: std::string GetLocalHostname() const;

      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->inbound_header.resize(HEADER_LENGTH);
                boost::asio::async_read(*this->socket,
                    boost::asio::buffer(this->inbound_header),
                    boost::bind(f, this, 
                                boost::asio::placeholders::error,
                                boost::make_tuple(handler)) );
              }

      // Handle a completed read of a message header. The handler is passed
      // using a tuple since boost::bind seems to have trouble binding
      // a function object created using boost::bind as a parameter
      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->Close();
                    // This will occur when the other side closes the
                    // connection
                  }
                }
                else
                {
                  std::size_t inbound_data_size = 0;
                  std::string header(&this->inbound_header[0], 
                                      this->inbound_header.size());
                  this->inbound_header.clear();

                  inbound_data_size = this->ParseHeader(header);

                 if (inbound_data_size > 0)
                  {
                    // Start the asynchronous call to receive data
                    this->inbound_data.resize(inbound_data_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->inbound_data), 
                        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->inbound_header.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->inbound_header),
                    //    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)
                  gzerr << "Error Reading data!\n";

                // Inform caller that data has been received
                std::string data(&this->inbound_data[0], 
                                  this->inbound_data.size());
                this->inbound_data.clear();

                if (data.empty())
                  gzerr << "OnReadData got empty data!!!\n";

                if (!e && !transport::is_stopped()) 
                {
                  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();

     private: void OnWrite(const boost::system::error_code &e,
                  boost::asio::streambuf *_b);
           //std::list<boost::asio::const_buffer> *_buffer);

     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: boost::asio::ip::tcp::endpoint GetLocalEndpoint() const;

     private: boost::asio::ip::tcp::endpoint GetRemoteEndpoint() const;

     private: static std::string GetHostname(boost::asio::ip::tcp::endpoint ep);

      private: boost::asio::ip::tcp::socket *socket;
      private: boost::asio::ip::tcp::acceptor *acceptor;

      private: std::deque<std::string> writeQueue;
      private: std::deque<unsigned int> writeCounts;
      private: boost::recursive_mutex *writeMutex;
      private: boost::recursive_mutex *readMutex;

      // Called when a new connection is received
      private: AcceptCallback acceptCB;

      //private: char inbound_header[HEADER_LENGTH];
      private: std::vector<char> inbound_header;
      private: std::vector<char> inbound_data;

      private: boost::thread *readThread;
      private: bool readQuit;

      public: unsigned int id;
      private: static unsigned int idCounter;
      private: ConnectionPtr acceptConn;

      private: event::EventT<void()> shutdown;
      private: static IOManager *iomanager;

      public: unsigned int writeCount;

      private: std::string localURI;
      private: std::string localAddress;
      private: std::string remoteURI;
      private: std::string remoteAddress;
    };
  }
}

#endif