buffered_reader.hh

Go to the documentation of this file.

//
// Created by Jonathan Hanks on 4/15/17.
//
 
#ifndef NDS_PROXY_BUFFERED_READER_HH
#define NDS_PROXY_BUFFERED_READER_HH
 
#include <algorithm>
#include <stdexcept>
#include <vector>
 
namespace nds_impl
{
    namespace Socket
    {
        template < typename Reader >
        class BufferedReader
        {
        public:
            explicit BufferedReader( Reader& r ) : r_( r ), buf_{}
            {
            }
            // BufferedReader(Reader&& r): r_{std::move(r)}, buf_{} {}
            // BufferedReader(Reader&& r): r_{std::move(r)} {}
 
            // BufferedReader(BufferedReader<Reader, ReaderCaptureType>&&
            // other): r_{std::move(other.r_)}, buf_{std::move(other.buf)} {}
 
            void
            write_all( const char* start, const char* end )
            {
                r_.write_all( start, end );
            }
 
            char*
            read_available( char* start, char* end )
            {
                size_t len_requested = end - start;
 
                if ( buf_.empty( ) )
                    fill_buffer( );
                if ( buf_.size( ) < len_requested )
                    len_requested = buf_.size( );
                std::copy( buf_.data( ), buf_.data( ) + len_requested, start );
                consume( len_requested );
                return start + len_requested;
            }
 
            char*
            read_exactly( char* start, const char* end )
            {
                size_t len_requested = end - start;
 
                auto result = peek( start, end );
                consume( len_requested );
                return result;
            }
 
            char*
            peek( char* start, const char* end )
            {
                size_t len_requested = end - start;
 
                while ( buf_.size( ) < len_requested )
                    fill_buffer( );
                std::copy( buf_.data( ), buf_.data( ) + len_requested, start );
                return start + len_requested;
            }
 
            template < typename It >
            std::vector< char >
            read_until( It begin_set, It end_set )
            {
                std::vector< char > result;
 
                decltype( buf_.size( ) ) cur = 0;
                auto remaining = buf_.size( );
                for ( bool match = false; !match; ++cur, --remaining )
                {
                    if ( remaining == 0 )
                    {
                        fill_buffer( );
                        remaining = buf_.size( ) - cur;
                    }
 
                    if ( std::find( begin_set, end_set, buf_.at( cur ) ) !=
                         end_set )
                    {
                        match = true;
                    }
                }
                result.resize( cur );
                std::copy(
                    buf_.begin( ), buf_.begin( ) + cur, result.begin( ) );
                consume( cur );
                return result;
            }
 
            template < typename It, typename OutIt >
            OutIt
            read_until( It    begin_set,
                        It    end_set,
                        OutIt dest_start,
                        OutIt dest_end )
            {
                decltype( buf_.size( ) ) cur = 0;
                auto  remaining = buf_.size( );
                OutIt dest_cur = dest_start;
                bool  match = false;
                for ( ; !match && dest_cur != dest_end;
                      ++cur, --remaining, ++dest_cur )
                {
                    if ( remaining == 0 )
                    {
                        fill_buffer( );
                        remaining = buf_.size( ) - cur;
                    }
 
                    if ( std::find( begin_set, end_set, buf_.at( cur ) ) !=
                         end_set )
                    {
                        match = true;
                    }
                }
                if ( match )
                {
                    std::copy( buf_.begin( ), buf_.begin( ) + cur, dest_start );
                }
                else
                {
                    throw std::range_error(
                        "Unable to find matching sequence" );
                }
                consume( cur );
                return dest_cur;
            }
 
            std::vector< char >::size_type
            __buffered_bytes( ) const
            {
                return buf_.size( );
            }
 
        private:
            Reader&             r_;
            std::vector< char > buf_;
 
            void
            fill_buffer( )
            {
                const decltype( buf_.size( ) ) chunk_size = 1024;
                auto data_size = buf_.size( );
 
                buf_.resize( data_size + chunk_size );
                auto data_start = buf_.data( ) + data_size;
 
                auto result =
                    r_.read_available( data_start, data_start + chunk_size );
 
                auto data_read_in =
                    static_cast< size_t >( result - data_start );
                if ( data_read_in > 0 )
                {
                    buf_.resize( data_size + data_read_in );
                }
                else
                {
                    buf_.resize( data_size );
                    throw std::runtime_error( "io operation failed" );
                }
            }
 
