core/impl/buffered_read_stream.hpp

0001 //
0002 // Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com)
0003 //
0004 // Distributed under the Boost Software License, Version 1.0. (See accompanying
0005 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
0006 //
0007 // Official repository: https://github.com/boostorg/beast
0008 //
0009
0010 #ifndef BOOST_BEAST_IMPL_BUFFERED_READ_STREAM_HPP
0011 #define BOOST_BEAST_IMPL_BUFFERED_READ_STREAM_HPP
0012
0013 #include <boost/beast/core/async_base.hpp>
0014 #include <boost/beast/core/error.hpp>
0015 #include <boost/beast/core/read_size.hpp>
0016 #include <boost/beast/core/stream_traits.hpp>
0017 #include <boost/beast/core/detail/is_invocable.hpp>
0018 #include <boost/asio/append.hpp>
0019 #include <boost/asio/dispatch.hpp>
0020 #include <boost/throw_exception.hpp>
0021
0022 namespace boost {
0023 namespace beast {
0024
0025
0026 template<class Stream, class DynamicBuffer>
0027 struct buffered_read_stream<Stream, DynamicBuffer>::ops
0028 {
0029
0030 template<class MutableBufferSequence, class Handler>
0031 class read_op
0032     : public async_base<Handler,
0033     beast::executor_type<buffered_read_stream>>
0034 {
0035     buffered_read_stream& s_;
0036     MutableBufferSequence b_;
0037     int step_ = 0;
0038
0039 public:
0040     read_op(read_op&&) = default;
0041     read_op(read_op const&) = delete;
0042
0043     template<class Handler_>
0044     read_op(
0045         Handler_&& h,
0046         buffered_read_stream& s,
0047         MutableBufferSequence const& b)
0048         : async_base<
0049             Handler, beast::executor_type<buffered_read_stream>>(
0050                 std::forward<Handler_>(h), s.get_executor())
0051         , s_(s)
0052         , b_(b)
0053     {
0054         (*this)({}, 0);
0055     }
0056
0057     void
0058     operator()(
0059         error_code ec,
0060         std::size_t bytes_transferred)
0061     {
0062         // VFALCO TODO Rewrite this using reenter/yield
0063         switch(step_)
0064         {
0065         case 0:
0066             if(s_.buffer_.size() == 0)
0067             {
0068                 if(s_.capacity_ == 0)
0069                 {
0070                     // read (unbuffered)
0071                     step_ = 1;
0072                     return s_.next_layer_.async_read_some(
0073                         b_, std::move(*this));
0074                 }
0075                 // read
0076                 step_ = 2;
0077                 return s_.next_layer_.async_read_some(
0078                     s_.buffer_.prepare(read_size(
0079                         s_.buffer_, s_.capacity_)),
0080                             std::move(*this));
0081             }
0082             step_ = 3;
0083             {
0084                 const auto ex = this->get_immediate_executor();
0085                 return net::dispatch(
0086                     ex,
0087                     net::append(std::move(*this), ec, 0));
0088             }
0089         case 1:
0090             // upcall
0091             break;
0092
0093         case 2:
0094             s_.buffer_.commit(bytes_transferred);
0095             BOOST_FALLTHROUGH;
0096
0097         case 3:
0098             bytes_transferred =
0099                 net::buffer_copy(b_, s_.buffer_.data());
0100             s_.buffer_.consume(bytes_transferred);
0101             break;
0102         }
0103         this->complete_now(ec, bytes_transferred);
0104     }
0105 };
0106
0107 struct run_read_op
0108 {
0109     buffered_read_stream* self;
0110
0111     using executor_type = typename buffered_read_stream::executor_type;
0112
0113     executor_type
0114     get_executor() const noexcept
0115     {
0116         return self->get_executor();
0117     }
0118
0119     template<class ReadHandler, class Buffers>
0120     void
0121     operator()(
0122         ReadHandler&& h,
0123         Buffers const* b)
0124     {
0125         // If you get an error on the following line it means
0126         // that your handler does not meet the documented type
0127         // requirements for the handler.
0128
0129         static_assert(
0130             beast::detail::is_invocable<ReadHandler,
0131             void(error_code, std::size_t)>::value,
0132             "ReadHandler type requirements not met");
0133
0134         read_op<
0135             Buffers,
0136             typename std::decay<ReadHandler>::type>(
0137                 std::forward<ReadHandler>(h), *self, *b);
0138     }
0139 };
0140
0141 };
0142
0143 //------------------------------------------------------------------------------
0144
0145 template<class Stream, class DynamicBuffer>
0146 template<class... Args>
0147 buffered_read_stream<Stream, DynamicBuffer>::
0148 buffered_read_stream(Args&&... args)
0149     : next_layer_(std::forward<Args>(args)...)
0150 {
0151 }
0152
0153 template<class Stream, class DynamicBuffer>
0154 template<class ConstBufferSequence, BOOST_BEAST_ASYNC_TPARAM2 WriteHandler>
0155 BOOST_BEAST_ASYNC_RESULT2(WriteHandler)
0156 buffered_read_stream<Stream, DynamicBuffer>::
0157 async_write_some(
0158     ConstBufferSequence const& buffers,
0159     WriteHandler&& handler)
0160 {
0161     static_assert(is_async_write_stream<next_layer_type>::value,
0162         "AsyncWriteStream type requirements not met");
0163     static_assert(net::is_const_buffer_sequence<
0164         ConstBufferSequence>::value,
0165             "ConstBufferSequence type requirements not met");
0166     static_assert(detail::is_completion_token_for<WriteHandler,
0167         void(error_code, std::size_t)>::value,
0168             "WriteHandler type requirements not met");
0169     return next_layer_.async_write_some(buffers,
0170         std::forward<WriteHandler>(handler));
0171 }
0172
0173 template<class Stream, class DynamicBuffer>
0174 template<class MutableBufferSequence>
0175 std::size_t
0176 buffered_read_stream<Stream, DynamicBuffer>::
0177 read_some(
0178     MutableBufferSequence const& buffers)
0179 {
0180     static_assert(is_sync_read_stream<next_layer_type>::value,
0181         "SyncReadStream type requirements not met");
0182     static_assert(net::is_mutable_buffer_sequence<
0183         MutableBufferSequence>::value,
0184             "MutableBufferSequence type requirements not met");
0185     error_code ec;
0186     auto n = read_some(buffers, ec);
0187     if(ec)
0188         BOOST_THROW_EXCEPTION(system_error{ec});
0189     return n;
0190 }
0191
0192 template<class Stream, class DynamicBuffer>
0193 template<class MutableBufferSequence>
0194 std::size_t
0195 buffered_read_stream<Stream, DynamicBuffer>::
0196 read_some(MutableBufferSequence const& buffers,
0197     error_code& ec)
0198 {
0199     static_assert(is_sync_read_stream<next_layer_type>::value,
0200         "SyncReadStream type requirements not met");
0201     static_assert(net::is_mutable_buffer_sequence<
0202         MutableBufferSequence>::value,
0203             "MutableBufferSequence type requirements not met");
0204     if(buffer_.size() == 0)
0205     {
0206         if(capacity_ == 0)
0207             return next_layer_.read_some(buffers, ec);
0208         buffer_.commit(next_layer_.read_some(
0209             buffer_.prepare(read_size(buffer_,
0210                 capacity_)), ec));
0211         if(ec)
0212             return 0;
0213     }
0214     else
0215     {
0216         ec = {};
0217     }
0218     auto bytes_transferred =
0219         net::buffer_copy(buffers, buffer_.data());
0220     buffer_.consume(bytes_transferred);
0221     return bytes_transferred;
0222 }
0223
0224 template<class Stream, class DynamicBuffer>
0225 template<class MutableBufferSequence, BOOST_BEAST_ASYNC_TPARAM2 ReadHandler>
0226 BOOST_BEAST_ASYNC_RESULT2(ReadHandler)
0227 buffered_read_stream<Stream, DynamicBuffer>::
0228 async_read_some(
0229     MutableBufferSequence const& buffers,
0230     ReadHandler&& handler)
0231 {
0232     static_assert(is_async_read_stream<next_layer_type>::value,
0233         "AsyncReadStream type requirements not met");
0234     static_assert(net::is_mutable_buffer_sequence<
0235         MutableBufferSequence>::value,
0236             "MutableBufferSequence type requirements not met");
0237     if(buffer_.size() == 0 && capacity_ == 0)
0238         return next_layer_.async_read_some(buffers,
0239             std::forward<ReadHandler>(handler));
0240     return net::async_initiate<
0241         ReadHandler,
0242         void(error_code, std::size_t)>(
0243             typename ops::run_read_op{this},
0244             handler,
0245             &buffers);
0246 }
0247
0248 } // beast
0249 } // boost
0250
0251 #endif