ceph/src/boost/boost/beast/websocket/impl/close.hpp

   1 //
   2 // Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com)
   3 //
   4 // Distributed under the Boost Software License, Version 1.0. (See accompanying
   5 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
   6 //
   7 // Official repository: https://github.com/boostorg/beast
   8 //
   9
  10 #ifndef BOOST_BEAST_WEBSOCKET_IMPL_CLOSE_HPP
  11 #define BOOST_BEAST_WEBSOCKET_IMPL_CLOSE_HPP
  12
  13 #include <boost/beast/websocket/teardown.hpp>
  14 #include <boost/beast/websocket/detail/mask.hpp>
  15 #include <boost/beast/websocket/impl/stream_impl.hpp>
  16 #include <boost/beast/core/async_base.hpp>
  17 #include <boost/beast/core/flat_static_buffer.hpp>
  18 #include <boost/beast/core/stream_traits.hpp>
  19 #include <boost/beast/core/detail/bind_continuation.hpp>
  20 #include <boost/asio/coroutine.hpp>
  21 #include <boost/asio/post.hpp>
  22 #include <boost/throw_exception.hpp>
  23 #include <memory>
  24
  25 namespace boost {
  26 namespace beast {
  27 namespace websocket {
  28
  29 /*  Close the WebSocket Connection
  30
  31     This composed operation sends the close frame if it hasn't already
  32     been sent, then reads and discards frames until receiving a close
  33     frame. Finally it invokes the teardown operation to shut down the
  34     underlying connection.
  35 */
  36 template<class NextLayer, bool deflateSupported>
  37 template<class Handler>
  38 class stream<NextLayer, deflateSupported>::close_op
  39     : public beast::stable_async_base<
  40         Handler, beast::executor_type<stream>>
  41     , public asio::coroutine
  42 {
  43     boost::weak_ptr<impl_type> wp_;
  44     error_code ev_;
  45     detail::frame_buffer& fb_;
  46
  47 public:
  48     static constexpr int id = 5; // for soft_mutex
  49
  50     template<class Handler_>
  51     close_op(
  52         Handler_&& h,
  53         boost::shared_ptr<impl_type> const& sp,
  54         close_reason const& cr)
  55         : stable_async_base<Handler,
  56             beast::executor_type<stream>>(
  57                 std::forward<Handler_>(h),
  58                     sp->stream().get_executor())
  59         , wp_(sp)
  60         , fb_(beast::allocate_stable<
  61             detail::frame_buffer>(*this))
  62     {
  63         // Serialize the close frame
  64         sp->template write_close<
  65             flat_static_buffer_base>(fb_, cr);
  66         (*this)({}, 0, false);
  67     }
  68
  69     void
  70     operator()(
  71         error_code ec = {},
  72         std::size_t bytes_transferred = 0,
  73         bool cont = true)
  74     {
  75         using beast::detail::clamp;
  76         auto sp = wp_.lock();
  77         if(! sp)
  78         {
  79             ec = net::error::operation_aborted;
  80             return this->complete(cont, ec);
  81         }
  82         auto& impl = *sp;
  83         BOOST_ASIO_CORO_REENTER(*this)
  84         {
  85             // Acquire the write lock
  86             if(! impl.wr_block.try_lock(this))
  87             {
  88                 BOOST_ASIO_CORO_YIELD
  89                 impl.op_close.emplace(std::move(*this));
  90                 impl.wr_block.lock(this);
  91                 BOOST_ASIO_CORO_YIELD
  92                 net::post(std::move(*this));
  93                 BOOST_ASSERT(impl.wr_block.is_locked(this));
  94             }
  95             if(impl.check_stop_now(ec))
  96                 goto upcall;
  97
  98             // Can't call close twice
  99             // TODO return a custom error code
 100             BOOST_ASSERT(! impl.wr_close);
 101
 102             // Send close frame
 103             impl.wr_close = true;
 104             impl.change_status(status::closing);
 105             impl.update_timer(this->get_executor());
 106             BOOST_ASIO_CORO_YIELD
 107             net::async_write(impl.stream(), fb_.data(),
 108                 beast::detail::bind_continuation(std::move(*this)));
 109             if(impl.check_stop_now(ec))
 110                 goto upcall;
 111
 112             if(impl.rd_close)
 113             {
 114                 // This happens when the read_op gets a close frame
 115                 // at the same time close_op is sending the close frame.
 116                 // The read_op will be suspended on the write block.
 117                 goto teardown;
 118             }
 119
 120             // Acquire the read lock
 121             if(! impl.rd_block.try_lock(this))
 122             {
 123                 BOOST_ASIO_CORO_YIELD
 124                 impl.op_r_close.emplace(std::move(*this));
 125                 impl.rd_block.lock(this);
 126                 BOOST_ASIO_CORO_YIELD
 127                 net::post(std::move(*this));
 128                 BOOST_ASSERT(impl.rd_block.is_locked(this));
 129                 if(impl.check_stop_now(ec))
 130                     goto upcall;
 131                 BOOST_ASSERT(! impl.rd_close);
 132             }
 133
 134             // Read until a receiving a close frame
 135             // TODO There should be a timeout on this
 136             if(impl.rd_remain > 0)
 137                 goto read_payload;
 138             for(;;)
 139             {
 140                 // Read frame header
 141                 while(! impl.parse_fh(
 142                     impl.rd_fh, impl.rd_buf, ev_))
 143                 {
 144                     if(ev_)
 145                         goto teardown;
 146                     BOOST_ASIO_CORO_YIELD
 147                     impl.stream().async_read_some(
 148                         impl.rd_buf.prepare(read_size(
 149                             impl.rd_buf, impl.rd_buf.max_size())),
 150                         beast::detail::bind_continuation(std::move(*this)));
 151                     impl.rd_buf.commit(bytes_transferred);
 152                     if(impl.check_stop_now(ec))
 153                         goto upcall;
 154                 }
 155                 if(detail::is_control(impl.rd_fh.op))
 156                 {
 157                     // Discard ping or pong frame
 158                     if(impl.rd_fh.op != detail::opcode::close)
 159                     {
 160                         impl.rd_buf.consume(clamp(impl.rd_fh.len));
 161                         continue;
 162                     }
 163
 164                     // Process close frame
 165                     // TODO Should we invoke the control callback?
 166                     BOOST_ASSERT(! impl.rd_close);
 167                     impl.rd_close = true;
 168                     auto const mb = buffers_prefix(
 169                         clamp(impl.rd_fh.len),
 170                         impl.rd_buf.data());
 171                     if(impl.rd_fh.len > 0 && impl.rd_fh.mask)
 172                         detail::mask_inplace(mb, impl.rd_key);
 173                     detail::read_close(impl.cr, mb, ev_);
 174                     if(ev_)
 175                         goto teardown;
 176                     impl.rd_buf.consume(clamp(impl.rd_fh.len));
 177                     goto teardown;
 178                 }
 179
 180             read_payload:
 181                 // Discard message frame
 182                 while(impl.rd_buf.size() < impl.rd_remain)
 183                 {
 184                     impl.rd_remain -= impl.rd_buf.size();
 185                     impl.rd_buf.consume(impl.rd_buf.size());
 186                     BOOST_ASIO_CORO_YIELD
 187                     impl.stream().async_read_some(
 188                         impl.rd_buf.prepare(read_size(
 189                             impl.rd_buf, impl.rd_buf.max_size())),
 190                         beast::detail::bind_continuation(std::move(*this)));
 191                     impl.rd_buf.commit(bytes_transferred);
 192                     if(impl.check_stop_now(ec))
 193                         goto upcall;
 194                 }
 195                 BOOST_ASSERT(impl.rd_buf.size() >= impl.rd_remain);
 196                 impl.rd_buf.consume(clamp(impl.rd_remain));
 197                 impl.rd_remain = 0;
 198             }
 199
 200         teardown:
 201             // Teardown
 202             BOOST_ASSERT(impl.wr_block.is_locked(this));
 203             using beast::websocket::async_teardown;
 204             BOOST_ASIO_CORO_YIELD
 205             async_teardown(impl.role, impl.stream(),
 206                 beast::detail::bind_continuation(std::move(*this)));
 207             BOOST_ASSERT(impl.wr_block.is_locked(this));
 208             if(ec == net::error::eof)
 209             {
 210                 // Rationale:
 211                 // http://stackoverflow.com/questions/25587403/boost-asio-ssl-async-shutdown-always-finishes-with-an-error
 212                 ec = {};
 213             }
 214             if(! ec)
 215                 ec = ev_;
 216             if(ec)
 217                 impl.change_status(status::failed);
 218             else
 219                 impl.change_status(status::closed);
 220             impl.close();
 221
 222         upcall:
 223             impl.wr_block.unlock(this);
 224             impl.rd_block.try_unlock(this)
 225                 && impl.op_r_rd.maybe_invoke();
 226             impl.op_rd.maybe_invoke()
 227                 || impl.op_idle_ping.maybe_invoke()
 228                 || impl.op_ping.maybe_invoke()
 229                 || impl.op_wr.maybe_invoke();
 230             this->complete(cont, ec);
 231         }
 232     }
 233 };
 234
 235 template<class NextLayer, bool deflateSupported>
 236 struct stream<NextLayer, deflateSupported>::
 237     run_close_op
 238 {
 239     template<class CloseHandler>
 240     void
 241     operator()(
 242         CloseHandler&& h,
 243         boost::shared_ptr<impl_type> const& sp,
 244         close_reason const& cr)
 245     {
 246         // If you get an error on the following line it means
 247         // that your handler does not meet the documented type
 248         // requirements for the handler.
 249
 250         static_assert(
 251             beast::detail::is_invocable<CloseHandler,
 252                 void(error_code)>::value,
 253             "CloseHandler type requirements not met");
 254
 255         close_op<
 256             typename std::decay<CloseHandler>::type>(
 257                 std::forward<CloseHandler>(h),
 258                 sp,
 259                 cr);
 260     }
 261 };
 262
 263 //------------------------------------------------------------------------------
 264
 265 template<class NextLayer, bool deflateSupported>
 266 void
 267 stream<NextLayer, deflateSupported>::
 268 close(close_reason const& cr)
 269 {
 270     static_assert(is_sync_stream<next_layer_type>::value,
 271         "SyncStream type requirements not met");
 272     error_code ec;
 273     close(cr, ec);
 274     if(ec)
 275         BOOST_THROW_EXCEPTION(system_error{ec});
 276 }
 277
 278 template<class NextLayer, bool deflateSupported>
 279 void
 280 stream<NextLayer, deflateSupported>::
 281 close(close_reason const& cr, error_code& ec)
 282 {
 283     static_assert(is_sync_stream<next_layer_type>::value,
 284         "SyncStream type requirements not met");
 285     using beast::detail::clamp;
 286     auto& impl = *impl_;
 287     ec = {};
 288     if(impl.check_stop_now(ec))
 289         return;
 290     BOOST_ASSERT(! impl.rd_close);
 291
 292     // Can't call close twice
 293     // TODO return a custom error code
 294     BOOST_ASSERT(! impl.wr_close);
 295
 296     // Send close frame
 297     {
 298         impl.wr_close = true;
 299         impl.change_status(status::closing);
 300         detail::frame_buffer fb;
 301         impl.template write_close<flat_static_buffer_base>(fb, cr);
 302         net::write(impl.stream(), fb.data(), ec);
 303         if(impl.check_stop_now(ec))
 304             return;
 305     }
 306
 307     // Read until a receiving a close frame
 308     error_code ev;
 309     if(impl.rd_remain > 0)
 310         goto read_payload;
 311     for(;;)
 312     {
 313         // Read frame header
 314         while(! impl.parse_fh(
 315             impl.rd_fh, impl.rd_buf, ev))
 316         {
 317             if(ev)
 318             {
 319                 // Protocol violation
 320                 return do_fail(close_code::none, ev, ec);
 321             }
 322             impl.rd_buf.commit(impl.stream().read_some(
 323                 impl.rd_buf.prepare(read_size(
 324                     impl.rd_buf, impl.rd_buf.max_size())), ec));
 325             if(impl.check_stop_now(ec))
 326                 return;
 327         }
 328
 329         if(detail::is_control(impl.rd_fh.op))
 330         {
 331             // Discard ping/pong frame
 332             if(impl.rd_fh.op != detail::opcode::close)
 333             {
 334                 impl.rd_buf.consume(clamp(impl.rd_fh.len));
 335                 continue;
 336             }
 337
 338             // Handle close frame
 339             // TODO Should we invoke the control callback?
 340             BOOST_ASSERT(! impl.rd_close);
 341             impl.rd_close = true;
 342             auto const mb = buffers_prefix(
 343                 clamp(impl.rd_fh.len),
 344                 impl.rd_buf.data());
 345             if(impl.rd_fh.len > 0 && impl.rd_fh.mask)
 346                 detail::mask_inplace(mb, impl.rd_key);
 347             detail::read_close(impl.cr, mb, ev);
 348             if(ev)
 349             {
 350                 // Protocol violation
 351                 return do_fail(close_code::none, ev, ec);
 352             }
 353             impl.rd_buf.consume(clamp(impl.rd_fh.len));
 354             break;
 355         }
 356
 357     read_payload:
 358         // Discard message frame
 359         while(impl.rd_buf.size() < impl.rd_remain)
 360         {
 361             impl.rd_remain -= impl.rd_buf.size();
 362             impl.rd_buf.consume(impl.rd_buf.size());
 363             impl.rd_buf.commit(
 364                 impl.stream().read_some(
 365                     impl.rd_buf.prepare(
 366                         read_size(
 367                             impl.rd_buf,
 368                             impl.rd_buf.max_size())),
 369                     ec));
 370             if(impl.check_stop_now(ec))
 371                 return;
 372         }
 373         BOOST_ASSERT(
 374             impl.rd_buf.size() >= impl.rd_remain);
 375         impl.rd_buf.consume(clamp(impl.rd_remain));
 376         impl.rd_remain = 0;
 377     }
 378     // _Close the WebSocket Connection_
 379     do_fail(close_code::none, error::closed, ec);
 380     if(ec == error::closed)
 381         ec = {};
 382 }
 383
 384 template<class NextLayer, bool deflateSupported>
 385 template<class CloseHandler>
 386 BOOST_BEAST_ASYNC_RESULT1(CloseHandler)
 387 stream<NextLayer, deflateSupported>::
 388 async_close(close_reason const& cr, CloseHandler&& handler)
 389 {
 390     static_assert(is_async_stream<next_layer_type>::value,
 391         "AsyncStream type requirements not met");
 392     return net::async_initiate<
 393         CloseHandler,
 394         void(error_code)>(
 395             run_close_op{},
 396             handler,
 397             impl_,
 398             cr);
 399 }
 400
 401 } // websocket
 402 } // beast
 403 } // boost
 404
 405 #endif