code

1 files changed, 121 insertions, 0 deletions

diff --git a/src/spectate/server.rs b/src/spectate/server.rs
new file mode 100644
index 0000000..8fa1a3d
--- /dev/null
+++ b/src/spectate/server.rs
@@ -0,0 +1,121 @@
+use crate::game::protocol::Packet;
+use crate::State;
+
+use super::Config;
+use anyhow::Result;
+use axum::extract;
+use axum::extract::connect_info::ConnectInfo;
+use axum::extract::ws::Message;
+use axum::response::Html;
+use axum::{
+    extract::ws::{WebSocket, WebSocketUpgrade},
+    response::IntoResponse,
+    routing::get,
+    Router,
+};
+use log::{info, warn};
+use std::net::SocketAddr;
+use std::sync::Arc;
+use tokio::spawn;
+use tokio::sync::{broadcast, RwLock};
+
+struct SpectateState {
+    past_events: RwLock<Vec<Packet>>,
+    events: broadcast::Sender<Packet>,
+}
+
+pub async fn spectate_server(config: Config, state: Arc<State>) -> Result<()> {
+    let sstate = Arc::new(SpectateState {
+        past_events: Default::default(),
+        events: broadcast::channel(16).0,
+    });
+    spawn(broadcaster(sstate.clone(), state));
+    let app = Router::new()
+        .route("/", get(index))
+        .route("/events", get(ws_handler))
+        .with_state(sstate);
+    let listener = tokio::net::TcpListener::bind(config.bind).await.unwrap();
+    info!("listening on {}", listener.local_addr()?);
+    axum::serve(
+        listener,
+        app.into_make_service_with_connect_info::<SocketAddr>(),
+    )
+    .await?;
+    Ok(())
+}
+
+async fn index() -> Html<&'static str> {
+    Html(include_str!("index.html"))
+}
+
+async fn broadcaster(sstate: Arc<SpectateState>, state: Arc<State>) {
+    let mut ticks = state.tick.subscribe();
+    while let Ok(new_game) = ticks.recv().await {
+        let mut events = Vec::new();
+
+        {
+            let g = state.game.read().await;
+            if new_game {
+                sstate.past_events.write().await.clear();
+                events.push(Packet::Game {
+                    my_id: 0,
+                    width: g.size.x as usize,
+                    height: g.size.y as usize,
+                });
+                for (player, (_, _, name)) in &g.heads {
+                    events.push(Packet::Player {
+                        id: *player,
+                        name: name.to_owned(),
+                    })
+                }
+            }
+            for (player, (_, pos, _)) in &g.heads {
+                events.push(Packet::Pos {
+                    id: *player,
+                    x: pos.x,
+                    y: pos.y,
+                })
+            }
+            if !g.dead.is_empty() {
+                events.push(Packet::Die(g.dead.clone()));
+            }
+        }
+
+        sstate.past_events.write().await.extend(events.clone());
+        for ev in events {
+            let _ = sstate.events.send(ev);
+        }
+    }
+}
+
+async fn ws_handler(
+    ws: WebSocketUpgrade,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    extract::State(state): extract::State<Arc<SpectateState>>,
+) -> impl IntoResponse {
+    ws.on_upgrade(move |socket| async move {
+        if let Err(e) = handle_socket(socket, addr, state).await {
+            warn!("client error {e}")
+        }
+    })
+}
+
+async fn handle_socket(
+    mut socket: WebSocket,
+    _addr: SocketAddr,
+    state: Arc<SpectateState>,
+) -> anyhow::Result<()> {
+    let past = state.past_events.read().await.clone();
+    for p in past {
+        socket
+            .send(Message::Text(serde_json::to_string(&p)?))
+            .await?;
+    }
+    let mut live = state.events.subscribe();
+    while let Ok(p) = live.recv().await {
+        socket
+            .send(Message::Text(serde_json::to_string(&p)?))
+            .await?;
+    }
+    Ok(())
+}