asio.hpp

Go to the documentation of this file.

/*
 *  @brief defines wrappers for boost::asio functions
 */
#pragma once
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
#include <vector>
#include <fc/thread/future.hpp>
#include <fc/io/iostream.hpp>

namespace fc { 
namespace asio {
    namespace detail {

        class read_write_handler
        {
        public:
          read_write_handler(const promise<size_t>::ptr& p);
          void operator()(const boost::system::error_code& ec, size_t bytes_transferred);
        private:
          promise<size_t>::ptr _completion_promise;
        };

        class read_write_handler_with_buffer
        {
        public:
          read_write_handler_with_buffer(const promise<size_t>::ptr& p, 
                                         const std::shared_ptr<const char>& buffer);
          void operator()(const boost::system::error_code& ec, size_t bytes_transferred);
        private:
          promise<size_t>::ptr _completion_promise;
          std::shared_ptr<const char> _buffer;
        };

        //void read_write_handler( const promise<size_t>::ptr& p, 
        //                         const boost::system::error_code& ec, 
        //                         size_t bytes_transferred );
        void read_write_handler_ec( promise<size_t>* p, 
                                    boost::system::error_code* oec, 
                                    const boost::system::error_code& ec, 
                                    size_t bytes_transferred );
        void error_handler( const promise<void>::ptr& p, 
                              const boost::system::error_code& ec );
        void error_handler_ec( promise<boost::system::error_code>* p, 
                              const boost::system::error_code& ec ); 

        template<typename C>
        struct non_blocking { 
          bool operator()( C& c ) { return c.non_blocking(); } 
          bool operator()( C& c, bool s ) { c.non_blocking(s); return true; } 
        };

#if WIN32  // windows stream handles do not support non blocking!
           template<>
         struct non_blocking<boost::asio::windows::stream_handle> { 
              typedef boost::asio::windows::stream_handle C;
            bool operator()( C& ) { return false; } 
            bool operator()( C&, bool ) { return false; } 
        };
#endif
    } // end of namespace detail

    /***
     * A structure for holding the boost io service and associated
     * threads
     */
    class default_io_service_scope
    {
       public:
          default_io_service_scope();
          ~default_io_service_scope();
          static void     set_num_threads(uint16_t num_threads);
          static uint16_t get_num_threads();
          boost::asio::io_service*          io;
       private:
          std::vector<boost::thread*>       asio_threads;
          boost::asio::io_service::work*    the_work;
       protected:
          static uint16_t num_io_threads; // marked protected to help with testing
    };

    boost::asio::io_service& default_io_service();

    template<typename AsyncReadStream, typename MutableBufferSequence>
    size_t read( AsyncReadStream& s, const MutableBufferSequence& buf ) {
        promise<size_t>::ptr p = promise<size_t>::create("fc::asio::read");
        boost::asio::async_read( s, buf, detail::read_write_handler(p) );
        return p->wait();
    }
    template<typename AsyncReadStream, typename MutableBufferSequence>
    future<size_t> read_some(AsyncReadStream& s, const MutableBufferSequence& buf)
    {
      promise<size_t>::ptr completion_promise = promise<size_t>::create("fc::asio::async_read_some");
      s.async_read_some(buf, detail::read_write_handler(completion_promise));
      return completion_promise;//->wait();
    }

    template<typename AsyncReadStream>
    future<size_t> read_some(AsyncReadStream& s, char* buffer, size_t length, size_t offset = 0)
    {
      promise<size_t>::ptr completion_promise = promise<size_t>::create("fc::asio::async_read_some");
      s.async_read_some(boost::asio::buffer(buffer + offset, length), 
                        detail::read_write_handler(completion_promise));
      return completion_promise;//->wait();
    }

    template<typename AsyncReadStream>
    future<size_t> read_some(AsyncReadStream& s, const std::shared_ptr<char>& buffer, size_t length, size_t offset)
    {
      promise<size_t>::ptr completion_promise = promise<size_t>::create("fc::asio::async_read_some");
      s.async_read_some(boost::asio::buffer(buffer.get() + offset, length), 
                        detail::read_write_handler_with_buffer(completion_promise, buffer));
      return completion_promise;//->wait();
    }

    template<typename AsyncReadStream, typename MutableBufferSequence>
    void async_read_some(AsyncReadStream& s, const MutableBufferSequence& buf, promise<size_t>::ptr completion_promise)
    {
      s.async_read_some(buf, detail::read_write_handler(completion_promise));
    }

    template<typename AsyncReadStream>
    void async_read_some(AsyncReadStream& s, char* buffer,
                         size_t length, promise<size_t>::ptr completion_promise)
    {
      s.async_read_some(boost::asio::buffer(buffer, length), detail::read_write_handler(completion_promise));
    }

    template<typename AsyncReadStream>
    void async_read_some(AsyncReadStream& s, const std::shared_ptr<char>& buffer,
                         size_t length, size_t offset, promise<size_t>::ptr completion_promise)
    {
      s.async_read_some(boost::asio::buffer(buffer.get() + offset, length), detail::read_write_handler_with_buffer(completion_promise, buffer));
    }

    template<typename AsyncReadStream>
    size_t read_some( AsyncReadStream& s, boost::asio::streambuf& buf )
    {
        char buffer[1024];
        size_t bytes_read = read_some( s, boost::asio::buffer( buffer, sizeof(buffer) ) );
        buf.sputn( buffer, bytes_read );
        return bytes_read;
    }

