Migrate to tokio

Author: jsurrea

antares/Cargo.lock

@@ -1,10 +1,15 @@
 [package]
-name = "naval-radar-simulator"
-version = "0.1.0"
+name = "antares"
+version = "1.0.0"
 edition = "2021"
 
 [dependencies]
 chrono = "0.4.38"
 rand = "0.8.5"
-serde = { version = "1.0.214", features = ["derive"] }
+tokio = { version = "1", features = ["full"] }
+serde = { version = "1", features = ["derive"] }
 toml = "0.8.19"
+async-trait = "0.1.88"
+tokio-tungstenite = "0.26.2"
+futures-util = "0.3.31"
+futures = "0.3.31"

antares/Cargo.toml

@@ -4,7 +4,7 @@
 //!
 
 use super::{Config, Radar, Simulation};
-use std::sync::mpsc::channel;
+use tokio::sync::mpsc;
 
 pub struct Controller {
     radar: Radar,
@@ -19,10 +19,15 @@ impl Controller {
         }
     }
 
-    pub fn run(self) {
-        let (wave_sender, wave_receiver) = channel();
-        self.simulation.start(wave_sender);
-        self.radar.start(wave_receiver);
-        loop {}
+    pub async fn run(self) {
+        let (wave_sender, wave_receiver) = mpsc::channel(100);
+        let Controller { radar, simulation } = self;
+
+        tokio::spawn(simulation.start(wave_sender));
+        tokio::spawn(radar.start(wave_receiver));
+
+        tokio::signal::ctrl_c()
+            .await
+            .expect("failed to listen for Ctrl+C");
     }
 }

antares/src/controller.rs

@@ -1,13 +1,12 @@
-//! # Radar Simulation Library
+//! # ANTARES Radar Simulation Library
 //!
-//! `radar-simulation` is a library for simulating radar data.
+//! `antares` is a library for simulating radar data.
 //!
 
 mod config;
 mod controller;
 mod radar;
 mod simulation;
-mod utils;
 
 use radar::{Radar, RadarConfig};
 use simulation::{Simulation, SimulationConfig, Wave};

antares/src/lib.rs

@@ -1,11 +1,11 @@
-use naval_radar_simulator::{Config, Controller};
+use antares::{Config, Controller};
 use std::{env, fs, process};
 
 fn main() {
     let args: Vec<String> = env::args().collect();
 
     if args.len() != 2 {
-        eprintln!("Usage: naval_radar <config-file>");
+        eprintln!("Usage: antares <config-file>");
         process::exit(1);
     }
 

antares/src/main.rs

@@ -0,0 +1,10 @@
+use super::super::Track;
+use async_trait::async_trait;
+use tokio::sync::broadcast::Sender;
+
+/// A trait representing a broadcast interface that sends track data to clients.
+#[async_trait]
+pub trait Broadcaster: Send + Sync {
+    async fn start(&self);
+    fn sender(&self) -> Sender<Track>;
+}

antares/src/radar/broadcaster/broadcaster.rs

@@ -0,0 +1,7 @@
+mod broadcaster;
+mod tcp;
+mod websocket;
+
+pub use broadcaster::Broadcaster;
+pub use tcp::TcpBroadcaster;
+pub use websocket::WebSocketBroadcaster;

antares/src/radar/broadcaster/mod.rs

@@ -0,0 +1,51 @@
+use super::super::Track;
+use super::Broadcaster;
+use async_trait::async_trait;
+use tokio::io::AsyncWriteExt;
+use tokio::net::{TcpListener, TcpStream};
+use tokio::sync::broadcast::{self, Sender};
+
+pub struct TcpBroadcaster {
+    bind_addr: String,
+    sender: Sender<Track>,
+}
+
+impl TcpBroadcaster {
+    pub fn new(bind_addr: String) -> Self {
+        let (sender, _) = broadcast::channel(100);
+        Self { bind_addr, sender }
+    }
+
+    async fn handle_client(
+        mut stream: TcpStream,
+        mut receiver: tokio::sync::broadcast::Receiver<Track>,
+    ) {
+        while let Ok(track) = receiver.recv().await {
+            let csv_line = format!("{}\n", track.serialize());
+            if let Err(e) = stream.write_all(csv_line.as_bytes()).await {
+                eprintln!("Failed to send data to TCP client: {}", e);
+                break;
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl Broadcaster for TcpBroadcaster {
+    async fn start(&self) {
+        let listener = TcpListener::bind(&self.bind_addr).await.unwrap();
+        let sender = self.sender.clone();
+
+        println!("📡 TCP broadcaster listening on {}", self.bind_addr);
+
+        loop {
+            let (stream, _) = listener.accept().await.unwrap();
+            let receiver = sender.subscribe();
+            tokio::spawn(Self::handle_client(stream, receiver));
+        }
+    }
+
+    fn sender(&self) -> Sender<Track> {
+        self.sender.clone()
+    }
+}

antares/src/radar/broadcaster/tcp.rs

@@ -0,0 +1,53 @@
+use super::super::Track;
+use super::Broadcaster;
+use async_trait::async_trait;
+use futures_util::{SinkExt, StreamExt};
+use tokio::net::TcpListener;
+use tokio::sync::broadcast::{self, Sender};
+use tokio_tungstenite::accept_async;
+use tokio_tungstenite::tungstenite::Message;
+
+pub struct WebSocketBroadcaster {
+    bind_addr: String,
+    sender: Sender<Track>,
+}
+
+impl WebSocketBroadcaster {
+    pub fn new(bind_addr: String) -> Self {
+        let (sender, _) = broadcast::channel(100);
+        Self { bind_addr, sender }
+    }
+
+    async fn handle_client(
+        stream: tokio::net::TcpStream,
+        mut receiver: tokio::sync::broadcast::Receiver<Track>,
+    ) {
+        if let Ok(ws_stream) = accept_async(stream).await {
+            let (mut sink, _) = ws_stream.split();
+            while let Ok(track) = receiver.recv().await {
+                let csv = format!("{}\n", track.serialize());
+                let _ = sink.send(Message::Text(csv.into())).await;
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl Broadcaster for WebSocketBroadcaster {
+    async fn start(&self) {
+        let listener = TcpListener::bind(&self.bind_addr).await.unwrap();
+        let sender = self.sender.clone();
+
+        println!("📡 WebSocket broadcaster listening on {}", self.bind_addr);
+
+        loop {
+            let (stream, _) = listener.accept().await.unwrap();
+            let receiver = sender.subscribe();
+            tokio::spawn(Self::handle_client(stream, receiver));
+        }
+    }
+
+    fn sender(&self) -> Sender<Track> {
+        self.sender.clone()
+    }
+}

antares/src/radar/broadcaster/websocket.rs

@@ -3,14 +3,7 @@ use serde::Deserialize;
 #[derive(Debug, Deserialize)]
 pub struct RadarConfig {
     pub detector: DetectorConfig,
-    pub protocol: ProtocolConfig,
-}
-
-#[derive(Debug, Deserialize)]
-pub struct ProtocolConfig {
-    pub host: String,
-    pub num_workers_tci: usize,
-    pub num_workers_tdi: usize,
+    pub broadcast: BroadcastConfig,
 }
 
 #[derive(Debug, Deserialize, Clone)]
@@ -20,3 +13,10 @@ pub struct DetectorConfig {
     pub angle: f64,
     pub start_coordinates: (f64, f64),
 }
+
+#[derive(Debug, Deserialize)]
+#[serde(tag = "type", rename_all = "lowercase")]
+pub enum BroadcastConfig {
+    Tcp { bind_addr: String },
+    WebSocket { bind_addr: String },
+}