test/detail/stream_state.ipp

0001 //
0002 // Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com)
0003 // Copyright (c) 2020 Richard Hodges (hodges.r@gmail.com)
0004 //
0005 // Distributed under the Boost Software License, Version 1.0. (See accompanying
0006 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
0007 //
0008 // Official repository: https://github.com/boostorg/beast
0009 //
0010
0011 #ifndef BOOST_BEAST_TEST_DETAIL_STREAM_STATE_IPP
0012 #define BOOST_BEAST_TEST_DETAIL_STREAM_STATE_IPP
0013
0014 #include <boost/beast/_experimental/test/error.hpp>
0015 #include <boost/make_shared.hpp>
0016
0017 namespace boost {
0018 namespace beast {
0019 namespace test {
0020
0021 namespace detail {
0022
0023 //------------------------------------------------------------------------------
0024
0025 stream_service::
0026 stream_service(net::execution_context& ctx)
0027     : beast::detail::service_base<stream_service>(ctx)
0028     , sp_(boost::make_shared<stream_service_impl>())
0029 {
0030 }
0031
0032 void
0033 stream_service::
0034 shutdown()
0035 {
0036     std::vector<std::unique_ptr<detail::stream_read_op_base>> v;
0037     std::lock_guard<std::mutex> g1(sp_->m_);
0038     v.reserve(sp_->v_.size());
0039     for(auto p : sp_->v_)
0040     {
0041         std::lock_guard<std::mutex> g2(p->m);
0042         v.emplace_back(std::move(p->op));
0043         p->code = detail::stream_status::eof;
0044     }
0045 }
0046
0047 auto
0048 stream_service::
0049 make_impl(
0050     net::any_io_executor exec,
0051     test::fail_count* fc) ->
0052     boost::shared_ptr<detail::stream_state>
0053 {
0054 #if defined(BOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT)
0055     auto& ctx = exec.context();
0056 #else
0057     auto& ctx = net::query(
0058         exec,
0059         net::execution::context);
0060 #endif
0061     auto& svc = net::use_service<stream_service>(ctx);
0062     auto sp = boost::make_shared<detail::stream_state>(exec, svc.sp_, fc);
0063     std::lock_guard<std::mutex> g(svc.sp_->m_);
0064     svc.sp_->v_.push_back(sp.get());
0065     return sp;
0066 }
0067
0068 //------------------------------------------------------------------------------
0069
0070 void
0071 stream_service_impl::
0072 remove(stream_state& impl)
0073 {
0074     std::lock_guard<std::mutex> g(m_);
0075     *std::find(
0076         v_.begin(), v_.end(),
0077             &impl) = std::move(v_.back());
0078     v_.pop_back();
0079 }
0080
0081 //------------------------------------------------------------------------------
0082
0083 stream_state::
0084 stream_state(
0085     net::any_io_executor exec_,
0086     boost::weak_ptr<stream_service_impl> wp_,
0087     fail_count* fc_)
0088     : exec(std::move(exec_))
0089     , wp(std::move(wp_))
0090     , fc(fc_)
0091 {
0092 }
0093
0094 stream_state::
0095 ~stream_state()
0096 {
0097     // cancel outstanding read
0098     if(op != nullptr)
0099         (*op)(net::error::operation_aborted);
0100 }
0101
0102 void
0103 stream_state::
0104 remove() noexcept
0105 {
0106     auto sp = wp.lock();
0107
0108     // If this goes off, it means the lifetime of a test::stream object
0109     // extended beyond the lifetime of the associated execution context.
0110     BOOST_ASSERT(sp);
0111
0112     sp->remove(*this);
0113 }
0114
0115 void
0116 stream_state::
0117 notify_read()
0118 {
0119     if(op)
0120     {
0121         auto op_ = std::move(op);
0122         op_->operator()(error_code{});
0123     }
0124     else
0125     {
0126         cv.notify_all();
0127     }
0128 }
0129
0130 void
0131 stream_state::
0132 cancel_read()
0133 {
0134     std::unique_ptr<stream_read_op_base> p;
0135     {
0136         std::lock_guard<std::mutex> lock(m);
0137         code = stream_status::eof;
0138         p = std::move(op);
0139     }
0140     if(p != nullptr)
0141         (*p)(net::error::operation_aborted);
0142 }
0143
0144 } // detail
0145
0146 } // test
0147 } // beast
0148 } // boost
0149
0150 #endif // BOOST_BEAST_TEST_DETAIL_STREAM_STATE_IPP