dahua-camera-server 0.3.1

axum HTTP, SSE and WebSocket API for the dahua-camera stack
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

//! MJPEG and snapshot endpoints, served from the shared decode lane.
//!
//! Both require the `decode` feature. Neither is on the live-view path: they
//! exist for clients that cannot play fMP4 and for grabbing stills, and they
//! cost real CPU whenever they are open.

use crate::AppState;
use axum::body::Body;
use axum::extract::{Path, State};
use axum::http::{header, Response, StatusCode};
use bytes::Bytes;
use std::time::Duration;

/// Multipart boundary token for the MJPEG stream.
const BOUNDARY: &str = "frame";

/// How long a snapshot request waits for the lane's first frame.
const FIRST_FRAME_WAIT: Duration = Duration::from_secs(5);

/// `GET /snapshot/{id}` — the most recent decoded frame, as JPEG.
///
/// Returns immediately when the lane already has a frame. Otherwise it starts
/// the lane and waits, which is why the first request after an idle period is
/// noticeably slower than the rest.
pub async fn snapshot_handler(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<Response<Body>, StatusCode> {
    let camera = state.get(&id).ok_or(StatusCode::NOT_FOUND)?;
    let mut lane = camera.jpeg_lane().await;

    let jpeg = match lane.latest() {
        Some(jpeg) => jpeg,
        None => tokio::time::timeout(FIRST_FRAME_WAIT, lane.wait_next())
            .await
            .ok()
            .flatten()
            .ok_or(StatusCode::SERVICE_UNAVAILABLE)?,
    };

    Response::builder()
        .header(header::CONTENT_TYPE, "image/jpeg")
        .header(header::CACHE_CONTROL, "no-cache, no-store, must-revalidate")
        .body(Body::from(jpeg))
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
}

/// `GET /mjpeg/{id}` — `multipart/x-mixed-replace` of decoded frames.
///
/// Backed by a `watch` channel, so a slow client receives fewer, newer frames
/// rather than accumulating a backlog (invariant 3).
pub async fn mjpeg_handler(
    State(state): State<AppState>,
    Path(id): Path<String>,
) -> Result<Response<Body>, StatusCode> {
    let camera = state.get(&id).ok_or(StatusCode::NOT_FOUND)?;
    let lane = camera.jpeg_lane().await;

    let stream = futures::stream::unfold(lane, |mut lane| async move {
        let jpeg = lane.wait_next().await?;
        Some((Ok::<Bytes, std::convert::Infallible>(multipart_part(&jpeg)), lane))
    });

    Response::builder()
        .header(header::CONTENT_TYPE, format!("multipart/x-mixed-replace; boundary={BOUNDARY}"))
        .header(header::CACHE_CONTROL, "no-cache, no-store, must-revalidate")
        .body(Body::from_stream(stream))
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
}

/// Frame one JPEG as a multipart part.
fn multipart_part(jpeg: &[u8]) -> Bytes {
    let mut part = Vec::with_capacity(jpeg.len() + 96);
    part.extend_from_slice(format!("--{BOUNDARY}\r\n").as_bytes());
    part.extend_from_slice(b"Content-Type: image/jpeg\r\nContent-Length: ");
    part.extend_from_slice(jpeg.len().to_string().as_bytes());
    part.extend_from_slice(b"\r\n\r\n");
    part.extend_from_slice(jpeg);
    part.extend_from_slice(b"\r\n");
    Bytes::from(part)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_part_carries_the_boundary_type_and_exact_length() {
        let jpeg = [0xff, 0xd8, 0x00, 0x01, 0xff, 0xd9];
        let part = multipart_part(&jpeg);
        let text = String::from_utf8_lossy(&part);

        assert!(text.starts_with("--frame\r\n"), "{text:?}");
        assert!(text.contains("Content-Type: image/jpeg"), "{text:?}");
        assert!(text.contains("Content-Length: 6"), "{text:?}");
        assert!(part.ends_with(b"\r\n"));
    }

    #[test]
    fn the_payload_survives_framing_byte_for_byte() {
        let jpeg = [0xff, 0xd8, 0x0a, 0x0d, 0x2d, 0x2d, 0xff, 0xd9];
        let part = multipart_part(&jpeg);

        let header_end = part
            .windows(4)
            .position(|w| w == b"\r\n\r\n")
            .expect("headers are terminated")
            + 4;
        assert_eq!(&part[header_end..part.len() - 2], &jpeg, "bytes that look like framing must survive");
    }

    #[test]
    fn an_empty_payload_still_produces_a_valid_part() {
        let text = String::from_utf8_lossy(&multipart_part(&[])).to_string();
        assert!(text.contains("Content-Length: 0"), "{text:?}");
    }
}