Back to home page

EIC code displayed by LXR
 
 

    

File indexing completed on 2026-08-08 08:43:31 

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_WEBSOCKET_IMPL_READ_HPP
0011 #define BOOST_BEAST_WEBSOCKET_IMPL_READ_HPP
0012 
0013 #include <boost/beast/core/buffer_traits.hpp>
0014 #include <boost/beast/websocket/teardown.hpp>
0015 #include <boost/beast/websocket/detail/mask.hpp>
0016 #include <boost/beast/websocket/impl/stream_impl.hpp>
0017 #include <boost/beast/core/async_base.hpp>
0018 #include <boost/beast/core/buffers_prefix.hpp>
0019 #include <boost/beast/core/buffers_suffix.hpp>
0020 #include <boost/beast/core/flat_static_buffer.hpp>
0021 #include <boost/beast/core/read_size.hpp>
0022 #include <boost/beast/core/stream_traits.hpp>
0023 #include <boost/beast/core/detail/bind_continuation.hpp>
0024 #include <boost/beast/core/detail/buffer.hpp>
0025 #include <boost/beast/core/detail/clamp.hpp>
0026 #include <boost/beast/core/detail/config.hpp>
0027 #include <boost/asio/coroutine.hpp>
0028 #include <boost/assert.hpp>
0029 #include <boost/config.hpp>
0030 #include <boost/optional.hpp>
0031 #include <boost/throw_exception.hpp>
0032 #include <algorithm>
0033 #include <limits>
0034 #include <memory>
0035 
0036 namespace boost {
0037 namespace beast {
0038 namespace websocket {
0039 
0040 /*  Read some message data into a buffer sequence.
0041 
0042     Also reads and handles control frames.
0043 */
0044 template<class NextLayer, bool deflateSupported>
0045 template<class Handler, class MutableBufferSequence>
0046 class stream<NextLayer, deflateSupported>::read_some_op
0047     : public beast::async_base<
0048         Handler, beast::executor_type<stream>>
0049     , public asio::coroutine
0050 {
0051     boost::weak_ptr<impl_type> wp_;
0052     MutableBufferSequence bs_;
0053     buffers_suffix<MutableBufferSequence> cb_;
0054     std::size_t bytes_written_ = 0;
0055     error_code result_;
0056     close_code code_;
0057     bool did_read_ = false;
0058 
0059 public:
0060     static constexpr int id = 1; // for soft_mutex
0061 
0062     template<class Handler_>
0063     read_some_op(
0064         Handler_&& h,
0065         boost::shared_ptr<impl_type> const& sp,
0066         MutableBufferSequence const& bs)
0067         : async_base<
0068             Handler, beast::executor_type<stream>>(
0069                 std::forward<Handler_>(h),
0070                     sp->stream().get_executor())
0071         , wp_(sp)
0072         , bs_(bs)
0073         , cb_(bs)
0074         , code_(close_code::none)
0075     {
0076         (*this)({}, 0, false);
0077     }
0078 
0079     void operator()(
0080         error_code ec = {},
0081         std::size_t bytes_transferred = 0,
0082         bool cont = true)
0083     {
0084         using beast::detail::clamp;
0085         auto sp = wp_.lock();
0086         if(! sp)
0087         {
0088             BOOST_BEAST_ASSIGN_EC(ec, net::error::operation_aborted);
0089             bytes_written_ = 0;
0090             return this->complete(cont, ec, bytes_written_);
0091         }
0092         auto& impl = *sp;
0093         BOOST_ASIO_CORO_REENTER(*this)
0094         {
0095             impl.update_timer(this->get_executor());
0096 
0097         acquire_read_lock:
0098             // Acquire the read lock
0099             if(! impl.rd_block.try_lock(this))
0100             {
0101             do_suspend:
0102                 BOOST_ASIO_CORO_YIELD
0103                 {
0104                     BOOST_ASIO_HANDLER_LOCATION((
0105                         __FILE__, __LINE__,
0106                         "websocket::async_read_some"));
0107 
0108                     this->set_allowed_cancellation(net::cancellation_type::all);
0109                     impl.op_r_rd.emplace(std::move(*this), net::cancellation_type::all);
0110                 }
0111                 if (ec)
0112                     return this->complete(cont, ec, bytes_written_);
0113 
0114                 this->set_allowed_cancellation(net::cancellation_type::terminal);
0115 
0116                 impl.rd_block.lock(this);
0117                 BOOST_ASIO_CORO_YIELD
0118                 {
0119                     BOOST_ASIO_HANDLER_LOCATION((
0120                         __FILE__, __LINE__,
0121                         "websocket::async_read_some"));
0122 
0123                     const auto ex = this->get_immediate_executor();
0124                     net::dispatch(ex, std::move(*this));
0125                 }
0126                 BOOST_ASSERT(impl.rd_block.is_locked(this));
0127 
0128                 BOOST_ASSERT(!ec);
0129                 if(impl.check_stop_now(ec))
0130                 {
0131                     // Issue 2264 - There is no guarantee that the next
0132                     // error will be operation_aborted.
0133                     // The error could be a result of the peer resetting the 
0134                     // connection
0135                     // BOOST_ASSERT(ec == net::error::operation_aborted);
0136                     goto upcall;
0137                 }
0138                 // VFALCO Should never get here
0139 
0140                 // The only way to get read blocked is if
0141                 // a `close_op` wrote a close frame
0142                 BOOST_ASSERT(impl.wr_close);
0143                 BOOST_ASSERT(impl.status_ != status::open);
0144                 BOOST_BEAST_ASSIGN_EC(ec, net::error::operation_aborted);
0145                 goto upcall;
0146             }
0147             else
0148             {
0149                 // Make sure the stream is not closed
0150                 if( impl.status_ == status::closed ||
0151                     impl.status_ == status::failed)
0152                 {
0153                     BOOST_BEAST_ASSIGN_EC(ec, net::error::operation_aborted);
0154                     goto upcall;
0155                 }
0156             }
0157 
0158             // if status_ == status::closing, we want to suspend
0159             // the read operation until the close completes,
0160             // then finish the read with operation_aborted.
0161 
0162         loop:
0163             BOOST_ASSERT(impl.rd_block.is_locked(this));
0164             // See if we need to read a frame header. This
0165             // condition is structured to give the decompressor
0166             // a chance to emit the final empty deflate block
0167             //
0168             if(impl.rd_remain == 0 &&
0169                 (! impl.rd_fh.fin || impl.rd_done))
0170             {
0171                 // Read frame header
0172                 while(! impl.parse_fh(
0173                     impl.rd_fh, impl.rd_buf, result_))
0174                 {
0175                     if(result_)
0176                     {
0177                         // _Fail the WebSocket Connection_
0178                         if(result_ == error::message_too_big)
0179                             code_ = close_code::too_big;
0180                         else
0181                             code_ = close_code::protocol_error;
0182                         goto close;
0183                     }
0184                     BOOST_ASSERT(impl.rd_block.is_locked(this));
0185                     BOOST_ASIO_CORO_YIELD
0186                     {
0187                         BOOST_ASIO_HANDLER_LOCATION((
0188                             __FILE__, __LINE__,
0189                             "websocket::async_read_some"));
0190 
0191                         impl.stream().async_read_some(
0192                             impl.rd_buf.prepare(read_size(
0193                                 impl.rd_buf, impl.rd_buf.max_size())),
0194                                     std::move(*this));
0195                     }
0196                     BOOST_ASSERT(impl.rd_block.is_locked(this));
0197                     impl.rd_buf.commit(bytes_transferred);
0198                     if(impl.check_stop_now(ec))
0199                         goto upcall;
0200                     impl.reset_idle();
0201 
0202                     // Allow a close operation
0203                     // to acquire the read block
0204                     impl.rd_block.unlock(this);
0205                     if( impl.op_r_close.maybe_invoke())
0206                     {
0207                         // Suspend
0208                         BOOST_ASSERT(impl.rd_block.is_locked());
0209                         goto do_suspend;
0210                     }
0211                     // Acquire read block
0212                     impl.rd_block.lock(this);
0213                 }
0214                 // Immediately apply the mask to the portion
0215                 // of the buffer holding payload data.
0216                 if(impl.rd_fh.len > 0 && impl.rd_fh.mask)
0217                     detail::mask_inplace(buffers_prefix(
0218                         clamp(impl.rd_fh.len),
0219                             impl.rd_buf.data()),
0220                                 impl.rd_key);
0221                 if(detail::is_control(impl.rd_fh.op))
0222                 {
0223                     // Clear this otherwise the next
0224                     // frame will be considered final.
0225                     impl.rd_fh.fin = false;
0226 
0227                     // Handle ping frame
0228                     if(impl.rd_fh.op == detail::opcode::ping)
0229                     {
0230                         impl.update_timer(this->get_executor());
0231 
0232                         if(impl.ctrl_cb)
0233                         {
0234                             if(! cont)
0235                             {
0236                                 BOOST_ASIO_CORO_YIELD
0237                                 {
0238                                     BOOST_ASIO_HANDLER_LOCATION((
0239                                         __FILE__, __LINE__,
0240                                         "websocket::async_read_some"));
0241 
0242                                     const auto ex = this->get_immediate_executor();
0243                                     net::dispatch(ex, std::move(*this));
0244                                 }
0245                                 BOOST_ASSERT(cont);
0246                                 // VFALCO call check_stop_now() here?
0247                             }
0248                         }
0249                         {
0250                             auto const b = buffers_prefix(
0251                                 clamp(impl.rd_fh.len),
0252                                     impl.rd_buf.data());
0253                             auto const len = buffer_bytes(b);
0254                             BOOST_ASSERT(len == impl.rd_fh.len);
0255                             ping_data payload;
0256                             detail::read_ping(payload, b);
0257                             impl.rd_buf.consume(len);
0258                             // Ignore ping when closing
0259                             if(impl.status_ == status::closing)
0260                                 goto loop;
0261                             if(impl.ctrl_cb)
0262                                 impl.ctrl_cb(
0263                                     frame_type::ping, to_string_view(payload));
0264                             impl.rd_fb.clear();
0265                             impl.template write_ping<
0266                                 flat_static_buffer_base>(impl.rd_fb,
0267                                     detail::opcode::pong, payload);
0268                         }
0269 
0270                         // Allow a close operation
0271                         // to acquire the read block
0272                         impl.rd_block.unlock(this);
0273                         impl.op_r_close.maybe_invoke();
0274 
0275                         // Acquire the write lock
0276                         if(! impl.wr_block.try_lock(this))
0277                         {
0278                             BOOST_ASIO_CORO_YIELD
0279                             {
0280                                 BOOST_ASIO_HANDLER_LOCATION((
0281                                     __FILE__, __LINE__,
0282                                     "websocket::async_read_some"));
0283 
0284                                 impl.op_rd.emplace(std::move(*this));
0285                             }
0286                             if (ec)
0287                                 return this->complete(cont, ec, bytes_written_);
0288 
0289                             impl.wr_block.lock(this);
0290                             BOOST_ASIO_CORO_YIELD
0291                             {
0292                                 BOOST_ASIO_HANDLER_LOCATION((
0293                                     __FILE__, __LINE__,
0294                                     "websocket::async_read_some"));
0295 
0296                                 const auto ex = this->get_immediate_executor();
0297                                 net::dispatch(ex, std::move(*this));
0298                             }
0299                             BOOST_ASSERT(impl.wr_block.is_locked(this));
0300                             if(impl.check_stop_now(ec))
0301                                 goto upcall;
0302                         }
0303 
0304                         // Send pong
0305                         BOOST_ASSERT(impl.wr_block.is_locked(this));
0306                         BOOST_ASIO_CORO_YIELD
0307                         {
0308                             BOOST_ASIO_HANDLER_LOCATION((
0309                                 __FILE__, __LINE__,
0310                                 "websocket::async_read_some"));
0311 
0312                             net::async_write(
0313                                 impl.stream(), net::const_buffer(impl.rd_fb.data()),
0314                                 beast::detail::bind_continuation(std::move(*this)));
0315                         }
0316                         BOOST_ASSERT(impl.wr_block.is_locked(this));
0317                         if(impl.check_stop_now(ec))
0318                             goto upcall;
0319                         impl.wr_block.unlock(this);
0320                         impl.op_close.maybe_invoke()
0321                             || impl.op_idle_ping.maybe_invoke()
0322                             || impl.op_ping.maybe_invoke()
0323                             || impl.op_wr.maybe_invoke();
0324                         goto acquire_read_lock;
0325                     }
0326 
0327                     // Handle pong frame
0328                     if(impl.rd_fh.op == detail::opcode::pong)
0329                     {
0330                         // Ignore pong when closing
0331                         if(! impl.wr_close && impl.ctrl_cb)
0332                         {
0333                             if(! cont)
0334                             {
0335                                 BOOST_ASIO_CORO_YIELD
0336                                 {
0337                                     BOOST_ASIO_HANDLER_LOCATION((
0338                                         __FILE__, __LINE__,
0339                                         "websocket::async_read_some"));
0340 
0341                                     const auto ex = this->get_immediate_executor();
0342                                     net::dispatch(ex, std::move(*this));
0343                                 }
0344                                 BOOST_ASSERT(cont);
0345                             }
0346                         }
0347                         auto const cb = buffers_prefix(clamp(
0348                             impl.rd_fh.len), impl.rd_buf.data());
0349                         auto const len = buffer_bytes(cb);
0350                         BOOST_ASSERT(len == impl.rd_fh.len);
0351                         ping_data payload;
0352                         detail::read_ping(payload, cb);
0353                         impl.rd_buf.consume(len);
0354                         // Ignore pong when closing
0355                         if(! impl.wr_close && impl.ctrl_cb)
0356                             impl.ctrl_cb(frame_type::pong, to_string_view(payload));
0357                         goto loop;
0358                     }
0359 
0360                     // Handle close frame
0361                     BOOST_ASSERT(impl.rd_fh.op == detail::opcode::close);
0362                     {
0363                         if(impl.ctrl_cb)
0364                         {
0365                             if(! cont)
0366                             {
0367                                 BOOST_ASIO_CORO_YIELD
0368                                 {
0369                                     BOOST_ASIO_HANDLER_LOCATION((
0370                                         __FILE__, __LINE__,
0371                                         "websocket::async_read_some"));
0372 
0373                                     const auto ex = this->get_immediate_executor();
0374                                     net::dispatch(ex, std::move(*this));
0375                                 }
0376                                 BOOST_ASSERT(cont);
0377                             }
0378                         }
0379                         auto const cb = buffers_prefix(clamp(
0380                             impl.rd_fh.len), impl.rd_buf.data());
0381                         auto const len = buffer_bytes(cb);
0382                         BOOST_ASSERT(len == impl.rd_fh.len);
0383                         BOOST_ASSERT(! impl.rd_close);
0384                         impl.rd_close = true;
0385                         close_reason cr;
0386                         detail::read_close(cr, cb, result_);
0387                         if(result_)
0388                         {
0389                             // _Fail the WebSocket Connection_
0390                             code_ = close_code::protocol_error;
0391                             goto close;
0392                         }
0393                         impl.cr = cr;
0394                         impl.rd_buf.consume(len);
0395                         if(impl.ctrl_cb)
0396                             impl.ctrl_cb(frame_type::close,
0397                                 to_string_view(impl.cr.reason));
0398                         // See if we are already closing
0399                         if(impl.status_ == status::closing)
0400                         {
0401                             // _Close the WebSocket Connection_
0402                             BOOST_ASSERT(impl.wr_close);
0403                             code_ = close_code::none;
0404                             result_ = error::closed;
0405                             goto close;
0406                         }
0407                         // _Start the WebSocket Closing Handshake_
0408                         code_ = cr.code == close_code::none ?
0409                             close_code::normal :
0410                             static_cast<close_code>(cr.code);
0411                         result_ = error::closed;
0412                         goto close;
0413                     }
0414                 }
0415                 if(impl.rd_fh.len == 0 && ! impl.rd_fh.fin)
0416                 {
0417                     // Empty non-final frame
0418                     goto loop;
0419                 }
0420                 impl.rd_done = false;
0421             }
0422             if(! impl.rd_deflated())
0423             {
0424                 if(impl.rd_remain > 0)
0425                 {
0426                     if(impl.rd_buf.size() == 0 && impl.rd_buf.max_size() >
0427                         (std::min)(clamp(impl.rd_remain),
0428                             buffer_bytes(cb_)))
0429                     {
0430                         // Fill the read buffer first, otherwise we
0431                         // get fewer bytes at the cost of one I/O.
0432                         BOOST_ASIO_CORO_YIELD
0433                         {
0434                             BOOST_ASIO_HANDLER_LOCATION((
0435                                 __FILE__, __LINE__,
0436                                 "websocket::async_read_some"));
0437 
0438                             impl.stream().async_read_some(
0439                                 impl.rd_buf.prepare(read_size(
0440                                     impl.rd_buf, impl.rd_buf.max_size())),
0441                                         std::move(*this));
0442                         }
0443                         impl.rd_buf.commit(bytes_transferred);
0444                         if(impl.check_stop_now(ec))
0445                             goto upcall;
0446                         impl.reset_idle();
0447                         if(impl.rd_fh.mask)
0448                             detail::mask_inplace(buffers_prefix(clamp(
0449                                 impl.rd_remain), impl.rd_buf.data()),
0450                                     impl.rd_key);
0451                     }
0452                     if(impl.rd_buf.size() > 0)
0453                     {
0454                         // Copy from the read buffer.
0455                         // The mask was already applied.
0456                         bytes_transferred = net::buffer_copy(cb_,
0457                             impl.rd_buf.data(), clamp(impl.rd_remain));
0458                         auto const mb = buffers_prefix(
0459                             bytes_transferred, cb_);
0460                         impl.rd_remain -= bytes_transferred;
0461                         if(impl.rd_op == detail::opcode::text)
0462                         {
0463                             if(! impl.rd_utf8.write(mb) ||
0464                                 (impl.rd_remain == 0 && impl.rd_fh.fin &&
0465                                     ! impl.rd_utf8.finish()))
0466                             {
0467                                 // _Fail the WebSocket Connection_
0468                                 code_ = close_code::bad_payload;
0469                                 result_ = error::bad_frame_payload;
0470                                 goto close;
0471                             }
0472                         }
0473                         bytes_written_ += bytes_transferred;
0474                         impl.rd_size += bytes_transferred;
0475                         impl.rd_buf.consume(bytes_transferred);
0476                     }
0477                     else
0478                     {
0479                         // Read into caller's buffer
0480                         BOOST_ASSERT(impl.rd_remain > 0);
0481                         BOOST_ASSERT(buffer_bytes(cb_) > 0);
0482                         BOOST_ASSERT(buffer_bytes(buffers_prefix(
0483                             clamp(impl.rd_remain), cb_)) > 0);
0484                         BOOST_ASIO_CORO_YIELD
0485                         {
0486                             BOOST_ASIO_HANDLER_LOCATION((
0487                                 __FILE__, __LINE__,
0488                                 "websocket::async_read_some"));
0489 
0490                             impl.stream().async_read_some(buffers_prefix(
0491                                 clamp(impl.rd_remain), cb_), std::move(*this));
0492                         }
0493                         if(impl.check_stop_now(ec))
0494                             goto upcall;
0495                         impl.reset_idle();
0496                         BOOST_ASSERT(bytes_transferred > 0);
0497                         auto const mb = buffers_prefix(
0498                             bytes_transferred, cb_);
0499                         impl.rd_remain -= bytes_transferred;
0500                         if(impl.rd_fh.mask)
0501                             detail::mask_inplace(mb, impl.rd_key);
0502                         if(impl.rd_op == detail::opcode::text)
0503                         {
0504                             if(! impl.rd_utf8.write(mb) ||
0505                                 (impl.rd_remain == 0 && impl.rd_fh.fin &&
0506                                     ! impl.rd_utf8.finish()))
0507                             {
0508                                 // _Fail the WebSocket Connection_
0509                                 code_ = close_code::bad_payload;
0510                                 result_ = error::bad_frame_payload;
0511                                 goto close;
0512                             }
0513                         }
0514                         bytes_written_ += bytes_transferred;
0515                         impl.rd_size += bytes_transferred;
0516                     }
0517                 }
0518                 BOOST_ASSERT( ! impl.rd_done );
0519                 if( impl.rd_remain == 0 && impl.rd_fh.fin )
0520                     impl.rd_done = true;
0521             }
0522             else
0523             {
0524                 // Read compressed message frame payload:
0525                 // inflate even if rd_fh_.len == 0, otherwise we
0526                 // never emit the end-of-stream deflate block.
0527                 while(buffer_bytes(cb_) > 0)
0528                 {
0529                     if( impl.rd_remain > 0 &&
0530                         impl.rd_buf.size() == 0 &&
0531                         ! did_read_)
0532                     {
0533                         // read new
0534                         BOOST_ASIO_CORO_YIELD
0535                         {
0536                             BOOST_ASIO_HANDLER_LOCATION((
0537                                 __FILE__, __LINE__,
0538                                 "websocket::async_read_some"));
0539 
0540                             impl.stream().async_read_some(
0541                                 impl.rd_buf.prepare(read_size(
0542                                     impl.rd_buf, impl.rd_buf.max_size())),
0543                                         std::move(*this));
0544                         }
0545                         if(impl.check_stop_now(ec))
0546                             goto upcall;
0547                         impl.reset_idle();
0548                         BOOST_ASSERT(bytes_transferred > 0);
0549                         impl.rd_buf.commit(bytes_transferred);
0550                         if(impl.rd_fh.mask)
0551                             detail::mask_inplace(
0552                                 buffers_prefix(clamp(impl.rd_remain),
0553                                     impl.rd_buf.data()), impl.rd_key);
0554                         did_read_ = true;
0555                     }
0556                     zlib::z_params zs;
0557                     {
0558                         auto const out = buffers_front(cb_);
0559                         zs.next_out = out.data();
0560                         zs.avail_out = out.size();
0561                         BOOST_ASSERT(zs.avail_out > 0);
0562                     }
0563                     if(impl.rd_remain > 0)
0564                     {
0565                         if(impl.rd_buf.size() > 0)
0566                         {
0567                             // use what's there
0568                             auto const in = buffers_prefix(
0569                                 clamp(impl.rd_remain), buffers_front(
0570                                     impl.rd_buf.data()));
0571                             zs.avail_in = in.size();
0572                             zs.next_in = in.data();
0573                         }
0574                         else
0575                         {
0576                             break;
0577                         }
0578                         impl.inflate(zs, ec);
0579                         if(impl.check_stop_now(ec))
0580                             goto upcall;
0581                         impl.rd_remain -= zs.total_in;
0582                         impl.rd_buf.consume(zs.total_in);
0583                     }
0584                     else if(impl.rd_fh.fin)
0585                     {
0586                         impl.inflate_with_eb(zs, ec);
0587                         if(impl.check_stop_now(ec))
0588                             goto upcall;
0589                         if(zs.total_out == 0)
0590                         {
0591                             impl.do_context_takeover_read(impl.role);
0592                             impl.rd_done = true;
0593                             break;
0594                         }
0595                     }
0596                     else
0597                     {
0598                         break;
0599                     }
0600                     if(impl.rd_msg_max && beast::detail::sum_exceeds(
0601                         impl.rd_size, zs.total_out, impl.rd_msg_max))
0602                     {
0603                         // _Fail the WebSocket Connection_
0604                         code_ = close_code::too_big;
0605                         result_ = error::message_too_big;
0606                         goto close;
0607                     }
0608                     cb_.consume(zs.total_out);
0609                     impl.rd_size += zs.total_out;
0610                     bytes_written_ += zs.total_out;
0611                 }
0612                 if(impl.rd_op == detail::opcode::text)
0613                 {
0614                     // check utf8
0615                     if(! impl.rd_utf8.write(
0616                         buffers_prefix(bytes_written_, bs_)) || (
0617                             impl.rd_done && ! impl.rd_utf8.finish()))
0618                     {
0619                         // _Fail the WebSocket Connection_
0620                         code_ = close_code::bad_payload;
0621                         result_ = error::bad_frame_payload;
0622                         goto close;
0623                     }
0624                 }
0625             }
0626             goto upcall;
0627 
0628         close:
0629             // Acquire the write lock
0630             if(! impl.wr_block.try_lock(this))
0631             {
0632                 BOOST_ASIO_CORO_YIELD
0633                 {
0634                     BOOST_ASIO_HANDLER_LOCATION((
0635                         __FILE__, __LINE__,
0636                         "websocket::async_read_some"));
0637 
0638                     impl.op_rd.emplace(std::move(*this));
0639                 }
0640                 if (ec)
0641                     return this->complete(cont, ec, bytes_written_);
0642 
0643                 impl.wr_block.lock(this);
0644                 BOOST_ASIO_CORO_YIELD
0645                 {
0646                     BOOST_ASIO_HANDLER_LOCATION((
0647                         __FILE__, __LINE__,
0648                         "websocket::async_read_some"));
0649 
0650                     const auto ex = this->get_immediate_executor();
0651                     net::dispatch(ex, std::move(*this));
0652                 }
0653                 BOOST_ASSERT(impl.wr_block.is_locked(this));
0654                 if(impl.check_stop_now(ec))
0655                     goto upcall;
0656             }
0657 
0658             impl.change_status(status::closing);
0659             impl.update_timer(this->get_executor());
0660 
0661             if(! impl.wr_close)
0662             {
0663                 impl.wr_close = true;
0664 
0665                 // Serialize close frame
0666                 impl.rd_fb.clear();
0667                 impl.template write_close<
0668                     flat_static_buffer_base>(
0669                         impl.rd_fb, code_);
0670 
0671                 // Send close frame
0672                 BOOST_ASSERT(impl.wr_block.is_locked(this));
0673                 BOOST_ASIO_CORO_YIELD
0674                 {
0675                     BOOST_ASIO_HANDLER_LOCATION((
0676                         __FILE__, __LINE__,
0677                         "websocket::async_read_some"));
0678 
0679                     net::async_write(impl.stream(), net::const_buffer(impl.rd_fb.data()),
0680                         beast::detail::bind_continuation(std::move(*this)));
0681                 }
0682                 BOOST_ASSERT(impl.wr_block.is_locked(this));
0683                 if(impl.check_stop_now(ec))
0684                     goto upcall;
0685             }
0686 
0687             // Teardown
0688             using beast::websocket::async_teardown;
0689             BOOST_ASSERT(impl.wr_block.is_locked(this));
0690             BOOST_ASIO_CORO_YIELD
0691             {
0692                 BOOST_ASIO_HANDLER_LOCATION((
0693                     __FILE__, __LINE__,
0694                     "websocket::async_read_some"));
0695 
0696                 async_teardown(impl.role, impl.stream(),
0697                     beast::detail::bind_continuation(std::move(*this)));
0698             }
0699             BOOST_ASSERT(impl.wr_block.is_locked(this));
0700             if(ec == net::error::eof)
0701             {
0702                 // Rationale:
0703                 // http://stackoverflow.com/questions/25587403/boost-asio-ssl-async-shutdown-always-finishes-with-an-error
0704                 ec = {};
0705             }
0706             if(! ec)
0707             {
0708                 BOOST_BEAST_ASSIGN_EC(ec, result_);
0709             }
0710             if(ec && ec != error::closed)
0711                 impl.change_status(status::failed);
0712             else
0713                 impl.change_status(status::closed);
0714             impl.close();
0715 
0716         upcall:
0717             impl.rd_block.try_unlock(this);
0718             impl.op_r_close.maybe_invoke();
0719             if(impl.wr_block.try_unlock(this))
0720                 impl.op_close.maybe_invoke()
0721                     || impl.op_idle_ping.maybe_invoke()
0722                     || impl.op_ping.maybe_invoke()
0723                     || impl.op_wr.maybe_invoke();
0724             this->complete(cont, ec, bytes_written_);
0725         }
0726     }
0727 };
0728 
0729 //------------------------------------------------------------------------------
0730 
0731 template<class NextLayer, bool deflateSupported>
0732 template<class Handler,  class DynamicBuffer>
0733 class stream<NextLayer, deflateSupported>::read_op
0734     : public beast::async_base<
0735         Handler, beast::executor_type<stream>>
0736     , public asio::coroutine
0737 {
0738     boost::weak_ptr<impl_type> wp_;
0739     DynamicBuffer& b_;
0740     std::size_t limit_;
0741     std::size_t bytes_written_ = 0;
0742     bool some_;
0743 
0744 public:
0745     template<class Handler_>
0746     read_op(
0747         Handler_&& h,
0748         boost::shared_ptr<impl_type> const& sp,
0749         DynamicBuffer& b,
0750         std::size_t limit,
0751         bool some)
0752         : async_base<Handler,
0753             beast::executor_type<stream>>(
0754                 std::forward<Handler_>(h),
0755                     sp->stream().get_executor())
0756         , wp_(sp)
0757         , b_(b)
0758         , limit_(limit ? limit : (
0759             std::numeric_limits<std::size_t>::max)())
0760         , some_(some)
0761     {
0762         (*this)({}, 0, false);
0763     }
0764 
0765     void operator()(
0766         error_code ec = {},
0767         std::size_t bytes_transferred = 0,
0768         bool cont = true)
0769     {
0770         using beast::detail::clamp;
0771         auto sp = wp_.lock();
0772         if(! sp)
0773         {
0774             BOOST_BEAST_ASSIGN_EC(ec, net::error::operation_aborted);
0775             bytes_written_ = 0;
0776             return this->complete(cont, ec, bytes_written_);
0777         }
0778         auto& impl = *sp;
0779         using mutable_buffers_type = typename
0780             DynamicBuffer::mutable_buffers_type;
0781         BOOST_ASIO_CORO_REENTER(*this)
0782         {
0783             do
0784             {
0785                 // VFALCO TODO use boost::beast::bind_continuation
0786                 BOOST_ASIO_CORO_YIELD
0787                 {
0788                     auto mb = beast::detail::dynamic_buffer_prepare(b_,
0789                         clamp(impl.read_size_hint_db(b_), limit_),
0790                             ec, error::buffer_overflow);
0791                     if(impl.check_stop_now(ec))
0792                         goto upcall;
0793 
0794                     BOOST_ASIO_HANDLER_LOCATION((
0795                         __FILE__, __LINE__,
0796                         "websocket::async_read"));
0797 
0798                     read_some_op<read_op, mutable_buffers_type>(
0799                         std::move(*this), sp, *mb);
0800                 }
0801 
0802                 b_.commit(bytes_transferred);
0803                 bytes_written_ += bytes_transferred;
0804                 if(ec)
0805                     goto upcall;
0806             }
0807             while(! some_ && ! impl.rd_done);
0808 
0809         upcall:
0810             this->complete(cont, ec, bytes_written_);
0811         }
0812     }
0813 };
0814 
0815 template<class NextLayer, bool deflateSupported>
0816 struct stream<NextLayer, deflateSupported>::
0817     run_read_some_op
0818 {
0819     boost::shared_ptr<impl_type> const& self;
0820 
0821     using executor_type = typename stream::executor_type;
0822 
0823     executor_type
0824     get_executor() const noexcept
0825     {
0826         return self->stream().get_executor();
0827     }
0828 
0829     template<
0830         class ReadHandler,
0831         class MutableBufferSequence>
0832     void
0833     operator()(
0834         ReadHandler&& h,
0835         MutableBufferSequence const& b)
0836     {
0837         // If you get an error on the following line it means
0838         // that your handler does not meet the documented type
0839         // requirements for the handler.
0840 
0841         static_assert(
0842             beast::detail::is_invocable<ReadHandler,
0843                 void(error_code, std::size_t)>::value,
0844             "ReadHandler type requirements not met");
0845 
0846         read_some_op<
0847             typename std::decay<ReadHandler>::type,
0848             MutableBufferSequence>(
0849                 std::forward<ReadHandler>(h),
0850                 self,
0851                 b);
0852     }
0853 };
0854 
0855 template<class NextLayer, bool deflateSupported>
0856 struct stream<NextLayer, deflateSupported>::
0857     run_read_op
0858 {
0859     boost::shared_ptr<impl_type> const& self;
0860 
0861     using executor_type = typename stream::executor_type;
0862 
0863     executor_type
0864     get_executor() const noexcept
0865     {
0866         return self->stream().get_executor();
0867     }
0868 
0869     template<
0870         class ReadHandler,
0871         class DynamicBuffer>
0872     void
0873     operator()(
0874         ReadHandler&& h,
0875         DynamicBuffer* b,
0876         std::size_t limit,
0877         bool some)
0878     {
0879         // If you get an error on the following line it means
0880         // that your handler does not meet the documented type
0881         // requirements for the handler.
0882 
0883         static_assert(
0884             beast::detail::is_invocable<ReadHandler,
0885                 void(error_code, std::size_t)>::value,
0886             "ReadHandler type requirements not met");
0887 
0888         read_op<
0889             typename std::decay<ReadHandler>::type,
0890             DynamicBuffer>(
0891                 std::forward<ReadHandler>(h),
0892                 self,
0893                 *b,
0894                 limit,
0895                 some);
0896     }
0897 };
0898 
0899 //------------------------------------------------------------------------------
0900 
0901 template<class NextLayer, bool deflateSupported>
0902 template<class DynamicBuffer>
0903 std::size_t
0904 stream<NextLayer, deflateSupported>::
0905 read(DynamicBuffer& buffer)
0906 {
0907     static_assert(is_sync_stream<next_layer_type>::value,
0908         "SyncStream type requirements not met");
0909     static_assert(
0910         net::is_dynamic_buffer<DynamicBuffer>::value,
0911         "DynamicBuffer type requirements not met");
0912     error_code ec;
0913     auto const bytes_written = read(buffer, ec);
0914     if(ec)
0915         BOOST_THROW_EXCEPTION(system_error{ec});
0916     return bytes_written;
0917 }
0918 
0919 template<class NextLayer, bool deflateSupported>
0920 template<class DynamicBuffer>
0921 std::size_t
0922 stream<NextLayer, deflateSupported>::
0923 read(DynamicBuffer& buffer, error_code& ec)
0924 {
0925     static_assert(is_sync_stream<next_layer_type>::value,
0926         "SyncStream type requirements not met");
0927     static_assert(
0928         net::is_dynamic_buffer<DynamicBuffer>::value,
0929         "DynamicBuffer type requirements not met");
0930     std::size_t bytes_written = 0;
0931     do
0932     {
0933         bytes_written += read_some(buffer, 0, ec);
0934         if(ec)
0935             return bytes_written;
0936     }
0937     while(! is_message_done());
0938     return bytes_written;
0939 }
0940 
0941 template<class NextLayer, bool deflateSupported>
0942 template<class DynamicBuffer, BOOST_BEAST_ASYNC_TPARAM2 ReadHandler>
0943 BOOST_BEAST_ASYNC_RESULT2(ReadHandler)
0944 stream<NextLayer, deflateSupported>::
0945 async_read(DynamicBuffer& buffer, ReadHandler&& handler)
0946 {
0947     static_assert(is_async_stream<next_layer_type>::value,
0948         "AsyncStream type requirements not met");
0949     static_assert(
0950         net::is_dynamic_buffer<DynamicBuffer>::value,
0951         "DynamicBuffer type requirements not met");
0952     return net::async_initiate<
0953         ReadHandler,
0954         void(error_code, std::size_t)>(
0955             run_read_op{impl_},
0956             handler,
0957             &buffer,
0958             0,
0959             false);
0960 }
0961 
0962 //------------------------------------------------------------------------------
0963 
0964 template<class NextLayer, bool deflateSupported>
0965 template<class DynamicBuffer>
0966 std::size_t
0967 stream<NextLayer, deflateSupported>::
0968 read_some(
0969     DynamicBuffer& buffer,
0970     std::size_t limit)
0971 {
0972     static_assert(is_sync_stream<next_layer_type>::value,
0973         "SyncStream type requirements not met");
0974     static_assert(
0975         net::is_dynamic_buffer<DynamicBuffer>::value,
0976         "DynamicBuffer type requirements not met");
0977     error_code ec;
0978     auto const bytes_written =
0979         read_some(buffer, limit, ec);
0980     if(ec)
0981         BOOST_THROW_EXCEPTION(system_error{ec});
0982     return bytes_written;
0983 }
0984 
0985 template<class NextLayer, bool deflateSupported>
0986 template<class DynamicBuffer>
0987 std::size_t
0988 stream<NextLayer, deflateSupported>::
0989 read_some(
0990     DynamicBuffer& buffer,
0991     std::size_t limit,
0992     error_code& ec)
0993 {
0994     static_assert(is_sync_stream<next_layer_type>::value,
0995         "SyncStream type requirements not met");
0996     static_assert(
0997         net::is_dynamic_buffer<DynamicBuffer>::value,
0998         "DynamicBuffer type requirements not met");
0999     using beast::detail::clamp;
1000     if(! limit)
1001         limit = (std::numeric_limits<std::size_t>::max)();
1002     auto const size =
1003         clamp(impl_->read_size_hint_db(buffer), limit);
1004     BOOST_ASSERT(size > 0);
1005     auto mb = beast::detail::dynamic_buffer_prepare(
1006         buffer, size, ec, error::buffer_overflow);
1007     if(impl_->check_stop_now(ec))
1008         return 0;
1009     auto const bytes_written = read_some(*mb, ec);
1010     buffer.commit(bytes_written);
1011     return bytes_written;
1012 }
1013 
1014 template<class NextLayer, bool deflateSupported>
1015 template<class DynamicBuffer, BOOST_BEAST_ASYNC_TPARAM2 ReadHandler>
1016 BOOST_BEAST_ASYNC_RESULT2(ReadHandler)
1017 stream<NextLayer, deflateSupported>::
1018 async_read_some(
1019     DynamicBuffer& buffer,
1020     std::size_t limit,
1021     ReadHandler&& handler)
1022 {
1023     static_assert(is_async_stream<next_layer_type>::value,
1024         "AsyncStream type requirements not met");
1025     static_assert(
1026         net::is_dynamic_buffer<DynamicBuffer>::value,
1027         "DynamicBuffer type requirements not met");
1028     return net::async_initiate<
1029         ReadHandler,
1030         void(error_code, std::size_t)>(
1031             run_read_op{impl_},
1032             handler,
1033             &buffer,
1034             limit,
1035             true);
1036 }
1037 
1038 //------------------------------------------------------------------------------
1039 
1040 template<class NextLayer, bool deflateSupported>
1041 template<class MutableBufferSequence>
1042 std::size_t
1043 stream<NextLayer, deflateSupported>::
1044 read_some(
1045     MutableBufferSequence const& buffers)
1046 {
1047     static_assert(is_sync_stream<next_layer_type>::value,
1048         "SyncStream type requirements not met");
1049     static_assert(net::is_mutable_buffer_sequence<
1050             MutableBufferSequence>::value,
1051         "MutableBufferSequence type requirements not met");
1052     error_code ec;
1053     auto const bytes_written = read_some(buffers, ec);
1054     if(ec)
1055         BOOST_THROW_EXCEPTION(system_error{ec});
1056     return bytes_written;
1057 }
1058 
1059 template<class NextLayer, bool deflateSupported>
1060 template<class MutableBufferSequence>
1061 std::size_t
1062 stream<NextLayer, deflateSupported>::
1063 read_some(
1064     MutableBufferSequence const& buffers,
1065     error_code& ec)
1066 {
1067     static_assert(is_sync_stream<next_layer_type>::value,
1068         "SyncStream type requirements not met");
1069     static_assert(net::is_mutable_buffer_sequence<
1070             MutableBufferSequence>::value,
1071         "MutableBufferSequence type requirements not met");
1072     using beast::detail::clamp;
1073     auto& impl = *impl_;
1074     close_code code{};
1075     std::size_t bytes_written = 0;
1076     ec = {};
1077     // Make sure the stream is open
1078     if(impl.check_stop_now(ec))
1079         return bytes_written;
1080 loop:
1081     // See if we need to read a frame header. This
1082     // condition is structured to give the decompressor
1083     // a chance to emit the final empty deflate block
1084     //
1085     if(impl.rd_remain == 0 && (
1086         ! impl.rd_fh.fin || impl.rd_done))
1087     {
1088         // Read frame header
1089         error_code result;
1090         while(! impl.parse_fh(impl.rd_fh, impl.rd_buf, result))
1091         {
1092             if(result)
1093             {
1094                 // _Fail the WebSocket Connection_
1095                 if(result == error::message_too_big)
1096                     code = close_code::too_big;
1097                 else
1098                     code = close_code::protocol_error;
1099                 do_fail(code, result, ec);
1100                 return bytes_written;
1101             }
1102             auto const bytes_transferred =
1103                 impl.stream().read_some(
1104                     impl.rd_buf.prepare(read_size(
1105                         impl.rd_buf, impl.rd_buf.max_size())),
1106                     ec);
1107             impl.rd_buf.commit(bytes_transferred);
1108             if(impl.check_stop_now(ec))
1109                 return bytes_written;
1110         }
1111         // Immediately apply the mask to the portion
1112         // of the buffer holding payload data.
1113         if(impl.rd_fh.len > 0 && impl.rd_fh.mask)
1114             detail::mask_inplace(buffers_prefix(
1115                 clamp(impl.rd_fh.len), impl.rd_buf.data()),
1116                     impl.rd_key);
1117         if(detail::is_control(impl.rd_fh.op))
1118         {
1119             // Get control frame payload
1120             auto const b = buffers_prefix(
1121                 clamp(impl.rd_fh.len), impl.rd_buf.data());
1122             auto const len = buffer_bytes(b);
1123             BOOST_ASSERT(len == impl.rd_fh.len);
1124 
1125             // Clear this otherwise the next
1126             // frame will be considered final.
1127             impl.rd_fh.fin = false;
1128 
1129             // Handle ping frame
1130             if(impl.rd_fh.op == detail::opcode::ping)
1131             {
1132                 ping_data payload;
1133                 detail::read_ping(payload, b);
1134                 impl.rd_buf.consume(len);
1135                 if(impl.wr_close)
1136                 {
1137                     // Ignore ping when closing
1138                     goto loop;
1139                 }
1140                 if(impl.ctrl_cb)
1141                     impl.ctrl_cb(frame_type::ping, to_string_view(payload));
1142                 detail::frame_buffer fb;
1143                 impl.template write_ping<flat_static_buffer_base>(fb,
1144                     detail::opcode::pong, payload);
1145                 net::write(impl.stream(), fb.data(), ec);
1146                 if(impl.check_stop_now(ec))
1147                     return bytes_written;
1148                 goto loop;
1149             }
1150             // Handle pong frame
1151             if(impl.rd_fh.op == detail::opcode::pong)
1152             {
1153                 ping_data payload;
1154                 detail::read_ping(payload, b);
1155                 impl.rd_buf.consume(len);
1156                 if(impl.ctrl_cb)
1157                     impl.ctrl_cb(frame_type::pong, to_string_view(payload));
1158                 goto loop;
1159             }
1160             // Handle close frame
1161             BOOST_ASSERT(impl.rd_fh.op == detail::opcode::close);
1162             {
1163                 BOOST_ASSERT(! impl.rd_close);
1164                 impl.rd_close = true;
1165                 close_reason cr;
1166                 detail::read_close(cr, b, result);
1167                 if(result)
1168                 {
1169                     // _Fail the WebSocket Connection_
1170                     do_fail(close_code::protocol_error,
1171                         result, ec);
1172                     return bytes_written;
1173                 }
1174                 impl.cr = cr;
1175                 impl.rd_buf.consume(len);
1176                 if(impl.ctrl_cb)
1177                     impl.ctrl_cb(frame_type::close, to_string_view(impl.cr.reason));
1178                 BOOST_ASSERT(! impl.wr_close);
1179                 // _Start the WebSocket Closing Handshake_
1180                 do_fail(
1181                     cr.code == close_code::none ?
1182                         close_code::normal :
1183                         static_cast<close_code>(cr.code),
1184                     error::closed, ec);
1185                 return bytes_written;
1186             }
1187         }
1188         if(impl.rd_fh.len == 0 && ! impl.rd_fh.fin)
1189         {
1190             // Empty non-final frame
1191             goto loop;
1192         }
1193         impl.rd_done = false;
1194     }
1195     else
1196     {
1197         ec = {};
1198     }
1199     if(! impl.rd_deflated())
1200     {
1201         if(impl.rd_remain > 0)
1202         {
1203             if(impl.rd_buf.size() == 0 && impl.rd_buf.max_size() >
1204                 (std::min)(clamp(impl.rd_remain),
1205                     buffer_bytes(buffers)))
1206             {
1207                 // Fill the read buffer first, otherwise we
1208                 // get fewer bytes at the cost of one I/O.
1209                 impl.rd_buf.commit(impl.stream().read_some(
1210                     impl.rd_buf.prepare(read_size(impl.rd_buf,
1211                         impl.rd_buf.max_size())), ec));
1212                 if(impl.check_stop_now(ec))
1213                     return bytes_written;
1214                 if(impl.rd_fh.mask)
1215                     detail::mask_inplace(
1216                         buffers_prefix(clamp(impl.rd_remain),
1217                             impl.rd_buf.data()), impl.rd_key);
1218             }
1219             if(impl.rd_buf.size() > 0)
1220             {
1221                 // Copy from the read buffer.
1222                 // The mask was already applied.
1223                 auto const bytes_transferred = net::buffer_copy(
1224                     buffers, impl.rd_buf.data(),
1225                         clamp(impl.rd_remain));
1226                 auto const mb = buffers_prefix(
1227                     bytes_transferred, buffers);
1228                 impl.rd_remain -= bytes_transferred;
1229                 if(impl.rd_op == detail::opcode::text)
1230                 {
1231                     if(! impl.rd_utf8.write(mb) ||
1232                         (impl.rd_remain == 0 && impl.rd_fh.fin &&
1233                             ! impl.rd_utf8.finish()))
1234                     {
1235                         // _Fail the WebSocket Connection_
1236                         do_fail(close_code::bad_payload,
1237                             error::bad_frame_payload, ec);
1238                         return bytes_written;
1239                     }
1240                 }
1241                 bytes_written += bytes_transferred;
1242                 impl.rd_size += bytes_transferred;
1243                 impl.rd_buf.consume(bytes_transferred);
1244             }
1245             else
1246             {
1247                 // Read into caller's buffer
1248                 BOOST_ASSERT(impl.rd_remain > 0);
1249                 BOOST_ASSERT(buffer_bytes(buffers) > 0);
1250                 BOOST_ASSERT(buffer_bytes(buffers_prefix(
1251                     clamp(impl.rd_remain), buffers)) > 0);
1252                 auto const bytes_transferred =
1253                     impl.stream().read_some(buffers_prefix(
1254                         clamp(impl.rd_remain), buffers), ec);
1255                 // VFALCO What if some bytes were written?
1256                 if(impl.check_stop_now(ec))
1257                     return bytes_written;
1258                 BOOST_ASSERT(bytes_transferred > 0);
1259                 auto const mb = buffers_prefix(
1260                     bytes_transferred, buffers);
1261                 impl.rd_remain -= bytes_transferred;
1262                 if(impl.rd_fh.mask)
1263                     detail::mask_inplace(mb, impl.rd_key);
1264                 if(impl.rd_op == detail::opcode::text)
1265                 {
1266                     if(! impl.rd_utf8.write(mb) ||
1267                         (impl.rd_remain == 0 && impl.rd_fh.fin &&
1268                             ! impl.rd_utf8.finish()))
1269                     {
1270                         // _Fail the WebSocket Connection_
1271                         do_fail(close_code::bad_payload,
1272                             error::bad_frame_payload, ec);
1273                         return bytes_written;
1274                     }
1275                 }
1276                 bytes_written += bytes_transferred;
1277                 impl.rd_size += bytes_transferred;
1278             }
1279         }
1280         BOOST_ASSERT( ! impl.rd_done );
1281         if( impl.rd_remain == 0 && impl.rd_fh.fin )
1282             impl.rd_done = true;
1283     }
1284     else
1285     {
1286         // Read compressed message frame payload:
1287         // inflate even if rd_fh_.len == 0, otherwise we
1288         // never emit the end-of-stream deflate block.
1289         //
1290         bool did_read = false;
1291         buffers_suffix<MutableBufferSequence> cb(buffers);
1292         while(buffer_bytes(cb) > 0)
1293         {
1294             zlib::z_params zs;
1295             {
1296                 auto const out = beast::buffers_front(cb);
1297                 zs.next_out = out.data();
1298                 zs.avail_out = out.size();
1299                 BOOST_ASSERT(zs.avail_out > 0);
1300             }
1301             if(impl.rd_remain > 0)
1302             {
1303                 if(impl.rd_buf.size() > 0)
1304                 {
1305                     // use what's there
1306                     auto const in = buffers_prefix(
1307                         clamp(impl.rd_remain), beast::buffers_front(
1308                             impl.rd_buf.data()));
1309                     zs.avail_in = in.size();
1310                     zs.next_in = in.data();
1311                 }
1312                 else if(! did_read)
1313                 {
1314                     // read new
1315                     auto const bytes_transferred =
1316                         impl.stream().read_some(
1317                             impl.rd_buf.prepare(read_size(
1318                                 impl.rd_buf, impl.rd_buf.max_size())),
1319                             ec);
1320                     if(impl.check_stop_now(ec))
1321                         return bytes_written;
1322                     BOOST_ASSERT(bytes_transferred > 0);
1323                     impl.rd_buf.commit(bytes_transferred);
1324                     if(impl.rd_fh.mask)
1325                         detail::mask_inplace(
1326                             buffers_prefix(clamp(impl.rd_remain),
1327                                 impl.rd_buf.data()), impl.rd_key);
1328                     auto const in = buffers_prefix(
1329                         clamp(impl.rd_remain), buffers_front(
1330                             impl.rd_buf.data()));
1331                     zs.avail_in = in.size();
1332                     zs.next_in = in.data();
1333                     did_read = true;
1334                 }
1335                 else
1336                 {
1337                     break;
1338                 }
1339                 impl.inflate(zs, ec);
1340                 if(impl.check_stop_now(ec))
1341                     return bytes_written;
1342                 impl.rd_remain -= zs.total_in;
1343                 impl.rd_buf.consume(zs.total_in);
1344             }
1345             else if(impl.rd_fh.fin)
1346             {
1347                 impl.inflate_with_eb(zs, ec);
1348                 if(impl.check_stop_now(ec))
1349                     return bytes_written;
1350                 if(zs.total_out == 0)
1351                 {
1352                     impl.do_context_takeover_read(impl.role);
1353                     impl.rd_done = true;
1354                     break;
1355                 }
1356             }
1357             else
1358             {
1359                 break;
1360             }
1361             if(impl.rd_msg_max && beast::detail::sum_exceeds(
1362                 impl.rd_size, zs.total_out, impl.rd_msg_max))
1363             {
1364                 do_fail(close_code::too_big,
1365                     error::message_too_big, ec);
1366                 return bytes_written;
1367             }
1368             cb.consume(zs.total_out);
1369             impl.rd_size += zs.total_out;
1370             bytes_written += zs.total_out;
1371         }
1372         if(impl.rd_op == detail::opcode::text)
1373         {
1374             // check utf8
1375             if(! impl.rd_utf8.write(beast::buffers_prefix(
1376                 bytes_written, buffers)) || (
1377                     impl.rd_done && ! impl.rd_utf8.finish()))
1378             {
1379                 // _Fail the WebSocket Connection_
1380                 do_fail(close_code::bad_payload,
1381                     error::bad_frame_payload, ec);
1382                 return bytes_written;
1383             }
1384         }
1385     }
1386     return bytes_written;
1387 }
1388 
1389 template<class NextLayer, bool deflateSupported>
1390 template<class MutableBufferSequence, BOOST_BEAST_ASYNC_TPARAM2 ReadHandler>
1391 BOOST_BEAST_ASYNC_RESULT2(ReadHandler)
1392 stream<NextLayer, deflateSupported>::
1393 async_read_some(
1394     MutableBufferSequence const& buffers,
1395     ReadHandler&& handler)
1396 {
1397     static_assert(is_async_stream<next_layer_type>::value,
1398         "AsyncStream type requirements not met");
1399     static_assert(net::is_mutable_buffer_sequence<
1400             MutableBufferSequence>::value,
1401         "MutableBufferSequence type requirements not met");
1402     return net::async_initiate<
1403         ReadHandler,
1404         void(error_code, std::size_t)>(
1405             run_read_some_op{impl_},
1406             handler,
1407             buffers);
1408 }
1409 
1410 } // websocket
1411 } // beast
1412 } // boost
1413 
1414 #endif
[ Source navigation ] [ Diff markup ] [ Identifier search ] [ general search ]

This page was automatically generated by the 2.3.7 LXR engine.

The LXR team
Valid CSS 2.1! Valid HTML 4.01!