libs/rpc/include/mace/rpc/http/detail/connection.hpp

#ifndef _MACE_RPC_HTTP_DETAIL_CONNECTION_HPP
#define _MACE_RPC_HTTP_DETAIL_CONNECTION_HPP
#include <mace/rpc/tcp/detail/connection.hpp>
#include <mace/rpc/raw/message.hpp>
#include <mace/rpc/raw/raw_io.hpp>
#include <mace/rpc/http/request.hpp>
#include <mace/rpc/http/request_parser.hpp>
#include <mace/rpc/http/reply.hpp>
#include <mace/rpc/error.hpp>
#include <boost/algorithm/string.hpp>

#include <mace/cmt/asio/tcp/socket.hpp>
#include <mace/cmt/asio.hpp>

namespace mace { namespace rpc { namespace http {   namespace detail {
  class connection : public mace::rpc::detail::connection_base  {
    public:
      connection(){}

      template<typename String>
      connection( String&& host, uint16_t port = 80 )
      :m_host( host + ":" + boost::lexical_cast<std::string>(port) ) {
        m_eps = mace::cmt::asio::tcp::resolve( host, port );
        if( !m_eps.size() ) {
          MACE_RPC_THROW( "Unable to resolve host '%1%'", %m_host );
        }
      }

      connection( const boost::asio::ip::tcp::endpoint& ep ) {
        m_host = ep.address().to_string() + ":" + boost::lexical_cast<std::string>(ep.port());
        m_eps.push_back(ep);
      }


      connection( const mace::cmt::asio::tcp::socket::ptr& sock )
      :m_sock(sock){
        if( m_sock ) {
          m_read_done = 
            mace::cmt::async( boost::bind( &connection::read_request_loop, this ) );
        }
      }

      virtual ~connection() {
        try { close(); } catch ( ... ) { }
      }

      void close() {
        if( m_sock ) {
          m_sock->close();
          if( m_read_done.valid() ) {
            m_read_done.wait();
          }
          m_sock.reset();
        }
      }

      virtual http::request pack_request( rpc::message&& m ) { return http::request(); }
      virtual rpc::message  unpack_reply( http::reply&& r )  { return rpc::message();  }

      virtual http::reply handle_request( http::request&& r ) {
        return http::reply();
      }

      http::reply send( http::request&& r ) {
        r.headers.push_back( header("Host", m_host) );
        if( !m_sock ) {
          m_sock.reset( new mace::cmt::asio::tcp::socket() );
          for( uint32_t i = 0; i < m_eps.size(); ++i ) {
            if( m_sock->connect( m_eps[i] ).wait() ) {
              std::swap( m_eps[0],m_eps[i]);
            }
          }
          MACE_RPC_THROW( "Unable to connect to host." );
        }
        *m_sock << r;
        return read_reply();
      }


      typedef mace::cmt::asio::tcp::socket::iterator itr_t;
      http::request read_request() {
        // read http request
        http::request r;
        http::request_parser p;
        boost::tribool result;
        itr_t end_parse;
        boost::tie(result,end_parse) = p.parse( r, itr_t(m_sock.get()), itr_t() );
        return r;
      }

      std::string read_line() {
         std::string l;
         itr_t i(m_sock.get());
         while( *i != '\n' ) {
          if( *i != '\r' )
              l+=*i;
          ++i;
         }
         return l;
      }

      http::reply read_reply() {
        http::reply r;
        std::string line = read_line();
        std::string ver  = line.substr(0,line.find(' '));
        std::stringstream ss(line.substr(ver.size()+1,line.size()) );
        int code; 
        ss >> code;
        r.status = (http::reply::status_type)code;
        line = read_line();
        std::string name, value;

        int content_len = 0;
        std::string content_length("content-length");
        while( line.size() ){
          if( line[0] != ' ' ) {
              name = line.substr(0,line.find(':') );
              boost::to_lower(name);
              value = line.substr(name.size()+1);
              boost::trim(value);
              r.headers.push_back(header(std::move(name),std::move(value)));
              if( name == content_length ) {
                content_len = boost::lexical_cast<int>(value);
              } else if( name == "connection" ) {
                boost::to_lower(value);
                r.keep_alive = (value == "keep-alive");
              }
          } else {
            if( r.headers.size() ) {
              r.headers.back().value += line;
            }
          }
          line = read_line();
        }
        if( content_len ) {
          r.content.resize(content_len);
          m_sock->read( &r.content.front(), r.content.size() );
        }

        if( !r.keep_alive ) {
          m_sock->close();
          m_sock.reset();
        }
        return r;
      }

      void read_rpc_reply() {
        // unpack it and send it to the handle method
        handle( unpack_reply( read_reply() ) );
      }

      virtual void send( rpc::message&& m ) {
        // assert message is a request..
        // delegate packing the message content to
        // a derived class
        send( pack_request( std::move(m) ) );

        // read reply
        m_read_done = cmt::async( boost::bind(&connection::read_rpc_reply, this ) );
      }

      void read_request_loop( ) {
        try {
          while ( true ) {
            handle_request( read_request() );
          }
        } catch ( ... ) {
          elog( "connection closed: %1%", 
          boost::current_exception_diagnostic_information() );
        }
        break_promises();
        m_sock.reset();
      }
      
      void handle_error( message::error_type e, const std::string& msg ) {
        elog( "%1%: %2%", int(e), msg );
      }

      std::string                                 m_host;
      std::vector<boost::asio::ip::tcp::endpoint> m_eps;
      cmt::future<void>                           m_read_done;
      mace::cmt::asio::tcp::socket::ptr           m_sock; 
  };

} } } } // mace::rpc::http::detail

#endif // _MACE_RPC_RAW_TCP_DETAIL_CONNECTION_HPP