LibCore: Implement TCPServer+UDPServer+LocalServer on Windows

[ayeteadoe]

Demo

To test my impls I made simple Marco-Polo servers where a Python or C++ client connects and starts a 5 entry long chain of "macro!" (Client) and "polo!" (Server) back and forth messages. After the 5th "polo!" is sent, the Server shuts down.

TCPServer

TCPServerWindows_Demo.mp4

Here is the code for both the Macro-Polo server and the python Client.

TestTCPServer.cpp

#include <AK/StringView.h>
#include <AK/Vector.h>
#include <LibCore/EventLoop.h>
#include <LibCore/Socket.h>
#include <LibCore/TCPServer.h>
#include <LibMain/Main.h>
#include <LibTest/TestCase.h>

using namespace Core;
class MarcoPoloServer : public EventReceiver {
    C_OBJECT(MarcoPoloServer)
public:
    static ErrorOr<NonnullRefPtr<MarcoPoloServer>> try_create()
    {
        auto server = TRY(TCPServer::try_create());
        return adopt_nonnull_ref_or_enomem(new (nothrow) MarcoPoloServer(move(server)));
    }

    ErrorOr<void> start_listening(u16 port = 8080)
    {
        TRY(m_server->listen(IPv4Address::from_string("127.0.0.1"sv).value(), port));

        m_server->on_ready_to_accept = [this]() {
            handle_new_connection();
        };

        outln("Server listening on localhost:{}", port);
        return {};
    }

    void shutdown()
    {
        outln("Shutting down server gracefully...");
        m_clients.clear();
        Core::EventLoop::current().quit(0);
    }

private:
    explicit MarcoPoloServer(NonnullRefPtr<TCPServer> server)
        : m_server(move(server))
    {
    }

    bool handle_new_connection()
    {
        auto socket_result = m_server->accept();
        if (socket_result.is_error()) {
            outln("Failed to accept connection: {}", socket_result.error());
            return false;
        }

        auto socket = socket_result.release_value();
        outln("New client connected!");

        auto client = make<Client>(move(socket), *this);
        m_clients.append(move(client));
        return true;
    }

    struct Client {
        NonnullOwnPtr<TCPSocket> socket;
        MarcoPoloServer& server;
        int exchange_count = 0;
        Vector<u8> buffer;

        Client(NonnullOwnPtr<TCPSocket> sock, MarcoPoloServer& srv)
            : socket(move(sock))
            , server(srv)
        {
            buffer.resize(1024);
            setup_socket_handlers();
        }

        void setup_socket_handlers()
        {
            socket->on_ready_to_read = [this]() {
                handle_read();
            };
        }

        void handle_read()
        {
            auto bytes_result = socket->read_some(buffer.span());
            if (bytes_result.is_error()) {
                outln("Error reading from client: {}", bytes_result.error());
                return;
            }

            auto bytes = bytes_result.value();
            if (bytes.is_empty()) {
                outln("Client closed connection");
                return;
            }

            StringView message(reinterpret_cast<char const*>(bytes.data()), bytes.size());
            message = message.trim_whitespace();

            outln("Received: '{}'", message);

            if (message == "marco!"sv) {
                exchange_count++;
                outln("Exchange {}/5 - Sending: 'polo!'", exchange_count);

                auto response = "polo!\n"sv;
                auto write_result = socket->write_some(response.bytes());
                if (write_result.is_error()) {
                    outln("Error writing to client: {}", write_result.error());
                    return;
                }

                if (exchange_count >= 5) {
                    outln("Completed 5 exchanges! Shutting down server.");
                    server.deferred_invoke([this]() {
                        server.shutdown();
                    });
                }
            } else {
                outln("Unexpected message from client: '{}'", message);
            }
        }
    };

    NonnullRefPtr<TCPServer> m_server;
    Vector<NonnullOwnPtr<Client>> m_clients;
};

TEST_CASE(macro_polo_server)
{
    EventLoop event_loop;

    auto server = TRY_OR_FAIL(MarcoPoloServer::try_create());
    TRY_OR_FAIL(server->start_listening(8000));

    outln("Marco/Polo TCP Server started!");
    outln("Run the Python client script to connect and play Marco/Polo");
    outln("The server will shutdown after 5 exchanges.");

    event_loop.exec();
}