    template<typename AsyncWriteStream, typename ConstBufferSequence>
    size_t write( AsyncWriteStream& s, const ConstBufferSequence& buf ) {
        promise<size_t>::ptr p = promise<size_t>::create("fc::asio::write");
        boost::asio::async_write(s, buf, detail::read_write_handler(p));
        return p->wait();
    }

    template<typename AsyncWriteStream, typename ConstBufferSequence>
    future<size_t> write_some( AsyncWriteStream& s, const ConstBufferSequence& buf ) {
        promise<size_t>::ptr p = promise<size_t>::create("fc::asio::write_some");
        s.async_write_some( buf, detail::read_write_handler(p));
        return p; //->wait();
    }

    template<typename AsyncWriteStream>
    future<size_t> write_some( AsyncWriteStream& s, const char* buffer, 
                               size_t length, size_t offset = 0) {
        promise<size_t>::ptr p = promise<size_t>::create("fc::asio::write_some");
        s.async_write_some( boost::asio::buffer(buffer + offset, length), detail::read_write_handler(p));
        return p; //->wait();
    }

    template<typename AsyncWriteStream>
    future<size_t> write_some( AsyncWriteStream& s, const std::shared_ptr<const char>& buffer, 
                               size_t length, size_t offset ) {
        promise<size_t>::ptr p = promise<size_t>::create("fc::asio::write_some");
        s.async_write_some( boost::asio::buffer(buffer.get() + offset, length), detail::read_write_handler_with_buffer(p, buffer));
        return p; //->wait();
    }

    template<typename AsyncWriteStream, typename ConstBufferSequence>
    void async_write_some(AsyncWriteStream& s, const ConstBufferSequence& buf, promise<size_t>::ptr completion_promise) {
      s.async_write_some(buf, detail::read_write_handler(completion_promise));
    }

    template<typename AsyncWriteStream>
    void async_write_some(AsyncWriteStream& s, const char* buffer, 
                          size_t length, promise<size_t>::ptr completion_promise) {
      s.async_write_some(boost::asio::buffer(buffer, length), 
                         detail::read_write_handler(completion_promise));
    }

    template<typename AsyncWriteStream>
    void async_write_some(AsyncWriteStream& s, const std::shared_ptr<const char>& buffer, 
                          size_t length, size_t offset, promise<size_t>::ptr completion_promise) {
      s.async_write_some(boost::asio::buffer(buffer.get() + offset, length), 
                         detail::read_write_handler_with_buffer(completion_promise, buffer));
    }

    namespace tcp {
        typedef boost::asio::ip::tcp::endpoint endpoint;
        typedef boost::asio::ip::tcp::resolver::iterator resolver_iterator;
        typedef boost::asio::ip::tcp::resolver resolver;
        std::vector<endpoint> resolve( const std::string& hostname, const std::string& port );

        template<typename SocketType, typename AcceptorType>
        void accept( AcceptorType& acc, SocketType& sock ) {
            promise<void>::ptr p = promise<void>::create("fc::asio::tcp::accept");
            acc.async_accept( sock, boost::bind( fc::asio::detail::error_handler, p, _1 ) );
            p->wait();
            //if( ec ) BOOST_THROW_EXCEPTION( boost::system::system_error(ec) );
        }

        template<typename AsyncSocket, typename EndpointType>
        void connect( AsyncSocket& sock, const EndpointType& ep ) {
            promise<void>::ptr p = promise<void>::create("fc::asio::tcp::connect");
            sock.async_connect( ep, boost::bind( fc::asio::detail::error_handler, p, _1 ) );
            p->wait();
            //if( ec ) BOOST_THROW_EXCEPTION( boost::system::system_error(ec) );
        }
    }
    namespace udp {
        typedef boost::asio::ip::udp::endpoint endpoint;
        typedef boost::asio::ip::udp::resolver::iterator resolver_iterator;
        typedef boost::asio::ip::udp::resolver resolver;
        std::vector<endpoint> resolve( resolver& r, const std::string& hostname, 
                                                         const std::string& port );
    }

    template<typename AsyncReadStream>
    class istream : public virtual fc::istream
    {
       public:
          istream( std::shared_ptr<AsyncReadStream> str )
          :_stream( std::move(str) ){}

          virtual size_t readsome( char* buf, size_t len )
          {
             return fc::asio::read_some(*_stream, buf, len).wait();
          }
          virtual size_t readsome( const std::shared_ptr<char>& buf, size_t len, size_t offset )
          {
             return fc::asio::read_some(*_stream, buf, len, offset).wait();
          }
    
       private:
          std::shared_ptr<AsyncReadStream> _stream;
    };

    template<typename AsyncWriteStream>
    class ostream : public virtual fc::ostream
    {
       public:
          ostream( std::shared_ptr<AsyncWriteStream> str )
          :_stream( std::move(str) ){}

          virtual size_t writesome( const char* buf, size_t len )
          {
             return fc::asio::write_some(*_stream, buf, len).wait();
          }
    
          virtual size_t     writesome( const std::shared_ptr<const char>& buf, size_t len, size_t offset )
          {
             return fc::asio::write_some(*_stream, buf, len, offset).wait();
          }
    
          virtual void       close(){ _stream->close(); }
          virtual void       flush() {}
       private:
          std::shared_ptr<AsyncWriteStream> _stream;
    };


} } // namespace fc::asio