Library Usage

Signal Fish Server is published as both a binary and a library crate. Embed the signaling server into your own Rust application.

EnhancedGameServer::new returns Arc<EnhancedGameServer>. Call handle_provide_connection_info through that Arc, rather than a borrowed &EnhancedGameServer. Its owned metadata transaction continues if the calling task is canceled. A running Direct host endpoint update publishes a fresh SessionPlan without a membership change; a rejected publication before commit leaves the old metadata intact and returns an error to the player.

Add Dependency

TOML

[dependencies]
signal-fish-server = "0.11.0"
tokio = { version = "1", features = ["full"] }
anyhow = "1"
axum = "0.8"

Basic Embedded Server

Run this complete example with cargo run --example embedded_server from the repository. For another application, use the dependencies above and copy the code below. Use the configuration guide to provide valid settings. The default metrics policy needs a configured token. Press Ctrl+C to send the shutdown advisory, close sockets, and wait for cleanup before the runtime exits. Signal setup failures return an error after cleanup.

The source is examples/embedded_server.rs. The example and this block use the same code. The tests run its startup and shutdown functions against an in-memory server.

Rust

use signal_fish_server::{
    config,
    database::DatabaseConfig,
    server::{run_drain_choreography, EnhancedGameServer, ServerConfig},
    websocket,
};
use std::{future::Future, net::SocketAddr, sync::Arc, time::Duration};

#[cfg(not(test))]
#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let cfg = config::load()?;
    let game_server = build_server(&cfg).await?;
    let listener =
        tokio::net::TcpListener::bind(SocketAddr::from(([0, 0, 0, 0], cfg.port))).await?;
    println!("Server listening on {}", listener.local_addr()?);
    serve_until(
        listener,
        game_server,
        &cfg.security.cors_origins,
        tokio::signal::ctrl_c(),
    )
    .await
}

pub async fn build_server(cfg: &config::Config) -> anyhow::Result<Arc<EnhancedGameServer>> {
    config::validate_config_security(cfg)?;
    let server_config = ServerConfig {
        default_max_players: cfg.server.default_max_players,
        ping_timeout: Duration::from_secs(cfg.server.ping_timeout),
        room_cleanup_interval: Duration::from_secs(cfg.server.room_cleanup_interval),
        drain_grace: Duration::from_secs(cfg.server.drain_grace_secs),
        max_rooms_per_game: cfg.server.max_rooms_per_game,
        max_rooms: cfg.server.max_rooms,
        rate_limit_config: signal_fish_server::rate_limit::RateLimitConfig {
            max_room_creations: cfg.rate_limit.max_room_creations,
            time_window: Duration::from_secs(cfg.rate_limit.time_window),
            max_join_attempts: cfg.rate_limit.max_join_attempts,
            max_signals: cfg.rate_limit.max_signals,
            max_signal_errors: cfg.rate_limit.max_signal_errors,
            max_inbound_error_replies: cfg.rate_limit.max_inbound_error_replies,
            max_relay_bytes: cfg.rate_limit.max_relay_bytes,
            max_room_relay_bytes: cfg.rate_limit.max_room_relay_bytes,
        },
        empty_room_timeout: Duration::from_secs(cfg.server.empty_room_timeout),
        inactive_room_timeout: Duration::from_secs(cfg.server.inactive_room_timeout),
        max_message_size: cfg.security.max_message_size,
        max_outbound_message_size: cfg.security.max_outbound_message_size,
        max_signal_bytes: cfg.security.max_signal_bytes,
        max_connection_info_bytes: cfg.security.max_connection_info_bytes,
        max_game_data_bytes: cfg.security.max_game_data_bytes.clone(),
        max_connections_per_ip: cfg.security.max_connections_per_ip,
        max_connections: cfg.security.max_connections,
        require_metrics_auth: cfg.security.require_metrics_auth,
        metrics_auth_token: cfg.security.metrics_auth_token.clone(),
        reconnection_window: Duration::from_secs(cfg.server.reconnection_window),
        event_buffer_size: cfg.server.event_buffer_size,
        enable_reconnection: cfg.server.enable_reconnection,
        websocket_config: cfg.websocket.clone(),
        app_id_allowlist_enabled: cfg.security.enforce_app_id_allowlist,
        heartbeat_throttle: Duration::from_secs(cfg.server.heartbeat_throttle_secs),
        region_id: cfg.server.region_id.clone(),
        room_code_prefix: cfg.server.room_code_prefix.clone(),
        default_max_spectators: cfg.server.default_max_spectators,
    };
    let game_server = EnhancedGameServer::new(
        server_config,
        cfg.protocol.clone(),
        cfg.relay_types.clone(),
        cfg.session.clone(),
        cfg.turn.clone(),
        DatabaseConfig::InMemory,
        cfg.metrics.clone(),
        cfg.coordination.clone(),
        cfg.security.transport.clone(),
        cfg.security.allowed_apps.clone(),
    )
    .await?;
    game_server.install_connect_token_key(&cfg.security)?;
    Ok(game_server)
}

