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_STREAM_IMPL_HPP
11#define BOOST_BEAST_WEBSOCKET_IMPL_STREAM_IMPL_HPP
12
13#include <boost/beast/websocket/rfc6455.hpp>
14#include <boost/beast/websocket/detail/frame.hpp>
15#include <boost/beast/websocket/detail/hybi13.hpp>
16#include <boost/beast/websocket/detail/mask.hpp>
17#include <boost/beast/websocket/detail/pmd_extension.hpp>
18#include <boost/beast/websocket/detail/prng.hpp>
19#include <boost/beast/websocket/detail/service.hpp>
20#include <boost/beast/websocket/detail/soft_mutex.hpp>
21#include <boost/beast/websocket/detail/utf8_checker.hpp>
22#include <boost/beast/http/read.hpp>
23#include <boost/beast/http/write.hpp>
24#include <boost/beast/http/rfc7230.hpp>
25#include <boost/beast/core/buffers_cat.hpp>
26#include <boost/beast/core/buffers_prefix.hpp>
27#include <boost/beast/core/buffers_suffix.hpp>
28#include <boost/beast/core/flat_static_buffer.hpp>
29#include <boost/beast/core/saved_handler.hpp>
30#include <boost/beast/core/static_buffer.hpp>
31#include <boost/beast/core/stream_traits.hpp>
32#include <boost/beast/core/detail/clamp.hpp>
33#include <boost/asio/steady_timer.hpp>
34#include <boost/core/empty_value.hpp>
35#include <boost/enable_shared_from_this.hpp>
36#include <boost/shared_ptr.hpp>
37#include <boost/optional.hpp>
38
39namespace boost {
40namespace beast {
41namespace websocket {
42
43template<
44 class NextLayer, bool deflateSupported>
45struct stream<NextLayer, deflateSupported>::impl_type
46 : boost::empty_value<NextLayer>
47 , detail::service::impl_type
48 , detail::impl_base<deflateSupported>
49{
50 NextLayer& stream() noexcept
51 {
52 return this->boost::empty_value<
53 NextLayer>::get();
54 }
55
56 boost::weak_ptr<impl_type>
57 weak_from_this()
58 {
59 return boost::static_pointer_cast<
60 impl_type>(this->detail::service::
61 impl_type::shared_from_this());
62 }
63
64 boost::shared_ptr<impl_type>
65 shared_this()
66 {
67 return boost::static_pointer_cast<
68 impl_type>(this->detail::service::
69 impl_type::shared_from_this());
70 }
71 using executor_type = typename std::decay<NextLayer>::type::executor_type;
72 typename net::steady_timer::rebind_executor<executor_type>::other
73 timer; // used for timeouts
74 close_reason cr; // set from received close frame
75 control_cb_type ctrl_cb; // control callback
76
77 std::size_t rd_msg_max /* max message size */ = 16 * 1024 * 1024;
78 std::uint64_t rd_size /* total size of current message so far */ = 0;
79 std::uint64_t rd_remain /* message frame bytes left in current frame */ = 0;
80 detail::frame_header rd_fh; // current frame header
81 detail::prepared_key rd_key; // current stateful mask key
82 detail::frame_buffer rd_fb; // to write control frames (during reads)
83 detail::utf8_checker rd_utf8; // to validate utf8
84 static_buffer<
85 +tcp_frame_size> rd_buf; // buffer for reads
86 detail::opcode rd_op /* current message binary or text */ = detail::opcode::text;
87 bool rd_cont /* `true` if the next frame is a continuation */ = false;
88 bool rd_done /* set when a message is done */ = true;
89 bool rd_close /* did we read a close frame? */ = false;
90 detail::soft_mutex rd_block; // op currently reading
91
92 role_type role /* server or client */ = role_type::client;
93 status status_ /* state of the object */ = status::closed;
94
95 detail::soft_mutex wr_block; // op currently writing
96 bool wr_close /* did we write a close frame? */ = false;
97 bool wr_cont /* next write is a continuation */ = false;
98 bool wr_frag /* autofrag the current message */ = false;
99 bool wr_frag_opt /* autofrag option setting */ = true;
100 bool wr_compress; /* compress current message */
101 bool wr_compress_opt /* compress message setting */ = true;
102 detail::opcode wr_opcode /* message type */ = detail::opcode::text;
103 std::unique_ptr<
104 std::uint8_t[]> wr_buf; // write buffer
105 std::size_t wr_buf_size /* write buffer size (current message) */ = 0;
106 std::size_t wr_buf_opt /* write buffer size option setting */ = 4096;
107 detail::fh_buffer wr_fb; // header buffer used for writes
108
109 saved_handler op_rd; // paused read op
110 saved_handler op_wr; // paused write op
111 saved_handler op_ping; // paused ping op
112 saved_handler op_idle_ping; // paused idle ping op
113 saved_handler op_close; // paused close op
114 saved_handler op_r_rd; // paused read op (async read)
115 saved_handler op_r_close; // paused close op (async read)
116
117 bool idle_pinging = false;
118 bool secure_prng_ = true;
119 bool ec_delivered = false;
120 bool timed_out = false;
121 int idle_counter = 0;
122
123 detail::decorator decorator_opt; // Decorator for HTTP messages
124 timeout timeout_opt; // Timeout/idle settings
125
126 template<class... Args>
127 impl_type(Args&&... args)
128 : boost::empty_value<NextLayer>(
129 boost::empty_init_t{},
130 std::forward<Args>(args)...)
131 , detail::service::impl_type(
132 this->get_context(
133 this->boost::empty_value<NextLayer>::get().get_executor()))
134 , timer(this->boost::empty_value<NextLayer>::get().get_executor())
135 {
136 timeout_opt.handshake_timeout = none();
137 timeout_opt.idle_timeout = none();
138 timeout_opt.keep_alive_pings = false;
139 }
140
141 void
142 shutdown() override
143 {
144 op_rd.reset();
145 op_wr.reset();
146 op_ping.reset();
147 op_idle_ping.reset();
148 op_close.reset();
149 op_r_rd.reset();
150 op_r_close.reset();
151 }
152
153 void
154 open(role_type role_)
155 {
156 // VFALCO TODO analyze and remove dupe code in reset()
157 timer.expires_at(never());
158 timed_out = false;
159 cr.code = close_code::none;
160 role = role_;
161 status_ = status::open;
162 rd_remain = 0;
163 rd_cont = false;
164 rd_done = true;
165 // Can't clear this because accept uses it
166 //rd_buf.reset();
167 rd_fh.fin = false;
168 rd_close = false;
169 wr_close = false;
170 // These should not be necessary, because all completion
171 // handlers must be allowed to execute otherwise the
172 // stream exhibits undefined behavior.
173 wr_block.reset();
174 rd_block.reset();
175
176 wr_cont = false;
177 wr_buf_size = 0;
178
179 this->open_pmd(role);
180 }
181
182 void
183 close()
184 {
185 timer.cancel();
186 wr_buf.reset();
187 this->close_pmd();
188 }
189
190 void
191 reset()
192 {
193 BOOST_ASSERT(status_ != status::open);
194 timer.expires_at(never());
195 cr.code = close_code::none;
196 rd_remain = 0;
197 rd_cont = false;
198 rd_done = true;
199 rd_buf.consume(n: rd_buf.size());
200 rd_fh.fin = false;
201 rd_close = false;
202 wr_close = false;
203 wr_cont = false;
204 // These should not be necessary, because all completion
205 // handlers must be allowed to execute otherwise the
206 // stream exhibits undefined behavior.
207 wr_block.reset();
208 rd_block.reset();
209
210 // VFALCO Is this needed?
211 timer.cancel();
212 }
213
214 void
215 time_out()
216 {
217 timed_out = true;
218 change_status(new_status: status::closed);
219 close_socket(get_lowest_layer(stream()));
220 }
221
222 // Called just before sending
223 // the first frame of each message
224 void
225 begin_msg(std::size_t n_bytes)
226 {
227 wr_frag = wr_frag_opt;
228 wr_compress =
229 this->pmd_enabled() &&
230 wr_compress_opt &&
231 this->should_compress(n_bytes);
232
233 // Maintain the write buffer
234 if( this->pmd_enabled() ||
235 role == role_type::client)
236 {
237 if(! wr_buf ||
238 wr_buf_size != wr_buf_opt)
239 {
240 wr_buf_size = wr_buf_opt;
241 wr_buf = boost::make_unique_noinit<
242 std::uint8_t[]>(size: wr_buf_size);
243 }
244 }
245 else
246 {
247 wr_buf_size = wr_buf_opt;
248 wr_buf.reset();
249 }
250
251 //
252 }
253
254 //--------------------------------------------------------------------------
255
256 template<class Decorator>
257 request_type
258 build_request(
259 detail::sec_ws_key_type& key,
260 string_view host, string_view target,
261 Decorator const& decorator);
262
263 void
264 on_response(
265 response_type const& res,
266 detail::sec_ws_key_type const& key,
267 error_code& ec);
268
269 template<class Body, class Allocator, class Decorator>
270 response_type
271 build_response(
272 http::request<Body,
273 http::basic_fields<Allocator>> const& req,
274 Decorator const& decorator,
275 error_code& result);
276
277 // Attempt to read a complete frame header.
278 // Returns `false` if more bytes are needed
279 template<class DynamicBuffer>
280 bool
281 parse_fh(detail::frame_header& fh,
282 DynamicBuffer& b, error_code& ec);
283
284 std::uint32_t
285 create_mask()
286 {
287 auto g = detail::make_prng(secure: secure_prng_);
288 for(;;)
289 if(auto key = g())
290 return key;
291 }
292
293 template<class DynamicBuffer>
294 std::size_t
295 read_size_hint_db(DynamicBuffer& buffer) const
296 {
297 auto const initial_size = (std::min)(
298 +tcp_frame_size,
299 buffer.max_size() - buffer.size());
300 if(initial_size == 0)
301 return 1; // buffer is full
302 return this->read_size_hint_pmd(
303 initial_size, rd_done, rd_remain, rd_fh);
304 }
305
306 template<class DynamicBuffer>
307 void
308 write_ping(DynamicBuffer& db,
309 detail::opcode code, ping_data const& data);
310
311 template<class DynamicBuffer>
312 void
313 write_close(DynamicBuffer& db, close_reason const& cr);
314
315 //--------------------------------------------------------------------------
316
317 void
318 set_option(timeout const& opt)
319 {
320 if( opt.handshake_timeout == none() &&
321 opt.idle_timeout == none())
322 {
323 // turn timer off
324 timer.cancel();
325 timer.expires_at(never());
326 }
327
328 timeout_opt = opt;
329 }
330
331 // Determine if an operation should stop and
332 // deliver an error code to the completion handler.
333 //
334 // This function must be called at the beginning
335 // of every composed operation, and every time a
336 // composed operation receives an intermediate
337 // completion.
338 //
339 bool
340 check_stop_now(error_code& ec)
341 {
342 // Deliver the timeout to the first caller
343 if(timed_out)
344 {
345 timed_out = false;
346 BOOST_BEAST_ASSIGN_EC(ec, beast::error::timeout);
347 return true;
348 }
349
350 // If the stream is closed then abort
351 if( status_ == status::closed ||
352 status_ == status::failed)
353 {
354 //BOOST_ASSERT(ec_delivered);
355 BOOST_BEAST_ASSIGN_EC(ec, net::error::operation_aborted);
356 return true;
357 }
358
359 // If no error then keep going
360 if(! ec)
361 return false;
362
363 // Is this the first error seen?
364 if(ec_delivered)
365 {
366 // No, so abort
367 BOOST_BEAST_ASSIGN_EC(ec, net::error::operation_aborted);
368 return true;
369 }
370
371 // Deliver the error to the completion handler
372 ec_delivered = true;
373 if(status_ != status::closed)
374 status_ = status::failed;
375 return true;
376 }
377
378 // Change the status of the stream
379 void
380 change_status(status new_status)
381 {
382 switch(new_status)
383 {
384 case status::handshake:
385 break;
386
387 case status::open:
388 break;
389
390 case status::closing:
391 //BOOST_ASSERT(status_ == status::open);
392 break;
393
394 case status::failed:
395 case status::closed:
396 // this->close(); // Is this right?
397 break;
398
399 default:
400 break;
401 }
402 status_ = new_status;
403 }
404
405 // Called to disarm the idle timeout counter
406 void
407 reset_idle()
408 {
409 idle_counter = 0;
410 }
411
412 // Maintain the expiration timer
413 template<class Executor>
414 void
415 update_timer(Executor const& ex)
416 {
417 switch(status_)
418 {
419 case status::handshake:
420 BOOST_ASSERT(idle_counter == 0);
421 if(! is_timer_set() &&
422 timeout_opt.handshake_timeout != none())
423 {
424 timer.expires_after(
425 timeout_opt.handshake_timeout);
426
427 BOOST_ASIO_HANDLER_LOCATION((
428 __FILE__, __LINE__,
429 "websocket::check_stop_now"
430 ));
431
432 timer.async_wait(
433 timeout_handler<Executor>(
434 ex, this->weak_from_this()));
435 }
436 break;
437
438 case status::open:
439 if(timeout_opt.idle_timeout != none())
440 {
441 idle_counter = 0;
442 if(timeout_opt.keep_alive_pings)
443 timer.expires_after(
444 timeout_opt.idle_timeout / 2);
445 else
446 timer.expires_after(
447 timeout_opt.idle_timeout);
448
449 BOOST_ASIO_HANDLER_LOCATION((
450 __FILE__, __LINE__,
451 "websocket::check_stop_now"
452 ));
453
454 timer.async_wait(
455 timeout_handler<Executor>(
456 ex, this->weak_from_this()));
457 }
458 else
459 {
460 timer.cancel();
461 timer.expires_at(never());
462 }
463 break;
464
465 case status::closing:
466 if(timeout_opt.handshake_timeout != none())
467 {
468 idle_counter = 0;
469 timer.expires_after(
470 timeout_opt.handshake_timeout);
471
472 BOOST_ASIO_HANDLER_LOCATION((
473 __FILE__, __LINE__,
474 "websocket::check_stop_now"
475 ));
476
477 timer.async_wait(
478 timeout_handler<Executor>(
479 ex, this->weak_from_this()));
480 }
481 else
482 {
483 // VFALCO This assert goes off when there's also
484 // a pending read with the timer set. The bigger
485 // fix is to give close its own timeout, instead
486 // of using the handshake timeout.
487 // BOOST_ASSERT(! is_timer_set());
488 }
489 break;
490
491 case status::failed:
492 case status::closed:
493 // this->close(); // Is this right?
494 timer.cancel();
495 timer.expires_at(never());
496 break;
497 }
498 }
499
500private:
501 template<class Executor>
502 static net::execution_context&
503 get_context(Executor const& ex,
504 typename std::enable_if< net::execution::is_executor<Executor>::value >::type* = 0)
505 {
506 return net::query(ex, net::execution::context);
507 }
508
509 template<class Executor>
510 static net::execution_context&
511 get_context(Executor const& ex,
512 typename std::enable_if< !net::execution::is_executor<Executor>::value >::type* = 0)
513 {
514 return ex.context();
515 }
516
517 bool
518 is_timer_set() const
519 {
520 return timer.expiry() != never();
521 }
522
523 template<class Executor>
524 class timeout_handler
525 : boost::empty_value<Executor>
526 {
527 boost::weak_ptr<impl_type> wp_;
528
529 public:
530 timeout_handler(
531 Executor const& ex,
532 boost::weak_ptr<impl_type>&& wp)
533 : boost::empty_value<Executor>(
534 boost::empty_init_t{}, ex)
535 , wp_(std::move(wp))
536 {
537 }
538
539 using executor_type = Executor;
540
541 executor_type
542 get_executor() const noexcept
543 {
544 return this->get();
545 }
546
547 void
548 operator()(error_code ec)
549 {
550 // timer canceled?
551 if(ec == net::error::operation_aborted)
552 return;
553 BOOST_ASSERT(! ec);
554
555 // stream destroyed?
556 auto sp = wp_.lock();
557 if(! sp)
558 return;
559 auto& impl = *sp;
560
561 switch(impl.status_)
562 {
563 case status::handshake:
564 impl.time_out();
565 return;
566
567 case status::open:
568 // timeout was disabled
569 if(impl.timeout_opt.idle_timeout == none())
570 return;
571
572 if( impl.timeout_opt.keep_alive_pings &&
573 impl.idle_counter < 1)
574 {
575 {
576 BOOST_ASIO_HANDLER_LOCATION((
577 __FILE__, __LINE__,
578 "websocket::timeout_handler"
579 ));
580
581 idle_ping_op<Executor>(sp, get_executor());
582 }
583 ++impl.idle_counter;
584 impl.timer.expires_after(
585 impl.timeout_opt.idle_timeout / 2);
586
587 {
588 BOOST_ASIO_HANDLER_LOCATION((
589 __FILE__, __LINE__,
590 "websocket::timeout_handler"
591 ));
592
593 impl.timer.async_wait(std::move(*this));
594 }
595 return;
596 }
597
598 impl.time_out();
599 return;
600
601 case status::closing:
602 impl.time_out();
603 return;
604
605 case status::closed:
606 case status::failed:
607 // nothing to do?
608 return;
609 }
610 }
611 };
612};
613
614//--------------------------------------------------------------------------
615//
616// client
617//
618//--------------------------------------------------------------------------
619
620template<class NextLayer, bool deflateSupported>
621template<class Decorator>
622request_type
623stream<NextLayer, deflateSupported>::impl_type::
624build_request(
625 detail::sec_ws_key_type& key,
626 string_view host, string_view target,
627 Decorator const& decorator)
628{
629 request_type req;
630 req.target(s: target);
631 req.version(value: 11);
632 req.method(v: http::verb::get);
633 req.set(name: http::field::host, value: host);
634 req.set(name: http::field::upgrade, value: "websocket");
635 req.set(name: http::field::connection, value: "Upgrade");
636 detail::make_sec_ws_key(key);
637 req.set(name: http::field::sec_websocket_key, value: to_string_view(s: key));
638 req.set(name: http::field::sec_websocket_version, value: "13");
639 this->build_request_pmd(req);
640 decorator_opt(req);
641 decorator(req);
642 return req;
643}
644
645// Called when the WebSocket Upgrade response is received
646template<class NextLayer, bool deflateSupported>
647void
648stream<NextLayer, deflateSupported>::impl_type::
649on_response(
650 response_type const& res,
651 detail::sec_ws_key_type const& key,
652 error_code& ec)
653{
654 auto const err =
655 [&](error e)
656 {
657 BOOST_BEAST_ASSIGN_EC(ec, e);
658 };
659 if(res.result() != http::status::switching_protocols)
660 return err(error::upgrade_declined);
661 if(res.version() != 11)
662 return err(error::bad_http_version);
663 {
664 auto const it = res.find(name: http::field::connection);
665 if(it == res.end())
666 return err(error::no_connection);
667 if(! http::token_list{it->value()}.exists(s: "upgrade"))
668 return err(error::no_connection_upgrade);
669 }
670 {
671 auto const it = res.find(name: http::field::upgrade);
672 if(it == res.end())
673 return err(error::no_upgrade);
674 if(! http::token_list{it->value()}.exists(s: "websocket"))
675 return err(error::no_upgrade_websocket);
676 }
677 {
678 auto const it = res.find(
679 name: http::field::sec_websocket_accept);
680 if(it == res.end())
681 return err(error::no_sec_accept);
682 detail::sec_ws_accept_type acc;
683 detail::make_sec_ws_accept(accept&: acc, key: to_string_view(s: key));
684 if (to_string_view(s: acc).compare(str: it->value()) != 0)
685 return err(error::bad_sec_accept);
686 }
687
688 ec = {};
689 this->on_response_pmd(res);
690 this->open(role_: role_type::client);
691}
692
693//------------------------------------------------------------------------------
694
695// Attempt to read a complete frame header.
696// Returns `false` if more bytes are needed
697template<class NextLayer, bool deflateSupported>
698template<class DynamicBuffer>
699bool
700stream<NextLayer, deflateSupported>::impl_type::
701parse_fh(
702 detail::frame_header& fh,
703 DynamicBuffer& b,
704 error_code& ec)
705{
706 if(buffer_bytes(b.data()) < 2)
707 {
708 // need more bytes
709 ec = {};
710 return false;
711 }
712 buffers_suffix<typename
713 DynamicBuffer::const_buffers_type> cb{
714 b.data()};
715 std::size_t need;
716 {
717 std::uint8_t tmp[2];
718 cb.consume(net::buffer_copy(
719 net::buffer(data&: tmp), cb));
720 fh.len = tmp[1] & 0x7f;
721 switch(fh.len)
722 {
723 case 126: need = 2; break;
724 case 127: need = 8; break;
725 default:
726 need = 0;
727 }
728 fh.mask = (tmp[1] & 0x80) != 0;
729 if(fh.mask)
730 need += 4;
731 if(buffer_bytes(cb) < need)
732 {
733 // need more bytes
734 ec = {};
735 return false;
736 }
737 fh.op = static_cast<
738 detail::opcode>(tmp[0] & 0x0f);
739 fh.fin = (tmp[0] & 0x80) != 0;
740 fh.rsv1 = (tmp[0] & 0x40) != 0;
741 fh.rsv2 = (tmp[0] & 0x20) != 0;
742 fh.rsv3 = (tmp[0] & 0x10) != 0;
743 }
744 switch(fh.op)
745 {
746 case detail::opcode::binary:
747 case detail::opcode::text:
748 if(rd_cont)
749 {
750 // new data frame when continuation expected
751 BOOST_BEAST_ASSIGN_EC(ec, error::bad_data_frame);
752 return false;
753 }
754 if(fh.rsv2 || fh.rsv3 ||
755 ! this->rd_deflated(fh.rsv1))
756 {
757 // reserved bits not cleared
758 BOOST_BEAST_ASSIGN_EC(ec, error::bad_reserved_bits);
759 return false;
760 }
761 break;
762
763 case detail::opcode::cont:
764 if(! rd_cont)
765 {
766 // continuation without an active message
767 BOOST_BEAST_ASSIGN_EC(ec, error::bad_continuation);
768 return false;
769 }
770 if(fh.rsv1 || fh.rsv2 || fh.rsv3)
771 {
772 // reserved bits not cleared
773 BOOST_BEAST_ASSIGN_EC(ec, error::bad_reserved_bits);
774 return false;
775 }
776 break;
777
778 default:
779 if(detail::is_reserved(op: fh.op))
780 {
781 // reserved opcode
782 BOOST_BEAST_ASSIGN_EC(ec, error::bad_opcode);
783 return false;
784 }
785 if(! fh.fin)
786 {
787 // fragmented control message
788 BOOST_BEAST_ASSIGN_EC(ec, error::bad_control_fragment);
789 return false;
790 }
791 if(fh.len > 125)
792 {
793 // invalid length for control message
794 BOOST_BEAST_ASSIGN_EC(ec, error::bad_control_size);
795 return false;
796 }
797 if(fh.rsv1 || fh.rsv2 || fh.rsv3)
798 {
799 // reserved bits not cleared
800 BOOST_BEAST_ASSIGN_EC(ec, error::bad_reserved_bits);
801 return false;
802 }
803 break;
804 }
805 if(role == role_type::server && ! fh.mask)
806 {
807 // unmasked frame from client
808 BOOST_BEAST_ASSIGN_EC(ec, error::bad_unmasked_frame);
809 return false;
810 }
811 if(role == role_type::client && fh.mask)
812 {
813 // masked frame from server
814 BOOST_BEAST_ASSIGN_EC(ec, error::bad_masked_frame);
815 return false;
816 }
817 if(detail::is_control(op: fh.op) &&
818 buffer_bytes(cb) < need + fh.len)
819 {
820 // Make the entire control frame payload
821 // get read in before we return `true`
822 return false;
823 }
824 switch(fh.len)
825 {
826 case 126:
827 {
828
829 std::uint16_t len_be;
830 BOOST_ASSERT(buffer_bytes(cb) >= sizeof(len_be));
831 cb.consume(net::buffer_copy(
832 net::mutable_buffer(&len_be, sizeof(len_be)), cb));
833 fh.len = endian::big_to_native(x: len_be);
834 if(fh.len < 126)
835 {
836 // length not canonical
837 BOOST_BEAST_ASSIGN_EC(ec, error::bad_size);
838 return false;
839 }
840 break;
841 }
842 case 127:
843 {
844 std::uint64_t len_be;
845 BOOST_ASSERT(buffer_bytes(cb) >= sizeof(len_be));
846 cb.consume(net::buffer_copy(
847 net::mutable_buffer(&len_be, sizeof(len_be)), cb));
848 fh.len = endian::big_to_native(x: len_be);
849 if(fh.len < 65536)
850 {
851 // length not canonical
852 BOOST_BEAST_ASSIGN_EC(ec, error::bad_size);
853 return false;
854 }
855 break;
856 }
857 }
858 if(fh.mask)
859 {
860 std::uint32_t key_le;
861 BOOST_ASSERT(buffer_bytes(cb) >= sizeof(key_le));
862 cb.consume(net::buffer_copy(
863 net::mutable_buffer(&key_le, sizeof(key_le)), cb));
864 fh.key = endian::little_to_native(x: key_le);
865 detail::prepare_key(prepared&: rd_key, key: fh.key);
866 }
867 else
868 {
869 // initialize this otherwise operator== breaks
870 fh.key = 0;
871 }
872 if(! detail::is_control(op: fh.op))
873 {
874 if(fh.op != detail::opcode::cont)
875 {
876 rd_size = 0;
877 rd_op = fh.op;
878 }
879 else
880 {
881 if(rd_size > (std::numeric_limits<
882 std::uint64_t>::max)() - fh.len)
883 {
884 // message size exceeds configured limit
885 BOOST_BEAST_ASSIGN_EC(ec, error::message_too_big);
886 return false;
887 }
888 }
889 if(! this->rd_deflated())
890 {
891 if(rd_msg_max && beast::detail::sum_exceeds(
892 x: rd_size, y: fh.len, z: rd_msg_max))
893 {
894 // message size exceeds configured limit
895 BOOST_BEAST_ASSIGN_EC(ec, error::message_too_big);
896 return false;
897 }
898 }
899 rd_cont = ! fh.fin;
900 rd_remain = fh.len;
901 }
902 b.consume(b.size() - buffer_bytes(cb));
903 ec = {};
904 return true;
905}
906
907template<class NextLayer, bool deflateSupported>
908template<class DynamicBuffer>
909void
910stream<NextLayer, deflateSupported>::impl_type::
911write_ping(DynamicBuffer& db,
912 detail::opcode code, ping_data const& data)
913{
914 detail::frame_header fh;
915 fh.op = code;
916 fh.fin = true;
917 fh.rsv1 = false;
918 fh.rsv2 = false;
919 fh.rsv3 = false;
920 fh.len = data.size();
921 fh.mask = role == role_type::client;
922 if(fh.mask)
923 fh.key = create_mask();
924 detail::write(db, fh);
925 if(data.empty())
926 return;
927 detail::prepared_key key;
928 if(fh.mask)
929 detail::prepare_key(prepared&: key, key: fh.key);
930 auto mb = db.prepare(data.size());
931 net::buffer_copy(mb,
932 net::const_buffer(
933 data.data(), data.size()));
934 if(fh.mask)
935 detail::mask_inplace(mb, key);
936 db.commit(data.size());
937}
938
939template<class NextLayer, bool deflateSupported>
940template<class DynamicBuffer>
941void
942stream<NextLayer, deflateSupported>::impl_type::
943write_close(DynamicBuffer& db, close_reason const& cr)
944{
945 using namespace boost::endian;
946 detail::frame_header fh;
947 fh.op = detail::opcode::close;
948 fh.fin = true;
949 fh.rsv1 = false;
950 fh.rsv2 = false;
951 fh.rsv3 = false;
952 fh.len = cr.code == close_code::none ?
953 0 : 2 + cr.reason.size();
954 if(role == role_type::client)
955 {
956 fh.mask = true;
957 fh.key = create_mask();
958 }
959 else
960 {
961 fh.mask = false;
962 }
963 detail::write(db, fh);
964 if(cr.code != close_code::none)
965 {
966 detail::prepared_key key;
967 if(fh.mask)
968 detail::prepare_key(prepared&: key, key: fh.key);
969 {
970 auto code_be = endian::native_to_big<std::uint16_t>(x: cr.code);
971 auto mb = db.prepare(2);
972 net::buffer_copy(mb,
973 net::const_buffer(&code_be, sizeof(code_be)));
974 if(fh.mask)
975 detail::mask_inplace(mb, key);
976 db.commit(2);
977 }
978 if(! cr.reason.empty())
979 {
980 auto mb = db.prepare(cr.reason.size());
981 net::buffer_copy(mb,
982 net::const_buffer(
983 cr.reason.data(), cr.reason.size()));
984 if(fh.mask)
985 detail::mask_inplace(mb, key);
986 db.commit(cr.reason.size());
987 }
988 }
989}
990
991} // websocket
992} // beast
993} // boost
994
995#endif
996

source code of boost/libs/beast/include/boost/beast/websocket/impl/stream_impl.hpp