123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377 |
- //
- // 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)
- {
- stream from{to.get_executor().context()};
- from.connect(to);
- return from;
- }
- void
- connect(stream& s1, stream& s2)
- {
- s1.connect(s2);
- }
- } // test
- } // beast
- } // boost
- #endif
|