diff --git a/backend/src/main.cpp b/backend/src/main.cpp index 39099a5..e904c43 100644 --- a/backend/src/main.cpp +++ b/backend/src/main.cpp @@ -1,12 +1,9 @@ -#include "event_loop.hpp" +#include "server2.hpp" +// #include "event_loop.hpp" #include "json.hpp" #include "mqtt.hpp" -#include "server.hpp" -#include -#include -#include +// #include "server.hpp" #include -#include #include #include #include @@ -36,23 +33,23 @@ int main(void) mqtt_client.subscribe("/skateboard/update", [&](std::string_view text) { // - std::println("Skateboard: {}", text); - auto result = mst::json::parse(text); - - if (!result) { - std::println(stderr, - "error: {} at {}", - result.error().message, - result.error().loc.idx); - return; - } - try { - auto parsed = std::move(result.value()); - std::println(".rotation = {}", - parsed->query(".rotation").value()->get_f64()); - } catch (std::runtime_error& ex) { - std::println(stderr, "exception: {}", ex.what()); - } + // std::println("Skateboard: {}", text); + // auto result = mst::json::parse(text); + // + // if (!result) { + // std::println(stderr, + // "error: {} at {}", + // result.error().message, + // result.error().loc.idx); + // return; + // } + // try { + // auto parsed = std::move(result.value()); + // std::println(".rotation = {}", + // parsed->query(".rotation").value()->get_f64()); + // } catch (std::runtime_error& ex) { + // std::println(stderr, "exception: {}", ex.what()); + // } }); auto mqtt_thread = std::thread([&]() { @@ -64,16 +61,19 @@ int main(void) } }); - auto mgr = mst::event::Manager::create().value(); - auto x = mst::Server::bind(mgr, "0.0.0.0", PORT); - if (!x) { - std::println("{}", x.error()); - return 1; - } - std::println("starting"); - { - auto x = mgr.start(); - } + // auto mgr = mst::event::Manager::create().value(); + // auto x = mst::Server::bind(mgr, "0.0.0.0", PORT); + // if (!x) { + // std::println("{}", x.error()); + // return 1; + // } + // std::println("starting"); + // { + // auto x = mgr.start(); + // } + + auto server = mst::server2::Server(); + server.listen(); mqtt_thread.join(); return 0; diff --git a/backend/src/mqtt.cpp b/backend/src/mqtt.cpp index 5d3f86b..55188fb 100644 --- a/backend/src/mqtt.cpp +++ b/backend/src/mqtt.cpp @@ -218,7 +218,7 @@ void Client::cb_publish() void Client::cb_message(std::string_view topic, const void* data, size_t size) { - std::println("[MQTT] Message received"); + // std::println("[MQTT] Message received"); auto text = std::string_view(static_cast(data), size); diff --git a/backend/src/server2.cpp b/backend/src/server2.cpp new file mode 100644 index 0000000..24e39fd --- /dev/null +++ b/backend/src/server2.cpp @@ -0,0 +1,175 @@ +#include "server2.hpp" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace { +using namespace mst::server2; + +auto get_listener_socket() -> int +{ + int listener; // Listening socket descriptor + int status; + + struct addrinfo hints = { }; + hints.ai_family = AF_INET; + hints.ai_socktype = SOCK_STREAM; + hints.ai_flags = AI_PASSIVE; + + struct addrinfo* addr; + + if ((status = ::getaddrinfo(NULL, "8881", &hints, &addr)) != 0) + throw Error(std::format("getaddrinfo ({})", ::gai_strerror(status))); + + struct addrinfo* p; + for (p = addr; p != NULL; p = p->ai_next) { + listener = ::socket(p->ai_family, p->ai_socktype, p->ai_protocol); + if (listener < 0) { + continue; + } + + int reuseaddr_opt = 1; + ::setsockopt( + listener, SOL_SOCKET, SO_REUSEADDR, &reuseaddr_opt, sizeof(int)); + + if (::bind(listener, p->ai_addr, p->ai_addrlen) < 0) { + ::close(listener); + continue; + } + + break; + } + + if (p == NULL) + throw Error("didn't get bound"); + + ::freeaddrinfo(addr); + + if (::listen(listener, 10) == -1) + throw Error(std::format("could not listen ({})", strerror(errno))); + + return listener; +} + +} + +namespace mst::server2 { + +struct Server::State { + std::vector<::pollfd> pollfds; + std::vector<::pollfd> queued_insertions; + std::vector queued_deletions; +}; + +Server::Server() + : m_state(std::make_unique()) +{ +} +Server::~Server() = default; + +void Server::listen() +{ + m_listener_fd = get_listener_socket(); + m_state->pollfds.push_back(::pollfd { + .fd = m_listener_fd, + .events = POLLIN, + .revents = { }, + }); + + std::println("[mst::server2] listening for connections"); + + while (true) { + int poll_count + = ::poll(m_state->pollfds.data(), m_state->pollfds.size(), -1); + + if (poll_count == -1) + throw Error(std::format("poll (%s)", strerror(errno))); + + for (size_t i = 0; i < m_state->pollfds.size(); ++i) { + auto& fd = m_state->pollfds[i]; + if (!(fd.revents & (POLLIN | POLLHUP))) + continue; + + if (fd.fd == m_listener_fd) { + create_connection(); + } else { + try { + handle_request(i); + } catch (Error& ex) { + std::println(stderr, + "[mst::server2] exception in handler for client {}: {}", + fd.fd, + ex.what()); + m_state->queued_deletions.push_back(i); + } + } + } + + auto& fds = m_state->pollfds; + auto& deletions = m_state->queued_deletions; + + std::reverse(deletions.begin(), deletions.end()); + for (auto idx : deletions) { + fds.erase(std::next(fds.begin(), static_cast(idx))); + } + m_state->queued_deletions.clear(); + + for (auto& fd : m_state->queued_insertions) { + m_state->pollfds.push_back(fd); + } + m_state->queued_insertions.clear(); + } +} + +void Server::create_connection() +{ + struct sockaddr_storage remoteaddr; + socklen_t addrlen = sizeof remoteaddr; + + int client_fd + = ::accept(m_listener_fd, (struct sockaddr*)&remoteaddr, &addrlen); + + if (client_fd == -1) + throw Error(std::format("format ({})", strerror(errno))); + + std::println("[mst::server2] client {} connected", client_fd); + + m_state->queued_insertions.push_back(::pollfd { + .fd = client_fd, + .events = POLLIN, + .revents = { }, + }); +} + +void Server::handle_request(size_t i) +{ + auto& client_fd = m_state->pollfds[i].fd; + auto buffer = std::vector(512); + + ssize_t byte_count = ::recv(client_fd, buffer.data(), buffer.size(), 0); + + if (byte_count <= 0) + throw Error(std::format("recv: {}", strerror(errno))); + + if (byte_count == 0) { + std::println("[mst::server2] client {} disconnected", client_fd); + + m_state->queued_deletions.push_back(i); + return; + } + + std::println("[mst::server2] received: {:s}", buffer); +} + +} diff --git a/backend/src/server2.hpp b/backend/src/server2.hpp new file mode 100644 index 0000000..1a8feb5 --- /dev/null +++ b/backend/src/server2.hpp @@ -0,0 +1,29 @@ +#pragma once + +#include +#include + +namespace mst::server2 { + +struct Error : public std::runtime_error { + using std::runtime_error::runtime_error; +}; + +class Server { +public: + explicit Server(); + ~Server(); + + void listen(); + +private: + struct State; + + void create_connection(); + void handle_request(size_t i); + + int m_listener_fd { }; + std::unique_ptr m_state { nullptr }; +}; + +} diff --git a/backend/src/tcp.cpp b/backend/src/tcp.cpp index 07e78a9..0ec053a 100644 --- a/backend/src/tcp.cpp +++ b/backend/src/tcp.cpp @@ -42,8 +42,8 @@ auto TcpListener::bind(const std::string& host, uint16_t port) struct sockaddr_in address = { .sin_family = AF_INET, - .sin_port = htons(port), - .sin_addr = in_addr { .s_addr = inet_addr(host.c_str()) }, + .sin_port = ::htons(port), + .sin_addr = in_addr { .s_addr = ::inet_addr(host.c_str()) }, .sin_zero = { }, };