diff --git a/services/frontend-leptos/Cargo.lock b/services/frontend-leptos/Cargo.lock index 4112f50..e98e1ff 100644 --- a/services/frontend-leptos/Cargo.lock +++ b/services/frontend-leptos/Cargo.lock @@ -492,6 +492,7 @@ version = "0.1.0" dependencies = [ "console_error_panic_hook", "gloo-net", + "gloo-timers", "js-sys", "leptos 0.7.8", "leptos-use", diff --git a/services/frontend-leptos/frontend/Cargo.toml b/services/frontend-leptos/frontend/Cargo.toml index 79e5c63..dc44509 100644 --- a/services/frontend-leptos/frontend/Cargo.toml +++ b/services/frontend-leptos/frontend/Cargo.toml @@ -23,10 +23,16 @@ web-sys = { version = "0.3", features = [ "AudioContext", "AudioBuffer", "AudioBufferSourceNode", + "AudioDestinationNode", + "AudioNode", + "AudioProcessingEvent", + "MediaStreamAudioSourceNode", + "ScriptProcessorNode", "Window", "Document", "Element", "HtmlElement", + "HtmlSelectElement", "KeyboardEvent", "Storage", "IntersectionObserver", @@ -42,10 +48,13 @@ web-sys = { version = "0.3", features = [ "HtmlCanvasElement", "MediaDevices", "MediaStream", + "MediaStreamConstraints", + "MediaStreamTrack", "Navigator", "console", ] } gloo-net = "0.6" +gloo-timers = "0.3" serde = { version = "1", features = ["derive"] } serde_json = "1" serde-wasm-bindgen = "0.6" diff --git a/services/frontend-leptos/frontend/src/features/live/audio/mod.rs b/services/frontend-leptos/frontend/src/features/live/audio/mod.rs new file mode 100644 index 0000000..df0643c --- /dev/null +++ b/services/frontend-leptos/frontend/src/features/live/audio/mod.rs @@ -0,0 +1,2 @@ +pub mod ring_buffer; +pub mod pcm_decoder; diff --git a/services/frontend-leptos/frontend/src/features/live/audio/pcm_decoder.rs b/services/frontend-leptos/frontend/src/features/live/audio/pcm_decoder.rs new file mode 100644 index 0000000..c977ebc --- /dev/null +++ b/services/frontend-leptos/frontend/src/features/live/audio/pcm_decoder.rs @@ -0,0 +1,69 @@ +/// PCM Frame decoded from binary WebSocket data +/// Format: [u32 userId (4 bytes)][i16 samples (N bytes)] +pub struct PcmFrame { + pub user_id: u32, + pub samples: Vec, // Normalized to [-1.0, 1.0] +} + +/// Decode a binary WebSocket message into PCM frames +/// Returns None if data is too short or malformed +pub fn decode_pcm_frame(data: &[u8]) -> Option { + if data.len() < 4 { + return None; + } + + let user_id = u32::from_le_bytes([data[0], data[1], data[2], data[3]]); + let sample_bytes = &data[4..]; + let sample_count = sample_bytes.len() / 2; + + if sample_count == 0 { + return None; + } + + let samples = decode_i16_samples(sample_bytes); + Some(PcmFrame { user_id, samples }) +} + +/// Decode raw i16 PCM bytes to normalized f32 samples [-1.0, 1.0] +pub fn decode_i16_samples(data: &[u8]) -> Vec { + let count = data.len() / 2; + let mut out = Vec::with_capacity(count); + + for i in 0..count { + let offset = i * 2; + if offset + 1 < data.len() { + let sample = i16::from_le_bytes([data[offset], data[offset + 1]]); + out.push((sample as f32) / 32768.0); + } + } + + out +} + +/// Encode f32 samples [-1.0, 1.0] to base64 for WebSocket transmission +/// Uses JavaScript btoa for encoding +pub fn encode_samples_to_base64(samples: &[f32]) -> String { + // Convert f32 samples to i16 bytes + let mut bytes = Vec::with_capacity(samples.len() * 2); + for &sample in samples { + let clamped = sample.max(-1.0).min(1.0); + let int_sample = (clamped * 32767.0) as i16; + bytes.extend_from_slice(&int_sample.to_le_bytes()); + } + encode_bytes_base64(&bytes) +} + +/// Encode raw bytes to base64 using JavaScript's btoa +fn encode_bytes_base64(data: &[u8]) -> String { + // Build binary string for btoa + let binary: String = data.iter().map(|&b| b as char).collect(); + + // Call btoa from JavaScript via js_sys::eval + let js_code = format!("btoa('{}')", binary.replace('\'', "\\'")); + js_sys::eval(&js_code) + .ok() + .and_then(|r| r.as_string()) + .unwrap_or_default() +} + +use wasm_bindgen::prelude::*; diff --git a/services/frontend-leptos/frontend/src/features/live/audio/ring_buffer.rs b/services/frontend-leptos/frontend/src/features/live/audio/ring_buffer.rs new file mode 100644 index 0000000..a806b47 --- /dev/null +++ b/services/frontend-leptos/frontend/src/features/live/audio/ring_buffer.rs @@ -0,0 +1,124 @@ +use std::sync::{Arc, Mutex}; + +/// AudioRingBuffer — Fixed-size circular buffer for real-time PCM streaming +/// Provides thread-safe write/read with automatic overwrite protection +pub struct AudioRingBuffer { + buffer: Vec, + capacity: usize, + write_pos: usize, + read_pos: usize, + available: usize, +} + +impl AudioRingBuffer { + /// Create a new ring buffer with given capacity (in samples) + pub fn new(capacity: usize) -> Self { + Self { + buffer: vec![0.0; capacity], + capacity, + write_pos: 0, + read_pos: 0, + available: 0, + } + } + + /// Write samples to the ring buffer. Overwrites oldest data if full. + pub fn write(&mut self, samples: &[f32]) { + let mut written = 0; + while written < samples.len() { + let chunk = (samples.len() - written).min(self.capacity - self.write_pos); + let src = &samples[written..written + chunk]; + let dest = &mut self.buffer[self.write_pos..self.write_pos + chunk]; + dest.copy_from_slice(src); + written += chunk; + self.write_pos = (self.write_pos + chunk) % self.capacity; + self.available = (self.available + chunk).min(self.capacity); + // If we overwrote unread data, advance read_pos + if self.available == self.capacity { + self.read_pos = self.write_pos; + } + } + } + + /// Read up to `max_samples` from the buffer. Returns the samples read. + pub fn read(&mut self, max_samples: usize) -> Vec { + let to_read = max_samples.min(self.available); + let mut out = Vec::with_capacity(to_read); + let mut remaining = to_read; + + while remaining > 0 { + let chunk = remaining.min(self.capacity - self.read_pos); + out.extend_from_slice(&self.buffer[self.read_pos..self.read_pos + chunk]); + remaining -= chunk; + self.read_pos = (self.read_pos + chunk) % self.capacity; + } + + self.available -= to_read; + out + } + + /// Number of samples available to read + pub fn available_samples(&self) -> usize { + self.available + } + + /// Clear all buffered data + pub fn clear(&mut self) { + self.write_pos = 0; + self.read_pos = 0; + self.available = 0; + } +} + +/// Thread-safe wrapper around AudioRingBuffer +pub struct SharedRingBuffer { + inner: Arc>, +} + +impl SharedRingBuffer { + pub fn new(capacity: usize) -> Self { + Self { + inner: Arc::new(Mutex::new(AudioRingBuffer::new(capacity))), + } + } + + pub fn write(&self, samples: &[f32]) { + if let Ok(mut guard) = self.inner.lock() { + guard.write(samples); + } + } + + pub fn read(&self, max_samples: usize) -> Vec { + if let Ok(mut guard) = self.inner.lock() { + guard.read(max_samples) + } else { + Vec::new() + } + } + + pub fn available_samples(&self) -> usize { + if let Ok(guard) = self.inner.lock() { + guard.available_samples() + } else { + 0 + } + } + + pub fn clear(&self) { + if let Ok(mut guard) = self.inner.lock() { + guard.clear(); + } + } + + pub fn clone_inner(&self) -> Arc> { + self.inner.clone() + } +} + +impl Clone for SharedRingBuffer { + fn clone(&self) -> Self { + Self { + inner: self.inner.clone(), + } + } +} diff --git a/services/frontend-leptos/frontend/src/features/live/hooks/mod.rs b/services/frontend-leptos/frontend/src/features/live/hooks/mod.rs index 88d7ece..28dcb67 100644 --- a/services/frontend-leptos/frontend/src/features/live/hooks/mod.rs +++ b/services/frontend-leptos/frontend/src/features/live/hooks/mod.rs @@ -1,2 +1,4 @@ pub mod use_voice_control; pub mod use_media_control; +pub mod use_audio_playback; +pub mod use_audio_transmit; diff --git a/services/frontend-leptos/frontend/src/features/live/hooks/use_audio_playback.rs b/services/frontend-leptos/frontend/src/features/live/hooks/use_audio_playback.rs new file mode 100644 index 0000000..ffa7b92 --- /dev/null +++ b/services/frontend-leptos/frontend/src/features/live/hooks/use_audio_playback.rs @@ -0,0 +1,107 @@ +use leptos::prelude::*; +use std::sync::Arc; +use crate::features::live::audio::pcm_decoder::decode_pcm_frame; +use crate::features::live::audio::ring_buffer::SharedRingBuffer; + +/// AudioPlaybackState — Manages PCM audio playback from WebSocket binary frames +pub struct AudioPlaybackState { + /// Ring buffer for incoming PCM data + pub buffer: SharedRingBuffer, + /// Whether playback is active + pub active: RwSignal, + /// Volume level (0.0-1.0) + pub volume: RwSignal, +} + +/// Create and initialize audio playback state +pub fn use_audio_playback() -> AudioPlaybackState { + let buffer = SharedRingBuffer::new(44100 * 5); // 5 seconds at 44.1kHz + let active = create_rw_signal::(false); + let volume = create_rw_signal::(0.5); + + AudioPlaybackState { + buffer, + active, + volume, + } +} + +/// Process incoming binary data from WebSocket (PCM audio frame) +/// Format: [u32 userId (4 bytes)][i16 samples (N bytes)] +pub fn process_pcm_data(state: &AudioPlaybackState, data: Vec) { + if let Some(frame) = decode_pcm_frame(&data) { + state.buffer.write(&frame.samples); + } +} + +/// Start consuming the ring buffer and playing through AudioContext +pub fn start_playback(state: &AudioPlaybackState) { + if state.active.get() { + return; + } + state.active.set(true); + + let buffer = state.buffer.clone(); + let active = state.active; + + wasm_bindgen_futures::spawn_local(async move { + let ctx = match web_sys::AudioContext::new() { + Ok(ctx) => ctx, + Err(_) => { + active.set(false); + return; + } + }; + + let ctx_ref = &ctx; + let _ = ctx_ref.resume(); + + while active.get() { + let available = buffer.available_samples(); + if available >= 4410 { + // ~100ms worth at 44.1kHz + let samples = buffer.read(4410); + if !samples.is_empty() { + play_samples(&ctx, &samples); + } + } + let _ = gloo_timers::future::TimeoutFuture::new(50).await; + } + + let _ = ctx.close(); + }); +} + +/// Stop playback and clear buffer +pub fn stop_playback(state: &AudioPlaybackState) { + state.active.set(false); + state.buffer.clear(); +} + +/// Play a chunk of PCM samples through AudioContext using AudioBufferSourceNode +fn play_samples(ctx: &web_sys::AudioContext, samples: &[f32]) { + let frame_count = samples.len() as u32; + let Ok(audio_buffer) = ctx.create_buffer(1, frame_count, ctx.sample_rate()) else { + return; + }; + + // Write samples into the buffer channel + let Ok(channel_data) = audio_buffer.get_channel_data(0) else { + return; + }; + + let len = samples.len().min(channel_data.len() as usize); + if len == 0 { + return; + } + + // Copy samples directly to audio buffer channel + let _ = audio_buffer.copy_to_channel(&samples[..len], 0); + + // Create source and play + if let Ok(source) = ctx.create_buffer_source() { + source.set_buffer(Some(&audio_buffer)); + source.set_loop(false); + let _ = source.start(); + } +} diff --git a/services/frontend-leptos/frontend/src/features/live/hooks/use_audio_transmit.rs b/services/frontend-leptos/frontend/src/features/live/hooks/use_audio_transmit.rs new file mode 100644 index 0000000..b125f45 --- /dev/null +++ b/services/frontend-leptos/frontend/src/features/live/hooks/use_audio_transmit.rs @@ -0,0 +1,79 @@ +use leptos::prelude::*; +use wasm_bindgen::prelude::*; +use wasm_bindgen_futures::spawn_local; +use web_sys::{MediaStream, MediaStreamConstraints, MediaStreamTrack}; + +/// AudioTransmitState — Manages microphone capture state +pub struct AudioTransmitState { + pub active: RwSignal, + pub stream: StoredValue>, +} + +/// Create microphone transmit state +pub fn use_audio_transmit() -> AudioTransmitState { + let active = create_rw_signal::(false); + let stream = StoredValue::new(None::); + AudioTransmitState { active, stream } +} + +/// Start microphone capture - requests getUserMedia and stores the stream +pub fn start_transmit(state: &AudioTransmitState) { + if state.active.get() { + return; + } + state.active.set(true); + + let constraints = MediaStreamConstraints::new(); + let _ = js_sys::Reflect::set( + &constraints, + &JsValue::from_str("audio"), + &JsValue::from_bool(true), + ); + + let window = match web_sys::window() { + Some(w) => w, + None => return, + }; + let media_devices = match window.navigator().media_devices() { + Ok(md) => md, + Err(_) => return, + }; + + let promise = match media_devices.get_user_media_with_constraints(&constraints) { + Ok(p) => p, + Err(_) => return, + }; + + // Clone signals before spawning async task to avoid reference escaping + let active_signal = state.active; + let stream_signal = state.stream; + + spawn_local(async move { + match wasm_bindgen_futures::JsFuture::from(promise).await { + Ok(val) => { + if let Ok(s) = val.dyn_into::() { + stream_signal.set_value(Some(s)); + } + } + Err(_) => { + active_signal.set(false); + } + } + }); +} + +/// Stop microphone transmission +pub fn stop_transmit(state: &AudioTransmitState) { + state.active.set(false); + state.stream.update_value(|s| { + if let Some(stream) = s.take() { + let tracks = stream.get_tracks(); + for i in 0..tracks.length() { + let track_val = tracks.get(i); + if let Ok(track) = track_val.dyn_into::() { + track.stop(); + } + } + } + }); +} diff --git a/services/frontend-leptos/frontend/src/features/live/mod.rs b/services/frontend-leptos/frontend/src/features/live/mod.rs index 7f745eb..ac2d4d8 100644 --- a/services/frontend-leptos/frontend/src/features/live/mod.rs +++ b/services/frontend-leptos/frontend/src/features/live/mod.rs @@ -1,5 +1,6 @@ pub mod components; pub mod hooks; +pub mod audio; use leptos::prelude::*; use crate::ui::card::Card;