TestTCPServer.py

#!/usr/bin/env python3

import socket
import time
import sys

def main():
    HOST = 'localhost'
    PORT = 8000

    print(f"Connecting to {HOST}:{PORT}...")

    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.connect((HOST, PORT))
            print("Connected to server!")

            for round_num in range(1, 6):
                print(f"\n--- Round {round_num}/5 ---")

                message = "marco!\n"
                print(f"Sending: '{message.strip()}'")
                sock.sendall(message.encode('utf-8'))

                response = sock.recv(1024).decode('utf-8').strip()
                print(f"Received: '{response}'")

                if response != "polo!":
                    print(f"Unexpected response: '{response}'")
                    break

                if round_num < 5:
                    time.sleep(0.5)

            print("\n🎉 Successfully completed 5 rounds of Marco/Polo!")
            print("Server should shutdown gracefully now.")

    except ConnectionRefusedError:
        print("❌ Could not connect to server. Make sure the server is running.")
        sys.exit(1)
    except Exception as e:
        print(f"❌ An error occurred: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

UDPServer

UDPServerWindows_Demo.mp4

Here is the code for both the Macro-Polo server and the python Client.

TestUDPServer.cpp

#include <AK/HashMap.h>
#include <AK/StringView.h>
#include <LibCore/EventLoop.h>
#include <LibCore/UDPServer.h>
#include <LibMain/Main.h>
#include <LibTest/TestCase.h>

using namespace Core;

class MarcoPoloUDPServer : public EventReceiver {
    C_OBJECT(MarcoPoloUDPServer)
public:
    static ErrorOr<NonnullRefPtr<MarcoPoloUDPServer>> try_create()
    {
        auto server = TRY(UDPServer::try_create());
        return adopt_nonnull_ref_or_enomem(new (nothrow) MarcoPoloUDPServer(move(server)));
    }

    ErrorOr<void> start_listening(u16 port = 8080)
    {
        if (!m_server->bind(IPv4Address::from_string("127.0.0.1"sv).value(), port)) {
            return Error::from_string_literal("Failed to bind UDP server");
        }

        m_server->on_ready_to_receive = [this]() {
            handle_incoming_message();
        };

        outln("UDP Server listening on localhost:{}", port);
        return {};
    }

    void shutdown()
    {
        outln("Shutting down UDP server gracefully...");
        m_clients.clear();
        Core::EventLoop::current().quit(0);
    }

private:
    explicit MarcoPoloUDPServer(NonnullRefPtr<UDPServer> server)
        : m_server(move(server))
    {
    }

    void handle_incoming_message()
    {
        struct sockaddr_in from;
        auto buffer_result = m_server->receive(1024, from);
        if (buffer_result.is_error()) {
            outln("Error receiving UDP message: {}", buffer_result.error());
            return;
        }

        auto buffer = buffer_result.release_value();
        if (buffer.is_empty()) {
            outln("Received empty UDP message");
            return;
        }

        StringView message(reinterpret_cast<char const*>(buffer.data()), buffer.size());
        message = message.trim_whitespace();

        auto addr = TRY_OR_FAIL(IPv4Address(from.sin_addr.s_addr).to_string());
        String client_id = TRY_OR_FAIL(String::formatted("{}:{}",
            addr, from.sin_port));

        outln("Received from {}: '{}'", client_id, message);

        if (message == "marco!"sv) {
            auto& client_state = m_clients.ensure(client_id, []() {
                return ClientState {};
            });

            client_state.exchange_count++;
            outln("Client {} - Exchange {}/5 - Sending: 'polo!'", client_id, client_state.exchange_count);

            auto response = "polo!"sv;
            auto send_result = m_server->send(response.bytes(), from);
            if (send_result.is_error()) {
                outln("Error sending UDP response: {}", send_result.error());
                return;
            }

            if (client_state.exchange_count >= 5) {
                outln("Client {} completed 5 exchanges!", client_id);
                m_clients.remove(client_id);
                if (m_clients.is_empty()) {
                    outln("All clients completed! Shutting down server.");
                    deferred_invoke([this]() {
                        shutdown();
                    });
                }
            }
        } else {
            outln("Unexpected message from client {}: '{}'", client_id, message);
        }
    }

    struct ClientState {
        int exchange_count = 0;
    };

    NonnullRefPtr<UDPServer> m_server;
    HashMap<String, ClientState> m_clients;
};

TEST_CASE(marco_polo_udp_server)
{
    EventLoop event_loop;

    auto server = TRY_OR_FAIL(MarcoPoloUDPServer::try_create());
    TRY_OR_FAIL(server->start_listening(8001));

    outln("Marco/Polo UDP Server started!");
    outln("Run the Python UDP client script to connect and play Marco/Polo");
    outln("The server will shutdown after a client completes 5 exchanges.");

    event_loop.exec();
}

TestUDPServer.py

#!/usr/bin/env python3

import socket
import time
import sys

def main():
    HOST = 'localhost'
    PORT = 8001

    print(f"Connecting to UDP server at {HOST}:{PORT}...")

    try:
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
            sock.settimeout(5.0)

            print("Connected to UDP server!")

            for round_num in range(1, 6):
                print(f"\n--- Round {round_num}/5 ---")

                message = "marco!"
                print(f"Sending: '{message}'")
                sock.sendto(message.encode('utf-8'), (HOST, PORT))

                try:
                    response_data, server_address = sock.recvfrom(1024)
                    response = response_data.decode('utf-8').strip()
                    print(f"Received from {server_address}: '{response}'")

                    if response != "polo!":
                        print(f"Unexpected response: '{response}'")
                        break

                    if round_num < 5:
                        time.sleep(0.5)

                except socket.timeout:
                    print("❌ Timeout waiting for response from server")
                    break

            print("\n🎉 Successfully completed 5 rounds of Marco/Polo!")
            print("Server should shutdown gracefully now.")

    except Exception as e:
        print(f"❌ An error occurred: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

LocalServer

LocalServerWindows_Demo.mp4

Here is the code for both the Macro-Polo server and the C++ Client (Python doesn't seem to support connecting for Winsock2-based local sockets).

TestLocalServer.cpp

#include <AK/StringView.h>
#include <AK/Vector.h>
#include <LibCore/EventLoop.h>
#include <LibCore/LocalServer.h>
#include <LibCore/Socket.h>
#include <LibCore/System.h>
#include <LibTest/TestCase.h>

using namespace Core;

class MarcoPoloLocalServer : public EventReceiver {
    C_OBJECT(MarcoPoloLocalServer)
public:
    static ErrorOr<NonnullRefPtr<MarcoPoloLocalServer>> try_create()
    {
        auto server = TRY(LocalServer::try_create());
        return adopt_nonnull_ref_or_enomem(new (nothrow) MarcoPoloLocalServer(move(server)));
    }

    ErrorOr<void> start_listening(ByteString const& socket_path = "marco_polo_socket"sv)
    {
        (void)Core::System::unlink(socket_path);

        if (!m_server->listen(socket_path)) {
            return Error::from_string_literal("Failed to listen on local socket");
        }

        m_server->on_accept = [this](NonnullOwnPtr<LocalSocket> socket) {
            handle_new_connection(move(socket));
        };

        m_server->on_accept_error = [](Error error) {
            outln("Accept error: {}", error);
        };

        outln("Local server listening on socket: {}", socket_path);
        return {};
    }

    void shutdown()
    {
        outln("Shutting down local server gracefully...");
        m_clients.clear();
        Core::EventLoop::current().quit(0);
    }

private:
    explicit MarcoPoloLocalServer(NonnullRefPtr<LocalServer> server)
        : m_server(move(server))
    {
    }

    void handle_new_connection(NonnullOwnPtr<LocalSocket> socket)
    {
        outln("New client connected to local socket!");
        auto client = make<Client>(move(socket), *this);
        m_clients.append(move(client));
    }

    struct Client {
        NonnullOwnPtr<LocalSocket> socket;
        MarcoPoloLocalServer& server;
        int exchange_count = 0;
        Vector<u8> buffer;

        Client(NonnullOwnPtr<LocalSocket> sock, MarcoPoloLocalServer& srv)
            : socket(move(sock))
            , server(srv)
        {
            buffer.resize(1024);
            setup_socket_handlers();
        }

        void setup_socket_handlers()
        {
            socket->on_ready_to_read = [this]() {
                handle_read();
            };
        }

        void handle_read()
        {
            auto bytes_result = socket->read_some(buffer.span());
            if (bytes_result.is_error()) {
                outln("Error reading from local client: {}", bytes_result.error());
                return;
            }

            auto bytes = bytes_result.value();
            if (bytes.is_empty()) {
                outln("Local client closed connection");
                return;
            }

            StringView message(reinterpret_cast<char const*>(bytes.data()), bytes.size());
            message = message.trim_whitespace();

            outln("Received on local socket: '{}'", message);

            if (message == "marco!"sv) {
                exchange_count++;
                outln("Exchange {}/5 - Sending: 'polo!'", exchange_count);

                auto response = "polo!\n"sv;
                auto write_result = socket->write_some(response.bytes());
                if (write_result.is_error()) {
                    outln("Error writing to local client: {}", write_result.error());
                    return;
                }

                if (exchange_count >= 5) {
                    outln("Completed 5 exchanges! Shutting down local server.");
                    server.deferred_invoke([this]() {
                        server.shutdown();
                    });
                }
            } else {
                outln("Unexpected message from local client: '{}'", message);
            }
        }
    };

    NonnullRefPtr<LocalServer> m_server;
    Vector<NonnullOwnPtr<Client>> m_clients;
};

TEST_CASE(marco_polo_local_server)
{
    EventLoop event_loop;

    auto server = TRY_OR_FAIL(MarcoPoloLocalServer::try_create());
    TRY_OR_FAIL(server->start_listening("marco_polo_test_socket"));

    outln("Marco/Polo Local Socket Server started!");
    outln("Run the C++ client script to connect and play Marco/Polo");
    outln("The server will shutdown after 5 exchanges.");

    event_loop.exec();
}

TestLocalSocket.cpp.

#include <AK/StringView.h>
#include <AK/Vector.h>
#include <LibCore/Socket.h>
#include <LibCore/System.h>
#include <LibTest/TestCase.h>

using namespace Core;

TEST_CASE(marco_polo_local_socket)
{
    auto socket = TRY_OR_FAIL(LocalSocket::connect("marco_polo_test_socket"));
    TRY_OR_FAIL(socket->set_blocking(true));

    outln("Connected to local socket server!");

    for (int round_num = 1; round_num <= 5; ++round_num) {
        outln("\n--- Round {} /5 ---"sv, round_num, 5);

        ByteString message = "marco!\n";
        outln("Sending: '{}'", message.trim("\n"sv, TrimMode::Right));
        auto write_result = socket->write_until_depleted(message.bytes());
        if (write_result.is_error()) {
            outln("Failed to send message: {}"sv, write_result.error());
            VERIFY_NOT_REACHED();
        }

        u8 buffer[1024];
        Bytes bytes_read;
        while (true) {
            auto read_result = socket->read_some(buffer);
            if (read_result.is_error()) {
                TRY_OR_FAIL(System::sleep_ms(10));
                continue;
            }
            bytes_read = read_result.release_value();
            if (!bytes_read.is_empty()) {
                break;
            }
            TRY_OR_FAIL(System::sleep_ms(10));
        }

        StringView response(bytes_read);
        response = response.trim_whitespace();

        outln("Received: '{}'"sv, response);

        EXPECT_EQ(response, "polo!"sv);

        if (round_num < 5) {
            TRY_OR_FAIL(System::sleep_ms(500)); // 0.5 seconds
        }
    }

    outln("\n🎉 Successfully completed 5 rounds of Marco/Polo!"sv);
    outln("Server should shutdown gracefully now."sv);
}

[R-Goc]

Can we somehow make sure we are trying this on a socket fd? If someone didn't read the implementation of this they might think we have somehow gotten an ioctl equivalent which is not the case.

[R-Goc]

Or maybe not call this ioctl at all and just wrap ioctlsocket in a LibCore api.