            void
            consume( std::vector< char >::size_type count )
            {
                if ( count <= 0 )
                    return;
                if ( count >= buf_.size( ) )
                {
                    buf_.clear( );
                    return;
                }
                auto current_size = buf_.size( );
                auto new_head = buf_.data( ) + count;
 
                std::move(
                    new_head, buf_.data( ) + current_size, buf_.data( ) );
                buf_.resize( current_size - count );
            }
        };
 
        template < typename Reader >
        class OwningBufferedReader
        {
            Reader              r_;
            std::vector< char > buf_;
 
            void
            fill_buffer( )
            {
                const decltype( buf_.size( ) ) chunk_size = 1024;
                auto data_size = buf_.size( );
 
                buf_.resize( data_size + chunk_size );
                auto data_start = buf_.data( ) + data_size;
 
                auto result =
                    r_.read_available( data_start, data_start + chunk_size );
 
                auto data_read_in =
                    static_cast< size_t >( result - data_start );
                if ( data_read_in > 0 )
                {
                    buf_.resize( data_size + data_read_in );
                }
                else
                {
                    buf_.resize( data_size );
                    throw std::runtime_error( "io operation failed" );
                }
            }
 
            void
            consume( std::vector< char >::size_type count )
            {
                if ( count <= 0 )
                    return;
                if ( count >= buf_.size( ) )
                {
                    buf_.clear( );
                    return;
                }
                auto current_size = buf_.size( );
                auto new_head = buf_.data( ) + count;
 
                std::move(
                    new_head, buf_.data( ) + current_size, buf_.data( ) );
                buf_.resize( current_size - count );
            }
 
        public:
            explicit OwningBufferedReader( Reader&& r )
                : r_{ std::move( r ) }, buf_{}
            {
            }
            // BufferedReader(Reader&& r): r_{std::move(r)}, buf_{} {}
            // BufferedReader(Reader&& r): r_{std::move(r)} {}
 
            // BufferedReader(BufferedReader<Reader, ReaderCaptureType>&&
            // other): r_{std::move(other.r_)}, buf_{std::move(other.buf)} {}
 
            void
            write_all( const char* start, const char* end )
            {
                r_.write_all( start, end );
            }
 
            char*
            read_available( char* start, char* end )
            {
                size_t len_requested = end - start;
 
                if ( buf_.empty( ) )
                    fill_buffer( );
                if ( buf_.size( ) < len_requested )
                    len_requested = buf_.size( );
                std::copy( buf_.data( ), buf_.data( ) + len_requested, start );
                consume( len_requested );
                return start + len_requested;
            }
 
            char*
            read_exactly( char* start, const char* end )
            {
                size_t len_requested = end - start;
 
                while ( buf_.size( ) < len_requested )
                    fill_buffer( );
                std::copy( buf_.data( ), buf_.data( ) + len_requested, start );
                consume( len_requested );
                return start + len_requested;
            }
 
            template < typename It >
            std::vector< char >
            read_until( It begin_set, It end_set )
            {
                std::vector< char > result;
 
                decltype( buf_.size( ) ) cur = 0;
                auto remaining = buf_.size( );
                for ( bool match = false; !match; ++cur, --remaining )
                {
                    if ( remaining == 0 )
                    {
                        fill_buffer( );
                        remaining = buf_.size( ) - cur;
                    }
 
                    if ( std::find( begin_set, end_set, buf_.at( cur ) ) !=
                         end_set )
                    {
                        match = true;
                    }
                }
                result.resize( cur );
                std::copy(
                    buf_.begin( ), buf_.begin( ) + cur, result.begin( ) );
                consume( cur );
                return result;
            }
 
            std::vector< char >::size_type
            __buffered_bytes( ) const
            {
                return buf_.size( );
            }
        };
    }
}
 
// Tests only after this point.
 
#ifdef _NDS_IMPL_ENABLE_CATCH_TESTS_
 
