test/impl/stream.ipp

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_TEST_IMPL_STREAM_IPP
0011 #define BOOST_BEAST_TEST_IMPL_STREAM_IPP
0012
0013 #include <boost/beast/_experimental/test/stream.hpp>
0014 #include <boost/beast/core/bind_handler.hpp>
0015 #include <boost/beast/core/buffer_traits.hpp>
0016 #include <boost/make_shared.hpp>
0017 #include <stdexcept>
0018 #include <vector>
0019
0020 namespace boost {
0021 namespace beast {
0022 namespace test {
0023
0024 //------------------------------------------------------------------------------
0025
0026 template<class Executor>
0027 void basic_stream<Executor>::initiate_read(
0028     boost::shared_ptr<detail::stream_state> const& in_,
0029     std::unique_ptr<detail::stream_read_op_base>&& op,
0030     std::size_t buf_size)
0031 {
0032     std::unique_lock<std::mutex> lock(in_->m);
0033
0034     ++in_->nread;
0035     if(in_->op != nullptr)
0036         BOOST_THROW_EXCEPTION(
0037             std::logic_error{"in_->op != nullptr"});
0038
0039     // test failure
0040     error_code ec;
0041     if(in_->fc && in_->fc->fail(ec))
0042     {
0043         lock.unlock();
0044         (*op)(ec);
0045         return;
0046     }
0047
0048     // A request to read 0 bytes from a stream is a no-op.
0049     if(buf_size == 0 || buffer_bytes(in_->b.data()) > 0)
0050     {
0051         lock.unlock();
0052         (*op)(ec);
0053         return;
0054     }
0055
0056     // deliver error
0057     if(in_->code != detail::stream_status::ok)
0058     {
0059         lock.unlock();
0060         (*op)(net::error::eof);
0061         return;
0062     }
0063
0064     // complete when bytes available or closed
0065     in_->op = std::move(op);
0066 }
0067
0068 //------------------------------------------------------------------------------
0069
0070 template<class Executor>
0071 basic_stream<Executor>::
0072 ~basic_stream()
0073 {
0074     close();
0075     in_->remove();
0076 }
0077
0078 template<class Executor>
0079 basic_stream<Executor>::
0080 basic_stream(basic_stream&& other)
0081 {
0082     auto in = detail::stream_service::make_impl(
0083         other.in_->exec, other.in_->fc);
0084     in_ = std::move(other.in_);
0085     out_ = std::move(other.out_);
0086     other.in_ = in;
0087 }
0088
0089
0090 template<class Executor>
0091 basic_stream<Executor>&
0092 basic_stream<Executor>::
0093 operator=(basic_stream&& other)
0094 {
0095     close();
0096     auto in = detail::stream_service::make_impl(
0097         other.in_->exec, other.in_->fc);
0098     in_->remove();
0099     in_ = std::move(other.in_);
0100     out_ = std::move(other.out_);
0101     other.in_ = in;
0102     return *this;
0103 }
0104
0105 //------------------------------------------------------------------------------
0106
0107 template<class Executor>
0108 basic_stream<Executor>::
0109 basic_stream(executor_type exec)
0110     : in_(detail::stream_service::make_impl(std::move(exec), nullptr))
0111 {
0112 }
0113
0114 template<class Executor>
0115 basic_stream<Executor>::
0116 basic_stream(
0117     net::io_context& ioc,
0118     fail_count& fc)
0119     : in_(detail::stream_service::make_impl(ioc.get_executor(), &fc))
0120 {
0121 }
0122
0123 template<class Executor>
0124 basic_stream<Executor>::
0125 basic_stream(
0126     net::io_context& ioc,
0127     string_view s)
0128     : in_(detail::stream_service::make_impl(ioc.get_executor(), nullptr))
0129 {
0130     in_->b.commit(net::buffer_copy(
0131         in_->b.prepare(s.size()),
0132         net::buffer(s.data(), s.size())));
0133 }
0134
0135 template<class Executor>
0136 basic_stream<Executor>::
0137 basic_stream(
0138     net::io_context& ioc,
0139     fail_count& fc,
0140     string_view s)
0141     : in_(detail::stream_service::make_impl(ioc.get_executor(), &fc))
0142 {
0143     in_->b.commit(net::buffer_copy(
0144         in_->b.prepare(s.size()),
0145         net::buffer(s.data(), s.size())));
0146 }
0147
0148 template<class Executor>
0149 void
0150 basic_stream<Executor>::
0151 connect(basic_stream& remote)
0152 {
0153     BOOST_ASSERT(! out_.lock());
0154     BOOST_ASSERT(! remote.out_.lock());
0155     std::lock(in_->m, remote.in_->m);
0156     std::lock_guard<std::mutex> guard1{in_->m, std::adopt_lock};
0157     std::lock_guard<std::mutex> guard2{remote.in_->m, std::adopt_lock};
0158     out_ = remote.in_;
0159     remote.out_ = in_;
0160     in_->code = detail::stream_status::ok;
0161     remote.in_->code = detail::stream_status::ok;
0162 }
0163
0164 template<class Executor>
0165 string_view
0166 basic_stream<Executor>::
0167 str() const
0168 {
0169     auto const bs = in_->b.data();
0170     if(buffer_bytes(bs) == 0)
0171         return {};
0172     net::const_buffer const b = *net::buffer_sequence_begin(bs);
0173     return {static_cast<char const*>(b.data()), b.size()};
0174 }
0175
0176 template<class Executor>
0177 void
0178 basic_stream<Executor>::
0179 append(string_view s)
0180 {
0181     std::lock_guard<std::mutex> lock{in_->m};
0182     in_->b.commit(net::buffer_copy(
0183         in_->b.prepare(s.size()),
0184         net::buffer(s.data(), s.size())));
0185 }
0186
0187 template<class Executor>
0188 void
0189 basic_stream<Executor>::
0190 clear()
0191 {
0192     std::lock_guard<std::mutex> lock{in_->m};
0193     in_->b.consume(in_->b.size());
0194 }
0195
0196 template<class Executor>
0197 void
0198 basic_stream<Executor>::
0199 close()
0200 {
0201     in_->cancel_read();
0202
0203     // disconnect
0204     {
0205         auto out = out_.lock();
0206         out_.reset();
0207
0208         // notify peer
0209         if(out)
0210         {
0211             std::lock_guard<std::mutex> lock(out->m);
0212             if(out->code == detail::stream_status::ok)
0213             {
0214                 out->code = detail::stream_status::eof;
0215                 out->notify_read();
0216             }
0217         }
0218     }
0219 }
0220
0221 template<class Executor>
0222 void
0223 basic_stream<Executor>::
0224 close_remote()
0225 {
0226     std::lock_guard<std::mutex> lock{in_->m};
0227     if(in_->code == detail::stream_status::ok)
0228     {
0229         in_->code = detail::stream_status::eof;
0230         in_->notify_read();
0231     }
0232 }
0233
0234 template<class Executor>
0235 void
0236 teardown(
0237     role_type,
0238     basic_stream<Executor>& s,
0239     boost::system::error_code& ec)
0240 {
0241     if( s.in_->fc &&
0242         s.in_->fc->fail(ec))
0243         return;
0244
0245     s.close();
0246
0247     if( s.in_->fc &&
0248         s.in_->fc->fail(ec))
0249     {
0250         BOOST_BEAST_ASSIGN_EC(ec, net::error::eof);
0251     }
0252     else
0253         ec = {};
0254 }
0255
0256 //------------------------------------------------------------------------------
0257
0258 template<class Executor>
0259 basic_stream<Executor>
0260 connect(basic_stream<Executor>& to)
0261 {
0262     basic_stream<Executor> from(to.get_executor());
0263     from.connect(to);
0264     return from;
0265 }
0266
0267 template<class Executor>
0268 void
0269 connect(basic_stream<Executor>& s1, basic_stream<Executor>& s2)
0270 {
0271     s1.connect(s2);
0272 }
0273
0274 } // test
0275 } // beast
0276 } // boost
0277
0278 #endif