// // Copyright (c) 2016-2019 Vinnie Falco (vinnie dot falco at gmail dot com) // // Distributed under the Boost Software License, Version 1.0. (See accompanying // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) // // Official repository: https://github.com/boostorg/beast // //------------------------------------------------------------------------------ // // Example: WebSocket server, stackless coroutine // //------------------------------------------------------------------------------ #include #include #include #include #include #include #include #include #include #include #include #include #include namespace beast = boost::beast; // from namespace http = beast::http; // from namespace websocket = beast::websocket; // from namespace net = boost::asio; // from using tcp = boost::asio::ip::tcp; // from //------------------------------------------------------------------------------ // Report a failure void fail(beast::error_code ec, char const* what) { std::cerr << what << ": " << ec.message() << "\n"; } // Echoes back all received WebSocket messages class session : public boost::asio::coroutine , public std::enable_shared_from_this { websocket::stream ws_; beast::flat_buffer buffer_; public: // Take ownership of the socket explicit session(tcp::socket socket) : ws_(std::move(socket)) { } // Start the asynchronous operation void run() { // We need to be executing within a strand to perform async operations // on the I/O objects in this session. Although not strictly necessary // for single-threaded contexts, this example code is written to be // thread-safe by default. net::dispatch(ws_.get_executor(), beast::bind_front_handler(&session::loop, shared_from_this(), beast::error_code{}, 0)); } #include void loop( beast::error_code ec, std::size_t bytes_transferred) { boost::ignore_unused(bytes_transferred); reenter(*this) { // Set suggested timeout settings for the websocket ws_.set_option( websocket::stream_base::timeout::suggested( beast::role_type::server)); // Set a decorator to change the Server of the handshake ws_.set_option(websocket::stream_base::decorator( [](websocket::response_type& res) { res.set(http::field::server, std::string(BOOST_BEAST_VERSION_STRING) + " websocket-server-stackless"); })); // Accept the websocket handshake yield ws_.async_accept( std::bind( &session::loop, shared_from_this(), std::placeholders::_1, 0)); if(ec) return fail(ec, "accept"); for(;;) { // Read a message into our buffer yield ws_.async_read( buffer_, std::bind( &session::loop, shared_from_this(), std::placeholders::_1, std::placeholders::_2)); if(ec == websocket::error::closed) { // This indicates that the session was closed return; } if(ec) fail(ec, "read"); // Echo the message ws_.text(ws_.got_text()); yield ws_.async_write( buffer_.data(), std::bind( &session::loop, shared_from_this(), std::placeholders::_1, std::placeholders::_2)); if(ec) return fail(ec, "write"); // Clear the buffer buffer_.consume(buffer_.size()); } } } #include }; //------------------------------------------------------------------------------ // Accepts incoming connections and launches the sessions class listener : public boost::asio::coroutine , public std::enable_shared_from_this { net::io_context& ioc_; tcp::acceptor acceptor_; tcp::socket socket_; public: listener( net::io_context& ioc, tcp::endpoint endpoint) : ioc_(ioc) , acceptor_(net::make_strand(ioc)) , socket_(net::make_strand(ioc)) { beast::error_code ec; // Open the acceptor acceptor_.open(endpoint.protocol(), ec); if(ec) { fail(ec, "open"); return; } // Allow address reuse acceptor_.set_option(net::socket_base::reuse_address(true), ec); if(ec) { fail(ec, "set_option"); return; } // Bind to the server address acceptor_.bind(endpoint, ec); if(ec) { fail(ec, "bind"); return; } // Start listening for connections acceptor_.listen( net::socket_base::max_listen_connections, ec); if(ec) { fail(ec, "listen"); return; } } // Start accepting incoming connections void run() { loop(); } private: #include void loop(beast::error_code ec = {}) { reenter(*this) { for(;;) { yield acceptor_.async_accept( socket_, std::bind( &listener::loop, shared_from_this(), std::placeholders::_1)); if(ec) { fail(ec, "accept"); } else { // Create the session and run it std::make_shared(std::move(socket_))->run(); } // Make sure each session gets its own strand socket_ = tcp::socket(net::make_strand(ioc_)); } } } #include }; //------------------------------------------------------------------------------ int main(int argc, char* argv[]) { // Check command line arguments. if (argc != 4) { std::cerr << "Usage: websocket-server-stackless
\n" << "Example:\n" << " websocket-server-stackless 0.0.0.0 8080 1\n"; return EXIT_FAILURE; } auto const address = net::ip::make_address(argv[1]); auto const port = static_cast(std::atoi(argv[2])); auto const threads = std::max(1, std::atoi(argv[3])); // The io_context is required for all I/O net::io_context ioc{threads}; // Create and launch a listening port std::make_shared(ioc, tcp::endpoint{address, port})->run(); // Run the I/O service on the requested number of threads std::vector v; v.reserve(threads - 1); for(auto i = threads - 1; i > 0; --i) v.emplace_back( [&ioc] { ioc.run(); }); ioc.run(); return EXIT_SUCCESS; }