1//
2// Copyright (c) 2022 Klemens D. Morgenstern (klemens dot morgenstern at gmx dot net)
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//------------------------------------------------------------------------------
11//
12// Example: Advanced server, flex (plain + SSL)
13//
14//------------------------------------------------------------------------------
15
16#include "example/common/server_certificate.hpp"
17
18#include <boost/beast/core.hpp>
19#include <boost/beast/http.hpp>
20#include <boost/beast/ssl.hpp>
21#include <boost/beast/websocket.hpp>
22#include <boost/beast/version.hpp>
23#include <boost/asio/as_tuple.hpp>
24#include <boost/asio/awaitable.hpp>
25#include <boost/asio/bind_executor.hpp>
26#include <boost/asio/bind_cancellation_slot.hpp>
27#include <boost/asio/co_spawn.hpp>
28#include <boost/asio/deferred.hpp>
29#include <boost/asio/detached.hpp>
30#include <boost/asio/dispatch.hpp>
31#include <boost/asio/experimental/parallel_group.hpp>
32#include <boost/asio/signal_set.hpp>
33#include <boost/asio/strand.hpp>
34#include <boost/make_unique.hpp>
35#include <boost/asio/use_awaitable.hpp>
36#include <boost/optional.hpp>
37#include <algorithm>
38#include <cstdlib>
39#include <functional>
40#include <iostream>
41#include <list>
42#include <memory>
43#include <queue>
44#include <string>
45#include <thread>
46#include <vector>
47
48#if defined(BOOST_ASIO_HAS_CO_AWAIT)
49
50
51namespace beast = boost::beast; // from <boost/beast.hpp>
52namespace http = beast::http; // from <boost/beast/http.hpp>
53namespace websocket = beast::websocket; // from <boost/beast/websocket.hpp>
54namespace net = boost::asio; // from <boost/asio.hpp>
55namespace ssl = boost::asio::ssl; // from <boost/asio/ssl.hpp>
56using tcp = boost::asio::ip::tcp; // from <boost/asio/ip/tcp.hpp>
57
58using executor_type = net::io_context::executor_type;
59using executor_with_default = net::as_tuple_t<net::use_awaitable_t<executor_type>>::executor_with_default<executor_type>;
60
61
62
63// Return a reasonable mime type based on the extension of a file.
64beast::string_view
65mime_type(beast::string_view path)
66{
67 using beast::iequals;
68 auto const ext = [&path]
69 {
70 auto const pos = path.rfind(".");
71 if(pos == beast::string_view::npos)
72 return beast::string_view{};
73 return path.substr(pos);
74 }();
75 if(iequals(ext, ".htm")) return "text/html";
76 if(iequals(ext, ".html")) return "text/html";
77 if(iequals(ext, ".php")) return "text/html";
78 if(iequals(ext, ".css")) return "text/css";
79 if(iequals(ext, ".txt")) return "text/plain";
80 if(iequals(ext, ".js")) return "application/javascript";
81 if(iequals(ext, ".json")) return "application/json";
82 if(iequals(ext, ".xml")) return "application/xml";
83 if(iequals(ext, ".swf")) return "application/x-shockwave-flash";
84 if(iequals(ext, ".flv")) return "video/x-flv";
85 if(iequals(ext, ".png")) return "image/png";
86 if(iequals(ext, ".jpe")) return "image/jpeg";
87 if(iequals(ext, ".jpeg")) return "image/jpeg";
88 if(iequals(ext, ".jpg")) return "image/jpeg";
89 if(iequals(ext, ".gif")) return "image/gif";
90 if(iequals(ext, ".bmp")) return "image/bmp";
91 if(iequals(ext, ".ico")) return "image/vnd.microsoft.icon";
92 if(iequals(ext, ".tiff")) return "image/tiff";
93 if(iequals(ext, ".tif")) return "image/tiff";
94 if(iequals(ext, ".svg")) return "image/svg+xml";
95 if(iequals(ext, ".svgz")) return "image/svg+xml";
96 return "application/text";
97}
98
99// Append an HTTP rel-path to a local filesystem path.
100// The returned path is normalized for the platform.
101std::string
102path_cat(
103 beast::string_view base,
104 beast::string_view path)
105{
106 if(base.empty())
107 return std::string(path);
108 std::string result(base);
109#ifdef BOOST_MSVC
110 char constexpr path_separator = '\\';
111 if(result.back() == path_separator)
112 result.resize(result.size() - 1);
113 result.append(path.data(), path.size());
114 for(auto& c : result)
115 if(c == '/')
116 c = path_separator;
117#else
118 char constexpr path_separator = '/';
119 if(result.back() == path_separator)
120 result.resize(result.size() - 1);
121 result.append(path.data(), path.size());
122#endif
123 return result;
124}
125
126// Return a response for the given request.
127//
128// The concrete type of the response message (which depends on the
129// request), is type-erased in message_generator.
130template<class Body, class Allocator>
131http::message_generator
132handle_request(
133 beast::string_view doc_root,
134 http::request<Body, http::basic_fields<Allocator>>&& req)
135{
136 // Returns a bad request response
137 auto const bad_request =
138 [&req](beast::string_view why)
139 {
140 http::response<http::string_body> res{http::status::bad_request, req.version()};
141 res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
142 res.set(http::field::content_type, "text/html");
143 res.keep_alive(req.keep_alive());
144 res.body() = std::string(why);
145 res.prepare_payload();
146 return res;
147 };
148
149 // Returns a not found response
150 auto const not_found =
151 [&req](beast::string_view target)
152 {
153 http::response<http::string_body> res{http::status::not_found, req.version()};
154 res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
155 res.set(http::field::content_type, "text/html");
156 res.keep_alive(req.keep_alive());
157 res.body() = "The resource '" + std::string(target) + "' was not found.";
158 res.prepare_payload();
159 return res;
160 };
161
162 // Returns a server error response
163 auto const server_error =
164 [&req](beast::string_view what)
165 {
166 http::response<http::string_body> res{http::status::internal_server_error, req.version()};
167 res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
168 res.set(http::field::content_type, "text/html");
169 res.keep_alive(req.keep_alive());
170 res.body() = "An error occurred: '" + std::string(what) + "'";
171 res.prepare_payload();
172 return res;
173 };
174
175 // Make sure we can handle the method
176 if( req.method() != http::verb::get &&
177 req.method() != http::verb::head)
178 return bad_request("Unknown HTTP-method");
179
180 // Request path must be absolute and not contain "..".
181 if( req.target().empty() ||
182 req.target()[0] != '/' ||
183 req.target().find("..") != beast::string_view::npos)
184 return bad_request("Illegal request-target");
185
186 // Build the path to the requested file
187 std::string path = path_cat(doc_root, req.target());
188 if(req.target().back() == '/')
189 path.append("index.html");
190
191 // Attempt to open the file
192 beast::error_code ec;
193 http::file_body::value_type body;
194 body.open(path.c_str(), beast::file_mode::scan, ec);
195
196 // Handle the case where the file doesn't exist
197 if(ec == beast::errc::no_such_file_or_directory)
198 return not_found(req.target());
199
200 // Handle an unknown error
201 if(ec)
202 return server_error(ec.message());
203
204 // Cache the size since we need it after the move
205 auto const size = body.size();
206
207 // Respond to HEAD request
208 if(req.method() == http::verb::head)
209 {
210 http::response<http::empty_body> res{http::status::ok, req.version()};
211 res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
212 res.set(http::field::content_type, mime_type(path));
213 res.content_length(size);
214 res.keep_alive(req.keep_alive());
215 return res;
216 }
217
218 // Respond to GET request
219 http::response<http::file_body> res{
220 std::piecewise_construct,
221 std::make_tuple(std::move(body)),
222 std::make_tuple(http::status::ok, req.version())};
223 res.set(http::field::server, BOOST_BEAST_VERSION_STRING);
224 res.set(http::field::content_type, mime_type(path));
225 res.content_length(size);
226 res.keep_alive(req.keep_alive());
227 return res;
228}
229
230//------------------------------------------------------------------------------
231
232// Report a failure
233void
234fail(beast::error_code ec, char const* what)
235{
236 // ssl::error::stream_truncated, also known as an SSL "short read",
237 // indicates the peer closed the connection without performing the
238 // required closing handshake (for example, Google does this to
239 // improve performance). Generally this can be a security issue,
240 // but if your communication protocol is self-terminated (as
241 // it is with both HTTP and WebSocket) then you may simply
242 // ignore the lack of close_notify.
243 //
244 // https://github.com/boostorg/beast/issues/38
245 //
246 // https://security.stackexchange.com/questions/91435/how-to-handle-a-malicious-ssl-tls-shutdown
247 //
248 // When a short read would cut off the end of an HTTP message,
249 // Beast returns the error beast::http::error::partial_message.
250 // Therefore, if we see a short read here, it has occurred
251 // after the message has been completed, so it is safe to ignore it.
252
253 if(ec == net::ssl::error::stream_truncated)
254 return;
255
256 std::cerr << what << ": " << ec.message() << "\n";
257}
258
259// A simple helper for cancellation_slot
260struct cancellation_signals
261{
262 std::list<net::cancellation_signal> sigs;
263 std::mutex mtx;
264 void emit(net::cancellation_type ct = net::cancellation_type::all)
265 {
266 std::lock_guard<std::mutex> _(mtx);
267
268 for (auto & sig : sigs)
269 sig.emit(ct);
270 }
271
272 net::cancellation_slot slot()
273 {
274 std::lock_guard<std::mutex> _(mtx);
275
276 auto itr = std::find_if(sigs.begin(), sigs.end(),
277 [](net::cancellation_signal & sig)
278 {
279 return !sig.slot().has_handler();
280 });
281
282 if (itr != sigs.end())
283 return itr->slot();
284 else
285 return sigs.emplace_back().slot();
286 }
287};
288
289//------------------------------------------------------------------------------
290
291// Echoes back all received WebSocket messages.
292// This uses the Curiously Recurring Template Pattern so that
293// the same code works with both SSL streams and regular sockets.
294template<class Derived>
295class websocket_session
296{
297 // Access the derived class, this is part of
298 // the Curiously Recurring Template Pattern idiom.
299 Derived&
300 derived()
301 {
302 return static_cast<Derived&>(*this);
303 }
304
305 beast::flat_buffer buffer_;
306
307 // Start the asynchronous operation
308 template<class Body, class Allocator>
309 void
310 do_accept(http::request<Body, http::basic_fields<Allocator>> req)
311 {
312 // Set suggested timeout settings for the websocket
313 derived().ws().set_option(
314 websocket::stream_base::timeout::suggested(
315 beast::role_type::server));
316
317 // Set a decorator to change the Server of the handshake
318 derived().ws().set_option(
319 websocket::stream_base::decorator(
320 [](websocket::response_type& res)
321 {
322 res.set(http::field::server,
323 std::string(BOOST_BEAST_VERSION_STRING) +
324 " advanced-server-flex");
325 }));
326
327 // Accept the websocket handshake
328 derived().ws().async_accept(
329 req,
330 beast::bind_front_handler(
331 &websocket_session::on_accept,
332 derived().shared_from_this()));
333 }
334
335private:
336 void
337 on_accept(beast::error_code ec)
338 {
339 if(ec)
340 return fail(ec, "accept");
341
342 // Read a message
343 do_read();
344 }
345
346 void
347 do_read()
348 {
349 // Read a message into our buffer
350 derived().ws().async_read(
351 buffer_,
352 beast::bind_front_handler(
353 &websocket_session::on_read,
354 derived().shared_from_this()));
355 }
356
357 void
358 on_read(
359 beast::error_code ec,
360 std::size_t bytes_transferred)
361 {
362 boost::ignore_unused(bytes_transferred);
363
364 // This indicates that the websocket_session was closed
365 if(ec == websocket::error::closed)
366 return;
367
368 if(ec)
369 return fail(ec, "read");
370
371 // Echo the message
372 derived().ws().text(derived().ws().got_text());
373 derived().ws().async_write(
374 buffer_.data(),
375 beast::bind_front_handler(
376 &websocket_session::on_write,
377 derived().shared_from_this()));
378 }
379
380 void
381 on_write(
382 beast::error_code ec,
383 std::size_t bytes_transferred)
384 {
385 boost::ignore_unused(bytes_transferred);
386
387 if(ec)
388 return fail(ec, "write");
389
390 // Clear the buffer
391 buffer_.consume(buffer_.size());
392
393 // Do another read
394 do_read();
395 }
396
397public:
398 // Start the asynchronous operation
399 template<class Body, class Allocator>
400 void
401 run(http::request<Body, http::basic_fields<Allocator>> req)
402 {
403 // Accept the WebSocket upgrade request
404 do_accept(std::move(req));
405 }
406};
407
408//------------------------------------------------------------------------------
409
410// Handles a plain WebSocket connection
411class plain_websocket_session
412 : public websocket_session<plain_websocket_session>
413 , public std::enable_shared_from_this<plain_websocket_session>
414{
415 websocket::stream<beast::tcp_stream> ws_;
416
417public:
418 // Create the session
419 explicit
420 plain_websocket_session(
421 beast::tcp_stream&& stream)
422 : ws_(std::move(stream))
423 {
424 }
425
426 // Called by the base class
427 websocket::stream<beast::tcp_stream>&
428 ws()
429 {
430 return ws_;
431 }
432};
433
434//------------------------------------------------------------------------------
435
436// Handles an SSL WebSocket connection
437class ssl_websocket_session
438 : public websocket_session<ssl_websocket_session>
439 , public std::enable_shared_from_this<ssl_websocket_session>
440{
441 websocket::stream<
442 beast::ssl_stream<beast::tcp_stream>> ws_;
443
444public:
445 // Create the ssl_websocket_session
446 explicit
447 ssl_websocket_session(
448 beast::ssl_stream<beast::tcp_stream>&& stream)
449 : ws_(std::move(stream))
450 {
451 }
452
453 // Called by the base class
454 websocket::stream<
455 beast::ssl_stream<beast::tcp_stream>>&
456 ws()
457 {
458 return ws_;
459 }
460};
461
462//------------------------------------------------------------------------------
463
464template<class Body, class Allocator>
465void
466make_websocket_session(
467 beast::tcp_stream stream,
468 http::request<Body, http::basic_fields<Allocator>> req)
469{
470 std::make_shared<plain_websocket_session>(
471 std::move(stream))->run(std::move(req));
472}
473
474template<class Body, class Allocator>
475void
476make_websocket_session(
477 beast::ssl_stream<beast::tcp_stream> stream,
478 http::request<Body, http::basic_fields<Allocator>> req)
479{
480 std::make_shared<ssl_websocket_session>(
481 std::move(stream))->run(std::move(req));
482}
483
484//------------------------------------------------------------------------------
485
486// Handles an HTTP server connection.
487// This uses the Curiously Recurring Template Pattern so that
488// the same code works with both SSL streams and regular sockets.
489template<class Derived>
490class http_session
491{
492 std::shared_ptr<std::string const> doc_root_;
493
494 // Access the derived class, this is part of
495 // the Curiously Recurring Template Pattern idiom.
496 Derived&
497 derived()
498 {
499 return static_cast<Derived&>(*this);
500 }
501
502 static constexpr std::size_t queue_limit = 8; // max responses
503 std::queue<http::message_generator> response_queue_;
504
505 // The parser is stored in an optional container so we can
506 // construct it from scratch it at the beginning of each new message.
507 boost::optional<http::request_parser<http::string_body>> parser_;
508
509protected:
510 beast::flat_buffer buffer_;
511
512public:
513 // Construct the session
514 http_session(
515 beast::flat_buffer buffer,
516 std::shared_ptr<std::string const> const& doc_root)
517 : doc_root_(doc_root)
518 , buffer_(std::move(buffer))
519 {
520 }
521
522 void
523 do_read()
524 {
525 // Construct a new parser for each message
526 parser_.emplace();
527
528 // Apply a reasonable limit to the allowed size
529 // of the body in bytes to prevent abuse.
530 parser_->body_limit(10000);
531
532 // Set the timeout.
533 beast::get_lowest_layer(
534 derived().stream()).expires_after(std::chrono::seconds(30));
535
536 // Read a request using the parser-oriented interface
537 http::async_read(
538 derived().stream(),
539 buffer_,
540 *parser_,
541 beast::bind_front_handler(
542 &http_session::on_read,
543 derived().shared_from_this()));
544 }
545
546 void
547 on_read(beast::error_code ec, std::size_t bytes_transferred)
548 {
549 boost::ignore_unused(bytes_transferred);
550
551 // This means they closed the connection
552 if(ec == http::error::end_of_stream)
553 return derived().do_eof();
554
555 if(ec)
556 return fail(ec, "read");
557
558 // See if it is a WebSocket Upgrade
559 if(websocket::is_upgrade(parser_->get()))
560 {
561 // Disable the timeout.
562 // The websocket::stream uses its own timeout settings.
563 beast::get_lowest_layer(derived().stream()).expires_never();
564
565 // Create a websocket session, transferring ownership
566 // of both the socket and the HTTP request.
567 return make_websocket_session(
568 derived().release_stream(),
569 parser_->release());
570 }
571
572 // Send the response
573 queue_write(handle_request(*doc_root_, parser_->release()));
574
575 // If we aren't at the queue limit, try to pipeline another request
576 if (response_queue_.size() < queue_limit)
577 do_read();
578 }
579
580 void
581 queue_write(http::message_generator response)
582 {
583 // Allocate and store the work
584 response_queue_.push(std::move(response));
585
586 // If there was no previous work, start the write loop
587 if (response_queue_.size() == 1)
588 do_write();
589 }
590
591 // Called to start/continue the write-loop. Should not be called when
592 // write_loop is already active.
593 void
594 do_write()
595 {
596 if(! response_queue_.empty())
597 {
598 bool keep_alive = response_queue_.front().keep_alive();
599
600 beast::async_write(
601 derived().stream(),
602 std::move(response_queue_.front()),
603 beast::bind_front_handler(
604 &http_session::on_write,
605 derived().shared_from_this(),
606 keep_alive));
607 }
608 }
609
610 void
611 on_write(
612 bool keep_alive,
613 beast::error_code ec,
614 std::size_t bytes_transferred)
615 {
616 boost::ignore_unused(bytes_transferred);
617
618 if(ec)
619 return fail(ec, "write");
620
621 if(! keep_alive)
622 {
623 // This means we should close the connection, usually because
624 // the response indicated the "Connection: close" semantic.
625 return derived().do_eof();
626 }
627
628 // Resume the read if it has been paused
629 if(response_queue_.size() == queue_limit)
630 do_read();
631
632 response_queue_.pop();
633
634 do_write();
635 }
636};
637
638//------------------------------------------------------------------------------
639
640// Handles a plain HTTP connection
641class plain_http_session
642 : public http_session<plain_http_session>
643 , public std::enable_shared_from_this<plain_http_session>
644{
645 beast::tcp_stream stream_;
646
647public:
648 // Create the session
649 plain_http_session(
650 beast::tcp_stream&& stream,
651 beast::flat_buffer&& buffer,
652 std::shared_ptr<std::string const> const& doc_root)
653 : http_session<plain_http_session>(
654 std::move(buffer),
655 doc_root)
656 , stream_(std::move(stream))
657 {
658 }
659
660 // Start the session
661 void
662 run()
663 {
664 this->do_read();
665 }
666
667 // Called by the base class
668 beast::tcp_stream&
669 stream()
670 {
671 return stream_;
672 }
673
674 // Called by the base class
675 beast::tcp_stream
676 release_stream()
677 {
678 return std::move(stream_);
679 }
680
681 // Called by the base class
682 void
683 do_eof()
684 {
685 // Send a TCP shutdown
686 beast::error_code ec;
687 stream_.socket().shutdown(tcp::socket::shutdown_send, ec);
688
689 // At this point the connection is closed gracefully
690 }
691};
692
693//------------------------------------------------------------------------------
694
695// Handles an SSL HTTP connection
696class ssl_http_session
697 : public http_session<ssl_http_session>
698 , public std::enable_shared_from_this<ssl_http_session>
699{
700 beast::ssl_stream<beast::tcp_stream> stream_;
701
702public:
703 // Create the http_session
704 ssl_http_session(
705 beast::tcp_stream&& stream,
706 ssl::context& ctx,
707 beast::flat_buffer&& buffer,
708 std::shared_ptr<std::string const> const& doc_root)
709 : http_session<ssl_http_session>(
710 std::move(buffer),
711 doc_root)
712 , stream_(std::move(stream), ctx)
713 {
714 }
715
716 // Start the session
717 void
718 run()
719 {
720 // Set the timeout.
721 beast::get_lowest_layer(stream_).expires_after(std::chrono::seconds(30));
722
723 // Perform the SSL handshake
724 // Note, this is the buffered version of the handshake.
725 stream_.async_handshake(
726 ssl::stream_base::server,
727 buffer_.data(),
728 beast::bind_front_handler(
729 &ssl_http_session::on_handshake,
730 shared_from_this()));
731 }
732
733 // Called by the base class
734 beast::ssl_stream<beast::tcp_stream>&
735 stream()
736 {
737 return stream_;
738 }
739
740 // Called by the base class
741 beast::ssl_stream<beast::tcp_stream>
742 release_stream()
743 {
744 return std::move(stream_);
745 }
746
747 // Called by the base class
748 void
749 do_eof()
750 {
751 // Set the timeout.
752 beast::get_lowest_layer(stream_).expires_after(std::chrono::seconds(30));
753
754 // Perform the SSL shutdown
755 stream_.async_shutdown(
756 beast::bind_front_handler(
757 &ssl_http_session::on_shutdown,
758 shared_from_this()));
759 }
760
761private:
762 void
763 on_handshake(
764 beast::error_code ec,
765 std::size_t bytes_used)
766 {
767 if(ec)
768 return fail(ec, "handshake");
769
770 // Consume the portion of the buffer used by the handshake
771 buffer_.consume(bytes_used);
772
773 do_read();
774 }
775
776 void
777 on_shutdown(beast::error_code ec)
778 {
779 if(ec)
780 return fail(ec, "shutdown");
781
782 // At this point the connection is closed gracefully
783 }
784};
785
786//------------------------------------------------------------------------------
787
788// Detects SSL handshakes
789class detect_session : public std::enable_shared_from_this<detect_session>
790{
791 beast::tcp_stream stream_;
792 ssl::context& ctx_;
793 std::shared_ptr<std::string const> doc_root_;
794 beast::flat_buffer buffer_;
795
796public:
797 explicit
798 detect_session(
799 tcp::socket&& socket,
800 ssl::context& ctx,
801 std::shared_ptr<std::string const> const& doc_root)
802 : stream_(std::move(socket))
803 , ctx_(ctx)
804 , doc_root_(doc_root)
805 {
806 }
807
808 // Launch the detector
809 void
810 run()
811 {
812 // We need to be executing within a strand to perform async operations
813 // on the I/O objects in this session. Although not strictly necessary
814 // for single-threaded contexts, this example code is written to be
815 // thread-safe by default.
816 net::dispatch(
817 stream_.get_executor(),
818 beast::bind_front_handler(
819 &detect_session::on_run,
820 this->shared_from_this()));
821 }
822
823 void
824 on_run()
825 {
826 // Set the timeout.
827 stream_.expires_after(std::chrono::seconds(30));
828
829 beast::async_detect_ssl(
830 stream_,
831 buffer_,
832 beast::bind_front_handler(
833 &detect_session::on_detect,
834 this->shared_from_this()));
835 }
836
837 void
838 on_detect(beast::error_code ec, bool result)
839 {
840 if(ec)
841 return fail(ec, "detect");
842
843 if(result)
844 {
845 // Launch SSL session
846 std::make_shared<ssl_http_session>(
847 std::move(stream_),
848 ctx_,
849 std::move(buffer_),
850 doc_root_)->run();
851 return;
852 }
853
854 // Launch plain session
855 std::make_shared<plain_http_session>(
856 std::move(stream_),
857 std::move(buffer_),
858 doc_root_)->run();
859 }
860};
861
862template<typename Stream>
863net::awaitable<void, executor_type> do_eof(Stream & stream)
864{
865 beast::error_code ec;
866 stream.socket().shutdown(tcp::socket::shutdown_send, ec);
867 co_return ;
868}
869
870template<typename Stream>
871BOOST_ASIO_NODISCARD net::awaitable<void, executor_type>
872do_eof(beast::ssl_stream<Stream> & stream)
873{
874 co_await stream.async_shutdown();
875}
876
877
878template<typename Stream, typename Body, typename Allocator>
879BOOST_ASIO_NODISCARD net::awaitable<void, executor_type>
880run_websocket_session(Stream & stream,
881 beast::flat_buffer & buffer,
882 http::request<Body, http::basic_fields<Allocator>> req,
883 const std::shared_ptr<std::string const> & doc_root)
884{
885 beast::websocket::stream<Stream&> ws{stream};
886
887 // Set suggested timeout settings for the websocket
888 ws.set_option(
889 websocket::stream_base::timeout::suggested(
890 beast::role_type::server));
891
892 // Set a decorator to change the Server of the handshake
893 ws.set_option(
894 websocket::stream_base::decorator(
895 [](websocket::response_type& res)
896 {
897 res.set(http::field::server,
898 std::string(BOOST_BEAST_VERSION_STRING) +
899 " advanced-server-flex");
900 }));
901
902 // Accept the websocket handshake
903 auto [ec] = co_await ws.async_accept(req);
904 if (ec)
905 co_return fail(ec, "accept");
906
907 while (true)
908 {
909
910
911 // Read a message
912 std::size_t bytes_transferred = 0u;
913 std::tie(ec, bytes_transferred) = co_await ws.async_read(buffer);
914
915 // This indicates that the websocket_session was closed
916 if (ec == websocket::error::closed)
917 co_return;
918 if (ec)
919 co_return fail(ec, "read");
920
921 ws.text(ws.got_text());
922 std::tie(ec, bytes_transferred) = co_await ws.async_write(buffer.data());
923
924 if (ec)
925 co_return fail(ec, "write");
926
927 // Clear the buffer
928 buffer.consume(buffer.size());
929 }
930}
931
932
933template<typename Stream>
934BOOST_ASIO_NODISCARD net::awaitable<void, executor_type>
935run_session(Stream & stream, beast::flat_buffer & buffer, const std::shared_ptr<std::string const> & doc_root)
936{
937 // a new parser must be used for every message
938 // so we use an optional to reconstruct it every time.
939 std::optional<http::request_parser<http::string_body>> parser;
940 parser.emplace();
941 // Apply a reasonable limit to the allowed size
942 // of the body in bytes to prevent abuse.
943 parser->body_limit(10000);
944
945 auto [ec, bytes_transferred] = co_await http::async_read(stream, buffer, *parser);
946
947 if(ec == http::error::end_of_stream)
948 co_await do_eof(stream);
949
950 if(ec)
951 co_return fail(ec, "read");
952
953 // this can be
954 // while ((co_await net::this_coro::cancellation_state).cancelled() == net::cancellation_type::none)
955 // on most compilers
956 for (auto cs = co_await net::this_coro::cancellation_state;
957 cs.cancelled() == net::cancellation_type::none;
958 cs = co_await net::this_coro::cancellation_state)
959 {
960 if(websocket::is_upgrade(parser->get()))
961 {
962 // Disable the timeout.
963 // The websocket::stream uses its own timeout settings.
964 beast::get_lowest_layer(stream).expires_never();
965
966 co_await run_websocket_session(stream, buffer, parser->release(), doc_root);
967 co_return ;
968 }
969
970 // we follow a different strategy then the other example: instead of queue responses,
971 // we always to one read & write in parallel.
972
973 auto res = handle_request(*doc_root, parser->release());
974 if (!res.keep_alive())
975 {
976 http::message_generator msg{std::move(res)};
977 auto [ec, sz] = co_await beast::async_write(stream, std::move(msg));
978 if (ec)
979 fail(ec, "write");
980 co_return ;
981 }
982
983 // we must use a new parser for every async_read
984 parser.reset();
985 parser.emplace();
986 parser->body_limit(10000);
987
988 http::message_generator msg{std::move(res)};
989
990 auto [_, ec_r, sz_r, ec_w, sz_w ] =
991 co_await net::experimental::make_parallel_group(
992 http::async_read(stream, buffer, *parser, net::deferred),
993 beast::async_write(stream, std::move(msg), net::deferred))
994 .async_wait(net::experimental::wait_for_all(),
995 net::as_tuple(net::use_awaitable_t<executor_type>{}));
996
997 if (ec_r)
998 co_return fail(ec_r, "read");
999
1000 if (ec_w)
1001 co_return fail(ec_w, "write");
1002
1003 }
1004
1005
1006}
1007
1008BOOST_ASIO_NODISCARD net::awaitable<void, executor_type>
1009detect_session(typename beast::tcp_stream::rebind_executor<executor_with_default>::other stream,
1010 net::ssl::context & ctx,
1011 std::shared_ptr<std::string const> doc_root)
1012{
1013 beast::flat_buffer buffer;
1014
1015 // Set the timeout.
1016 stream.expires_after(std::chrono::seconds(30));
1017 // on_run
1018 auto [ec, result] = co_await beast::async_detect_ssl(stream, buffer);
1019 // on_detect
1020 if (ec)
1021 co_return fail(ec, "detect");
1022
1023 if(result)
1024 {
1025 using stream_type = typename beast::tcp_stream::rebind_executor<executor_with_default>::other;
1026 beast::ssl_stream<stream_type> ssl_stream{std::move(stream), ctx};
1027 auto [ec, bytes_used] = co_await ssl_stream.async_handshake(net::ssl::stream_base::server, buffer.data());
1028
1029 if(ec)
1030 co_return fail(ec, "handshake");
1031
1032 buffer.consume(bytes_used);
1033 co_await run_session(ssl_stream, buffer, doc_root);
1034 }
1035 else
1036 co_await run_session(stream, buffer, doc_root);
1037
1038
1039}
1040
1041bool init_listener(typename tcp::acceptor::rebind_executor<executor_with_default>::other & acceptor,
1042 const tcp::endpoint &endpoint)
1043{
1044 beast::error_code ec;
1045 // Open the acceptor
1046 acceptor.open(endpoint.protocol(), ec);
1047 if(ec)
1048 {
1049 fail(ec, "open");
1050 return false;
1051 }
1052
1053 // Allow address reuse
1054 acceptor.set_option(net::socket_base::reuse_address(true), ec);
1055 if(ec)
1056 {
1057 fail(ec, "set_option");
1058 return false;
1059 }
1060
1061 // Bind to the server address
1062 acceptor.bind(endpoint, ec);
1063 if(ec)
1064 {
1065 fail(ec, "bind");
1066 return false;
1067 }
1068
1069 // Start listening for connections
1070 acceptor.listen(
1071 net::socket_base::max_listen_connections, ec);
1072 if(ec)
1073 {
1074 fail(ec, "listen");
1075 return false;
1076 }
1077 return true;
1078
1079}
1080
1081// Accepts incoming connections and launches the sessions.
1082BOOST_ASIO_NODISCARD net::awaitable<void, executor_type>
1083 listen(ssl::context& ctx,
1084 tcp::endpoint endpoint,
1085 std::shared_ptr<std::string const> doc_root,
1086 cancellation_signals & sig)
1087{
1088 typename tcp::acceptor::rebind_executor<executor_with_default>::other acceptor{co_await net::this_coro::executor};
1089 if (!init_listener(acceptor, endpoint))
1090 co_return;
1091
1092 while ((co_await net::this_coro::cancellation_state).cancelled() == net::cancellation_type::none)
1093 {
1094 auto [ec, sock] = co_await acceptor.async_accept();
1095 const auto exec = sock.get_executor();
1096 using stream_type = typename beast::tcp_stream::rebind_executor<executor_with_default>::other;
1097 if (!ec)
1098 // We dont't need a strand, since the awaitable is an implicit strand.
1099 net::co_spawn(exec,
1100 detect_session(stream_type(std::move(sock)), ctx, doc_root),
1101 net::bind_cancellation_slot(sig.slot(), net::detached));
1102 }
1103}
1104
1105//------------------------------------------------------------------------------
1106
1107int main(int argc, char* argv[])
1108{
1109 // Check command line arguments.
1110 if (argc != 5)
1111 {
1112 std::cerr <<
1113 "Usage: advanced-server-flex-awaitable <address> <port> <doc_root> <threads>\n" <<
1114 "Example:\n" <<
1115 " advanced-server-flex-awaitable 0.0.0.0 8080 . 1\n";
1116 return EXIT_FAILURE;
1117 }
1118 auto const address = net::ip::make_address(argv[1]);
1119 auto const port = static_cast<unsigned short>(std::atoi(argv[2]));
1120 auto const doc_root = std::make_shared<std::string>(argv[3]);
1121 auto const threads = std::max<int>(1, std::atoi(argv[4]));
1122
1123 // The io_context is required for all I/O
1124 net::io_context ioc{threads};
1125
1126 // The SSL context is required, and holds certificates
1127 ssl::context ctx{ssl::context::tlsv12};
1128
1129 // This holds the self-signed certificate used by the server
1130 load_server_certificate(ctx);
1131
1132 // Cancellation-signal for SIGINT
1133 cancellation_signals cancellation;
1134
1135 // Create and launch a listening routine
1136 net::co_spawn(
1137 ioc,
1138 listen(ctx, tcp::endpoint{address, port}, doc_root, cancellation),
1139 net::bind_cancellation_slot(cancellation.slot(), net::detached));
1140
1141
1142 // Capture SIGINT and SIGTERM to perform a clean shutdown
1143 net::signal_set signals(ioc, SIGINT, SIGTERM);
1144 signals.async_wait(
1145 [&](beast::error_code const&, int sig)
1146 {
1147 if (sig == SIGINT)
1148 cancellation.emit(net::cancellation_type::all);
1149 else
1150 {
1151 // Stop the `io_context`. This will cause `run()`
1152 // to return immediately, eventually destroying the
1153 // `io_context` and all of the sockets in it.
1154 ioc.stop();
1155 }
1156 });
1157
1158 // Run the I/O service on the requested number of threads
1159 std::vector<std::thread> v;
1160 v.reserve(threads - 1);
1161 for(auto i = threads - 1; i > 0; --i)
1162 v.emplace_back(
1163 [&ioc]
1164 {
1165 ioc.run();
1166 });
1167 ioc.run();
1168
1169 // (If we get here, it means we got a SIGINT or SIGTERM)
1170
1171 // Block until all the threads exit
1172 for(auto& t : v)
1173 t.join();
1174
1175 return EXIT_SUCCESS;
1176}
1177
1178
1179#else
1180
1181int main(int, char * [])
1182{
1183 std::printf(format: "awaitables require C++20\n");
1184 return 1;
1185}
1186
1187#endif
1188

source code of boost/libs/beast/example/advanced/server-flex-awaitable/advanced_server_flex_awaitable.cpp