pub async fn serve_until(
    listener: tokio::net::TcpListener,
    game_server: Arc<EnhancedGameServer>,
    cors_origins: &str,
    shutdown: impl Future<Output = std::io::Result<()>> + Send + 'static,
) -> anyhow::Result<()> {
    let router = websocket::create_standalone_router(cors_origins).with_state(game_server.clone());
    let (shutdown_tx, mut shutdown_rx) = tokio::sync::watch::channel(false);
    let cleanup_server = game_server.clone();
    let cleanup = tokio::spawn(async move {
        cleanup_server
            .cleanup_task_until(async move {
                let _ = shutdown_rx.wait_for(|stopping| *stopping).await;
            })
            .await;
    });
    let drain_server = game_server.clone();
    let drain_shutdown_tx = shutdown_tx.clone();
    let (signal_result_tx, signal_result_rx) = tokio::sync::oneshot::channel();
    let result = axum::serve(
        listener,
        router.into_make_service_with_connect_info::<SocketAddr>(),
    )
    .with_graceful_shutdown(async move {
        let signal_result = shutdown.await;
        run_drain_choreography(&drain_server, drain_shutdown_tx).await;
        let _ = signal_result_tx.send(signal_result);
    })
    .await;
    if !game_server.is_draining() {
        run_drain_choreography(&game_server, shutdown_tx).await;
    }
    cleanup.await?;
    result?;
    signal_result_rx.await??;
    Ok(())
}

Keep the owning Tokio runtime alive until the shutdown drain finishes. Canceling an individual socket caller leaves its owned supervisor running. Destroying the runtime cancels every task and cannot complete asynchronous connection cleanup. If you destroy the runtime before draining, discard the server instance. Do not reuse a surviving Arc<EnhancedGameServer> on another runtime: its registrations, room membership, and connection metrics can retain state from dead sockets.

The supplied WebSocket routers accept HTTP/1.1 GET and HTTP/2 extended CONNECT. If you mount a raw WebSocket handler, allow both methods and enable extended CONNECT on your HTTP/2 listener. Keep the required Origin policy extension.

The admission API now returns RegisterClientError::AdmissionFailed when admission unwinds or its owned task ends before handoff. It returns no connection identity for that attempt. This is a breaking Rust API change: add an AdmissionFailed arm to exhaustive matches on RegisterClientError. Handle it as a failed connection attempt. The WebSocket path reports close code 1011 with reason admission_failed. Panic payloads are not returned to the client.

Custom Storage Backend

The fragments below are schematic. They need your storage implementation and all remaining GameDatabase methods. Add async-trait and uuid to your application dependencies.

Implement the GameDatabase trait for custom room-record storage. This alone does not persist a live room across process restart; read the consistency and durability contract before designing a custom backend.

Override get_total_room_count with your storage's exact live-room count: its trait default fails closed (returns an error), and the server-wide room ceiling (server.max_rooms, on by default) consults it on every room creation, so a backend that does not implement the query refuses every creation:

Rust

use signal_fish_server::database::GameDatabase;
use signal_fish_server::protocol::{Room, RoomId, PlayerId, PlayerInfo, SpectatorInfo, ConnectionInfo, LobbyState};
use async_trait::async_trait;
use anyhow::Result;
use std::collections::HashMap;
use std::any::Any;
use uuid::Uuid;

pub struct MyCustomDatabase {
    // Your storage implementation (e.g., Redis client, PostgreSQL pool, etc.)
}

#[async_trait]
impl GameDatabase for MyCustomDatabase {
    // Required for the server-wide room ceiling (server.max_rooms): the
    // trait default fails closed and refuses every room creation.
    async fn get_total_room_count(&self) -> anyhow::Result<usize> {
        // Return the exact live-room row count from your storage.
        todo!("sum the live rooms your storage holds")
    }

    async fn initialize(&self) -> Result<()> {
        // Initialize database connection and run migrations
        Ok(())
    }

    async fn create_room(
        &self,
        game_name: String,
        room_code: Option<String>,
        max_players: u8,
        supports_authority: bool,
        creator_id: PlayerId,
        relay_type: String,
        region_id: String,
        application_id: Option<Uuid>,
    ) -> Result<Room> {
        // Create and store room in your database
        todo!("Implement room creation")
    }

    async fn get_room(&self, game_name: &str, room_code: &str) -> Result<Option<Room>> {
        // Retrieve room from your database by game name and room code
        Ok(None)
    }

    async fn get_room_by_id(&self, room_id: &RoomId) -> Result<Option<Room>> {
        // Retrieve room from your database by ID
        Ok(None)
    }

    async fn add_player_to_room(&self, room_id: &RoomId, player: PlayerInfo) -> Result<bool> {
        // Add player to room (returns false if room is full)
        Ok(false)
    }

