v1.0.0 — Hexagonal architecture rewrite
All checks were successful
CI / ci (push) Successful in 7m16s
All checks were successful
CI / ci (push) Successful in 7m16s
Restructure the monolithic 252-line main.rs into a 10-crate workspace with clean hexagonal architecture, swappable adapters, and a production- ready deployment pipeline. Backend architecture: - domain: Canvas, Color/Position/PixelUpdate value objects, port traits (CanvasStore, CanvasPersistence, EventBroadcaster), BroadcastEvent - application: use cases (place_pixel, get_state, save/restore snapshot, connect/disconnect), AppState with Arc snapshot cache - config: AppConfig structs + ConfigSource trait - api-types: shared DTOs, event name constants - adapters: config-env, canvas-file, http-axum (rust-embed), socketio, websocket — all behind port traits, swappable via feature flags - server: composition root with graceful shutdown (SIGTERM/SIGINT) Frontend: - Transport abstraction: Socket.IO and native WebSocket via VITE_TRANSPORT - Canvas zoom/pan with mouse wheel, pinch-to-zoom, and +/- buttons - ImageData rendering (~50x faster than fillRect loop) - Touch support, responsive CSS scaling, mobile-friendly layout - OG/Twitter Card meta tags for rich link previews Production: - Docker: musl static build on scratch — 2.73MB image - CI workflows for Gitea and GitHub Actions (fmt, clippy, test, Docker push) - deploy.sh for private registry - 39 unit tests across domain and application - Zero unwraps, zero unsafe, graceful error handling with tracing - Periodic canvas snapshots with rotation, restored on startup - All config via environment variables with typed defaults
This commit is contained in:
14
crates/adapters/websocket/Cargo.toml
Normal file
14
crates/adapters/websocket/Cargo.toml
Normal file
@@ -0,0 +1,14 @@
|
||||
[package]
|
||||
name = "websocket"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
|
||||
[dependencies]
|
||||
application = { workspace = true }
|
||||
domain = { workspace = true }
|
||||
axum = { workspace = true, features = ["ws"] }
|
||||
futures = { workspace = true }
|
||||
serde = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
tokio = { workspace = true }
|
||||
tracing = { workspace = true }
|
||||
163
crates/adapters/websocket/src/handler.rs
Normal file
163
crates/adapters/websocket/src/handler.rs
Normal file
@@ -0,0 +1,163 @@
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::Ordering;
|
||||
|
||||
use axum::extract::State;
|
||||
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
|
||||
use axum::response::IntoResponse;
|
||||
use domain::{BroadcastEvent, BroadcastSubscription, Color, Position};
|
||||
use futures::{SinkExt, StreamExt, stream::SplitSink};
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::{error, info};
|
||||
|
||||
use crate::WsState;
|
||||
use crate::messages::{ClientMessage, ServerMessage};
|
||||
use application::AppState;
|
||||
use application::canvas::place_pixel;
|
||||
|
||||
type WsSender = SplitSink<WebSocket, Message>;
|
||||
|
||||
pub async fn ws_upgrade(
|
||||
ws: WebSocketUpgrade,
|
||||
State(state): State<Arc<WsState>>,
|
||||
) -> impl IntoResponse {
|
||||
ws.on_upgrade(|socket| handle_connection(socket, state))
|
||||
}
|
||||
|
||||
async fn handle_connection(socket: WebSocket, state: Arc<WsState>) {
|
||||
let (mut sender, mut receiver) = socket.split();
|
||||
|
||||
let connection_id = state
|
||||
.connection_counter
|
||||
.fetch_add(1, Ordering::Relaxed)
|
||||
.to_string();
|
||||
|
||||
info!("WebSocket connected: {connection_id}");
|
||||
|
||||
// Subscribe before snapshotting to avoid missing updates
|
||||
let subscription = state.app_state.broadcaster().subscribe();
|
||||
|
||||
if !send_canvas_snapshot(&mut sender, &state.app_state).await {
|
||||
return;
|
||||
}
|
||||
|
||||
application::soldiers::connect::execute(&state.app_state, connection_id.clone());
|
||||
|
||||
let (error_sender, error_receiver) = mpsc::unbounded_channel::<String>();
|
||||
|
||||
let mut send_task = tokio::spawn(run_send_loop(sender, subscription, error_receiver));
|
||||
|
||||
let app_state = state.app_state.clone();
|
||||
let recv_connection_id = connection_id.clone();
|
||||
let mut recv_task = tokio::spawn(async move {
|
||||
while let Some(Ok(message)) = receiver.next().await {
|
||||
if let Message::Text(text) = message {
|
||||
handle_client_message(&app_state, &error_sender, &recv_connection_id, &text);
|
||||
} else if let Message::Close(_) = message {
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
tokio::select! {
|
||||
_ = &mut send_task => recv_task.abort(),
|
||||
_ = &mut recv_task => send_task.abort(),
|
||||
}
|
||||
|
||||
info!("WebSocket disconnected: {connection_id}");
|
||||
application::soldiers::disconnect::execute(&state.app_state, &connection_id);
|
||||
}
|
||||
|
||||
async fn send_canvas_snapshot(sender: &mut WsSender, state: &AppState) -> bool {
|
||||
let pixels = application::canvas::get_state::execute(state);
|
||||
let bytes = Color::collect_as_bytes(&pixels);
|
||||
sender.send(Message::Binary(bytes.into())).await.is_ok()
|
||||
}
|
||||
|
||||
async fn run_send_loop(
|
||||
mut sender: WsSender,
|
||||
mut subscription: BroadcastSubscription,
|
||||
mut error_receiver: mpsc::UnboundedReceiver<String>,
|
||||
) {
|
||||
loop {
|
||||
tokio::select! {
|
||||
Some(event) = subscription.recv() => {
|
||||
let Some(json) = serialize_broadcast_event(&event) else { continue };
|
||||
if sender.send(Message::Text(json.into())).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Some(error_json) = error_receiver.recv() => {
|
||||
if sender.send(Message::Text(error_json.into())).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
else => break,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn serialize_broadcast_event(event: &BroadcastEvent) -> Option<String> {
|
||||
let message = match event {
|
||||
BroadcastEvent::PixelUpdated(update) => ServerMessage::from(*update),
|
||||
BroadcastEvent::SoldierCountChanged(count) => {
|
||||
ServerMessage::CurrentSoldiers { count: *count }
|
||||
}
|
||||
};
|
||||
serde_json::to_string(&message)
|
||||
.inspect_err(|err| error!("Failed to serialize broadcast event: {err}"))
|
||||
.ok()
|
||||
}
|
||||
|
||||
fn handle_client_message(
|
||||
state: &AppState,
|
||||
error_sender: &mpsc::UnboundedSender<String>,
|
||||
connection_id: &str,
|
||||
text: &str,
|
||||
) {
|
||||
let Ok(message) = serde_json::from_str::<ClientMessage>(text) else {
|
||||
return;
|
||||
};
|
||||
|
||||
match message {
|
||||
ClientMessage::PlacePixel { x, y, color } => {
|
||||
handle_place_pixel(state, error_sender, connection_id, x, y, color);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_place_pixel(
|
||||
state: &AppState,
|
||||
error_sender: &mpsc::UnboundedSender<String>,
|
||||
connection_id: &str,
|
||||
x: u32,
|
||||
y: u32,
|
||||
color: u32,
|
||||
) {
|
||||
let command = place_pixel::Command {
|
||||
user_id: connection_id,
|
||||
position: Position::new(x, y),
|
||||
color: Color::new(color),
|
||||
};
|
||||
|
||||
match place_pixel::execute(state, command) {
|
||||
Ok(place_pixel::Outcome::Placed(_)) => {}
|
||||
Ok(place_pixel::Outcome::CooldownActive) => {
|
||||
send_error(error_sender, domain::COOLDOWN_MESSAGE);
|
||||
}
|
||||
Err(err) => {
|
||||
send_error(error_sender, &err.to_string());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn send_error(error_sender: &mpsc::UnboundedSender<String>, message: &str) {
|
||||
let error_message = ServerMessage::Error {
|
||||
message: message.to_string(),
|
||||
};
|
||||
match serde_json::to_string(&error_message) {
|
||||
Ok(json) => {
|
||||
let _ = error_sender.send(json);
|
||||
}
|
||||
Err(err) => error!("Failed to serialize error message: {err}"),
|
||||
}
|
||||
}
|
||||
24
crates/adapters/websocket/src/lib.rs
Normal file
24
crates/adapters/websocket/src/lib.rs
Normal file
@@ -0,0 +1,24 @@
|
||||
mod handler;
|
||||
mod messages;
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::AtomicU64;
|
||||
|
||||
use application::AppState;
|
||||
use axum::{Router, routing::get};
|
||||
|
||||
pub(crate) struct WsState {
|
||||
app_state: Arc<AppState>,
|
||||
connection_counter: AtomicU64,
|
||||
}
|
||||
|
||||
pub fn build_router(state: Arc<AppState>) -> Router {
|
||||
let ws_state = Arc::new(WsState {
|
||||
app_state: state,
|
||||
connection_counter: AtomicU64::new(0),
|
||||
});
|
||||
|
||||
Router::new()
|
||||
.route("/ws", get(handler::ws_upgrade))
|
||||
.with_state(ws_state)
|
||||
}
|
||||
30
crates/adapters/websocket/src/messages.rs
Normal file
30
crates/adapters/websocket/src/messages.rs
Normal file
@@ -0,0 +1,30 @@
|
||||
use domain::PixelUpdate;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(tag = "type")]
|
||||
pub enum ClientMessage {
|
||||
#[serde(rename = "place-pixel")]
|
||||
PlacePixel { x: u32, y: u32, color: u32 },
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(tag = "type")]
|
||||
pub enum ServerMessage {
|
||||
#[serde(rename = "pixel-updated")]
|
||||
PixelUpdated { x: u32, y: u32, color: u32 },
|
||||
#[serde(rename = "current_soldiers")]
|
||||
CurrentSoldiers { count: usize },
|
||||
#[serde(rename = "error")]
|
||||
Error { message: String },
|
||||
}
|
||||
|
||||
impl From<PixelUpdate> for ServerMessage {
|
||||
fn from(update: PixelUpdate) -> Self {
|
||||
Self::PixelUpdated {
|
||||
x: update.position().x(),
|
||||
y: update.position().y(),
|
||||
color: update.color().as_u32(),
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user