manawenuz/wz-phone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add real audio pipeline with Opus + RaptorQ FEC

Browse Source 
- AudioPipeline: Kotlin AudioRecord/AudioTrack on JVM threads, PCM
  shuttled to Rust via lock-free ring buffers + JNI
- FEC: RaptorQ fountain codes on encode (5 frames/block, 20% repair
  ratio for GOOD profile), decoder feeds repair symbols for recovery
- Real audio level meter from mic RMS (replaces fake animation)
- Room name editable in UI (default: "android")
- Relay changed to pangolin.manko.yoga:4433
- Stats overlay shows FEC recovered count
- CallState now synced from polled stats (fixes "Connecting" stuck bug)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Claude
2026-04-05 12:33:59 +00:00
parent 81c756c076
commit bf91cf25bd
15 changed files with 663 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
crates/wzp-android/src/audio_ring.rs Normal file
View File

@@ -0,0 +1,91 @@
//! Lock-free SPSC ring buffers for audio PCM transfer between
//! Kotlin AudioRecord/AudioTrack threads and the Rust engine.
//!
//! These use a simple spin-free design: the producer writes and advances
//! a write cursor, the consumer reads and advances a read cursor.
//! Both cursors are atomic so no mutex is needed.
use std::sync::atomic::{AtomicUsize, Ordering};
/// Ring buffer capacity in i16 samples.
/// 960 samples * 10 frames = ~200ms of audio at 48kHz mono.
const RING_CAPACITY: usize = 960 * 10;
/// Lock-free single-producer single-consumer ring buffer for i16 PCM samples.
pub struct AudioRing {
buf: Box<[i16; RING_CAPACITY]>,
write_pos: AtomicUsize,
read_pos: AtomicUsize,
}
// SAFETY: AudioRing is designed for SPSC — one thread writes, one reads.
// The atomics ensure visibility. The buffer itself is never accessed
// from the same index by both threads simultaneously because the
// producer only writes to positions between write_pos and read_pos,
// and the consumer only reads from positions between read_pos and write_pos.
unsafe impl Send for AudioRing {}
unsafe impl Sync for AudioRing {}
impl AudioRing {
pub fn new() -> Self {
Self {
buf: Box::new([0i16; RING_CAPACITY]),
write_pos: AtomicUsize::new(0),
read_pos: AtomicUsize::new(0),
}
}
/// Number of samples available to read.
pub fn available(&self) -> usize {
let w = self.write_pos.load(Ordering::Acquire);
let r = self.read_pos.load(Ordering::Acquire);
w.wrapping_sub(r)
}
/// Number of samples that can be written without overwriting.
pub fn free_space(&self) -> usize {
RING_CAPACITY - self.available()
}
/// Write samples into the ring. Returns number of samples written.
/// Drops oldest samples if the ring is full.
pub fn write(&self, samples: &[i16]) -> usize {
let w = self.write_pos.load(Ordering::Relaxed);
let count = samples.len().min(RING_CAPACITY);
for i in 0..count {
let idx = (w + i) % RING_CAPACITY;
// SAFETY: We're the only writer, and the reader won't read
// past read_pos which we haven't advanced past yet.
unsafe {
let ptr = self.buf.as_ptr() as *mut i16;
*ptr.add(idx) = samples[i];
}
}
self.write_pos.store(w.wrapping_add(count), Ordering::Release);
// If we overwrote unread data, advance read_pos
if self.available() > RING_CAPACITY {
let new_read = self.write_pos.load(Ordering::Relaxed).wrapping_sub(RING_CAPACITY);
self.read_pos.store(new_read, Ordering::Release);
}
count
}
/// Read samples from the ring into `out`. Returns number of samples read.
pub fn read(&self, out: &mut [i16]) -> usize {
let avail = self.available();
let count = out.len().min(avail);
let r = self.read_pos.load(Ordering::Relaxed);
for i in 0..count {
let idx = (r + i) % RING_CAPACITY;
out[i] = unsafe { *self.buf.as_ptr().add(idx) };
}
self.read_pos.store(r.wrapping_add(count), Ordering::Release);
count
}
}

280
crates/wzp-android/src/engine.rs
View File

