connect server to game

backend/src/main.cpp

@@ -31,25 +31,17 @@ int main(void)
     auto mqtt_client = mst::mqtt::Client(
         BACKEND_MQTT_HOST, BACKEND_MQTT_PORT, "test", "1234");
 
+    auto server = mst::server2::Server();
+
     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 sent: {}", text);
+        try {
+            auto parsed = *mst::json::parse(text);
+            auto angle = parsed->query(".rotation").value()->get_f64();
+            server.notify_subscribers(angle);
+        } catch (std::runtime_error& ex) {
+            std::println(stderr, "exception: {}", ex.what());
+        }
     });
 
     auto mqtt_thread = std::thread([&]() {
@@ -72,7 +64,6 @@ int main(void)
     //     auto x = mgr.start();
     // }
 
-    auto server = mst::server2::Server();
     server.listen();
 
     mqtt_thread.join();

backend/src/server2.cpp

@@ -5,6 +5,7 @@
 #include <cstring>
 #include <format>
 #include <iterator>
+#include <mutex>
 #include <netdb.h>
 #include <poll.h>
 #include <print>
@@ -17,6 +18,18 @@
 namespace {
 using namespace mst::server2;
 
+struct ReqHeader {
+    enum ReqTy {
+        Subscribe = 0,
+    };
+
+    ReqTy ty;
+};
+
+struct Measurement {
+    double angle;
+};
+
 auto get_listener_socket() -> int
 {
     int listener; // Listening socket descriptor
@@ -29,7 +42,7 @@ auto get_listener_socket() -> int
 
     struct addrinfo* addr;
 
-    if ((status = ::getaddrinfo(NULL, "8881", &hints, &addr)) != 0)
+    if ((status = ::getaddrinfo(NULL, "8889", &hints, &addr)) != 0)
         throw Error(std::format("getaddrinfo ({})", ::gai_strerror(status)));
 
     struct addrinfo* p;
@@ -67,9 +80,11 @@ auto get_listener_socket() -> int
 namespace mst::server2 {
 
 struct Server::State {
+    std::mutex mx;
     std::vector<::pollfd> pollfds;
-    std::vector<::pollfd> queued_insertions;
     std::vector<std::size_t> queued_deletions;
+
+    std::vector<int> subscriber_fds;
 };
 
 Server::Server()
@@ -90,12 +105,16 @@ void Server::listen()
     std::println("[mst::server2] listening for connections");
 
     while (true) {
+        std::println(
+            "[mst::server2] polling on {} fds", m_state->pollfds.size());
         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)));
 
+        auto lock = std::lock_guard(m_state->mx);
+
         for (size_t i = 0; i < m_state->pollfds.size(); ++i) {
             auto& fd = m_state->pollfds[i];
             if (!(fd.revents & (POLLIN | POLLHUP)))
@@ -118,17 +137,41 @@ void Server::listen()
 
         auto& fds = m_state->pollfds;
         auto& deletions = m_state->queued_deletions;
+        auto& subs = m_state->subscriber_fds;
 
         std::reverse(deletions.begin(), deletions.end());
         for (auto idx : deletions) {
+
+            // hack to delete subscriber fds. subscribers should be handled
+            // in a better way.
+            for (auto iter = subs.begin(); iter != subs.end(); ++iter) {
+                if (fds[idx].fd == *iter) {
+                    subs.erase(iter);
+                    break;
+                }
+            }
+
+            std::println("deleting {}", idx);
             fds.erase(std::next(fds.begin(), static_cast<long>(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::notify_subscribers(double angle)
+{
+    auto m = Measurement { angle };
+
+    auto lock = std::lock_guard(m_state->mx);
+
+    for (auto& fd : m_state->subscriber_fds) {
+        if (fd == 0)
+            continue;
+
+        ssize_t bytes_send = ::send(fd, &m, sizeof(m), 0);
+
+        if (bytes_send == -1)
+            fd = 0;
     }
 }
 
@@ -145,7 +188,7 @@ void Server::create_connection()
 
     std::println("[mst::server2] client {} connected", client_fd);
 
-    m_state->queued_insertions.push_back(::pollfd {
+    m_state->pollfds.push_back(::pollfd {
         .fd = client_fd,
         .events = POLLIN,
         .revents = { },
@@ -155,11 +198,12 @@ void Server::create_connection()
 void Server::handle_request(size_t i)
 {
     auto& client_fd = m_state->pollfds[i].fd;
-    auto buffer = std::vector<char>(512);
+    auto header = ReqHeader { };
 
-    ssize_t byte_count = ::recv(client_fd, buffer.data(), buffer.size(), 0);
+    std::println("[mst::server2] waiting to receive", client_fd);
+    ssize_t byte_count = ::recv(client_fd, &header, sizeof(header), 0);
 
-    if (byte_count <= 0)
+    if (byte_count < 0)
         throw Error(std::format("recv: {}", strerror(errno)));
 
     if (byte_count == 0) {
@@ -169,7 +213,17 @@ void Server::handle_request(size_t i)
         return;
     }
 
-    std::println("[mst::server2] received: {:s}", buffer);
+    if (byte_count != sizeof(header))
+        throw Error(std::format("invalid request"));
+
+    switch (header.ty) {
+        case ReqHeader::Subscribe: {
+            m_state->subscriber_fds.push_back(client_fd);
+            break;
+        }
+        default:
+            throw Error(std::format("invalid request"));
+    }
 }
 
 }

backend/src/server2.hpp

@@ -15,6 +15,7 @@ public:
     ~Server();
 
     void listen();
+    void notify_subscribers(double angle);
 
 private:
     struct State;

game/src/main.rs

@@ -2,6 +2,7 @@
 
 mod engine;
 mod event_queue;
+mod server2;
 pub mod vermiparous;
 
 use core::panic;
@@ -9,12 +10,14 @@ use std::{
     collections::HashSet,
     f64::consts::PI,
     sync::{Arc, Mutex},
+    thread,
     time::Duration,
 };
 
 use crate::{
     engine::{Color, Key, Renderer, Scene, Shape, V2, V3},
     event_queue::EventQueue,
+    server2::Server2,
     vermiparous::Server,
 };
 
@@ -390,13 +393,19 @@ impl ShapeGroup {
 }
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let t = thread::spawn(|| {
+        let mut server = Server2::new().unwrap();
+        server
+            .subscribe(|measurement| {
+                println!("angle = {}", measurement.angle);
+            })
+            .unwrap();
+    });
+    t.join().unwrap();
+
     let mut sdl_io = engine::SdlIo::new()?;
     let event_queue = Arc::new(Mutex::new(EventQueue::new()));
     let mut game = Game::new(event_queue.clone());
-    std::thread::spawn(move || {
-        let server = Server::bind("10.133.51.127:5000");
-        server.start(event_queue);
-    });
     let segments: Vec<Segment> = vec![Segment::new(
         0,
         vec![Obstacle {

game/src/server2.rs

@@ -0,0 +1,52 @@
+use std::{
+    io::{self, BufReader, Read, Write},
+    net::TcpStream,
+    slice::{from_raw_parts, from_raw_parts_mut},
+    thread,
+    time::Duration,
+};
+
+#[repr(C)]
+enum ReqTy {
+    Subscribe = 0,
+}
+
+#[repr(C)]
+struct ReqHeader {
+    ty: ReqTy,
+}
+
+#[repr(C)]
+pub struct Measurement {
+    pub angle: f64,
+}
+
+pub struct Server2 {
+    stream: TcpStream,
+}
+
+impl Server2 {
+    pub fn new() -> io::Result<Self> {
+        Ok(Self {
+            stream: TcpStream::connect("127.0.0.1:8889")?,
+        })
+    }
+
+    pub fn subscribe(&mut self, cb: impl Fn(Measurement) -> ()) -> io::Result<()> {
+        let req = ReqHeader {
+            ty: ReqTy::Subscribe,
+        };
+        let req_as_bytes =
+            unsafe { from_raw_parts((&req as *const _) as *const u8, size_of::<ReqHeader>()) };
+        self.stream.write_all(req_as_bytes)?;
+
+        loop {
+            let mut packet = Measurement { angle: 0.0 };
+            let packet_as_bytes = unsafe {
+                from_raw_parts_mut(&mut packet as *mut _ as *mut u8, size_of::<Measurement>())
+            };
+            self.stream.read_exact(packet_as_bytes)?;
+            cb(packet);
+        }
+    }
+}