EchoServerCallback.cpp

// 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 <functional>
#include <iostream>
#include <string>
#include <vector>

#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;
}