//
|
// Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com)
|
//
|
// Distributed under the Boost Software License, Version 1.0. (See accompanying
|
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
|
//
|
// Official repository: https://github.com/boostorg/beast
|
//
|
|
#ifndef BOOST_BEAST_TEST_IMPL_STREAM_IPP
|
#define BOOST_BEAST_TEST_IMPL_STREAM_IPP
|
|
#include <boost/beast/_experimental/test/stream.hpp>
|
#include <boost/beast/core/bind_handler.hpp>
|
#include <boost/beast/core/buffer_traits.hpp>
|
#include <boost/make_shared.hpp>
|
#include <stdexcept>
|
#include <vector>
|
|
namespace boost {
|
namespace beast {
|
namespace test {
|
|
//------------------------------------------------------------------------------
|
|
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,
|
std::size_t buf_size)
|
{
|
std::unique_lock<std::mutex> lock(in_->m);
|
|
++in_->nread;
|
if(in_->op != nullptr)
|
BOOST_THROW_EXCEPTION(
|
std::logic_error{"in_->op != nullptr"});
|
|
// test failure
|
error_code ec;
|
if(in_->fc && in_->fc->fail(ec))
|
{
|
lock.unlock();
|
(*op)(ec);
|
return;
|
}
|
|
// A request to read 0 bytes from a stream is a no-op.
|
if(buf_size == 0 || buffer_bytes(in_->b.data()) > 0)
|
{
|
lock.unlock();
|
(*op)(ec);
|
return;
|
}
|
|
// deliver error
|
if(in_->code != status::ok)
|
{
|
lock.unlock();
|
(*op)(net::error::eof);
|
return;
|
}
|
|
// complete when bytes available or closed
|
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()
|
{
|
close();
|
in_->remove();
|
}
|
|
stream::
|
stream(stream&& other)
|
{
|
auto in = service::make_impl(
|
other.in_->ioc, other.in_->fc);
|
in_ = std::move(other.in_);
|
out_ = std::move(other.out_);
|
other.in_ = in;
|
}
|
|
stream&
|
stream::
|
operator=(stream&& other)
|
{
|
close();
|
auto in = service::make_impl(
|
other.in_->ioc, other.in_->fc);
|
in_->remove();
|
in_ = std::move(other.in_);
|
out_ = std::move(other.out_);
|
other.in_ = in;
|
return *this;
|
}
|
|
//------------------------------------------------------------------------------
|
|
stream::
|
stream(net::io_context& ioc)
|
: in_(service::make_impl(ioc, nullptr))
|
{
|
}
|
|
stream::
|
stream(
|
net::io_context& ioc,
|
fail_count& fc)
|
: in_(service::make_impl(ioc, &fc))
|
{
|
}
|
|
stream::
|
stream(
|
net::io_context& ioc,
|
string_view s)
|
: in_(service::make_impl(ioc, nullptr))
|
{
|
in_->b.commit(net::buffer_copy(
|
in_->b.prepare(s.size()),
|
net::buffer(s.data(), s.size())));
|
}
|
|
stream::
|
stream(
|
net::io_context& ioc,
|
fail_count& fc,
|
string_view s)
|
: in_(service::make_impl(ioc, &fc))
|
{
|
in_->b.commit(net::buffer_copy(
|
in_->b.prepare(s.size()),
|
net::buffer(s.data(), s.size())));
|
}
|
|
void
|
stream::
|
connect(stream& remote)
|
{
|
BOOST_ASSERT(! out_.lock());
|
BOOST_ASSERT(! remote.out_.lock());
|
std::lock(in_->m, remote.in_->m);
|
std::lock_guard<std::mutex> guard1{in_->m, std::adopt_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;
|
}
|
|
string_view
|
stream::
|
str() const
|
{
|
auto const bs = in_->b.data();
|
if(buffer_bytes(bs) == 0)
|
return {};
|
net::const_buffer const b = *net::buffer_sequence_begin(bs);
|
return {static_cast<char const*>(b.data()), b.size()};
|
}
|
|
void
|
stream::
|
append(string_view s)
|
{
|
std::lock_guard<std::mutex> lock{in_->m};
|
in_->b.commit(net::buffer_copy(
|
in_->b.prepare(s.size()),
|
net::buffer(s.data(), s.size())));
|
}
|
|
void
|
stream::
|
clear()
|
{
|
std::lock_guard<std::mutex> lock{in_->m};
|
in_->b.consume(in_->b.size());
|
}
|
|
void
|
stream::
|
close()
|
{
|
in_->cancel_read();
|
|
// disconnect
|
{
|
auto out = out_.lock();
|
out_.reset();
|
|
// notify peer
|
if(out)
|
{
|
std::lock_guard<std::mutex> lock(out->m);
|
if(out->code == status::ok)
|
{
|
out->code = status::eof;
|
out->notify_read();
|
}
|
}
|
}
|
}
|
|
void
|
stream::
|
close_remote()
|
{
|
std::lock_guard<std::mutex> lock{in_->m};
|
if(in_->code == status::ok)
|
{
|
in_->code = status::eof;
|
in_->notify_read();
|
}
|
}
|
|
void
|
teardown(
|
role_type,
|
stream& s,
|
boost::system::error_code& ec)
|
{
|
if( s.in_->fc &&
|
s.in_->fc->fail(ec))
|
return;
|
|
s.close();
|
|
if( s.in_->fc &&
|
s.in_->fc->fail(ec))
|
ec = net::error::eof;
|
else
|
ec = {};
|
}
|
|
//------------------------------------------------------------------------------
|
|
stream
|
connect(stream& 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)
|
from.connect(to);
|
return from;
|
}
|
|
void
|
connect(stream& s1, stream& s2)
|
{
|
s1.connect(s2);
|
}
|
|
} // test
|
} // beast
|
} // boost
|
|
#endif
|
