slige/runtime/rpc_server.cpp

#include "rpc_server.hpp"
#include "json.hpp"
#include <netinet/in.h>
#include <sys/socket.h>
#include <unistd.h>
#include <utility>

auto create_address(uint16_t port) -> sockaddr_in
{
    return {
        .sin_family = AF_INET,
        .sin_port = htons(port),
        .sin_addr = { .s_addr = inet_addr("127.0.0.1") },
        .sin_zero = { 0 },
    };
}

template <typename Functor>
auto slige_rpc::RpcServer<Functor>::listen() -> Res<Unit>
{
    int socket_fd = ::socket(AF_INET, SOCK_STREAM, 0);
    if (socket_fd < 0) {
        return Err { "could not get socket" };
    };

    sockaddr_in address = create_address(13370);
    unsigned int address_size = sizeof(this->address);

    if (::bind(socket_fd, (struct sockaddr*)&address, sizeof(address)) < 0) {
        return Err { "could not bind" };
    };

    if (::listen(socket_fd, 0) < 0) {
        return Err { "could not listen" };
    }
    while (true) {
        int client = ::accept(
            this->fd, (struct sockaddr*)&this->address, &address_size);
        if (client < 0) {
            return Err { "could not accept" };
        }
        const size_t buf_len = 1024;
        int8_t buffer[buf_len] = {};
        auto bracket_finder = BracketFinder();
        std::string message = {};
        while (true) {
            ssize_t bytes_read = read(client, buffer, buf_len);
            if (bytes_read < 0) {
                return Err { "could not read" };
            } else if (bytes_read == 0) {
                break;
            }
            for (size_t i; i < (size_t)bytes_read; ++i) {
                message += buffer[i];
                bracket_finder.feed(buffer[i]);
                if (!bracket_finder.bracket_closed()) {
                    continue;
                }
                auto req = sliger::json::parse_json(message);
                auto writer = BufferedWriter(client);
                this->functor(req, std::move(writer));
                message.clear();
                break;
            }
        }
    }
    std::unreachable();
}

auto slige_rpc::BufferedWriter::write(std::string message) -> Res<Unit>
{
    for (size_t i = 0; i < message.length(); ++i) {
        auto res = this->write((uint8_t)message[i]);
        if (!res.is_ok()) {
            return res.err();
        }
    }
    return Unit {};
}

auto slige_rpc::BufferedWriter::write(uint8_t byte) -> Res<Unit>
{
    if (this->occupied >= length) {
        auto res = this->flush();
        if (!res.is_ok()) {
            return res.err();
        }
    }
    this->buffer[this->occupied] = byte;
    return Unit {};
}

auto slige_rpc::BufferedWriter::flush() -> Res<size_t>
{
    auto result = ::write(this->fd, this->buffer, this->occupied);
    if (result < 0) {
        return { { "unable to write" } };
    }
    this->occupied = 0;
    return (size_t)result;
}