    async fn remove_player_from_room(
        &self,
        room_id: &RoomId,
        player_id: &PlayerId,
    ) -> Result<Option<PlayerInfo>> {
        // Remove player from room
        Ok(None)
    }

    async fn get_room_players(&self, room_id: &RoomId) -> Result<Vec<PlayerInfo>> {
        // Get all players in a room
        Ok(Vec::new())
    }

    async fn delete_room(&self, room_id: &RoomId) -> Result<bool> {
        // Delete a specific room by ID
        Ok(false)
    }

    async fn health_check(&self) -> bool {
        // Health check for your database
        true
    }

    fn as_any(&self) -> &(dyn Any + Send + Sync) {
        self
    }

    // ... implement all other required methods from the GameDatabase trait
    // See src/database/mod.rs for the complete trait definition
}

Use your custom database:

Rust

let game_server = EnhancedGameServer::new(
    server_config,
    protocol_config,
    relay_types_config,
    session_config,
    turn_config,
    DatabaseConfig::Custom(Arc::new(MyCustomDatabase::new())),
    metrics_config,
    coordination_config,
    transport_config,
    allowed_apps,
)
.await?;

Custom Router Integration

This fragment needs game_server from the complete example, a SocketAddr import, and your api_status handler. Use the complete example's shutdown path when you serve the merged router.

Integrate Signal Fish into an existing Axum application:

Rust

use axum::{Router, routing::get};
use signal_fish_server::websocket;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Your existing routes
    let app = Router::new()
        .route("/", get(|| async { "Hello, world!" }))
        .route("/api/status", get(api_status));

    // Add Signal Fish routes
    let signal_fish_routes = websocket::create_standalone_router("*")
        .with_state(game_server);

    // Merge routers
    let app = app.merge(signal_fish_routes);

    // Start server
    let addr = SocketAddr::from(([0, 0, 0, 0], 3000));
    let listener = tokio::net::TcpListener::bind(addr).await?;
    axum::serve(listener, app).await?;

    Ok(())
}

Programmatic Configuration

Build configuration programmatically instead of from files:

Rust

use signal_fish_server::config::{
    Config, ServerConfig, ProtocolConfig, SecurityConfig,
};

let config = Config {
    port: 3536,
    server: ServerConfig {
        default_max_players: 8,
        ping_timeout: 30,
        room_cleanup_interval: 60,
        ..Default::default()
    },
    protocol: ProtocolConfig {
        max_game_name_length: 64,
        room_code_length: 6,
        ..Default::default()
    },
    security: SecurityConfig {
        cors_origins: "*".to_string(),
        enforce_app_id_allowlist: false,
        ..Default::default()
    },
    ..Default::default()
};

Message Handling

Note: The EnhancedGameServer does not currently expose public APIs for directly sending messages to players. The server automatically handles message routing based on client connections and room membership. This is an internal implementation detail.

If you need to send custom server-initiated messages, you would need to extend the server implementation or use the internal message coordinator interfaces.

Event Hooks

Note: The EnhancedGameServer does not currently expose a public event subscription API. Server events are handled internally for metrics, logging, and coordination.

If you need to monitor server events, consider:

• Using the metrics endpoints (/metrics or /metrics/prom) to track room and player counts
• Implementing custom logging by extending the server modules
• Monitoring structured logs with a log aggregation system (all events are logged with tracing)

Feature Flags

Enable optional features:

TOML

[dependencies]
signal-fish-server = { version = "0.11.0", features = ["tls", "legacy-fullmesh"] }

Available features:

• tls - Built-in TLS/mTLS support
• legacy-fullmesh - Upstream matchbox full-mesh signaling mode. Dangerous if exposed: it carries none of the main service's controls — no authentication, no app allowlist, no rate limits, no telemetry, and no graceful shutdown (a matchbox 0.14 limitation). It binds the main port + 1. It must never face the public internet in hosted deployments; use it only for local interop testing against legacy matchbox clients.

Testing

This fragment is schematic. It needs constructor arguments and a room created through the WebSocket API. For a runnable embedding test, see the starter tests.

Use the library in your tests:

Rust

#[cfg(test)]
mod tests {
    use signal_fish_server::{
        server::EnhancedGameServer,
        protocol::{ClientMessage, ServerMessage},
    };

    #[tokio::test]
    async fn test_room_creation() {
        // Note: This is a conceptual example. The actual server does not expose
        // a public handle_message API. Message handling is internal to the WebSocket
        // connection lifecycle. For testing, see tests/integration_tests.rs for
        // examples of how to test via the WebSocket interface.

        let server = EnhancedGameServer::new(
            /* config */
        ).await.unwrap();

        // Example: Test via database state instead
        let room = server.database().get_room("test-game", "ABC123").await.unwrap();
        assert!(room.is_some());
    }
}

API Documentation

Generate full API docs:

Bash

cargo doc --open --no-deps

Next Steps

• Protocol Reference - Message types and flow
• Development - Building and testing