// SPDX-License-Identifier: LGPL-3.0-or-later // Copyright © 2020-2021 Florian Fischer #include <sys/socket.h> #include <array> #include <atomic> #include <cstdint> #include <cstdlib> #include <cstring> #include <iostream> #include <string> #include "Common.hpp" #include "Debug.hpp" #include "Runtime.hpp" #include "io.hpp" #include "io/Future.hpp" using SendFuture = emper::io::SendFuture; using RecvFuture = emper::io::RecvFuture; const std::string HOST = "::"; const std::string PORT = "12345"; static const int BACKLOG = 1024; const size_t BUF_SIZE = 1024; static std::atomic<bool> quit = false; class Client { int sockfd; std::array<char, BUF_SIZE> buf; void terminate() { emper::io::closeAndForget(sockfd); delete this; } void submitSend(int32_t bytes_recv) { SendFuture sf(sockfd, buf.data(), bytes_recv, MSG_NOSIGNAL); sf.setCallback( [this, bytes_recv](int32_t bytes_send) { this->onSend(bytes_send, bytes_recv); }); sf.submit(); } void onSend(int32_t bytes_send, int32_t bytes_recv) { if (unlikely(bytes_send != bytes_recv)) { LOGE("server send failed: " << strerror(-bytes_send)); this->terminate(); return; } if (!quit.load(std::memory_order_consume)) { submitRecv(); } else { this->terminate(); } } void onRecv(int32_t bytes_recv) { if (unlikely(bytes_recv <= 0)) { // socket was shutdown if (bytes_recv < 0) { LOGE("server read failed:" << strerror(-bytes_recv)); } this->terminate(); return; } if (unlikely(bytes_recv == 5 && strncmp("quit\n", buf.data(), bytes_recv) == 0)) { quit = true; Runtime::getRuntime()->initiateTermination(); this->terminate(); return; } submitSend(bytes_recv); } public: Client(int socket) : sockfd(socket) {} void submitRecv() { RecvFuture rf(sockfd, buf.data(), BUF_SIZE, 0); rf.setCallback([this](int32_t bytes_recv) { this->onRecv(bytes_recv); }); rf.submit(); } }; auto main(int argc, char* argv[]) -> int { std::string host = HOST; std::string port = PORT; if (argc > 2) { std::cerr << "Usage: " << argv[0] << " [port]" << std::endl; exit(EXIT_FAILURE); } if (argc > 1) { port = std::string(argv[1]); } std::cout << "Echoserver listening on " << host << ":" << port << std::endl; Runtime runtime; auto* listener = emper::io::tcp_listener(host, port, [](int socket) { auto* client = new Client(socket); client->submitRecv(); }, BACKLOG, {emper::io::SockOpt::ReusePort}); if (!listener) { exit(EXIT_FAILURE); } runtime.scheduleFromAnywhere(*listener); runtime.waitUntilFinished(); return EXIT_FAILURE; }