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