ceph/src/boost/boost/beast/_experimental/test/detail/stream_state.ipp

   1 //
   2 // Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com)
   3 // Copyright (c) 2020 Richard Hodges (hodges.r@gmail.com)
   4 //
   5 // Distributed under the Boost Software License, Version 1.0. (See accompanying
   6 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
   7 //
   8 // Official repository: https://github.com/boostorg/beast
   9 //
  10
  11 #ifndef BOOST_BEAST_TEST_DETAIL_STREAM_STATE_IPP
  12 #define BOOST_BEAST_TEST_DETAIL_STREAM_STATE_IPP
  13
  14 #include <boost/beast/_experimental/test/error.hpp>
  15 #include <boost/make_shared.hpp>
  16
  17 namespace boost {
  18 namespace beast {
  19 namespace test {
  20
  21 namespace detail {
  22
  23 //------------------------------------------------------------------------------
  24
  25 stream_service::
  26 stream_service(net::execution_context& ctx)
  27     : beast::detail::service_base<stream_service>(ctx)
  28     , sp_(boost::make_shared<stream_service_impl>())
  29 {
  30 }
  31
  32 void
  33 stream_service::
  34 shutdown()
  35 {
  36     std::vector<std::unique_ptr<detail::stream_read_op_base>> v;
  37     std::lock_guard<std::mutex> g1(sp_->m_);
  38     v.reserve(sp_->v_.size());
  39     for(auto p : sp_->v_)
  40     {
  41         std::lock_guard<std::mutex> g2(p->m);
  42         v.emplace_back(std::move(p->op));
  43         p->code = detail::stream_status::eof;
  44     }
  45 }
  46
  47 auto
  48 stream_service::
  49 make_impl(
  50     net::any_io_executor exec,
  51     test::fail_count* fc) ->
  52     boost::shared_ptr<detail::stream_state>
  53 {
  54 #if defined(BOOST_ASIO_USE_TS_EXECUTOR_AS_DEFAULT)
  55     auto& ctx = exec.context();
  56 #else
  57     auto& ctx = net::query(
  58         exec,
  59         net::execution::context);
  60 #endif
  61     auto& svc = net::use_service<stream_service>(ctx);
  62     auto sp = boost::make_shared<detail::stream_state>(exec, svc.sp_, fc);
  63     std::lock_guard<std::mutex> g(svc.sp_->m_);
  64     svc.sp_->v_.push_back(sp.get());
  65     return sp;
  66 }
  67
  68 //------------------------------------------------------------------------------
  69
  70 void
  71 stream_service_impl::
  72 remove(stream_state& impl)
  73 {
  74     std::lock_guard<std::mutex> g(m_);
  75     *std::find(
  76         v_.begin(), v_.end(),
  77             &impl) = std::move(v_.back());
  78     v_.pop_back();
  79 }
  80
  81 //------------------------------------------------------------------------------
  82
  83 stream_state::
  84 stream_state(
  85     net::any_io_executor exec_,
  86     boost::weak_ptr<stream_service_impl> wp_,
  87     fail_count* fc_)
  88     : exec(std::move(exec_))
  89     , wp(std::move(wp_))
  90     , fc(fc_)
  91 {
  92 }
  93
  94 stream_state::
  95 ~stream_state()
  96 {
  97     // cancel outstanding read
  98     if(op != nullptr)
  99         (*op)(net::error::operation_aborted);
 100 }
 101
 102 void
 103 stream_state::
 104 remove() noexcept
 105 {
 106     auto sp = wp.lock();
 107
 108     // If this goes off, it means the lifetime of a test::stream object
 109     // extended beyond the lifetime of the associated execution context.
 110     BOOST_ASSERT(sp);
 111
 112     sp->remove(*this);
 113 }
 114
 115 void
 116 stream_state::
 117 notify_read()
 118 {
 119     if(op)
 120     {
 121         auto op_ = std::move(op);
 122         op_->operator()(error_code{});
 123     }
 124     else
 125     {
 126         cv.notify_all();
 127     }
 128 }
 129
 130 void
 131 stream_state::
 132 cancel_read()
 133 {
 134     std::unique_ptr<stream_read_op_base> p;
 135     {
 136         std::lock_guard<std::mutex> lock(m);
 137         code = stream_status::eof;
 138         p = std::move(op);
 139     }
 140     if(p != nullptr)
 141         (*p)(net::error::operation_aborted);
 142 }
 143
 144 } // detail
 145
 146 } // test
 147 } // beast
 148 } // boost
 149
 150 #endif // BOOST_BEAST_TEST_DETAIL_STREAM_STATE_IPP