//------------------------------------------------------------------------------
-stream::
-service::
-service(net::execution_context& ctx)
- : beast::detail::service_base<service>(ctx)
- , sp_(boost::make_shared<service_impl>())
-{
-}
-
-void
-stream::
-service::
-shutdown()
-{
- std::vector<std::unique_ptr<read_op_base>> v;
- std::lock_guard<std::mutex> g1(sp_->m_);
- v.reserve(sp_->v_.size());
- for(auto p : sp_->v_)
- {
- std::lock_guard<std::mutex> g2(p->m);
- v.emplace_back(std::move(p->op));
- p->code = status::eof;
- }
-}
-
-auto
-stream::
-service::
-make_impl(
- net::io_context& ctx,
- test::fail_count* fc) ->
- boost::shared_ptr<state>
-{
- auto& svc = net::use_service<service>(ctx);
- auto sp = boost::make_shared<state>(ctx, svc.sp_, fc);
- std::lock_guard<std::mutex> g(svc.sp_->m_);
- svc.sp_->v_.push_back(sp.get());
- return sp;
-}
-
-void
-stream::
-service_impl::
-remove(state& impl)
-{
- std::lock_guard<std::mutex> g(m_);
- *std::find(
- v_.begin(), v_.end(),
- &impl) = std::move(v_.back());
- v_.pop_back();
-}
-
-//------------------------------------------------------------------------------
-
-void stream::initiate_read(
- boost::shared_ptr<state> const& in_,
- std::unique_ptr<stream::read_op_base>&& op,
+template<class Executor>
+void basic_stream<Executor>::initiate_read(
+ boost::shared_ptr<detail::stream_state> const& in_,
+ std::unique_ptr<detail::stream_read_op_base>&& op,
std::size_t buf_size)
{
std::unique_lock<std::mutex> lock(in_->m);
}
// deliver error
- if(in_->code != status::ok)
+ if(in_->code != detail::stream_status::ok)
{
lock.unlock();
(*op)(net::error::eof);
in_->op = std::move(op);
}
-stream::
-state::
-state(
- net::io_context& ioc_,
- boost::weak_ptr<service_impl> wp_,
- fail_count* fc_)
- : ioc(ioc_)
- , wp(std::move(wp_))
- , fc(fc_)
-{
-}
-
-stream::
-state::
-~state()
-{
- // cancel outstanding read
- if(op != nullptr)
- (*op)(net::error::operation_aborted);
-}
-
-void
-stream::
-state::
-remove() noexcept
-{
- auto sp = wp.lock();
-
- // If this goes off, it means the lifetime of a test::stream object
- // extended beyond the lifetime of the associated execution context.
- BOOST_ASSERT(sp);
-
- sp->remove(*this);
-}
-
-void
-stream::
-state::
-notify_read()
-{
- if(op)
- {
- auto op_ = std::move(op);
- op_->operator()(error_code{});
- }
- else
- {
- cv.notify_all();
- }
-}
-
-void
-stream::
-state::
-cancel_read()
-{
- std::unique_ptr<read_op_base> p;
- {
- std::lock_guard<std::mutex> lock(m);
- code = status::eof;
- p = std::move(op);
- }
- if(p != nullptr)
- (*p)(net::error::operation_aborted);
-}
-
//------------------------------------------------------------------------------
-stream::
-~stream()
+template<class Executor>
+basic_stream<Executor>::
+~basic_stream()
{
close();
in_->remove();
}
-stream::
-stream(stream&& other)
+template<class Executor>
+basic_stream<Executor>::
+basic_stream(basic_stream&& other)
{
- auto in = service::make_impl(
- other.in_->ioc, other.in_->fc);
+ auto in = detail::stream_service::make_impl(
+ other.in_->exec, other.in_->fc);
in_ = std::move(other.in_);
out_ = std::move(other.out_);
other.in_ = in;
}
-stream&
-stream::
-operator=(stream&& other)
+
+template<class Executor>
+basic_stream<Executor>&
+basic_stream<Executor>::
+operator=(basic_stream&& other)
{
close();
- auto in = service::make_impl(
- other.in_->ioc, other.in_->fc);
+ auto in = detail::stream_service::make_impl(
+ other.in_->exec, other.in_->fc);
in_->remove();
in_ = std::move(other.in_);
out_ = std::move(other.out_);
//------------------------------------------------------------------------------
-stream::
-stream(net::io_context& ioc)
- : in_(service::make_impl(ioc, nullptr))
+template<class Executor>
+basic_stream<Executor>::
+basic_stream(executor_type exec)
+ : in_(detail::stream_service::make_impl(std::move(exec), nullptr))
{
}
-stream::
-stream(
+template<class Executor>
+basic_stream<Executor>::
+basic_stream(
net::io_context& ioc,
fail_count& fc)
- : in_(service::make_impl(ioc, &fc))
+ : in_(detail::stream_service::make_impl(ioc.get_executor(), &fc))
{
}
-stream::
-stream(
+template<class Executor>
+basic_stream<Executor>::
+basic_stream(
net::io_context& ioc,
string_view s)
- : in_(service::make_impl(ioc, nullptr))
+ : in_(detail::stream_service::make_impl(ioc.get_executor(), nullptr))
{
in_->b.commit(net::buffer_copy(
in_->b.prepare(s.size()),
net::buffer(s.data(), s.size())));
}
-stream::
-stream(
+template<class Executor>
+basic_stream<Executor>::
+basic_stream(
net::io_context& ioc,
fail_count& fc,
string_view s)
- : in_(service::make_impl(ioc, &fc))
+ : in_(detail::stream_service::make_impl(ioc.get_executor(), &fc))
{
in_->b.commit(net::buffer_copy(
in_->b.prepare(s.size()),
net::buffer(s.data(), s.size())));
}
+template<class Executor>
void
-stream::
-connect(stream& remote)
+basic_stream<Executor>::
+connect(basic_stream& remote)
{
BOOST_ASSERT(! out_.lock());
BOOST_ASSERT(! remote.out_.lock());
std::lock_guard<std::mutex> guard2{remote.in_->m, std::adopt_lock};
out_ = remote.in_;
remote.out_ = in_;
- in_->code = status::ok;
- remote.in_->code = status::ok;
+ in_->code = detail::stream_status::ok;
+ remote.in_->code = detail::stream_status::ok;
}
+template<class Executor>
string_view
-stream::
+basic_stream<Executor>::
str() const
{
auto const bs = in_->b.data();
return {static_cast<char const*>(b.data()), b.size()};
}
+template<class Executor>
void
-stream::
+basic_stream<Executor>::
append(string_view s)
{
std::lock_guard<std::mutex> lock{in_->m};
net::buffer(s.data(), s.size())));
}
+template<class Executor>
void
-stream::
+basic_stream<Executor>::
clear()
{
std::lock_guard<std::mutex> lock{in_->m};
in_->b.consume(in_->b.size());
}
+template<class Executor>
void
-stream::
+basic_stream<Executor>::
close()
{
in_->cancel_read();
if(out)
{
std::lock_guard<std::mutex> lock(out->m);
- if(out->code == status::ok)
+ if(out->code == detail::stream_status::ok)
{
- out->code = status::eof;
+ out->code = detail::stream_status::eof;
out->notify_read();
}
}
}
}
+template<class Executor>
void
-stream::
+basic_stream<Executor>::
close_remote()
{
std::lock_guard<std::mutex> lock{in_->m};
- if(in_->code == status::ok)
+ if(in_->code == detail::stream_status::ok)
{
- in_->code = status::eof;
+ in_->code = detail::stream_status::eof;
in_->notify_read();
}
}
+template<class Executor>
void
teardown(
role_type,
- stream& s,
+ basic_stream<Executor>& s,
boost::system::error_code& ec)
{
if( s.in_->fc &&
//------------------------------------------------------------------------------
-stream
-connect(stream& to)
+template<class Executor>
+basic_stream<Executor>
+connect(basic_stream<Executor>& to)
{
-#if defined(BOOST_ASIO_NO_TS_EXECUTORS)
- stream from{net::query(to.get_executor(), net::execution::context)};
-#else // defined(BOOST_ASIO_NO_TS_EXECUTORS)
- stream from{to.get_executor().context()};
-#endif // defined(BOOST_ASIO_NO_TS_EXECUTORS)
+ basic_stream<Executor> from(to.get_executor());
from.connect(to);
return from;
}
+template<class Executor>
void
-connect(stream& s1, stream& s2)
+connect(basic_stream<Executor>& s1, basic_stream<Executor>& s2)
{
s1.connect(s2);
}
projects / ceph.git / blobdiff