#pragma once namespace server { namespace core { template struct async_read_frame { socket_type& socket; buffer_type buffer; asio::error_code error_code; async_read_frame(socket_type& socket, buffer_type&& buffer) : socket(socket), buffer(std::move(buffer)) { this->buffer; } ~async_read_frame() { this->buffer; } asio::awaitable async_read() { try { std::size_t n = co_await socket.async_read_some(buffer, asio::use_awaitable); //printf("%s\n", /*this_thread::get_debug_name().c_str(),*/ buffer); //rdata.erase(rdata.begin(), rdata.begin() + n); //for (;;) { //std::size_t n; //socket.async_receive(asio::buffer(rdata, 12), [this, &n](auto error_code, auto bytes_read) { n = bytes_read; }); //std::size_t n = co_await asio::async_read_until(socket, asio::dynamic_buffer(rdata, 12), "\n", asio::use_awaitable); } } catch (std::exception & e) { printf("exception: %s", e.what()); } } bool await_ready() { return false; } void await_suspend(std::experimental::coroutine_handle<> coro) { /* asio::co_spawn(io_context, [this]() { return async_read(); }, asio::detached); co_await socket.async_receive(buffer, [this, coro](auto error_code, auto bytes_read) { coro.resume(); }); */ /* asio::async_read_until(socket, buffer, "\n", [this, coro](auto error_code, auto bytes_read) { this->error_code = error_code; coro.resume(); }); */ /* asio::async_read(socket, buffer, [this, coro](auto error_code, auto bytes_read) { this->error_code = error_code; coro.resume(); }); */ socket.async_read_some(buffer, [this, coro](auto error_code, auto bytes_read) { this->error_code = error_code; coro.resume(); }); //socket.async_receive(buffer, [this, coro](auto error_code) { this->error_code = error_code; coro.resume(); }); } void await_resume() { if (error_code) { throw asio::system_error(error_code); } } }; struct tcp_connection { asio::ip::tcp::socket socket; std::vector rdata; std::vector wdata; int id; asio::ip::address address; uint16 port; bool marked_for_delete; tcp_connection(asio::ip::tcp::socket&& new_socket) : socket(std::move(new_socket)), id(-1), address(socket.remote_endpoint().address()), port(socket.remote_endpoint().port()) { rdata.resize(12); //rdata.reserve(12); } void close() { asio::error_code error_code; socket.shutdown(asio::ip::tcp::socket::shutdown_both, error_code); if (error_code) printf("network: tcp_connection::close: %s errored when shutting down socket: %i (%s)", address.to_string().c_str(), error_code.value(), error_code.message().c_str()); socket.close(); marked_for_delete = true; } //* template auto async_recv_helper2(socket_type& socket, buffer_type&& buffer) { return async_read_frame(socket, std::forward(buffer)); } template auto async_recv_helper(socket_type& socket, buffer_type&& buffer) -> std::future { try { while (socket.is_open()) { co_await async_recv_helper2(socket, std::forward(buffer)); } } catch (const std::exception & exception) { printf("exception: %s", exception.what()); } } auto async_recv() { auto buffer = asio::buffer(rdata, 5); async_recv_helper(socket, std::move(buffer)); } //*/ asio::awaitable async_read() { try { char data[1024]; //for (rdata;;) for (;;) { //std::size_t n; //socket.async_receive(asio::buffer(rdata, 12), [this, &n](auto error_code, auto bytes_read) { n = bytes_read; }); std::size_t n = co_await socket.async_read_some(asio::buffer(rdata, 12), asio::use_awaitable); //std::size_t n = co_await asio::async_read_until(socket, asio::dynamic_buffer(rdata, 12), "\n", asio::use_awaitable); printf("%s\n", /*this_thread::get_debug_name().c_str(),*/ rdata.data()); rdata.erase(rdata.begin(), rdata.begin() + n); } } catch (std::exception & e) { printf("exception: %s", e.what()); } } }; static_assert(std::is_nothrow_move_constructible::value); struct connection_pool { std::vector> connections; bool allow_overflow; void init(uint32 max_connection_count, bool allow_overflow = false) { //connections.reserve(max_connection_count); this->allow_overflow = allow_overflow; } tcp_connection* allocate(asio::ip::tcp::socket&& socket) { tcp_connection* test = nullptr; tcp_connection* taken = test + 1; int id = 0; for (auto& i : connections) { bool exchanged = i.compare_exchange_weak(test, taken); if (exchanged) { //auto new_connection = new tcp_connection(std::move(socket)); //new_connection->id = id; //i = new_connection; //return new_connection; } id++; } return nullptr; } void release(int id) { //auto connection = connections[id].load(); //delete connection; //connections[id].store(nullptr); } }; } }