#include <algorithm>
#include <iterator>
#include <sstream>
 
#include "socket/socket.hh"
#include "tests/dummy_socket.hh"
#include "catch.hpp"
 
TEST_CASE( "Create a buffered socket", "[buffered,create]" )
{
    nds_impl::Socket::FullSocket                                     s1;
    nds_impl::Socket::BufferedReader< nds_impl::Socket::FullSocket > b1{ s1 };
}
 
TEST_CASE( "You can write through a buffered reader", "[buffered,write]" )
{
    using namespace nds_testing;
 
    RecordingDummySocket                                     s;
    nds_impl::Socket::BufferedReader< RecordingDummySocket > b{ s };
 
    auto hello = std::string( "hello" );
    b.write_all( hello.data( ), hello.data( ) + hello.length( ) );
    REQUIRE( s.str( ) == hello );
 
    auto world = std::string( " world!" );
    b.write_all( world.data( ), world.data( ) + world.length( ) );
    REQUIRE( s.str( ) == hello + world );
}
 
TEST_CASE( "You can read from a buffered writter", "[buffered,read]" )
{
    using namespace nds_testing;
 
    std::vector< char > dest( 20 );
 
    auto hello = std::string{ "hello world!" };
    auto s = DummySocket( hello );
 
    nds_impl::Socket::BufferedReader< DummySocket > b( s );
 
    REQUIRE( b.__buffered_bytes( ) == 0 );
 
    auto end = b.read_available( dest.data( ), dest.data( ) + dest.size( ) );
    REQUIRE( b.__buffered_bytes( ) == 0 );
    REQUIRE( end != dest.data( ) );
    REQUIRE( end == dest.data( ) + hello.size( ) );
    auto output = std::string( dest.data( ), end );
    REQUIRE( output == hello );
}
 
TEST_CASE( "You can read the data in small segments from a buffered reader",
           "[buffered,read]" )
{
    using namespace nds_testing;
 
    std::vector< char > dest( 5 );
 
    auto hello = std::string{ "hello world!" };
    auto s = DummySocket( hello );
 
    nds_impl::Socket::BufferedReader< DummySocket > b( s );
 
    REQUIRE( b.__buffered_bytes( ) == 0 );
 
    auto end = b.read_available( dest.data( ), dest.data( ) + dest.size( ) );
    REQUIRE( b.__buffered_bytes( ) == ( hello.size( ) - dest.size( ) ) );
    REQUIRE( end != dest.data( ) );
    REQUIRE( end == dest.data( ) + dest.size( ) );
    auto output = std::string( dest.data( ), end );
    REQUIRE( output == hello.substr( 0, dest.size( ) ) );
 
    end = b.read_available( dest.data( ), dest.data( ) + dest.size( ) );
    REQUIRE( b.__buffered_bytes( ) == ( hello.size( ) - 2 * dest.size( ) ) );
    REQUIRE( end != dest.data( ) );
    REQUIRE( end == dest.data( ) + dest.size( ) );
    output = std::string( dest.data( ), end );
    REQUIRE( output == hello.substr( dest.size( ), dest.size( ) ) );
}
 
TEST_CASE( "You can ask for an exact amount of bytes to be returned",
           "[buffered,read]" )
{
    using namespace nds_testing;
 
    std::vector< char > dest( 10 );
 
    auto input = std::string{ "0123456789" };
    auto s = DummySocket( input );
    nds_impl::Socket::BufferedReader< DummySocket > b( s );
 
    auto stride = 4;
    auto end = b.read_exactly( dest.data( ), dest.data( ) + stride );
    REQUIRE( b.__buffered_bytes( ) == ( input.size( ) - stride ) );
    REQUIRE( end == dest.data( ) + stride );
 
    REQUIRE_THROWS(
        b.read_exactly( dest.data( ), dest.data( ) + dest.size( ) ) );
}
 