@@ -4,6 +4,9 @@
//! static bionic stubs in the Rust std prebuilt rlibs. ALL work must happen
//! on the JNI calling thread or via the tokio current_thread runtime.
//! No std::thread::spawn or tokio multi_thread allowed.
//!
//! Audio capture and playout happen on Kotlin JVM threads via AudioRecord
//! and AudioTrack. PCM samples are transferred through lock-free ring buffers.
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, AtomicU16, AtomicU32, Ordering};
@@ -11,15 +14,23 @@ use std::sync::{Arc, Mutex};
use std::time::Instant;
use bytes::Bytes;
use tracing::{error, info};
use tracing::{error, info, warn};
use wzp_codec::opus_dec::OpusDecoder;
use wzp_codec::opus_enc::OpusEncoder;
use wzp_crypto::{KeyExchange, WarzoneKeyExchange};
use wzp_fec::{RaptorQFecDecoder, RaptorQFecEncoder};
use wzp_proto::{
CodecId, MediaHeader, MediaPacket, MediaTransport, QualityProfile, SignalMessage,
AudioDecoder, AudioEncoder, CodecId, FecDecoder, FecEncoder,
MediaHeader, MediaPacket, MediaTransport, QualityProfile, SignalMessage,
};
use crate::audio_ring::AudioRing;
use crate::commands::EngineCommand;
use crate::stats::{CallState, CallStats};
/// Opus frame size at 48kHz mono, 20ms = 960 samples.
const FRAME_SAMPLES: usize = 960;
/// Configuration to start a call.
pub struct CallStartConfig {
pub profile: QualityProfile,
@@ -41,16 +52,22 @@ impl Default for CallStartConfig {
}
}
struct EngineState {
running: AtomicBool,
muted: AtomicBool,
stats: Mutex<CallStats>,
command_tx: std::sync::mpsc::Sender<EngineCommand>,
command_rx: Mutex<Option<std::sync::mpsc::Receiver<EngineCommand>>>,
pub(crate) struct EngineState {
pub running: AtomicBool,
pub muted: AtomicBool,
pub stats: Mutex<CallStats>,
pub command_tx: std::sync::mpsc::Sender<EngineCommand>,
pub command_rx: Mutex<Option<std::sync::mpsc::Receiver<EngineCommand>>>,
/// Ring buffer: Kotlin AudioRecord → Rust encoder
pub capture_ring: AudioRing,
/// Ring buffer: Rust decoder → Kotlin AudioTrack
pub playout_ring: AudioRing,
/// Current audio level (RMS) for UI display, updated by capture path.
pub audio_level_rms: AtomicU32,
}
pub struct WzpEngine {
state: Arc<EngineState>,
pub(crate) state: Arc<EngineState>,
tokio_runtime: Option<tokio::runtime::Runtime>,
call_start: Option<Instant>,
}
@@ -64,6 +81,9 @@ impl WzpEngine {
stats: Mutex::new(CallStats::default()),
command_tx: tx,
command_rx: Mutex::new(Some(rx)),
capture_ring: AudioRing::new(),
playout_ring: AudioRing::new(),
audio_level_rms: AtomicU32::new(0),
});
Self {
state,
@@ -85,8 +105,6 @@ impl WzpEngine {
};
}
// Create single-threaded tokio runtime — NO thread spawning.
// On Android, pthread_create crashes due to static bionic stubs.
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
@@ -97,17 +115,16 @@ impl WzpEngine {
let room = config.room.clone();
let identity_seed = config.identity_seed;
let profile = config.profile;
let state = self.state.clone();
self.state.running.store(true, Ordering::Release);
self.call_start = Some(Instant::now());
// Run the entire call on the current thread's tokio runtime.
// This blocks the JNI thread until the call ends, so Kotlin
// must call startCall from a background coroutine.
let state_clone = state.clone();
runtime.block_on(async move {
if let Err(e) = run_call(relay_addr, &room, &identity_seed, state_clone).await {
if let Err(e) = run_call(relay_addr, &room, &identity_seed, profile, state_clone).await
{
error!("call failed: {e}");
}
});
@@ -135,19 +152,17 @@ impl WzpEngine {
self.state.muted.store(muted, Ordering::Relaxed);
}
pub fn set_speaker(&self, _enabled: bool) {
// TODO: route audio via AudioManager on Kotlin side
}
pub fn set_speaker(&self, _enabled: bool) {}
pub fn force_profile(&self, _profile: QualityProfile) {
// TODO: wire to pipeline when codec thread is re-enabled
}
pub fn force_profile(&self, _profile: QualityProfile) {}
pub fn get_stats(&self) -> CallStats {
let mut stats = self.state.stats.lock().unwrap().clone();
if let Some(start) = self.call_start {
stats.duration_secs = start.elapsed().as_secs_f64();
}
// Include current audio level
stats.audio_level = self.state.audio_level_rms.load(Ordering::Relaxed);
stats
}
@@ -155,6 +170,23 @@ impl WzpEngine {
self.state.running.load(Ordering::Acquire)
}
pub fn write_audio(&self, samples: &[i16]) -> usize {
if self.state.muted.load(Ordering::Relaxed) {
return samples.len();
}
// Compute RMS for audio level display
if !samples.is_empty() {
let sum_sq: f64 = samples.iter().map(|&s| (s as f64) * (s as f64)).sum();
let rms = (sum_sq / samples.len() as f64).sqrt() as u32;
self.state.audio_level_rms.store(rms, Ordering::Relaxed);
}
self.state.capture_ring.write(samples)
}
pub fn read_audio(&self, out: &mut [i16]) -> usize {
self.state.playout_ring.read(out)
}
pub fn destroy(mut self) {
self.stop_call();
}
@@ -166,22 +198,19 @@ impl Drop for WzpEngine {
}
}
/// Run the full call lifecycle: connect, handshake, send/recv media.
/// All async, no thread spawning.
/// Run the full call lifecycle: connect, handshake, send/recv media with Opus + FEC.
async fn run_call(
relay_addr: SocketAddr,
room: &str,
identity_seed: &[u8; 32],
profile: QualityProfile,
state: Arc<EngineState>,
) -> Result<(), anyhow::Error> {
// Install rustls crypto provider
let _ = rustls::crypto::ring::default_provider().install_default();
// Create QUIC endpoint
let bind_addr: SocketAddr = "0.0.0.0:0".parse().unwrap();
let endpoint = wzp_transport::create_endpoint(bind_addr, None)?;
// Connect to relay with room as SNI
let sni = if room.is_empty() { "android" } else { room };
info!(%relay_addr, sni, "connecting to relay...");
let client_cfg = wzp_transport::client_config();
@@ -236,58 +265,223 @@ async fn run_call(
stats.state = CallState::Active;
}
// Simple media loop: send silence, recv and count frames.
// No codec thread, no Oboe — just network I/O to verify connectivity.
// Audio pipeline will be added once native threading is resolved.
// Initialize Opus codec
let mut encoder =
OpusEncoder::new(profile).map_err(|e| anyhow::anyhow!("opus encoder init: {e}"))?;
let mut decoder =
OpusDecoder::new(profile).map_err(|e| anyhow::anyhow!("opus decoder init: {e}"))?;
// Initialize FEC encoder/decoder
let mut fec_enc = wzp_fec::create_encoder(&profile);
let mut fec_dec = wzp_fec::create_decoder(&profile);
info!(
fec_ratio = profile.fec_ratio,
frames_per_block = profile.frames_per_block,
"codec + FEC initialized (48kHz mono, 20ms frames, RaptorQ)"
);
let seq = AtomicU16::new(0);
let ts = AtomicU32::new(0);
let transport_recv = transport.clone();
// Pre-allocate buffers
let mut capture_buf = vec![0i16; FRAME_SAMPLES];
let mut encode_buf = vec![0u8; encoder.max_frame_bytes()];
let mut frame_in_block: u8 = 0;
let mut block_id: u8 = 0;
// Send task: capture ring → Opus encode → FEC → MediaPackets
let send_task = async {
let silence = vec![0u8; 20]; // minimal opus silence frame
info!("send task started (Opus + RaptorQ FEC)");
loop {
if !state.running.load(Ordering::Relaxed) {
break;
}
let avail = state.capture_ring.available();
if avail < FRAME_SAMPLES {
tokio::time::sleep(std::time::Duration::from_millis(5)).await;
continue;
}
let read = state.capture_ring.read(&mut capture_buf);
if read < FRAME_SAMPLES {
continue;
}
// Opus encode
let encoded_len = match encoder.encode(&capture_buf, &mut encode_buf) {
Ok(n) => n,
Err(e) => {
warn!("opus encode error: {e}");
continue;
}
};
let encoded = &encode_buf[..encoded_len];
// Build source packet
let s = seq.fetch_add(1, Ordering::Relaxed);
let t = ts.fetch_add(20, Ordering::Relaxed);
let packet = MediaPacket {
let t = ts.fetch_add(FRAME_SAMPLES as u32, Ordering::Relaxed);
let source_pkt = MediaPacket {
header: MediaHeader {
version: 0,
is_repair: false,
codec_id: CodecId::Opus24k,
codec_id: profile.codec,
has_quality_report: false,
fec_ratio_encoded: 0,
fec_ratio_encoded: MediaHeader::encode_fec_ratio(profile.fec_ratio),
seq: s,
timestamp: t,
fec_block: 0,
fec_symbol: 0,
fec_block: block_id,
fec_symbol: frame_in_block,
reserved: 0,
csrc_count: 0,
},
payload: Bytes::from(silence.clone()),
payload: Bytes::copy_from_slice(encoded),
quality_report: None,
};
if let Err(e) = transport.send_media(&packet).await {
// Send source packet
if let Err(e) = transport.send_media(&source_pkt).await {
error!("send error: {e}");
break;
}
// 20ms frame interval
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
// Feed encoded frame to FEC encoder
if let Err(e) = fec_enc.add_source_symbol(encoded) {
warn!("fec add_source error: {e}");
}
frame_in_block += 1;
// When block is full, generate repair packets
if frame_in_block >= profile.frames_per_block {
match fec_enc.generate_repair(profile.fec_ratio) {
Ok(repairs) => {
let repair_count = repairs.len();
for (sym_idx, repair_data) in repairs {
let rs = seq.fetch_add(1, Ordering::Relaxed);
let repair_pkt = MediaPacket {
header: MediaHeader {
version: 0,
is_repair: true,
codec_id: profile.codec,
has_quality_report: false,
fec_ratio_encoded: MediaHeader::encode_fec_ratio(
profile.fec_ratio,
),
seq: rs,
timestamp: t,
fec_block: block_id,
fec_symbol: sym_idx,
reserved: 0,
csrc_count: 0,
},
payload: Bytes::from(repair_data),
quality_report: None,
};
if let Err(e) = transport.send_media(&repair_pkt).await {
error!("send repair error: {e}");
break;
}
}
if repair_count > 0 && (block_id % 50 == 0 || block_id == 0) {
info!(
block_id,
repair_count,
fec_ratio = profile.fec_ratio,
"FEC block complete"
);
}
}
Err(e) => {
warn!("fec generate_repair error: {e}");
}
}
let _ = fec_enc.finalize_block();
block_id = block_id.wrapping_add(1);
frame_in_block = 0;
}
if s % 500 == 0 {
info!(seq = s, block_id, frame_in_block, "sending");
}
}
};
// Pre-allocate decode buffer
let mut decode_buf = vec![0i16; FRAME_SAMPLES];
// Recv task: MediaPackets → FEC decode → Opus decode → playout ring
let recv_task = async {
let mut frames_decoded: u64 = 0;
let mut fec_recovered: u64 = 0;
info!("recv task started (Opus + RaptorQ FEC)");
loop {
if !state.running.load(Ordering::Relaxed) {
break;
}
match transport_recv.recv_media().await {
Ok(Some(_pkt)) => {
frames_decoded += 1;
Ok(Some(pkt)) => {
let is_repair = pkt.header.is_repair;
let pkt_block = pkt.header.fec_block;
let pkt_symbol = pkt.header.fec_symbol;
// Feed every packet (source + repair) to FEC decoder
let _ = fec_dec.add_symbol(
pkt_block,
pkt_symbol,
is_repair,
&pkt.payload,
);
// Source packets: decode directly
if !is_repair {
match decoder.decode(&pkt.payload, &mut decode_buf) {
Ok(samples) => {
state.playout_ring.write(&decode_buf[..samples]);
frames_decoded += 1;
}
Err(e) => {
warn!("opus decode error: {e}");
if let Ok(samples) = decoder.decode_lost(&mut decode_buf) {
state.playout_ring.write(&decode_buf[..samples]);
}
}
}
}
// Try FEC recovery for this block
// (useful when source packets were lost but repair arrived)
if let Ok(Some(recovered_frames)) = fec_dec.try_decode(pkt_block) {
// FEC recovered the block — any previously missing frames
// are now available. In a full jitter buffer implementation,
// we'd insert recovered frames at the right position.
// For now, log recovery for telemetry.
fec_recovered += recovered_frames.len() as u64;
if fec_recovered % 50 == 1 {
info!(
fec_recovered,
block = pkt_block,
frames = recovered_frames.len(),
"FEC block recovered"
);
}
}
// Expire old blocks to prevent memory growth
if pkt_block > 3 {
fec_dec.expire_before(pkt_block.wrapping_sub(3));
}
if frames_decoded == 1 || frames_decoded % 500 == 0 {
info!(frames_decoded, fec_recovered, "recv stats");
}
let mut stats = state.stats.lock().unwrap();
stats.frames_decoded = frames_decoded;
stats.fec_recovered = fec_recovered;
}
Ok(None) => {
info!("relay disconnected");
@@ -301,7 +495,7 @@ async fn run_call(
}
};
// Update encoded frame count in send task
// Stats task
let stats_task = async {
loop {
if !state.running.load(Ordering::Relaxed) {

50
crates/wzp-android/src/jni_bridge.rs
View File

@@ -174,6 +174,56 @@ pub unsafe extern "system" fn Java_com_wzp_engine_WzpEngine_nativeForceProfile(
}));
}
/// Write captured PCM samples from Kotlin AudioRecord into the engine's capture ring.
/// pcm is a Java short[] array.
#[unsafe(no_mangle)]
pub unsafe extern "system" fn Java_com_wzp_engine_WzpEngine_nativeWriteAudio(
env: JNIEnv,
_class: JClass,
handle: jlong,
pcm: jni::objects::JShortArray,
) -> jint {
let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
let h = unsafe { handle_ref(handle) };
let len = env.get_array_length(&pcm).unwrap_or(0) as usize;
if len == 0 {
return 0;
}
let mut buf = vec![0i16; len];
// GetShortArrayRegion copies Java array into our buffer
if env.get_short_array_region(&pcm, 0, &mut buf).is_err() {
return 0;
}
h.engine.write_audio(&buf) as jint
}));
result.unwrap_or(0)
}
/// Read decoded PCM samples from the engine's playout ring for Kotlin AudioTrack.
/// pcm is a Java short[] array to fill. Returns number of samples actually read.
#[unsafe(no_mangle)]
pub unsafe extern "system" fn Java_com_wzp_engine_WzpEngine_nativeReadAudio(
env: JNIEnv,
_class: JClass,
handle: jlong,
pcm: jni::objects::JShortArray,
) -> jint {
let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {
let h = unsafe { handle_ref(handle) };
let len = env.get_array_length(&pcm).unwrap_or(0) as usize;
if len == 0 {
return 0;
}
let mut buf = vec![0i16; len];
let read = h.engine.read_audio(&mut buf);
if read > 0 {
let _ = env.set_short_array_region(&pcm, 0, &buf[..read]);
}
read as jint
}));
result.unwrap_or(0)
}
#[unsafe(no_mangle)]
pub unsafe extern "system" fn Java_com_wzp_engine_WzpEngine_nativeDestroy(
_env: JNIEnv,

1
crates/wzp-android/src/lib.rs
View File

@@ -10,6 +10,7 @@
//! allowing `cargo check` and unit tests on the host.
pub mod audio_android;
pub mod audio_ring;
pub mod commands;
pub mod engine;
pub mod pipeline;

26
crates/wzp-android/src/stats.rs
View File

@@ -1,21 +1,31 @@
//! Call statistics for the Android engine.
/// State of the call.
#[derive(Clone, Debug, Default, serde::Serialize, PartialEq, Eq)]
/// Serializes as integer for easy parsing on the Kotlin side:
/// 0=Idle, 1=Connecting, 2=Active, 3=Reconnecting, 4=Closed
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub enum CallState {
/// Engine is idle, no active call.
#[default]
Idle,
/// Establishing connection to the relay.
Connecting,
/// Call is active with audio flowing.
Active,
/// Temporarily lost connection, attempting to recover.
Reconnecting,
/// Call has ended.
Closed,
}
impl serde::Serialize for CallState {
fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
let n: u8 = match self {
CallState::Idle => 0,
CallState::Connecting => 1,
CallState::Active => 2,
CallState::Reconnecting => 3,
CallState::Closed => 4,
};
serializer.serialize_u8(n)
}
}
/// Aggregated call statistics, serializable for JNI bridge.
#[derive(Clone, Debug, Default, serde::Serialize)]
pub struct CallStats {
@@ -39,4 +49,8 @@ pub struct CallStats {
pub frames_decoded: u64,
/// Number of playout underruns (buffer empty when audio needed).
pub underruns: u64,
/// Frames recovered by FEC.
pub fec_recovered: u64,
/// Current mic audio level (RMS of i16 samples, 0-32767).
pub audio_level: u32,
}