diff --git a/backend/src/main.cpp b/backend/src/main.cpp index e904c43..343ae8f 100644 --- a/backend/src/main.cpp +++ b/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(); diff --git a/backend/src/server2.cpp b/backend/src/server2.cpp index 24e39fd..98c54ce 100644 --- a/backend/src/server2.cpp +++ b/backend/src/server2.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -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 queued_deletions; + + std::vector 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(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(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")); + } } } diff --git a/backend/src/server2.hpp b/backend/src/server2.hpp index 1a8feb5..afab7e3 100644 --- a/backend/src/server2.hpp +++ b/backend/src/server2.hpp @@ -15,6 +15,7 @@ public: ~Server(); void listen(); + void notify_subscribers(double angle); private: struct State; diff --git a/game/src/main.rs b/game/src/main.rs index 7864c5e..b0fe332 100644 --- a/game/src/main.rs +++ b/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> { + 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 = vec![Segment::new( 0, vec![Obstacle { diff --git a/game/src/server2.rs b/game/src/server2.rs new file mode 100644 index 0000000..a1b4c6e --- /dev/null +++ b/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 { + 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::()) }; + 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::()) + }; + self.stream.read_exact(packet_as_bytes)?; + cb(packet); + } + } +}