TEST_CASE( "You can peek at an exact amount of data without consuming it",
           "[buffered,read]" )
{
    using namespace nds_testing;
 
    std::vector< char > dest( 10 );
    auto                input = std::string{ "0123456789" };
 
    auto                                            s = DummySocket( input );
    nds_impl::Socket::BufferedReader< DummySocket > b( s );
 
    auto stride = 4;
    auto end = b.peek( dest.data( ), dest.data( ) + stride );
    REQUIRE( b.__buffered_bytes( ) == input.size( ) );
    REQUIRE( end == dest.data( ) + stride );
    REQUIRE( std::equal( dest.data( ), dest.data( ) + stride, input.data( ) ) );
 
    // clear the buffer
    std::fill( dest.begin( ), dest.end( ), 0 );
    REQUIRE(
        !std::equal( dest.data( ), dest.data( ) + stride, input.data( ) ) );
 
    end = b.read_exactly( dest.data( ), dest.data( ) + stride );
    REQUIRE( b.__buffered_bytes( ) == input.size( ) - stride );
    REQUIRE( end == dest.data( ) + stride );
    REQUIRE( std::equal( dest.data( ), dest.data( ) + stride, input.data( ) ) );
}
 
TEST_CASE( "You can ask a buffered read to read until a sequence is found",
           "[buffered,read_until]" )
{
    using namespace nds_testing;
 
    auto input = std::string{ "0123456789" };
    auto s = DummySocket( input );
    nds_impl::Socket::BufferedReader< DummySocket > b( s );
 
    auto terminators = std::string( "59" );
    auto result = b.read_until( terminators.begin( ), terminators.end( ) );
    REQUIRE( std::string( result.begin( ), result.end( ) ) ==
             std::string( "012345" ) );
}
 
TEST_CASE( "You can ask a buffered read to read until a sequence is found, "
           "with a bound",
           "[buffered,read_until]" )
{
    using namespace nds_testing;
 
    std::vector< char > dest( 5 );
    auto                input = std::string{ "0123456789abcdef" };
    auto                s = DummySocket( input );
    nds_impl::Socket::BufferedReader< DummySocket > b( s );
 
    std::fill( dest.begin( ), dest.end( ), 0 );
    auto terminators = std::string( "29" );
    auto end = b.read_until(
        terminators.begin( ), terminators.end( ), dest.begin( ), dest.end( ) );
    auto length = std::distance( dest.begin( ), end );
    REQUIRE( length == 3 );
    REQUIRE( std::string( dest.data( ), length ) == std::string( "012" ) );
}
 
TEST_CASE( "A bounded buffered read_until will throw an exception if it cannot "
           "find a sequence",
           "[buffered,read_until]" )
{
    using namespace nds_testing;
 
    std::vector< char > dest( 5 );
    auto                input = std::string{ "0123456789abcdef" };
    auto                s = DummySocket( input );
    nds_impl::Socket::BufferedReader< DummySocket > b( s );
 
    auto terminators = std::string( "Z" );
    REQUIRE_THROWS_AS( b.read_until( terminators.begin( ),
                                     terminators.end( ),
                                     dest.begin( ),
                                     dest.end( ) ),
                       std::range_error );
}
 
TEST_CASE( "A bounded buffered read_until will throw an exception if it cannot "
           "find a sequence when input is empty",
           "[buffered,read_until]" )
{
    using namespace nds_testing;
 
    std::vector< char >                             dest( 5 );
    auto                                            input = std::string{ "" };
    auto                                            s = DummySocket( input );
    nds_impl::Socket::BufferedReader< DummySocket > b( s );
 
    auto terminators = std::string( "Z" );
    REQUIRE_THROWS_AS( b.read_until( terminators.begin( ),
                                     terminators.end( ),
                                     dest.begin( ),
                                     dest.end( ) ),
                       std::runtime_error );
}
 
TEST_CASE( "Create an owning buffered socket", "[buffered,create]" )
{
    nds_impl::Socket::FullSocket                                           s1;
    nds_impl::Socket::OwningBufferedReader< nds_impl::Socket::FullSocket > b1{
        std::move( s1 )
    };
}
 
