core/impl/basic_stream.hpp

0001 //
0002 // Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com)
0003 //
0004 // Distributed under the Boost Software License, Version 1.0. (See accompanying
0005 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
0006 //
0007 // Official repository: https://github.com/boostorg/beast
0008 //
0009
0010 #ifndef BOOST_BEAST_CORE_IMPL_BASIC_STREAM_HPP
0011 #define BOOST_BEAST_CORE_IMPL_BASIC_STREAM_HPP
0012
0013 #include <boost/beast/core/async_base.hpp>
0014 #include <boost/beast/core/buffer_traits.hpp>
0015 #include <boost/beast/core/buffers_prefix.hpp>
0016 #include <boost/beast/websocket/teardown.hpp>
0017 #include <boost/asio/coroutine.hpp>
0018 #include <boost/assert.hpp>
0019 #include <boost/make_shared.hpp>
0020 #include <boost/core/exchange.hpp>
0021 #include <cstdlib>
0022 #include <type_traits>
0023 #include <utility>
0024
0025 namespace boost {
0026 namespace beast {
0027
0028 //------------------------------------------------------------------------------
0029
0030 template<class Protocol, class Executor, class RatePolicy>
0031 template<class... Args>
0032 basic_stream<Protocol, Executor, RatePolicy>::
0033 impl_type::
0034 impl_type(std::false_type, Args&&... args)
0035     : socket(std::forward<Args>(args)...)
0036     , read(ex())
0037     , write(ex())
0038     , timer(ex())
0039 {
0040     reset();
0041 }
0042
0043 template<class Protocol, class Executor, class RatePolicy>
0044 template<class RatePolicy_, class... Args>
0045 basic_stream<Protocol, Executor, RatePolicy>::
0046 impl_type::
0047 impl_type(std::true_type,
0048     RatePolicy_&& policy, Args&&... args)
0049     : boost::empty_value<RatePolicy>(
0050         boost::empty_init_t{},
0051         std::forward<RatePolicy_>(policy))
0052     , socket(std::forward<Args>(args)...)
0053     , read(ex())
0054     , write(ex())
0055     , timer(ex())
0056 {
0057     reset();
0058 }
0059
0060 template<class Protocol, class Executor, class RatePolicy>
0061 template<class Executor2>
0062 void
0063 basic_stream<Protocol, Executor, RatePolicy>::
0064 impl_type::
0065 on_timer(Executor2 const& ex2)
0066 {
0067     BOOST_ASSERT(waiting > 0);
0068
0069     // the last waiter starts the new slice
0070     if(--waiting > 0)
0071         return;
0072
0073     // update the expiration time
0074     BOOST_VERIFY(timer.expires_after(
0075         std::chrono::seconds(1)) == 0);
0076
0077     rate_policy_access::on_timer(policy());
0078
0079     struct handler : boost::empty_value<Executor2>
0080     {
0081         boost::weak_ptr<impl_type> wp;
0082
0083         using executor_type = Executor2;
0084
0085         executor_type
0086         get_executor() const noexcept
0087         {
0088             return this->get();
0089         }
0090
0091         handler(
0092             Executor2 const& ex2,
0093             boost::shared_ptr<impl_type> const& sp)
0094             : boost::empty_value<Executor2>(
0095                 boost::empty_init_t{}, ex2)
0096             , wp(sp)
0097         {
0098         }
0099
0100         void
0101         operator()(error_code ec)
0102         {
0103             auto sp = wp.lock();
0104             if(! sp)
0105                 return;
0106             if(ec == net::error::operation_aborted)
0107                 return;
0108             BOOST_ASSERT(! ec);
0109             if(ec)
0110                 return;
0111             sp->on_timer(this->get());
0112         }
0113     };
0114
0115     // wait on the timer again
0116     ++waiting;
0117     timer.async_wait(handler(ex2, this->shared_from_this()));
0118 }
0119
0120 template<class Protocol, class Executor, class RatePolicy>
0121 void
0122 basic_stream<Protocol, Executor, RatePolicy>::
0123 impl_type::
0124 reset()
0125 {
0126     // If assert goes off, it means that there are
0127     // already read or write (or connect) operations
0128     // outstanding, so there is nothing to apply
0129     // the expiration time to!
0130     //
0131     BOOST_ASSERT(! read.pending || ! write.pending);
0132
0133     if(! read.pending)
0134         BOOST_VERIFY(
0135             read.timer.expires_at(never()) == 0);
0136
0137     if(! write.pending)
0138         BOOST_VERIFY(
0139             write.timer.expires_at(never()) == 0);
0140 }
0141
0142 template<class Protocol, class Executor, class RatePolicy>
0143 void
0144 basic_stream<Protocol, Executor, RatePolicy>::
0145 impl_type::
0146 close() noexcept
0147 {
0148     {
0149         error_code ec;
0150         socket.close(ec);
0151     }
0152 #if !defined(BOOST_NO_EXCEPTIONS)
0153     try
0154     {
0155         timer.cancel();
0156     }
0157     catch(...)
0158     {
0159     }
0160 #else
0161     timer.cancel();
0162 #endif
0163 }
0164
0165 //------------------------------------------------------------------------------
0166
0167 template<class Protocol, class Executor, class RatePolicy>
0168 template<class Executor2>
0169 struct basic_stream<Protocol, Executor, RatePolicy>::
0170     timeout_handler
0171 {
0172     using executor_type = Executor2;
0173
0174     op_state& state;
0175     boost::weak_ptr<impl_type> wp;
0176     tick_type tick;
0177     executor_type ex;
0178
0179     executor_type get_executor() const noexcept
0180     {
0181         return ex;
0182     }
0183
0184     void
0185     operator()(error_code ec)
0186     {
0187         // timer canceled
0188         if(ec == net::error::operation_aborted)
0189             return;
0190         BOOST_ASSERT(! ec);
0191
0192         auto sp = wp.lock();
0193
0194         // stream destroyed
0195         if(! sp)
0196             return;
0197
0198         // stale timer
0199         if(tick < state.tick)
0200             return;
0201         BOOST_ASSERT(tick == state.tick);
0202
0203         // timeout
0204         BOOST_ASSERT(! state.timeout);
0205         sp->close();
0206         state.timeout = true;
0207     }
0208 };
0209
0210 //------------------------------------------------------------------------------
0211
0212 template<class Protocol, class Executor, class RatePolicy>
0213 struct basic_stream<Protocol, Executor, RatePolicy>::ops
0214 {
0215
0216 template<bool isRead, class Buffers, class Handler>
0217 class transfer_op
0218     : public async_base<Handler, Executor>
0219     , public boost::asio::coroutine
0220 {
0221     boost::shared_ptr<impl_type> impl_;
0222     pending_guard pg_;
0223     Buffers b_;
0224
0225     using is_read = std::integral_constant<bool, isRead>;
0226
0227     op_state&
0228     state()
0229     {
0230         if (isRead)
0231             return impl_->read;
0232         else
0233             return impl_->write;
0234     }
0235
0236     std::size_t
0237     available_bytes()
0238     {
0239         if (isRead)
0240             return rate_policy_access::
0241                 available_read_bytes(impl_->policy());
0242         else
0243             return rate_policy_access::
0244                 available_write_bytes(impl_->policy());
0245     }
0246
0247     void
0248     transfer_bytes(std::size_t n)
0249     {
0250         if (isRead)
0251             rate_policy_access::
0252                 transfer_read_bytes(impl_->policy(), n);
0253         else
0254             rate_policy_access::
0255                 transfer_write_bytes(impl_->policy(), n);
0256     }
0257
0258     void
0259     async_perform(
0260         std::size_t amount, std::true_type)
0261     {
0262         impl_->socket.async_read_some(
0263             beast::buffers_prefix(amount, b_),
0264                 std::move(*this));
0265     }
0266
0267     void
0268     async_perform(
0269         std::size_t amount, std::false_type)
0270     {
0271         impl_->socket.async_write_some(
0272             beast::buffers_prefix(amount, b_),
0273                 std::move(*this));
0274     }
0275
0276     static bool never_pending_;
0277
0278 public:
0279     template<class Handler_>
0280     transfer_op(
0281         Handler_&& h,
0282         basic_stream& s,
0283         Buffers const& b)
0284         : async_base<Handler, Executor>(
0285             std::forward<Handler_>(h), s.get_executor())
0286         , impl_(s.impl_)
0287         , pg_()
0288         , b_(b)
0289     {
0290         this->set_allowed_cancellation(net::cancellation_type::all);
0291         if (buffer_bytes(b_) == 0 && state().pending)
0292         {
0293             // Workaround:
0294             // Corner case discovered in https://github.com/boostorg/beast/issues/2065
0295             // Enclosing SSL stream wishes to complete a 0-length write early by
0296             // executing a 0-length read against the underlying stream.
0297             // This can occur even if an existing async_read is in progress.
0298             // In this specific case, we will complete the async op with no error
0299             // in order to prevent assertions and/or internal corruption of the basic_stream
0300             this->complete(false, error_code(), std::size_t{0});
0301         }
0302         else
0303         {
0304             pg_.assign(state().pending);
0305             (*this)({});
0306         }
0307     }
0308
0309     void
0310     operator()(
0311         error_code ec,
0312         std::size_t bytes_transferred = 0)
0313     {
0314         BOOST_ASIO_CORO_REENTER(*this)
0315         {
0316             // handle empty buffers
0317             if(detail::buffers_empty(b_))
0318             {
0319                 // make sure we perform the no-op
0320                 BOOST_ASIO_CORO_YIELD
0321                 {
0322                     BOOST_ASIO_HANDLER_LOCATION((
0323                         __FILE__, __LINE__,
0324                         (isRead ? "basic_stream::async_read_some"
0325                             : "basic_stream::async_write_some")));
0326
0327                     async_perform(0, is_read{});
0328                 }
0329                 // apply the timeout manually, otherwise
0330                 // behavior varies across platforms.
0331                 if(state().timer.expiry() <= clock_type::now())
0332                 {
0333                     impl_->close();
0334                     BOOST_BEAST_ASSIGN_EC(ec, beast::error::timeout);
0335                 }
0336                 goto upcall;
0337             }
0338
0339             // if a timeout is active, wait on the timer
0340             if(state().timer.expiry() != never())
0341             {
0342                 BOOST_ASIO_HANDLER_LOCATION((
0343                     __FILE__, __LINE__,
0344                     (isRead ? "basic_stream::async_read_some"
0345                         : "basic_stream::async_write_some")));
0346
0347                 state().timer.async_wait(
0348                     timeout_handler<decltype(this->get_executor())>{
0349                         state(),
0350                         impl_,
0351                         state().tick,
0352                         this->get_executor()});
0353             }
0354
0355             // check rate limit, maybe wait
0356             std::size_t amount;
0357             amount = available_bytes();
0358             if(amount == 0)
0359             {
0360                 ++impl_->waiting;
0361                 BOOST_ASIO_CORO_YIELD
0362                 {
0363                     BOOST_ASIO_HANDLER_LOCATION((
0364                         __FILE__, __LINE__,
0365                         (isRead ? "basic_stream::async_read_some"
0366                             : "basic_stream::async_write_some")));
0367
0368                     impl_->timer.async_wait(std::move(*this));
0369                 }
0370                 if(ec)
0371                 {
0372                     // socket was closed, or a timeout
0373                     BOOST_ASSERT(ec ==
0374                         net::error::operation_aborted);
0375                     // timeout handler invoked?
0376                     if(state().timeout)
0377                     {
0378                         // yes, socket already closed
0379                         BOOST_BEAST_ASSIGN_EC(ec, beast::error::timeout);
0380                         state().timeout = false;
0381                     }
0382                     goto upcall;
0383                 }
0384                 impl_->on_timer(this->get_executor());
0385
0386                 // Allow at least one byte, otherwise
0387                 // bytes_transferred could be 0.
0388                 amount = std::max<std::size_t>(
0389                     available_bytes(), 1);
0390             }
0391
0392             BOOST_ASIO_CORO_YIELD
0393             {
0394                 BOOST_ASIO_HANDLER_LOCATION((
0395                     __FILE__, __LINE__,
0396                     (isRead ? "basic_stream::async_read_some"
0397                         : "basic_stream::async_write_some")));
0398
0399                 async_perform(amount, is_read{});
0400             }
0401
0402             if(state().timer.expiry() != never())
0403             {
0404                 ++state().tick;
0405
0406                 // try cancelling timer
0407                 auto const n =
0408                     state().timer.cancel();
0409                 if(n == 0)
0410                 {
0411                     // timeout handler invoked?
0412                     if(state().timeout)
0413                     {
0414                         // yes, socket already closed
0415                         BOOST_BEAST_ASSIGN_EC(ec, beast::error::timeout);
0416                         state().timeout = false;
0417                     }
0418                 }
0419                 else
0420                 {
0421                     BOOST_ASSERT(n == 1);
0422                     BOOST_ASSERT(! state().timeout);
0423                 }
0424             }
0425
0426         upcall:
0427             pg_.reset();
0428             transfer_bytes(bytes_transferred);
0429             this->complete_now(ec, bytes_transferred);
0430         }
0431     }
0432 };
0433
0434 template<class Handler>
0435 class connect_op
0436     : public async_base<Handler, Executor>
0437 {
0438     boost::shared_ptr<impl_type> impl_;
0439     pending_guard pg0_;
0440     pending_guard pg1_;
0441
0442     op_state&
0443     state() noexcept
0444     {
0445         return impl_->write;
0446     }
0447
0448 public:
0449     template<class Handler_>
0450     connect_op(
0451         Handler_&& h,
0452         basic_stream& s,
0453         endpoint_type ep)
0454         : async_base<Handler, Executor>(
0455             std::forward<Handler_>(h), s.get_executor())
0456         , impl_(s.impl_)
0457         , pg0_(impl_->read.pending)
0458         , pg1_(impl_->write.pending)
0459     {
0460         this->set_allowed_cancellation(net::cancellation_type::all);
0461         if(state().timer.expiry() != stream_base::never())
0462         {
0463             BOOST_ASIO_HANDLER_LOCATION((
0464                 __FILE__, __LINE__,
0465                 "basic_stream::async_connect"));
0466
0467             impl_->write.timer.async_wait(
0468                 timeout_handler<decltype(this->get_executor())>{
0469                     state(),
0470                     impl_,
0471                     state().tick,
0472                     this->get_executor()});
0473         }
0474
0475         BOOST_ASIO_HANDLER_LOCATION((
0476             __FILE__, __LINE__,
0477             "basic_stream::async_connect"));
0478
0479         impl_->socket.async_connect(
0480             ep, std::move(*this));
0481         // *this is now moved-from
0482     }
0483
0484     template<
0485         class Endpoints, class Condition,
0486         class Handler_>
0487     connect_op(
0488         Handler_&& h,
0489         basic_stream& s,
0490         Endpoints const& eps,
0491         Condition const& cond)
0492         : async_base<Handler, Executor>(
0493             std::forward<Handler_>(h), s.get_executor())
0494         , impl_(s.impl_)
0495         , pg0_(impl_->read.pending)
0496         , pg1_(impl_->write.pending)
0497     {
0498         this->set_allowed_cancellation(net::cancellation_type::all);
0499         if(state().timer.expiry() != stream_base::never())
0500         {
0501             BOOST_ASIO_HANDLER_LOCATION((
0502                 __FILE__, __LINE__,
0503                 "basic_stream::async_connect"));
0504
0505             impl_->write.timer.async_wait(
0506                 timeout_handler<decltype(this->get_executor())>{
0507                     state(),
0508                     impl_,
0509                     state().tick,
0510                     this->get_executor()});
0511         }
0512
0513         BOOST_ASIO_HANDLER_LOCATION((
0514             __FILE__, __LINE__,
0515             "basic_stream::async_connect"));
0516
0517         net::async_connect(impl_->socket,
0518             eps, cond, std::move(*this));
0519         // *this is now moved-from
0520     }
0521
0522     template<
0523         class Iterator, class Condition,
0524         class Handler_>
0525     connect_op(
0526         Handler_&& h,
0527         basic_stream& s,
0528         Iterator begin, Iterator end,
0529         Condition const& cond)
0530         : async_base<Handler, Executor>(
0531             std::forward<Handler_>(h), s.get_executor())
0532         , impl_(s.impl_)
0533         , pg0_(impl_->read.pending)
0534         , pg1_(impl_->write.pending)
0535     {
0536         this->set_allowed_cancellation(net::cancellation_type::all);
0537         if(state().timer.expiry() != stream_base::never())
0538         {
0539             BOOST_ASIO_HANDLER_LOCATION((
0540                 __FILE__, __LINE__,
0541                 "basic_stream::async_connect"));
0542
0543             impl_->write.timer.async_wait(
0544                 timeout_handler<decltype(this->get_executor())>{
0545                     state(),
0546                     impl_,
0547                     state().tick,
0548                     this->get_executor()});
0549         }
0550
0551         BOOST_ASIO_HANDLER_LOCATION((
0552             __FILE__, __LINE__,
0553             "basic_stream::async_connect"));
0554
0555         net::async_connect(impl_->socket,
0556             begin, end, cond, std::move(*this));
0557         // *this is now moved-from
0558     }
0559
0560     template<class... Args>
0561     void
0562     operator()(error_code ec, Args&&... args)
0563     {
0564         if(state().timer.expiry() != stream_base::never())
0565         {
0566             ++state().tick;
0567
0568             // try cancelling timer
0569             auto const n =
0570                 impl_->write.timer.cancel();
0571             if(n == 0)
0572             {
0573                 // timeout handler invoked?
0574                 if(state().timeout)
0575                 {
0576                     // yes, socket already closed
0577                     BOOST_BEAST_ASSIGN_EC(ec, beast::error::timeout);
0578                     state().timeout = false;
0579                 }
0580             }
0581             else
0582             {
0583                 BOOST_ASSERT(n == 1);
0584                 BOOST_ASSERT(! state().timeout);
0585             }
0586         }
0587
0588         pg0_.reset();
0589         pg1_.reset();
0590         this->complete_now(ec, std::forward<Args>(args)...);
0591     }
0592 };
0593
0594 struct run_read_op
0595 {
0596     template<class ReadHandler, class Buffers>
0597     void
0598     operator()(
0599         ReadHandler&& h,
0600         basic_stream* s,
0601         Buffers const& b)
0602     {
0603         // If you get an error on the following line it means
0604         // that your handler does not meet the documented type
0605         // requirements for the handler.
0606
0607         static_assert(
0608             detail::is_invocable<ReadHandler,
0609                 void(error_code, std::size_t)>::value,
0610             "ReadHandler type requirements not met");
0611
0612         transfer_op<
0613             true,
0614             Buffers,
0615             typename std::decay<ReadHandler>::type>(
0616                 std::forward<ReadHandler>(h), *s, b);
0617     }
0618 };
0619
0620 struct run_write_op
0621 {
0622     template<class WriteHandler, class Buffers>
0623     void
0624     operator()(
0625         WriteHandler&& h,
0626         basic_stream* s,
0627         Buffers const& b)
0628     {
0629         // If you get an error on the following line it means
0630         // that your handler does not meet the documented type
0631         // requirements for the handler.
0632
0633         static_assert(
0634             detail::is_invocable<WriteHandler,
0635                 void(error_code, std::size_t)>::value,
0636             "WriteHandler type requirements not met");
0637
0638         transfer_op<
0639             false,
0640             Buffers,
0641             typename std::decay<WriteHandler>::type>(
0642                 std::forward<WriteHandler>(h), *s, b);
0643     }
0644 };
0645
0646 struct run_connect_op
0647 {
0648     template<class ConnectHandler>
0649     void
0650     operator()(
0651         ConnectHandler&& h,
0652         basic_stream* s,
0653         endpoint_type const& ep)
0654     {
0655         // If you get an error on the following line it means
0656         // that your handler does not meet the documented type
0657         // requirements for the handler.
0658
0659         static_assert(
0660             detail::is_invocable<ConnectHandler,
0661                 void(error_code)>::value,
0662             "ConnectHandler type requirements not met");
0663
0664         connect_op<typename std::decay<ConnectHandler>::type>(
0665             std::forward<ConnectHandler>(h), *s, ep);
0666     }
0667 };
0668
0669 struct run_connect_range_op
0670 {
0671     template<
0672         class RangeConnectHandler,
0673         class EndpointSequence,
0674         class Condition>
0675     void
0676     operator()(
0677         RangeConnectHandler&& h,
0678         basic_stream* s,
0679         EndpointSequence const& eps,
0680         Condition const& cond)
0681     {
0682         // If you get an error on the following line it means
0683         // that your handler does not meet the documented type
0684         // requirements for the handler.
0685
0686         static_assert(
0687             detail::is_invocable<RangeConnectHandler,
0688                 void(error_code, typename Protocol::endpoint)>::value,
0689             "RangeConnectHandler type requirements not met");
0690
0691         connect_op<typename std::decay<RangeConnectHandler>::type>(
0692             std::forward<RangeConnectHandler>(h), *s, eps, cond);
0693     }
0694 };
0695
0696 struct run_connect_iter_op
0697 {
0698     template<
0699         class IteratorConnectHandler,
0700         class Iterator,
0701         class Condition>
0702     void
0703     operator()(
0704         IteratorConnectHandler&& h,
0705         basic_stream* s,
0706         Iterator begin, Iterator end,
0707         Condition const& cond)
0708     {
0709         // If you get an error on the following line it means
0710         // that your handler does not meet the documented type
0711         // requirements for the handler.
0712
0713         static_assert(
0714             detail::is_invocable<IteratorConnectHandler,
0715                 void(error_code, Iterator)>::value,
0716             "IteratorConnectHandler type requirements not met");
0717
0718         connect_op<typename std::decay<IteratorConnectHandler>::type>(
0719             std::forward<IteratorConnectHandler>(h), *s, begin, end, cond);
0720     }
0721 };
0722
0723 };
0724
0725 //------------------------------------------------------------------------------
0726
0727 template<class Protocol, class Executor, class RatePolicy>
0728 basic_stream<Protocol, Executor, RatePolicy>::
0729 ~basic_stream()
0730 {
0731     // the shared object can outlive *this,
0732     // cancel any operations so the shared
0733     // object is destroyed as soon as possible.
0734     impl_->close();
0735 }
0736
0737 template<class Protocol, class Executor, class RatePolicy>
0738 template<class Arg0, class... Args, class>
0739 basic_stream<Protocol, Executor, RatePolicy>::
0740 basic_stream(Arg0&& arg0, Args&&... args)
0741     : impl_(boost::make_shared<impl_type>(
0742         std::false_type{},
0743         std::forward<Arg0>(arg0),
0744         std::forward<Args>(args)...))
0745 {
0746 }
0747
0748 template<class Protocol, class Executor, class RatePolicy>
0749 template<class RatePolicy_, class Arg0, class... Args, class>
0750 basic_stream<Protocol, Executor, RatePolicy>::
0751 basic_stream(
0752     RatePolicy_&& policy, Arg0&& arg0, Args&&... args)
0753     : impl_(boost::make_shared<impl_type>(
0754         std::true_type{},
0755         std::forward<RatePolicy_>(policy),
0756         std::forward<Arg0>(arg0),
0757         std::forward<Args>(args)...))
0758 {
0759 }
0760
0761 template<class Protocol, class Executor, class RatePolicy>
0762 basic_stream<Protocol, Executor, RatePolicy>::
0763 basic_stream(basic_stream&& other)
0764     : impl_(boost::make_shared<impl_type>(
0765         std::move(*other.impl_)))
0766 {
0767     // Explainer: Asio's sockets provide the guarantee that a moved-from socket
0768     // will be in a state as-if newly created. i.e.:
0769     // * having the same (valid) executor
0770     // * the socket shall not be open
0771     // We provide the same guarantee by moving the impl rather than the pointer
0772     // controlling its lifetime.
0773 }
0774
0775 template<class Protocol, class Executor, class RatePolicy>
0776 template<class Executor_>
0777 basic_stream<Protocol, Executor, RatePolicy>::
0778 basic_stream(basic_stream<Protocol, Executor_, RatePolicy> && other)
0779     : impl_(boost::make_shared<impl_type>(std::false_type{}, std::move(other.impl_->socket)))
0780 {
0781 }
0782
0783 //------------------------------------------------------------------------------
0784
0785 template<class Protocol, class Executor, class RatePolicy>
0786 auto
0787 basic_stream<Protocol, Executor, RatePolicy>::
0788 release_socket() ->
0789     socket_type
0790 {
0791     this->cancel();
0792     return std::move(impl_->socket);
0793 }
0794
0795 template<class Protocol, class Executor, class RatePolicy>
0796 void
0797 basic_stream<Protocol, Executor, RatePolicy>::
0798 expires_after(net::steady_timer::duration expiry_time)
0799 {
0800     // If assert goes off, it means that there are
0801     // already read or write (or connect) operations
0802     // outstanding, so there is nothing to apply
0803     // the expiration time to!
0804     //
0805     BOOST_ASSERT(
0806         ! impl_->read.pending ||
0807         ! impl_->write.pending);
0808
0809     if(! impl_->read.pending)
0810         BOOST_VERIFY(
0811             impl_->read.timer.expires_after(
0812                 expiry_time) == 0);
0813
0814     if(! impl_->write.pending)
0815         BOOST_VERIFY(
0816             impl_->write.timer.expires_after(
0817                 expiry_time) == 0);
0818 }
0819
0820 template<class Protocol, class Executor, class RatePolicy>
0821 void
0822 basic_stream<Protocol, Executor, RatePolicy>::
0823 expires_at(
0824     net::steady_timer::time_point expiry_time)
0825 {
0826     // If assert goes off, it means that there are
0827     // already read or write (or connect) operations
0828     // outstanding, so there is nothing to apply
0829     // the expiration time to!
0830     //
0831     BOOST_ASSERT(
0832         ! impl_->read.pending ||
0833         ! impl_->write.pending);
0834
0835     if(! impl_->read.pending)
0836         BOOST_VERIFY(
0837             impl_->read.timer.expires_at(
0838                 expiry_time) == 0);
0839
0840     if(! impl_->write.pending)
0841         BOOST_VERIFY(
0842             impl_->write.timer.expires_at(
0843                 expiry_time) == 0);
0844 }
0845
0846 template<class Protocol, class Executor, class RatePolicy>
0847 void
0848 basic_stream<Protocol, Executor, RatePolicy>::
0849 expires_never()
0850 {
0851     impl_->reset();
0852 }
0853
0854 template<class Protocol, class Executor, class RatePolicy>
0855 void
0856 basic_stream<Protocol, Executor, RatePolicy>::
0857 cancel()
0858 {
0859     error_code ec;
0860     impl_->socket.cancel(ec);
0861     impl_->timer.cancel();
0862 }
0863
0864 template<class Protocol, class Executor, class RatePolicy>
0865 void
0866 basic_stream<Protocol, Executor, RatePolicy>::
0867 close()
0868 {
0869     impl_->close();
0870 }
0871
0872 //------------------------------------------------------------------------------
0873
0874 template<class Protocol, class Executor, class RatePolicy>
0875 template<BOOST_BEAST_ASYNC_TPARAM1 ConnectHandler>
0876 BOOST_BEAST_ASYNC_RESULT1(ConnectHandler)
0877 basic_stream<Protocol, Executor, RatePolicy>::
0878 async_connect(
0879     endpoint_type const& ep,
0880     ConnectHandler&& handler)
0881 {
0882     return net::async_initiate<
0883         ConnectHandler,
0884         void(error_code)>(
0885             typename ops::run_connect_op{},
0886             handler,
0887             this,
0888             ep);
0889 }
0890
0891 template<class Protocol, class Executor, class RatePolicy>
0892 template<
0893     class EndpointSequence,
0894     BOOST_ASIO_COMPLETION_TOKEN_FOR(void(error_code, typename Protocol::endpoint)) RangeConnectHandler,
0895     class>
0896 BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(RangeConnectHandler,void(error_code, typename Protocol::endpoint))
0897 basic_stream<Protocol, Executor, RatePolicy>::
0898 async_connect(
0899     EndpointSequence const& endpoints,
0900     RangeConnectHandler&& handler)
0901 {
0902     return net::async_initiate<
0903         RangeConnectHandler,
0904         void(error_code, typename Protocol::endpoint)>(
0905             typename ops::run_connect_range_op{},
0906             handler,
0907             this,
0908             endpoints,
0909             detail::any_endpoint{});
0910 }
0911
0912 template<class Protocol, class Executor, class RatePolicy>
0913 template<
0914     class EndpointSequence,
0915     class ConnectCondition,
0916     BOOST_ASIO_COMPLETION_TOKEN_FOR(void(error_code, typename Protocol::endpoint)) RangeConnectHandler,
0917     class>
0918 BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(RangeConnectHandler,void (error_code, typename Protocol::endpoint))
0919 basic_stream<Protocol, Executor, RatePolicy>::
0920 async_connect(
0921     EndpointSequence const& endpoints,
0922     ConnectCondition connect_condition,
0923     RangeConnectHandler&& handler)
0924 {
0925     return net::async_initiate<
0926         RangeConnectHandler,
0927         void(error_code, typename Protocol::endpoint)>(
0928             typename ops::run_connect_range_op{},
0929             handler,
0930             this,
0931             endpoints,
0932             connect_condition);
0933 }
0934
0935 template<class Protocol, class Executor, class RatePolicy>
0936 template<
0937     class Iterator,
0938     BOOST_ASIO_COMPLETION_TOKEN_FOR(void(error_code, Iterator)) IteratorConnectHandler>
0939 BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(IteratorConnectHandler,void (error_code, Iterator))
0940 basic_stream<Protocol, Executor, RatePolicy>::
0941 async_connect(
0942     Iterator begin, Iterator end,
0943     IteratorConnectHandler&& handler)
0944 {
0945     return net::async_initiate<
0946         IteratorConnectHandler,
0947         void(error_code, Iterator)>(
0948             typename ops::run_connect_iter_op{},
0949             handler,
0950             this,
0951             begin, end,
0952             detail::any_endpoint{});
0953 }
0954
0955 template<class Protocol, class Executor, class RatePolicy>
0956 template<
0957     class Iterator,
0958     class ConnectCondition,
0959     BOOST_ASIO_COMPLETION_TOKEN_FOR(void(error_code, Iterator)) IteratorConnectHandler>
0960 BOOST_ASIO_INITFN_AUTO_RESULT_TYPE(IteratorConnectHandler,void (error_code, Iterator))
0961 basic_stream<Protocol, Executor, RatePolicy>::
0962 async_connect(
0963     Iterator begin, Iterator end,
0964     ConnectCondition connect_condition,
0965     IteratorConnectHandler&& handler)
0966 {
0967     return net::async_initiate<
0968         IteratorConnectHandler,
0969         void(error_code, Iterator)>(
0970             typename ops::run_connect_iter_op{},
0971             handler,
0972             this,
0973             begin, end,
0974             connect_condition);
0975 }
0976
0977 //------------------------------------------------------------------------------
0978
0979 template<class Protocol, class Executor, class RatePolicy>
0980 template<class MutableBufferSequence, BOOST_BEAST_ASYNC_TPARAM2 ReadHandler>
0981 BOOST_BEAST_ASYNC_RESULT2(ReadHandler)
0982 basic_stream<Protocol, Executor, RatePolicy>::
0983 async_read_some(
0984     MutableBufferSequence const& buffers,
0985     ReadHandler&& handler)
0986 {
0987     static_assert(net::is_mutable_buffer_sequence<
0988         MutableBufferSequence>::value,
0989         "MutableBufferSequence type requirements not met");
0990     return net::async_initiate<
0991         ReadHandler,
0992         void(error_code, std::size_t)>(
0993             typename ops::run_read_op{},
0994             handler,
0995             this,
0996             buffers);
0997 }
0998
0999 template<class Protocol, class Executor, class RatePolicy>
1000 template<class ConstBufferSequence, BOOST_BEAST_ASYNC_TPARAM2 WriteHandler>
1001 BOOST_BEAST_ASYNC_RESULT2(WriteHandler)
1002 basic_stream<Protocol, Executor, RatePolicy>::
1003 async_write_some(
1004     ConstBufferSequence const& buffers,
1005     WriteHandler&& handler)
1006 {
1007     static_assert(net::is_const_buffer_sequence<
1008         ConstBufferSequence>::value,
1009         "ConstBufferSequence type requirements not met");
1010     return net::async_initiate<
1011         WriteHandler,
1012         void(error_code, std::size_t)>(
1013             typename ops::run_write_op{},
1014             handler,
1015             this,
1016             buffers);
1017 }
1018
1019 //------------------------------------------------------------------------------
1020 //
1021 // Customization points
1022 //
1023
1024 #if ! BOOST_BEAST_DOXYGEN
1025
1026 template<
1027     class Protocol, class Executor, class RatePolicy>
1028 void
1029 beast_close_socket(
1030     basic_stream<Protocol, Executor, RatePolicy>& stream)
1031 {
1032     error_code ec;
1033     stream.socket().close(ec);
1034 }
1035
1036 template<
1037     class Protocol, class Executor, class RatePolicy>
1038 void
1039 teardown(
1040     role_type role,
1041     basic_stream<Protocol, Executor, RatePolicy>& stream,
1042     error_code& ec)
1043 {
1044     using beast::websocket::teardown;
1045     teardown(role, stream.socket(), ec);
1046 }
1047
1048 template<
1049     class Protocol, class Executor, class RatePolicy,
1050     class TeardownHandler>
1051 void
1052 async_teardown(
1053     role_type role,
1054     basic_stream<Protocol, Executor, RatePolicy>& stream,
1055     TeardownHandler&& handler)
1056 {
1057     using beast::websocket::async_teardown;
1058     async_teardown(role, stream.socket(),
1059         std::forward<TeardownHandler>(handler));
1060 }
1061
1062 #endif
1063
1064 } // beast
1065 } // boost
1066
1067 #endif