From d7961cdc9891167af7c9195d62560e43fe7505ab Mon Sep 17 00:00:00 2001 From: enzotar Date: Wed, 25 Feb 2026 10:29:44 -0800 Subject: [PATCH] =?UTF-8?q?feat:=20universal=20bitstream=20streaming=20?= =?UTF-8?q?=E2=80=94=20any=20input=20=E2=86=92=20any=20output?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New crate: engine/ds-stream/ - Binary protocol: 16-byte header, typed frame/input enums - Frame types: Pixels, Delta, Audio, Signal, Neural (0x01-0x43) - Input types: Pointer, Key, Scroll, Gamepad, MIDI, BCI (0x01-0x90) - WebSocket relay server (tokio + tungstenite) - Source → receivers: frame broadcast - Receivers → source: input routing - Codec: encode/decode, XOR delta compression, RLE, convenience builders - 17 unit tests, all passing Streaming modes (stream-source.html): 1. Pixel mode: raw RGBA framebuffer (~28 MB/s) 2. Delta mode: XOR + RLE compression (~1-9 MB/s, 70-95% savings) 3. Signal mode: compact JSON signal diffs (~2 KB/s, 12000x reduction) 4. Neural mode: procedural SDF pixel generator (concept demo) 5. Audio channel: spring velocity→frequency synthesis 6. Multi-receiver: broadcast to all connected clients Thin receiver client (stream-receiver.html, ~300 lines): - Zero framework, zero build step - Renders any incoming bitstream mode - Local signal-diff renderer for signal mode - AudioContext playback for audio frames - Full input capture: click/drag, keyboard, scroll - Per-channel bitstream bus visualization DREAMSTACK.md: Phase 7 section with protocol spec --- Cargo.toml | 4 + DREAMSTACK.md | 61 ++- engine/ds-stream/Cargo.toml | 20 + engine/ds-stream/src/codec.rs | 247 ++++++++++ engine/ds-stream/src/lib.rs | 26 + engine/ds-stream/src/main.rs | 28 ++ engine/ds-stream/src/protocol.rs | 412 ++++++++++++++++ engine/ds-stream/src/relay.rs | 255 ++++++++++ examples/stream-receiver.html | 676 ++++++++++++++++++++++++++ examples/stream-source.html | 783 +++++++++++++++++++++++++++++++ 10 files changed, 2511 insertions(+), 1 deletion(-) create mode 100644 engine/ds-stream/Cargo.toml create mode 100644 engine/ds-stream/src/codec.rs create mode 100644 engine/ds-stream/src/lib.rs create mode 100644 engine/ds-stream/src/main.rs create mode 100644 engine/ds-stream/src/protocol.rs create mode 100644 engine/ds-stream/src/relay.rs create mode 100644 examples/stream-receiver.html create mode 100644 examples/stream-source.html diff --git a/Cargo.toml b/Cargo.toml index bcda22f..52f05e0 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,6 +7,8 @@ members = [ "compiler/ds-layout", "compiler/ds-types", "compiler/ds-cli", + "engine/ds-physics", + "engine/ds-stream", ] [workspace.package] @@ -20,3 +22,5 @@ ds-analyzer = { path = "compiler/ds-analyzer" } ds-codegen = { path = "compiler/ds-codegen" } ds-layout = { path = "compiler/ds-layout" } ds-types = { path = "compiler/ds-types" } +ds-physics = { path = "engine/ds-physics" } +ds-stream = { path = "engine/ds-stream" } diff --git a/DREAMSTACK.md b/DREAMSTACK.md index b6303fc..92e2693 100644 --- a/DREAMSTACK.md +++ b/DREAMSTACK.md @@ -34,6 +34,10 @@ DreamStack is **real and running** — 6 Rust crates, 34 tests, 8 examples, ~7KB | Types | `Signal`, `Derived` | ✅ Hindley-Milner | | Dev server | `dreamstack dev app.ds` | ✅ HMR | | Router | `route "/path" -> body` / `navigate` | ✅ Hash-based | +| Two-way binding | `input { bind: name }` | ✅ Signal ↔ input | +| Async resources | `DS.resource()` / `DS.fetchJSON()` | ✅ Loading/Ok/Err | +| Springs | `let x = spring(200)` | ✅ RK4 physics | +| Constraints | `constrain el.width = expr` | ✅ Reactive solver | ### DreamStack vs React @@ -46,6 +50,7 @@ DreamStack is **real and running** — 6 Rust crates, 34 tests, 8 examples, ~7KB | Conditional | `when x -> text "y"` | `{x && y}` | | Lists | `for item in items -> ...` | `{items.map(i => ...)}` | | Router | `route "/path" -> body` | `react-router` (external) | +| Forms | `input { bind: name }` | `useState` + `onChange` (manual) | | Animation | Built-in springs | framer-motion (external) | | Layout | Built-in Cassowary | CSS only | | Types | Native HM, `Signal` | TypeScript (external) | @@ -64,7 +69,7 @@ DreamStack is **real and running** — 6 Rust crates, 34 tests, 8 examples, ~7KB ### Examples -`counter.ds` · `list.ds` · `router.ds` · `todomvc.html` · `search.html` · `dashboard.html` · `playground.html` · `showcase.html` · `benchmarks.html` +`counter.ds` · `list.ds` · `router.ds` · `form.ds` · `springs.ds` · `todomvc.html` · `search.html` · `dashboard.html` · `playground.html` · `showcase.html` · `benchmarks.html` --- @@ -403,3 +408,57 @@ Nobody has unified them. That's the opportunity. 4. **Layout and animation are not afterthoughts.** They're core primitives, not CSS bolt-ons or third-party libraries. 5. **The editor and the runtime are the same thing.** Bidirectional editing collapses the design-develop gap entirely. 6. **UI is data, all the way down.** If you can't `map` over your UI structure, your abstraction is wrong. +7. **Any input bitstream → any output bitstream.** The UI is just one codec. Tomorrow's neural nets generate the pixels directly. + +--- + +## Phase 7: Universal Bitstream Streaming + +> *Stream the whole UI as bytes. Neural nets will generate the pixels, acoustics, and actuator commands.* + +DreamStack's `engine/ds-stream` crate implements a universal binary protocol for streaming any I/O: + +``` +┌──────────┐ WebSocket / WebRTC ┌──────────┐ +│ Source │ ──────frames (bytes)──► │ Receiver │ +│ (renders) │ ◄──────inputs (bytes)── │ (~250 LOC)│ +└──────────┘ └──────────┘ +``` + +### Binary Protocol (16-byte header) + +| Field | Size | Description | +|-------|------|-------------| +| type | u8 | Frame/input type (pixels, audio, haptic, neural, BCI) | +| flags | u8 | Input flag, keyframe flag, compression flag | +| seq | u16 | Sequence number | +| timestamp | u32 | Relative ms since stream start | +| width | u16 | Frame width or channel count | +| height | u16 | Frame height or sample rate | +| length | u32 | Payload length | + +### Output Types +- `Pixels` (0x01) — raw RGBA framebuffer +- `Audio` (0x10) — PCM audio samples +- `Haptic` (0x20) — vibration/actuator commands +- `NeuralFrame` (0x40) — neural-generated pixels *(future)* +- `NeuralAudio` (0x41) — neural speech/music synthesis *(future)* +- `NeuralActuator` (0x42) — learned motor control *(future)* + +### Input Types +- `Pointer` (0x01) — mouse/touch position + buttons +- `Key` (0x10) — keyboard events +- `Gamepad` (0x30) — controller axes + buttons +- `BciInput` (0x90) — brain-computer interface *(future)* + +### Demos +- `examples/stream-source.html` — Springs demo captures canvas → streams pixels at 30fps +- `examples/stream-receiver.html` — Thin client (~250 lines, no framework) renders bytes + +### Run It +```bash +cargo run -p ds-stream # start relay on :9100 +open examples/stream-source.html # source: renders + streams +open examples/stream-receiver.html # receiver: displays bytes +``` + diff --git a/engine/ds-stream/Cargo.toml b/engine/ds-stream/Cargo.toml new file mode 100644 index 0000000..0a4269b --- /dev/null +++ b/engine/ds-stream/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "ds-stream" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "Universal bitstream streaming — any input to any output" + +[[bin]] +name = "ds-stream-relay" +path = "src/main.rs" + +[lib] +path = "src/lib.rs" + +[dependencies] +tokio = { version = "1", features = ["full"] } +tokio-tungstenite = "0.24" +futures-util = "0.3" + +[dev-dependencies] diff --git a/engine/ds-stream/src/codec.rs b/engine/ds-stream/src/codec.rs new file mode 100644 index 0000000..c692b72 --- /dev/null +++ b/engine/ds-stream/src/codec.rs @@ -0,0 +1,247 @@ +//! Frame and Input Codec — encode/decode complete messages. +//! +//! A message = header (16 bytes) + payload (variable length). + +use crate::protocol::*; + +// ─── Frame Encoder ─── + +/// Encode a complete frame message: header + payload. +pub fn encode_frame( + frame_type: FrameType, + seq: u16, + timestamp: u32, + width: u16, + height: u16, + flags: u8, + payload: &[u8], +) -> Vec { + let header = FrameHeader { + frame_type: frame_type as u8, + flags, + seq, + timestamp, + width, + height, + length: payload.len() as u32, + }; + let mut buf = Vec::with_capacity(HEADER_SIZE + payload.len()); + buf.extend_from_slice(&header.encode()); + buf.extend_from_slice(payload); + buf +} + +/// Encode an input event message: header + payload, with FLAG_INPUT set. +pub fn encode_input( + input_type: InputType, + seq: u16, + timestamp: u32, + payload: &[u8], +) -> Vec { + let header = FrameHeader { + frame_type: input_type as u8, + flags: FLAG_INPUT, + seq, + timestamp, + width: 0, + height: 0, + length: payload.len() as u32, + }; + let mut buf = Vec::with_capacity(HEADER_SIZE + payload.len()); + buf.extend_from_slice(&header.encode()); + buf.extend_from_slice(payload); + buf +} + +// ─── Frame Decoder ─── + +/// Decoded message: header + payload reference. +#[derive(Debug)] +pub struct DecodedMessage<'a> { + pub header: FrameHeader, + pub payload: &'a [u8], +} + +/// Decode a complete message from bytes. +/// Returns None if the buffer is too short for the header or payload. +pub fn decode_message(buf: &[u8]) -> Option> { + let header = FrameHeader::decode(buf)?; + let payload_start = HEADER_SIZE; + let payload_end = payload_start + header.length as usize; + if buf.len() < payload_end { + return None; + } + Some(DecodedMessage { + header, + payload: &buf[payload_start..payload_end], + }) +} + +/// Total message size from a header. +pub fn message_size(header: &FrameHeader) -> usize { + HEADER_SIZE + header.length as usize +} + +// ─── Delta Compression (stub for future neural compression) ─── + +/// Compute XOR delta between two frames. +/// Returns a delta buffer where unchanged pixels are zero bytes. +/// This is the simplest possible delta — a neural compressor would replace this. +pub fn compute_delta(current: &[u8], previous: &[u8]) -> Vec { + assert_eq!(current.len(), previous.len(), "frames must be same size"); + current.iter().zip(previous.iter()).map(|(c, p)| c ^ p).collect() +} + +/// Apply XOR delta to reconstruct current frame from previous + delta. +pub fn apply_delta(previous: &[u8], delta: &[u8]) -> Vec { + assert_eq!(previous.len(), delta.len(), "frames must be same size"); + previous.iter().zip(delta.iter()).map(|(p, d)| p ^ d).collect() +} + +/// Check if a delta frame is worth sending (vs sending a keyframe). +/// Returns true if the delta is significantly smaller due to zero runs. +pub fn delta_is_worthwhile(delta: &[u8]) -> bool { + let zero_count = delta.iter().filter(|&&b| b == 0).count(); + // If more than 30% of bytes are zero, delta is worthwhile + zero_count > delta.len() * 3 / 10 +} + +// ─── Convenience Builders ─── + +/// Build a pixel frame message from raw RGBA data. +pub fn pixel_frame( + seq: u16, + timestamp: u32, + width: u16, + height: u16, + rgba_data: &[u8], +) -> Vec { + encode_frame( + FrameType::Pixels, + seq, + timestamp, + width, + height, + FLAG_KEYFRAME, + rgba_data, + ) +} + +/// Build a pointer input message. +pub fn pointer_input(seq: u16, timestamp: u32, event: &PointerEvent, input_type: InputType) -> Vec { + let header = FrameHeader { + frame_type: input_type as u8, + flags: FLAG_INPUT, + seq, + timestamp, + width: 0, + height: 0, + length: PointerEvent::SIZE as u32, + }; + let mut buf = Vec::with_capacity(HEADER_SIZE + PointerEvent::SIZE); + buf.extend_from_slice(&header.encode()); + buf.extend_from_slice(&event.encode()); + buf +} + +/// Build a ping/heartbeat message. +pub fn ping(seq: u16, timestamp: u32) -> Vec { + encode_frame(FrameType::Ping, seq, timestamp, 0, 0, 0, &[]) +} + +// ─── Tests ─── + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn frame_encode_decode_roundtrip() { + let payload = vec![0xFF; 100]; + let msg = encode_frame( + FrameType::Pixels, + 1, + 5000, + 320, + 240, + FLAG_KEYFRAME, + &payload, + ); + let decoded = decode_message(&msg).unwrap(); + assert_eq!(decoded.header.frame_type, FrameType::Pixels as u8); + assert_eq!(decoded.header.seq, 1); + assert_eq!(decoded.header.timestamp, 5000); + assert_eq!(decoded.header.width, 320); + assert_eq!(decoded.header.height, 240); + assert_eq!(decoded.header.length, 100); + assert!(decoded.header.is_keyframe()); + assert_eq!(decoded.payload, &payload[..]); + } + + #[test] + fn input_encode_decode_roundtrip() { + let ptr = PointerEvent { x: 100, y: 200, buttons: 1 }; + let msg = pointer_input(5, 1234, &ptr, InputType::Pointer); + let decoded = decode_message(&msg).unwrap(); + assert!(decoded.header.is_input()); + assert_eq!(decoded.header.frame_type, InputType::Pointer as u8); + let ptr2 = PointerEvent::decode(decoded.payload).unwrap(); + assert_eq!(ptr2.x, 100); + assert_eq!(ptr2.y, 200); + } + + #[test] + fn delta_compression() { + let frame1 = vec![10, 20, 30, 40, 50, 60, 70, 80]; + let frame2 = vec![10, 20, 30, 40, 55, 65, 70, 80]; // 2 pixels changed + let delta = compute_delta(&frame2, &frame1); + // Unchanged bytes should be 0 + assert_eq!(delta[0], 0); + assert_eq!(delta[1], 0); + assert_eq!(delta[6], 0); + assert_eq!(delta[7], 0); + // Reconstruct + let reconstructed = apply_delta(&frame1, &delta); + assert_eq!(reconstructed, frame2); + } + + #[test] + fn delta_worthwhile_check() { + // All zeros = definitely worthwhile (100% unchanged) + let all_zero = vec![0u8; 100]; + assert!(delta_is_worthwhile(&all_zero)); + // All changed = not worthwhile + let all_changed = vec![0xFF; 100]; + assert!(!delta_is_worthwhile(&all_changed)); + } + + #[test] + fn ping_message() { + let msg = ping(99, 5000); + let decoded = decode_message(&msg).unwrap(); + assert_eq!(decoded.header.frame_type, FrameType::Ping as u8); + assert_eq!(decoded.header.seq, 99); + assert_eq!(decoded.payload.len(), 0); + } + + #[test] + fn partial_buffer_returns_none() { + let msg = encode_frame(FrameType::Pixels, 0, 0, 10, 10, 0, &[1, 2, 3]); + // Truncate: header says 3 bytes payload but we only give 2 + assert!(decode_message(&msg[..HEADER_SIZE + 2]).is_none()); + } + + #[test] + fn message_size_calculation() { + let header = FrameHeader { + frame_type: 0, + flags: 0, + seq: 0, + timestamp: 0, + width: 0, + height: 0, + length: 1024, + }; + assert_eq!(message_size(&header), HEADER_SIZE + 1024); + } +} diff --git a/engine/ds-stream/src/lib.rs b/engine/ds-stream/src/lib.rs new file mode 100644 index 0000000..9ff87e0 --- /dev/null +++ b/engine/ds-stream/src/lib.rs @@ -0,0 +1,26 @@ +//! DreamStack Universal Bitstream +//! +//! Any input bitstream → any output bitstream. +//! Neural nets generate the pixels. The receiver just renders bytes. +//! +//! # Architecture +//! +//! ```text +//! ┌──────────┐ WebSocket ┌──────────┐ +//! │ Source │ ─────frames──► │ Receiver │ +//! │ (server) │ ◄────inputs─── │ (client) │ +//! └──────────┘ └──────────┘ +//! ``` +//! +//! The **source** runs the DreamStack signal graph, springs, and renderer. +//! It captures frames as bytes and streams them to connected receivers. +//! +//! The **receiver** is a thin client (~200 lines) that renders incoming +//! pixel frames and sends input events back to the source. +//! +//! The **relay** is a WebSocket server that routes frames and inputs +//! between source and receivers. + +pub mod protocol; +pub mod codec; +pub mod relay; diff --git a/engine/ds-stream/src/main.rs b/engine/ds-stream/src/main.rs new file mode 100644 index 0000000..67338e4 --- /dev/null +++ b/engine/ds-stream/src/main.rs @@ -0,0 +1,28 @@ +//! DreamStack Bitstream Relay Server +//! +//! Usage: `cargo run -p ds-stream` +//! +//! Starts a WebSocket relay on port 9100. +//! - Source connects to ws://localhost:9100/source +//! - Receivers connect to ws://localhost:9100/stream + +use ds_stream::relay::{run_relay, RelayConfig}; + +#[tokio::main] +async fn main() { + let port = std::env::args() + .nth(1) + .and_then(|s| s.parse::().ok()) + .unwrap_or(9100); + + let config = RelayConfig { + addr: format!("0.0.0.0:{}", port).parse().unwrap(), + ..Default::default() + }; + + eprintln!("Starting DreamStack Bitstream Relay on port {}...", port); + if let Err(e) = run_relay(config).await { + eprintln!("Relay error: {}", e); + std::process::exit(1); + } +} diff --git a/engine/ds-stream/src/protocol.rs b/engine/ds-stream/src/protocol.rs new file mode 100644 index 0000000..5f2d3d4 --- /dev/null +++ b/engine/ds-stream/src/protocol.rs @@ -0,0 +1,412 @@ +//! Universal Bitstream Protocol +//! +//! Binary wire format for streaming any I/O between source and receiver. +//! Designed to be the same protocol a neural renderer would emit. +//! +//! ## Header Format (16 bytes) +//! +//! ```text +//! ┌──────┬───────┬──────┬───────────┬───────┬────────┬────────┐ +//! │ type │ flags │ seq │ timestamp │ width │ height │ length │ +//! │ u8 │ u8 │ u16 │ u32 │ u16 │ u16 │ u32 │ +//! └──────┴───────┴──────┴───────────┴───────┴────────┴────────┘ +//! ``` + +/// Frame header size in bytes. +pub const HEADER_SIZE: usize = 16; + +/// Magic bytes for protocol identification. +pub const MAGIC: [u8; 2] = [0xD5, 0x7A]; // "DS" + "z" for stream + +// ─── Frame Types ─── + +/// Output frame types — what the source generates. +#[repr(u8)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FrameType { + /// Raw RGBA pixel data (width × height × 4 bytes) + Pixels = 0x01, + /// PNG/WebP compressed frame + CompressedPixels = 0x02, + /// Delta frame — XOR diff from previous keyframe + DeltaPixels = 0x03, + /// Audio PCM samples (f32, interleaved channels) + AudioPcm = 0x10, + /// Opus-compressed audio chunk + AudioCompressed = 0x11, + /// Haptic pulse command + Haptic = 0x20, + /// Actuator/motor command + Actuator = 0x21, + /// LED/display matrix data + LedMatrix = 0x22, + /// Signal graph state sync (DreamStack-native) + SignalSync = 0x30, + /// Signal diff — only changed signals + SignalDiff = 0x31, + + // ── Forward-thinking: Neural rendering ── + + /// Neural-generated frame (model output tensor as pixels) + NeuralFrame = 0x40, + /// Neural audio synthesis output + NeuralAudio = 0x41, + /// Neural actuator command (learned motor control) + NeuralActuator = 0x42, + /// Neural scene description (latent space representation) + NeuralLatent = 0x43, + + // ── Control ── + + /// Keyframe — receiver should reset state + Keyframe = 0xF0, + /// Heartbeat / keep-alive + Ping = 0xFE, + /// Stream end + End = 0xFF, +} + +impl FrameType { + pub fn from_u8(v: u8) -> Option { + match v { + 0x01 => Some(Self::Pixels), + 0x02 => Some(Self::CompressedPixels), + 0x03 => Some(Self::DeltaPixels), + 0x10 => Some(Self::AudioPcm), + 0x11 => Some(Self::AudioCompressed), + 0x20 => Some(Self::Haptic), + 0x21 => Some(Self::Actuator), + 0x22 => Some(Self::LedMatrix), + 0x30 => Some(Self::SignalSync), + 0x31 => Some(Self::SignalDiff), + 0x40 => Some(Self::NeuralFrame), + 0x41 => Some(Self::NeuralAudio), + 0x42 => Some(Self::NeuralActuator), + 0x43 => Some(Self::NeuralLatent), + 0xF0 => Some(Self::Keyframe), + 0xFE => Some(Self::Ping), + 0xFF => Some(Self::End), + _ => None, + } + } +} + +// ─── Input Types ─── + +/// Input event types — what the receiver sends back. +#[repr(u8)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum InputType { + /// Pointer/mouse: x(u16), y(u16), buttons(u8) + Pointer = 0x01, + /// Pointer down + PointerDown = 0x02, + /// Pointer up + PointerUp = 0x03, + /// Key down: keycode(u16), modifiers(u8) + KeyDown = 0x10, + /// Key up + KeyUp = 0x11, + /// Touch start/move: id(u8), x(u16), y(u16) + Touch = 0x20, + /// Touch end + TouchEnd = 0x21, + /// Gamepad axis: axis(u8), value(f32) + GamepadAxis = 0x30, + /// Gamepad button: button(u8), pressed(bool) + GamepadButton = 0x31, + /// MIDI message: status(u8), data1(u8), data2(u8) + Midi = 0x40, + /// Scroll/wheel: dx(i16), dy(i16) + Scroll = 0x50, + /// Resize: width(u16), height(u16) + Resize = 0x60, + + // ── Forward-thinking ── + + /// Voice/audio input chunk + VoiceInput = 0x70, + /// Camera frame from receiver + CameraInput = 0x71, + /// Sensor telemetry (accelerometer, gyro, etc.) + SensorInput = 0x80, + /// BCI/neural signal input + BciInput = 0x90, +} + +impl InputType { + pub fn from_u8(v: u8) -> Option { + match v { + 0x01 => Some(Self::Pointer), + 0x02 => Some(Self::PointerDown), + 0x03 => Some(Self::PointerUp), + 0x10 => Some(Self::KeyDown), + 0x11 => Some(Self::KeyUp), + 0x20 => Some(Self::Touch), + 0x21 => Some(Self::TouchEnd), + 0x30 => Some(Self::GamepadAxis), + 0x31 => Some(Self::GamepadButton), + 0x40 => Some(Self::Midi), + 0x50 => Some(Self::Scroll), + 0x60 => Some(Self::Resize), + 0x70 => Some(Self::VoiceInput), + 0x71 => Some(Self::CameraInput), + 0x80 => Some(Self::SensorInput), + 0x90 => Some(Self::BciInput), + _ => None, + } + } +} + +// ─── Frame Header ─── + +/// 16-byte frame header for all bitstream messages. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct FrameHeader { + /// Frame type (FrameType or InputType discriminant) + pub frame_type: u8, + /// Flags: bit 0 = is_input (1) vs is_output (0), bits 1-7 reserved + pub flags: u8, + /// Sequence number (wraps at u16::MAX) + pub seq: u16, + /// Timestamp in milliseconds (relative to stream start) + pub timestamp: u32, + /// Width (for pixel frames) or channel count (for audio) + pub width: u16, + /// Height (for pixel frames) or sample rate / 100 (for audio) + pub height: u16, + /// Payload length in bytes + pub length: u32, +} + +/// Flag: this message carries input (receiver → source) +pub const FLAG_INPUT: u8 = 0x01; +/// Flag: this is a keyframe (full state, no delta) +pub const FLAG_KEYFRAME: u8 = 0x02; +/// Flag: payload is compressed +pub const FLAG_COMPRESSED: u8 = 0x04; + +impl FrameHeader { + /// Encode header to 16 bytes (little-endian). + pub fn encode(&self) -> [u8; HEADER_SIZE] { + let mut buf = [0u8; HEADER_SIZE]; + buf[0] = self.frame_type; + buf[1] = self.flags; + buf[2..4].copy_from_slice(&self.seq.to_le_bytes()); + buf[4..8].copy_from_slice(&self.timestamp.to_le_bytes()); + buf[8..10].copy_from_slice(&self.width.to_le_bytes()); + buf[10..12].copy_from_slice(&self.height.to_le_bytes()); + buf[12..16].copy_from_slice(&self.length.to_le_bytes()); + buf + } + + /// Decode header from 16 bytes (little-endian). + pub fn decode(buf: &[u8]) -> Option { + if buf.len() < HEADER_SIZE { + return None; + } + Some(Self { + frame_type: buf[0], + flags: buf[1], + seq: u16::from_le_bytes([buf[2], buf[3]]), + timestamp: u32::from_le_bytes([buf[4], buf[5], buf[6], buf[7]]), + width: u16::from_le_bytes([buf[8], buf[9]]), + height: u16::from_le_bytes([buf[10], buf[11]]), + length: u32::from_le_bytes([buf[12], buf[13], buf[14], buf[15]]), + }) + } + + /// Check if this is an input event (receiver → source). + pub fn is_input(&self) -> bool { + self.flags & FLAG_INPUT != 0 + } + + /// Check if this is a keyframe. + pub fn is_keyframe(&self) -> bool { + self.flags & FLAG_KEYFRAME != 0 + } +} + +// ─── Input Events ─── + +/// Compact binary input event (pointer). +#[derive(Debug, Clone, Copy)] +pub struct PointerEvent { + pub x: u16, + pub y: u16, + pub buttons: u8, +} + +impl PointerEvent { + pub const SIZE: usize = 5; + + pub fn encode(&self) -> [u8; Self::SIZE] { + let mut buf = [0u8; Self::SIZE]; + buf[0..2].copy_from_slice(&self.x.to_le_bytes()); + buf[2..4].copy_from_slice(&self.y.to_le_bytes()); + buf[4] = self.buttons; + buf + } + + pub fn decode(buf: &[u8]) -> Option { + if buf.len() < Self::SIZE { + return None; + } + Some(Self { + x: u16::from_le_bytes([buf[0], buf[1]]), + y: u16::from_le_bytes([buf[2], buf[3]]), + buttons: buf[4], + }) + } +} + +/// Compact binary input event (key). +#[derive(Debug, Clone, Copy)] +pub struct KeyEvent { + pub keycode: u16, + pub modifiers: u8, +} + +impl KeyEvent { + pub const SIZE: usize = 3; + + pub fn encode(&self) -> [u8; Self::SIZE] { + let mut buf = [0u8; Self::SIZE]; + buf[0..2].copy_from_slice(&self.keycode.to_le_bytes()); + buf[2] = self.modifiers; + buf + } + + pub fn decode(buf: &[u8]) -> Option { + if buf.len() < Self::SIZE { + return None; + } + Some(Self { + keycode: u16::from_le_bytes([buf[0], buf[1]]), + modifiers: buf[2], + }) + } +} + +/// Compact binary scroll event. +#[derive(Debug, Clone, Copy)] +pub struct ScrollEvent { + pub dx: i16, + pub dy: i16, +} + +impl ScrollEvent { + pub const SIZE: usize = 4; + + pub fn encode(&self) -> [u8; Self::SIZE] { + let mut buf = [0u8; Self::SIZE]; + buf[0..2].copy_from_slice(&self.dx.to_le_bytes()); + buf[2..4].copy_from_slice(&self.dy.to_le_bytes()); + buf + } + + pub fn decode(buf: &[u8]) -> Option { + if buf.len() < Self::SIZE { + return None; + } + Some(Self { + dx: i16::from_le_bytes([buf[0], buf[1]]), + dy: i16::from_le_bytes([buf[2], buf[3]]), + }) + } +} + +// ─── Tests ─── + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn header_roundtrip() { + let header = FrameHeader { + frame_type: FrameType::Pixels as u8, + flags: FLAG_KEYFRAME, + seq: 42, + timestamp: 1234567, + width: 1920, + height: 1080, + length: 1920 * 1080 * 4, + }; + let encoded = header.encode(); + let decoded = FrameHeader::decode(&encoded).unwrap(); + assert_eq!(header, decoded); + } + + #[test] + fn header_too_short() { + assert!(FrameHeader::decode(&[0u8; 15]).is_none()); + } + + #[test] + fn pointer_event_roundtrip() { + let evt = PointerEvent { x: 500, y: 300, buttons: 1 }; + let encoded = evt.encode(); + let decoded = PointerEvent::decode(&encoded).unwrap(); + assert_eq!(evt.x, decoded.x); + assert_eq!(evt.y, decoded.y); + assert_eq!(evt.buttons, decoded.buttons); + } + + #[test] + fn key_event_roundtrip() { + let evt = KeyEvent { keycode: 0x0041, modifiers: 0x03 }; + let encoded = evt.encode(); + let decoded = KeyEvent::decode(&encoded).unwrap(); + assert_eq!(evt.keycode, decoded.keycode); + assert_eq!(evt.modifiers, decoded.modifiers); + } + + #[test] + fn scroll_event_roundtrip() { + let evt = ScrollEvent { dx: -120, dy: 360 }; + let encoded = evt.encode(); + let decoded = ScrollEvent::decode(&encoded).unwrap(); + assert_eq!(evt.dx, decoded.dx); + assert_eq!(evt.dy, decoded.dy); + } + + #[test] + fn frame_type_roundtrip() { + for val in [0x01, 0x02, 0x03, 0x10, 0x11, 0x20, 0x30, 0x31, 0x40, 0x41, 0xF0, 0xFE, 0xFF] { + let ft = FrameType::from_u8(val); + assert!(ft.is_some(), "FrameType::from_u8({:#x}) should be Some", val); + assert_eq!(ft.unwrap() as u8, val); + } + assert!(FrameType::from_u8(0x99).is_none()); + } + + #[test] + fn input_type_roundtrip() { + for val in [0x01, 0x02, 0x03, 0x10, 0x11, 0x20, 0x21, 0x30, 0x31, 0x40, 0x50, 0x60, 0x70, 0x80, 0x90] { + let it = InputType::from_u8(val); + assert!(it.is_some(), "InputType::from_u8({:#x}) should be Some", val); + assert_eq!(it.unwrap() as u8, val); + } + assert!(InputType::from_u8(0xAA).is_none()); + } + + #[test] + fn header_flags() { + let h = FrameHeader { + frame_type: FrameType::Pixels as u8, + flags: FLAG_INPUT | FLAG_KEYFRAME, + seq: 0, + timestamp: 0, + width: 0, + height: 0, + length: 0, + }; + assert!(h.is_input()); + assert!(h.is_keyframe()); + + let h2 = FrameHeader { flags: 0, ..h }; + assert!(!h2.is_input()); + assert!(!h2.is_keyframe()); + } +} diff --git a/engine/ds-stream/src/relay.rs b/engine/ds-stream/src/relay.rs new file mode 100644 index 0000000..7514604 --- /dev/null +++ b/engine/ds-stream/src/relay.rs @@ -0,0 +1,255 @@ +//! WebSocket Relay Server +//! +//! Routes frames from source→receivers and inputs from receivers→source. +//! Roles are determined by the connection path: +//! - `/source` — the frame producer (DreamStack renderer) +//! - `/stream` — frame consumers (thin receivers) +//! +//! The relay is intentionally dumb — it just forwards bytes. +//! The protocol semantics live in the source and receiver. + + +use std::net::SocketAddr; +use std::sync::Arc; + +use futures_util::{SinkExt, StreamExt}; +use tokio::net::{TcpListener, TcpStream}; +use tokio::sync::{broadcast, mpsc, RwLock}; +use tokio_tungstenite::tungstenite::Message; + +/// Relay server configuration. +pub struct RelayConfig { + /// Address to bind to. + pub addr: SocketAddr, + /// Maximum number of receivers. + pub max_receivers: usize, + /// Frame broadcast channel capacity. + pub frame_buffer_size: usize, +} + +impl Default for RelayConfig { + fn default() -> Self { + Self { + addr: "0.0.0.0:9100".parse().unwrap(), + max_receivers: 64, + frame_buffer_size: 16, + } + } +} + +/// Stats tracked by the relay. +#[derive(Debug, Default, Clone)] +pub struct RelayStats { + pub frames_relayed: u64, + pub bytes_relayed: u64, + pub inputs_relayed: u64, + pub connected_receivers: usize, + pub source_connected: bool, +} + +/// Shared relay state. +struct RelayState { + /// Broadcast channel: source → all receivers (frames) + frame_tx: broadcast::Sender>, + /// Channel: receivers → source (input events) + input_tx: mpsc::Sender>, + input_rx: Option>>, + /// Live stats + stats: RelayStats, +} + +/// Run the WebSocket relay server. +pub async fn run_relay(config: RelayConfig) -> Result<(), Box> { + let listener = TcpListener::bind(&config.addr).await?; + eprintln!("╔══════════════════════════════════════════════════╗"); + eprintln!("║ DreamStack Bitstream Relay v0.1.0 ║"); + eprintln!("║ ║"); + eprintln!("║ Source: ws://{}/source ║", config.addr); + eprintln!("║ Receiver: ws://{}/stream ║", config.addr); + eprintln!("╚══════════════════════════════════════════════════╝"); + + let (frame_tx, _) = broadcast::channel(config.frame_buffer_size); + let (input_tx, input_rx) = mpsc::channel(256); + + let state = Arc::new(RwLock::new(RelayState { + frame_tx, + input_tx, + input_rx: Some(input_rx), + stats: RelayStats::default(), + })); + + while let Ok((stream, addr)) = listener.accept().await { + let state = state.clone(); + tokio::spawn(handle_connection(stream, addr, state)); + } + Ok(()) +} + +async fn handle_connection( + stream: TcpStream, + addr: SocketAddr, + state: Arc>, +) { + // Peek at the HTTP upgrade request to determine role + let ws_stream = match tokio_tungstenite::accept_hdr_async( + stream, + |_req: &tokio_tungstenite::tungstenite::handshake::server::Request, + res: tokio_tungstenite::tungstenite::handshake::server::Response| { + // We'll extract the path from the URI later via a different mechanism + Ok(res) + }, + ) + .await + { + Ok(ws) => ws, + Err(e) => { + eprintln!("[relay] WebSocket handshake failed from {}: {}", addr, e); + return; + } + }; + + // For simplicity in the PoC, first connection = source, subsequent = receivers. + // A production version would parse the URI path. + let is_source = { + let s = state.read().await; + !s.stats.source_connected + }; + + if is_source { + eprintln!("[relay] Source connected: {}", addr); + handle_source(ws_stream, addr, state).await; + } else { + eprintln!("[relay] Receiver connected: {}", addr); + handle_receiver(ws_stream, addr, state).await; + } +} + +async fn handle_source( + ws_stream: tokio_tungstenite::WebSocketStream, + addr: SocketAddr, + state: Arc>, +) { + let (mut ws_sink, mut ws_source) = ws_stream.split(); + + // Mark source as connected and take the input_rx + let input_rx = { + let mut s = state.write().await; + s.stats.source_connected = true; + s.input_rx.take() + }; + + let frame_tx = { + let s = state.read().await; + s.frame_tx.clone() + }; + + // Forward input events from receivers → source + if let Some(mut input_rx) = input_rx { + let state_clone = state.clone(); + tokio::spawn(async move { + while let Some(input_bytes) = input_rx.recv().await { + let msg = Message::Binary(input_bytes.into()); + if ws_sink.send(msg).await.is_err() { + break; + } + let mut s = state_clone.write().await; + s.stats.inputs_relayed += 1; + } + }); + } + + // Receive frames from source → broadcast to receivers + while let Some(Ok(msg)) = ws_source.next().await { + if let Message::Binary(data) = msg { + let data_vec: Vec = data.into(); + { + let mut s = state.write().await; + s.stats.frames_relayed += 1; + s.stats.bytes_relayed += data_vec.len() as u64; + } + // Broadcast to all receivers (ignore send errors = no receivers) + let _ = frame_tx.send(data_vec); + } + } + + // Source disconnected + eprintln!("[relay] Source disconnected: {}", addr); + let mut s = state.write().await; + s.stats.source_connected = false; +} + +async fn handle_receiver( + ws_stream: tokio_tungstenite::WebSocketStream, + addr: SocketAddr, + state: Arc>, +) { + let (mut ws_sink, mut ws_source) = ws_stream.split(); + + // Subscribe to frame broadcast + let mut frame_rx = { + let mut s = state.write().await; + s.stats.connected_receivers += 1; + s.frame_tx.subscribe() + }; + + let input_tx = { + let s = state.read().await; + s.input_tx.clone() + }; + + // Forward frames from broadcast → this receiver + let _state_clone = state.clone(); + let addr_clone = addr; + let send_task = tokio::spawn(async move { + loop { + match frame_rx.recv().await { + Ok(frame_bytes) => { + let msg = Message::Binary(frame_bytes.into()); + if ws_sink.send(msg).await.is_err() { + break; + } + } + Err(broadcast::error::RecvError::Lagged(n)) => { + eprintln!("[relay] Receiver {} lagged by {} frames", addr_clone, n); + } + Err(_) => break, + } + } + }); + + // Forward input events from this receiver → source + while let Some(Ok(msg)) = ws_source.next().await { + if let Message::Binary(data) = msg { + let _ = input_tx.send(data.into()).await; + } + } + + // Receiver disconnected + send_task.abort(); + eprintln!("[relay] Receiver disconnected: {}", addr); + let mut s = state.write().await; + s.stats.connected_receivers = s.stats.connected_receivers.saturating_sub(1); +} + +// ─── Tests ─── + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_config() { + let config = RelayConfig::default(); + assert_eq!(config.addr, "0.0.0.0:9100".parse::().unwrap()); + assert_eq!(config.max_receivers, 64); + assert_eq!(config.frame_buffer_size, 16); + } + + #[test] + fn stats_default() { + let stats = RelayStats::default(); + assert_eq!(stats.frames_relayed, 0); + assert_eq!(stats.bytes_relayed, 0); + assert!(!stats.source_connected); + } +} diff --git a/examples/stream-receiver.html b/examples/stream-receiver.html new file mode 100644 index 0000000..83d398f --- /dev/null +++ b/examples/stream-receiver.html @@ -0,0 +1,676 @@ + + + + + + + DreamStack — Stream Receiver + + + + + +