TEST_CASE( "You can write through a owning buffered reader",
           "[buffered,write]" )
{
    using namespace nds_testing;
 
    std::vector< char >                                            buf;
    RecordingDummySocket                                           s( buf );
    nds_impl::Socket::OwningBufferedReader< RecordingDummySocket > b{ std::move(
        s ) };
 
    auto hello = std::string( "hello" );
    b.write_all( hello.data( ), hello.data( ) + hello.length( ) );
    REQUIRE( std::string( buf.data( ), buf.size( ) ) == hello );
 
    auto world = std::string( " world!" );
    b.write_all( world.data( ), world.data( ) + world.length( ) );
    REQUIRE( std::string( buf.data( ), buf.size( ) ) == hello + world );
}
 
TEST_CASE( "You can read from a owning buffered writter", "[buffered,read]" )
{
    using namespace nds_testing;
 
    std::vector< char > dest( 20 );
 
    auto hello = std::string{ "hello world!" };
    auto s = DummySocket( hello );
 
    nds_impl::Socket::OwningBufferedReader< DummySocket > b( std::move( s ) );
 
    REQUIRE( b.__buffered_bytes( ) == 0 );
 
    auto end = b.read_available( dest.data( ), dest.data( ) + dest.size( ) );
    REQUIRE( b.__buffered_bytes( ) == 0 );
    REQUIRE( end != dest.data( ) );
    REQUIRE( end == dest.data( ) + hello.size( ) );
    auto output = std::string( dest.data( ), end );
    REQUIRE( output == hello );
}
 
TEST_CASE(
    "You can read the data in small segments from an owning buffered reader",
    "[buffered,read]" )
{
    using namespace nds_testing;
 
    std::vector< char > dest( 5 );
 
    auto hello = std::string{ "hello world!" };
    auto s = DummySocket( hello );
 
    nds_impl::Socket::OwningBufferedReader< DummySocket > b( std::move( s ) );
 
    REQUIRE( b.__buffered_bytes( ) == 0 );
 
    auto end = b.read_available( dest.data( ), dest.data( ) + dest.size( ) );
    REQUIRE( b.__buffered_bytes( ) == ( hello.size( ) - dest.size( ) ) );
    REQUIRE( end != dest.data( ) );
    REQUIRE( end == dest.data( ) + dest.size( ) );
    auto output = std::string( dest.data( ), end );
    REQUIRE( output == hello.substr( 0, dest.size( ) ) );
 
    end = b.read_available( dest.data( ), dest.data( ) + dest.size( ) );
    REQUIRE( b.__buffered_bytes( ) == ( hello.size( ) - 2 * dest.size( ) ) );
    REQUIRE( end != dest.data( ) );
    REQUIRE( end == dest.data( ) + dest.size( ) );
    output = std::string( dest.data( ), end );
    REQUIRE( output == hello.substr( dest.size( ), dest.size( ) ) );
}
 
TEST_CASE( "You can ask for an exact amount of bytes to be returned from an "
           "owning buffered reader",
           "[buffered,read]" )
{
    using namespace nds_testing;
 
    std::vector< char > dest( 10 );
 
    auto input = std::string{ "0123456789" };
    auto s = DummySocket( input );
    nds_impl::Socket::OwningBufferedReader< DummySocket > b( std::move( s ) );
 
    auto stride = 4;
    auto end = b.read_exactly( dest.data( ), dest.data( ) + stride );
    REQUIRE( b.__buffered_bytes( ) == ( input.size( ) - stride ) );
    REQUIRE( end == dest.data( ) + stride );
 
    REQUIRE_THROWS(
        b.read_exactly( dest.data( ), dest.data( ) + dest.size( ) ) );
}
 
TEST_CASE(
    "You can ask an owning buffered read to read until a sequence is found",
    "[buffered,read_until]" )
{
    using namespace nds_testing;
 
    std::vector< char > dest( 10 );
 
    auto input = std::string{ "0123456789" };
    auto s = DummySocket( input );
    nds_impl::Socket::OwningBufferedReader< DummySocket > b( std::move( s ) );
 
    auto terminators = std::string( "59" );
    auto result = b.read_until( terminators.begin( ), terminators.end( ) );
    REQUIRE( std::string( result.begin( ), result.end( ) ) ==
             std::string( "012345" ) );
}
 
#endif // _NDS_IMPL_ENABLE_CATCH_TESTS_
 
#endif // NDS_PROXY_BUFFERED_READER_HH