deregister disconnected clients

backend/deploy/mosquitto.local.conf

@@ -0,0 +1,4 @@
+listener 1883
+allow_anonymous true
+password_file mqtt_users
+

backend/src/event_loop.cpp

@@ -1,11 +1,36 @@
 #include "event_loop.hpp"
 #include "errno_shim.hpp"
 #include "server.hpp"
+#include <algorithm>
+#include <print>
+#include <stdexcept>
 #include <sys/epoll.h>
 #include <utility>
-#include <variant>
+
+#define CHECK(EXPR)                                                            \
+    do {                                                                       \
+        if (!(EXPR).has_value()) [[unlikely]] {                                \
+            return std::unexpected((EXPR).error());                            \
+        }                                                                      \
+    } while (false)
 
 namespace mst::event {
+auto Event::fd() -> int
+{
+    switch (this->kind) {
+        case mst::event::EventKind::Server: {
+            auto& ref = std::get<std::unique_ptr<mst::Server>>(this->data);
+            return ref->fd();
+            break;
+        }
+        case mst::event::EventKind::Client: {
+            auto& ref = std::get<std::unique_ptr<mst::Client>>(this->data);
+            return ref->fd();
+        }
+        default:
+            std::unreachable();
+    }
+}
 auto Manager::start() -> Result<void>
 {
     epoll_event events[128] = { };
@@ -16,22 +41,22 @@ auto Manager::start() -> Result<void>
         }
         for (int i = 0; i < events_len; ++i) {
             auto event = (mst::event::Event*)events[i].data.ptr;
-            switch (event->data.index()) {
+            switch (event->kind) {
-                case 0: {
+                case mst::event::EventKind::Server: {
                     auto& ref
                         = std::get<std::unique_ptr<mst::Server>>(event->data);
                     auto res = ref->wake(*this);
-                    if (!res) {
+                    CHECK(res);
-                        return std::unexpected(res.error());
-                    }
                     break;
                 }
-                case 1: {
+                case mst::event::EventKind::Client: {
                     auto& ref
                         = std::get<std::unique_ptr<mst::Client>>(event->data);
                     auto res = ref->wake();
-                    if (!res) {
+                    CHECK(res);
-                        return std::unexpected(res.error());
+                    auto do_now = res.value();
+                    if (do_now == Client::lllll::Disconnect) {
+                        CHECK(this->deregister_event(ref->fd()));
                     }
                     break;
                 }
@@ -42,6 +67,31 @@ auto Manager::start() -> Result<void>
     }
 }
 
+auto Manager::deregister_event(int fd) -> Result<void>
+{
+    auto poll_event = epoll_event { .events = EPOLLIN, .data = { } };
+    if (::epoll_ctl(this->epoll_fd, EPOLL_CTL_DEL, fd, &poll_event) < 0) {
+        return std::unexpected(
+            errno_shim("could not remove listener to epoll"));
+    }
+    size_t idx = this->events.size();
+    for (size_t i = 0; i < this->events.size(); ++i) {
+        if (this->events[i]->fd() == fd) {
+            idx = i;
+        }
+    }
+
+    if (idx == this->events.size()) {
+        throw std::runtime_error("contract broken");
+    }
+
+    this->events.erase(std::find_if(this->events.begin(),
+        this->events.end(),
+        [&](auto& e) { return e->fd() == fd; }));
+
+    return { };
+}
+
 auto Manager::create() -> Result<Manager>
 {
     auto epoll_fd = ::epoll_create1(0);

backend/src/event_loop.hpp

@@ -2,8 +2,10 @@
 #include "errno_shim.hpp"
 #include "result.hpp"
 #include "server.hpp"
+#include <algorithm>
 #include <memory>
 #include <sys/epoll.h>
+#include <utility>
 #include <variant>
 #include <vector>
 
@@ -11,9 +13,23 @@ namespace mst {
 class Server;
 class Client;
 namespace event {
-    struct Event {
+    enum class EventKind {
+        Server,
+        Client,
+    };
+    class Event {
+    public:
+        Event(std::variant<std::unique_ptr<mst::Server>,
+            std::unique_ptr<mst::Client>> data)
+            : data(std::move(data))
+            , kind(data.index() == 0 ? EventKind::Server : EventKind::Client)
+        {
+        }
+        auto fd() -> int;
+
         std::variant<std::unique_ptr<mst::Server>, std::unique_ptr<mst::Client>>
             data;
+        EventKind kind;
     };
 
     template <typename Data>
@@ -41,6 +57,7 @@ namespace event {
             }
             return { };
         }
+        auto deregister_event(int fd) -> Result<void>;
         static auto create() -> Result<Manager>;
 
     private:

backend/src/server.cpp

@@ -5,7 +5,7 @@
 
 namespace mst {
 
-auto Client::wake() -> Result<void>
+auto Client::wake() -> Result<Client::lllll>
 {
     uint8_t buffer[128] = { };
     auto x = this->connection.read(buffer, 128);
@@ -13,11 +13,14 @@ auto Client::wake() -> Result<void>
         return std::unexpected(x.error());
     }
     auto bytes_read = x.value();
+    if (bytes_read == 0) {
+        return { Client::lllll::Disconnect };
+    }
     for (size_t i = 0; i < bytes_read; ++i) {
         std::println("{:c}", buffer[i]);
     }
 
-    return { };
+    return { Client::lllll::Ok };
 }
 
 auto Server::bind(

backend/src/server.hpp

@@ -15,6 +15,10 @@ public:
     static auto bind(
         mst::event::Manager& mgr, const std::string& host, uint16_t port)
         -> Result<void>;
+    auto fd() -> int
+    {
+        return this->listener.fd;
+    };
 
 private:
     TcpListener listener;
@@ -26,8 +30,17 @@ private:
 };
 
 class Client {
+
 public:
-    auto wake() -> Result<void>;
+    enum class lllll {
+        Ok,
+        Disconnect,
+    };
+    auto wake() -> Result<lllll>;
+    auto fd() -> int
+    {
+        return this->connection.fd;
+    };
 
     Client(Server&, TcpConnection connection)
         : connection(connection)