vector_calls/
wire.rs

1//! The call's wire format over one Iroh connection: a reliable control stream for
2//! the few messages that must arrive, unreliable QUIC datagrams for audio, where a
3//! late frame is worth less than nothing, and one unidirectional QUIC stream per
4//! video frame, so a keyframe always lands whole and a stale delta can be reset.
5
6use bytes::{BufMut, Bytes, BytesMut};
7use serde::{Deserialize, Serialize};
8use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
9
10pub const CALL_ALPN: &[u8] = b"vector/call/1";
11
12/// Datagram header: sequence, milliseconds since the sender's media start, flags.
13pub const HEADER_LEN: usize = 8;
14
15pub struct Frame<'a> {
16    pub seq: u16,
17    pub flags: u16,
18    pub payload: &'a [u8],
19}
20
21pub fn pack(seq: u16, ts_ms: u32, flags: u16, payload: &[u8]) -> Bytes {
22    let mut b = BytesMut::with_capacity(HEADER_LEN + payload.len());
23    b.put_u16(seq);
24    b.put_u32(ts_ms);
25    b.put_u16(flags);
26    b.put_slice(payload);
27    b.freeze()
28}
29
30pub fn unpack(data: &[u8]) -> Option<Frame<'_>> {
31    if data.len() < HEADER_LEN {
32        return None;
33    }
34    Some(Frame {
35        seq: u16::from_be_bytes([data[0], data[1]]),
36        flags: u16::from_be_bytes([data[6], data[7]]),
37        payload: &data[HEADER_LEN..],
38    })
39}
40
41/// Control messages, length-prefixed JSON on the bidirectional stream.
42#[derive(Serialize, Deserialize, Debug, Clone, PartialEq)]
43#[serde(tag = "t", rename_all = "snake_case")]
44pub enum Control {
45    /// First message from the caller: which call this connection carries.
46    Hello { call_id: String },
47    Mute { on: bool },
48    Bye,
49    /// What I am sending on the video streams from now on: either, both or neither,
50    /// and whether the screen comes with its sound.
51    Video {
52        camera: bool,
53        screen: bool,
54        #[serde(default)]
55        audio: bool,
56    },
57    /// My decoder lost that chain: its next frame must be a keyframe.
58    KeyframeRequest { kind: VideoKind },
59    /// My view of your video is hidden; stop spending upload on it (or resume).
60    VideoPause { on: bool },
61    /// My decoder cannot take this codec after all; send with another.
62    VideoUnsupported { codec: VideoCodec },
63    /// A message from a newer peer: skipped, never fatal.
64    #[serde(other)]
65    Unknown,
66}
67
68/// The two pictures a side can send at once; a frame's flags say which it is.
69#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
70#[serde(rename_all = "snake_case")]
71pub enum VideoKind {
72    Camera,
73    Screen,
74}
75
76impl VideoKind {
77    pub fn index(self) -> usize {
78        match self {
79            Self::Camera => 0,
80            Self::Screen => 1,
81        }
82    }
83    pub fn from_flags(flags: u8) -> Self {
84        if flags & VIDEO_SCREEN != 0 { Self::Screen } else { Self::Camera }
85    }
86}
87
88/// Which of the two a side is sending.
89#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
90pub struct Tracks {
91    pub camera: bool,
92    pub screen: bool,
93}
94
95impl Tracks {
96    pub fn any(self) -> bool {
97        self.camera || self.screen
98    }
99    pub fn has(self, kind: VideoKind) -> bool {
100        match kind {
101            VideoKind::Camera => self.camera,
102            VideoKind::Screen => self.screen,
103        }
104    }
105    pub fn with(mut self, kind: VideoKind, on: bool) -> Self {
106        match kind {
107            VideoKind::Camera => self.camera = on,
108            VideoKind::Screen => self.screen = on,
109        }
110        self
111    }
112}
113
114/// The codec byte in a video frame header. A fixed table, so nothing from the wire
115/// is ever handed to a decoder as a string.
116#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
117#[serde(rename_all = "snake_case")]
118#[repr(u8)]
119pub enum VideoCodec {
120    H264 = 1,
121    Vp8 = 2,
122}
123
124impl VideoCodec {
125    pub fn from_byte(b: u8) -> Option<Self> {
126        match b {
127            1 => Some(Self::H264),
128            2 => Some(Self::Vp8),
129            _ => None,
130        }
131    }
132    /// The name the signalling tag and the webview use.
133    pub fn name(self) -> &'static str {
134        match self {
135            Self::H264 => "h264",
136            Self::Vp8 => "vp8",
137        }
138    }
139    pub fn from_name(s: &str) -> Option<Self> {
140        match s {
141            "h264" => Some(Self::H264),
142            "vp8" => Some(Self::Vp8),
143            _ => None,
144        }
145    }
146}
147
148/// Video frame flags.
149pub const VIDEO_KEY: u8 = 1;
150pub const VIDEO_SCREEN: u8 = 2;
151/// Header on every video frame, on the QUIC stream and on the link alike.
152pub const VIDEO_HEADER_LEN: usize = 16;
153/// A frame bigger than this is refused: the whole connection window is only 2 MB,
154/// and a bigger frame would sit in front of the control stream. A crisp 1440p
155/// keyframe at the top of the screen ladder is a few hundred kilobytes.
156pub const MAX_VIDEO_FRAME: usize = 1024 * 1024;
157
158#[derive(Debug, Clone, Copy, PartialEq, Eq)]
159pub struct VideoHeader {
160    pub seq: u32,
161    /// Capture time in the sender's microseconds.
162    pub ts_us: u64,
163    pub flags: u8,
164    pub codec: VideoCodec,
165}
166
167impl VideoHeader {
168    pub fn is_key(&self) -> bool {
169        self.flags & VIDEO_KEY != 0
170    }
171
172    pub fn write(&self, out: &mut [u8]) {
173        debug_assert!(out.len() >= VIDEO_HEADER_LEN);
174        out[0..4].copy_from_slice(&self.seq.to_be_bytes());
175        out[4..12].copy_from_slice(&self.ts_us.to_be_bytes());
176        out[12] = self.flags;
177        out[13] = self.codec as u8;
178        out[14] = 0;
179        out[15] = 0;
180    }
181
182    pub fn parse(data: &[u8]) -> Option<Self> {
183        if data.len() < VIDEO_HEADER_LEN {
184            return None;
185        }
186        Some(Self {
187            seq: u32::from_be_bytes([data[0], data[1], data[2], data[3]]),
188            ts_us: u64::from_be_bytes([data[4], data[5], data[6], data[7], data[8], data[9], data[10], data[11]]),
189            flags: data[12],
190            codec: VideoCodec::from_byte(data[13])?,
191        })
192    }
193}
194
195pub const MAX_CONTROL_LEN: usize = 4096;
196
197pub async fn write_control<W: AsyncWrite + Unpin>(send: &mut W, msg: &Control) -> Result<(), String> {
198    let body = serde_json::to_vec(msg).map_err(|e| e.to_string())?;
199    if body.len() > MAX_CONTROL_LEN {
200        return Err("control message too large".into());
201    }
202    let mut buf = Vec::with_capacity(2 + body.len());
203    buf.extend_from_slice(&(body.len() as u16).to_be_bytes());
204    buf.extend_from_slice(&body);
205    send.write_all(&buf).await.map_err(|e| e.to_string())
206}
207
208pub async fn read_control<R: AsyncRead + Unpin>(recv: &mut R) -> Result<Control, String> {
209    let mut len = [0u8; 2];
210    recv.read_exact(&mut len).await.map_err(|e| e.to_string())?;
211    let len = u16::from_be_bytes(len) as usize;
212    if len == 0 || len > MAX_CONTROL_LEN {
213        return Err("control message length out of range".into());
214    }
215    let mut body = vec![0u8; len];
216    recv.read_exact(&mut body).await.map_err(|e| e.to_string())?;
217    serde_json::from_slice(&body).map_err(|e| e.to_string())
218}
219
220#[cfg(test)]
221mod tests {
222    use super::*;
223
224    #[test]
225    fn a_video_header_round_trips_and_refuses_an_unknown_codec() {
226        let h = VideoHeader { seq: 70_000, ts_us: 1 << 40, flags: VIDEO_KEY | VIDEO_SCREEN, codec: VideoCodec::Vp8 };
227        let mut buf = [0u8; VIDEO_HEADER_LEN + 3];
228        h.write(&mut buf);
229        assert_eq!(VideoHeader::parse(&buf), Some(h));
230        assert!(h.is_key());
231        buf[13] = 9;
232        assert!(VideoHeader::parse(&buf).is_none());
233        assert!(VideoHeader::parse(&buf[..10]).is_none());
234    }
235
236    #[test]
237    fn an_unknown_control_message_parses_as_unknown() {
238        let c: Control = serde_json::from_str(r#"{"t":"hologram","on":true}"#).unwrap();
239        assert_eq!(c, Control::Unknown);
240        let v: Control = serde_json::from_str(r#"{"t":"video","camera":false,"screen":true}"#).unwrap();
241        assert_eq!(v, Control::Video { camera: false, screen: true, audio: false });
242        assert_eq!(serde_json::to_string(&Control::KeyframeRequest { kind: VideoKind::Camera }).unwrap(), r#"{"t":"keyframe_request","kind":"camera"}"#);
243        assert_eq!(VideoKind::from_flags(VIDEO_KEY | VIDEO_SCREEN), VideoKind::Screen);
244        assert!(Tracks::default().with(VideoKind::Screen, true).has(VideoKind::Screen));
245    }
246
247    #[test]
248    fn a_frame_round_trips() {
249        let b = pack(7, 1234, 1, &[9, 8, 7]);
250        let f = unpack(&b).unwrap();
251        assert_eq!((f.seq, f.flags, f.payload), (7, 1, &[9u8, 8, 7][..]));
252        assert!(unpack(&b[..5]).is_none());
253    }
254}
vector_calls/wire.rs

vector_calls/
wire.rs