📡 Bitstream Receiver

+

No framework. No runtime. Just bytes → pixels + audio + haptics.

+ +
+
+ Connecting to relay... +
+ +
+
+

Received Frame

+ +
+ +
+

Receiver Stats

+
+
+
FPS In
+
0
+
+
+
Latency
+
—
+
+
+
Mode
+
—
+
+
+
Bandwidth
+
0
+
+
+
Frames
+
0
+
+
+
Inputs
+
0
+
+
+
Delta Saved
+
—
+
+
+
Audio
+
—
+
+
+
Signals
+
—
+
+
+ +
+

Universal Bitstream Bus

+
+ pixels + ◄── + — + ● +
+
+ delta + ◄── + — + ● +
+
+ signals + ◄── + — + ● +
+
+ neural + ◄── + — + ● +
+
+ audio + ◄── + — + ● +
+
+ pointer + ──► + 0 B + ● +
+
+ keyboard + ──► + 0 B + ● +
+
+ scroll + ──► + 0 B + ● +
+
+ voice──►future +
+
+ camera──►future +
+
+ BCI──►future +
+
+
+
+ +
Built with DreamStack — zero-framework receiver
+
This client: ~300 lines · No framework · Just bytes
+ + + + + \ No newline at end of file diff --git a/examples/stream-source.html b/examples/stream-source.html new file mode 100644 index 0000000..f592c87 --- /dev/null +++ b/examples/stream-source.html @@ -0,0 +1,783 @@ + + + + + + + DreamStack — Stream Source + + + + + +

⚡ Bitstream Source

+

Renders the UI. Streams pixels/signals/audio. Receives inputs.

+ +
+
+ Connecting to relay... +
+ +
+
+

Rendered Scene

+ +
+
+ + + + + +
+
+ +
+

Stream Stats

+
+
+
FPS
+
0
+
+
+
Mode
+
pixel
+
+
+
Receivers
+
0
+
+
+
Frame Size
+
0
+
+
+
Bandwidth
+
0
+
+
+
Delta Ratio
+
—
+
+
+
Frames
+
0
+
+
+
Inputs
+
0
+
+
+
Audio
+
off
+
+
+
+
+
+
Built with DreamStack — universal bitstream
+ + + + + \ No newline at end of file