dahua-camera-server 0.3.1

//! `GET /stream/{id}` — live video as fMP4 over a WebSocket.
//!
//! This is the passthrough path and the reason eight cameras cost ~10% CPU:
//! samples arrive AVCC-framed from RTSP and are wrapped in fMP4 boxes without
//! ever being decoded. The browser's `MediaSource` does the work, on the GPU.

use crate::AppState;
use dahua_camera_rtsp::Fmp4Muxer;
use dahua_camera_rtsp::Camera;
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::IntoResponse;
use std::sync::Arc;
use std::time::Duration;

/// How long to wait for codec parameters before giving up on a client.
///
/// Generous because a lazy camera has to complete an RTSP handshake first,
/// and this is the request that triggers it.
const PARAMS_WAIT: Duration = Duration::from_secs(15);

/// Upgrade a request to the fMP4 stream socket.
pub async fn stream_handler(
    State(state): State<AppState>,
    Path(id): Path<String>,
    upgrade: WebSocketUpgrade,
) -> axum::response::Response {
    let Some(camera) = state.get(&id) else {
        return (StatusCode::NOT_FOUND, format!("no camera '{id}'")).into_response();
    };

    upgrade
        .max_message_size(4 * 1024 * 1024)
        .on_upgrade(move |socket| pump(socket, camera))
        .into_response()
}

async fn pump(mut socket: WebSocket, camera: Arc<Camera>) {
    let id = camera.id().to_owned();

    // Subscribe *before* waiting for parameters: on a lazy camera, this
    // subscription is what makes the supervisor connect at all (invariant 4).
    let mut frames = camera.subscribe();

    let Some(params) = camera.wait_for_params(PARAMS_WAIT).await else {
        tracing::warn!(camera = %id, "no codec parameters within timeout, closing");
        let _ = socket.send(Message::Close(None)).await;
        return;
    };

    let mut generation = params.generation;
    let mut muxer = new_muxer(&camera, &params);

    if socket.send(Message::Binary(muxer.init_segment().into())).await.is_err() {
        return;
    }

    loop {
        tokio::select! {
            incoming = socket.recv() => {
                match incoming {
                    // The client sends nothing meaningful; anything other than
                    // a close is ignored, and a close or error ends the stream.
                    None | Some(Err(_)) | Some(Ok(Message::Close(_))) => break,
                    _ => continue,
                }
            }
            sample = frames.recv_synced() => {
                let Some(sample) = sample else { break };

                // A parameter change invalidates the muxer's `avcC`: the
                // client needs a fresh init segment, and this sample may
                // predate the change, so drop it and resync.
                if let Some(current) = camera.params() {
                    if current.generation != generation {
                        generation = current.generation;
                        muxer = new_muxer(&camera, &current);
                        if socket.send(Message::Binary(muxer.init_segment().into())).await.is_err() {
                            break;
                        }
                        frames.resync();
                        tracing::debug!(camera = %id, generation, "re-initialised after parameter change");
                        continue;
                    }
                }

                let fragment = muxer.fragment(&sample.data, sample.dts, sample.is_idr);
                if socket.send(Message::Binary(fragment.into())).await.is_err() {
                    break;
                }
            }
        }
    }

    tracing::debug!(camera = %id, "stream client disconnected");
}

fn new_muxer(camera: &Camera, params: &dahua_camera_core::StreamParams) -> Fmp4Muxer {
    Fmp4Muxer::new(
        params.codec.avcc.to_vec(),
        params.codec.width,
        params.codec.height,
        params.codec.clock_rate,
        camera.config().fps_hint,
    )
}