edison/chanora
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(beta): wire voice in/out end-to-end with push-to-talk (v0.2.0-beta.1)

Browse Source 
Reaches the Internal Beta milestone of DEC-001's release sequence the
same day as Alpha. Adds voice capture and playback through the full
Flutter UI → FRB → Rust core → tsclientlib → server path.

Promotions from PoC:
  poc/audio-capture-playback-spike  →  crates/chanora_audio/

New product code:
  crates/chanora_audio/src/engine.rs — cpal capture and playback,
    audiopus Opus VoIP encoder (48 kHz mono 20 ms frames), tsclientlib
    AudioHandler for decode + jitter buffer + mix on playback,
    push-to-talk gate, graceful playback-only fallback when capture
    is unavailable.
  crates/chanora_protocol/src/adapter.rs — extended with
    voice_out_tx (clonable mpsc::Sender<OutPacket>) and
    take_voice_in() (one-shot mpsc::Receiver<InboundVoice>); main
    loop now interleaves outbound voice drain, event pumping, and
    control-request handling.
  crates/chanora_protocol/src/lib.rs — re-exports the few
    tsproto_packets types (OutAudio, OutPacket, InAudioBuf,
    AudioData, CodecType, Direction) that chanora_audio
    legitimately needs. Documented as the single deliberate
    cross-crate type re-export per SAD-067, justified by the
    performance cost of a parallel type hierarchy on the 20 ms
    voice frame.
  core/chanora_core/src/lib.rs — ChanoraSession::start_audio,
    set_ptt, audio_stats; disconnect now stops the engine first.
  crates/chanora_bridge/src/api.rs — startAudio, setPtt,
    audioStats commands and BridgeAudioStats DTO.
  apps/chanora_flutter/lib/main.dart — "Start audio" button +
    hold-to-talk PTT button with pressed/released visual state +
    live stats line (TX/RX/PTT). Stats polled every 500 ms.

ARB:
  Both en and zh-Hans gain startAudioAction, pttHoldToTalk,
  pttTransmitting, audioStatsLine. Banner updated to
  "Beta build — voice in/out wired; not production ready."

FRB config:
  flutter_rust_bridge.yaml gains local: true so codegen resolves
  the workspace member's library stem to "chanora_bridge" instead
  of falling back to "UNKNOWN".

Empirical verification (2026-05-14, against cn.teamspeak.app):
  cargo check + cargo test --workspace: all green.
  flutter analyze: 0 issues.
  flutter test: 4/4 passing including:
    - test/alpha_e2e_test.dart (regression: Alpha still works)
    - test/beta_e2e_test.dart (Beta: connect → startAudio →
      PTT cycle → disconnect against cn.teamspeak.app).
  Live smoke (cargo test alpha_smoke -- --ignored): 49 channels,
  37 clients retrieved.
  Capture stream open against the host PipeWire auto_null source
  refused (snd_pcm_hw_params); engine correctly logged the warning
  and continued in playback-only mode. TX=0 frames, RX=0 frames
  reflects the headless null-source environment; on a real mic
  host the encoder produces ~50 frames/second while PTT is held.

Honest Beta scope (NOT in this release):
  - AEC / AGC / NS / HPF DSP (DEC-007..010): AudioEffects exists
    as a struct but the filters are no-ops. Beta+ work.
  - Production-quality resampler: current code is linear
    interpolation. Beta+ work.
  - Identity persistence via chanora_storage: still ephemeral.
  - Push-to-Dart event stream: UI polls instead.
  - chanora_diagnostics tracing-layer wiring: still scaffold.
  - Mobile (Android) cdylib + UI: PoC-proven, not yet in product.
  - Reconnect / network-loss recovery for the voice path.

Docs updates:
  - docs/governance/product-decision-register.md bumped to v0.9.7
    (Beta-milestone change-history entry; no row changes).
  - docs/governance/poc-results-summary.md bumped to v0.6.0
    (RISK-PoC-005 updated with Beta progress).
This commit is contained in:
EdisonJwa
2026-05-14 22:43:57 +08:00
parent 53b176b722
commit 9790005c3e
28 changed files with 2056 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/chanora_audio/Cargo.toml
+15 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "chanora_audio"
description = "Chanora audio subsystem — capture, DSP (HPF/NS/AEC/AGC per DEC-007..010), Opus encode/decode, jitter buffer, mixer, playback. Crate selection: cpal for desktop and Android per DEC-011.1."
description = "Chanora audio subsystem — cpal-based capture/playback, audiopus encode, tsclientlib AudioHandler for decode + jitter buffer + mix. DEC-011, DEC-011.1."
version.workspace = true
edition.workspace = true
rust-version.workspace = true
@@ -10,5 +10,19 @@ repository.workspace = true
publish.workspace = true
[dependencies]
chanora_protocol = { path = "../chanora_protocol" }
thiserror.workspace = true
tracing.workspace = true
# Cross-platform audio I/O (DEC-011.1).
cpal = "0.16"
# Opus encoder. tsclientlib already pulls this; we depend explicitly so
# this crate can compile against it without going through tsclientlib.
audiopus = "0.3.0-rc.0"
# AudioHandler lives in the tsclientlib crate behind the `audio`
# feature. We import the crate just for the AudioHandler type; the
# Connection type stays inside chanora_protocol.
tsclientlib = { git = "https://github.com/ReSpeak/tsclientlib.git", rev = "04aa2491", default-features = false, features = ["default-tls", "audio"] }
tokio = { version = "1", features = ["sync", "rt", "macros", "time"] }
crates/chanora_audio/src/engine.rs
+533
View File
@@ -0,0 +1,533 @@
//! Audio engine — owns the cpal input/output streams, the Opus
//! encoder, and the tsclientlib `AudioHandler` for decode+mix.
//!
//! The engine is started after a protocol connection is established
//! and stopped before disconnect. It does not retry on device
//! change.
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::{Arc, Mutex};
use cpal::traits::{DeviceTrait, HostTrait, StreamTrait};
use cpal::{SampleFormat, SizedSample};
use tokio::sync::mpsc;
use tracing::{debug, error, info, warn};
use audiopus::coder::Encoder as OpusEncoder;
use audiopus::{Application as OpusApp, Channels as OpusChannels, SampleRate as OpusSampleRate};
use tsclientlib::audio::AudioHandler;
use chanora_protocol::{
AudioData, CodecType, InboundVoice, OutAudio, OutPacket,
};
use crate::AudioError;
/// Stable Chanora-side identifier for AudioHandler bookkeeping.
/// We only ever have one connection at a time (DEC-006), so this is
/// trivially unique.
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
pub struct SessionAudioId(pub u64);
/// Audio framing: 48 kHz mono, 20 ms = 960 samples per frame.
const SAMPLE_RATE: u32 = 48_000;
const FRAME_SAMPLES: usize = 48_000 / 50; // 960
const MAX_OPUS_FRAME: usize = 1275;
/// Engine configuration.
#[derive(Debug, Clone)]
pub struct AudioEngineConfig {
/// Input gain applied before encoding (1.0 = pass-through).
pub mic_gain: f32,
/// Initial PTT state. When false the encoder is bypassed and no
/// outbound packets are produced.
pub ptt_initial: bool,
}
impl Default for AudioEngineConfig {
fn default() -> Self {
Self {
mic_gain: 1.0,
ptt_initial: false,
}
}
}
/// Running audio engine. Drop = stop.
pub struct AudioEngine {
ptt: Arc<AtomicBool>,
frames_sent: Arc<AtomicU32>,
frames_received: Arc<AtomicU32>,
// Streams must be dropped to stop audio. Both are `!Send` because
// cpal's Stream isn't Send on some backends; we keep them in an
// Option wrapped by Mutex so stop() can move them out.
_input_stream: Mutex<Option<cpal::Stream>>,
_output_stream: Mutex<Option<cpal::Stream>>,
// Hand the inbound-voice forwarder task a shutdown signal.
shutdown_tx: Option<tokio::sync::oneshot::Sender<()>>,
/// True if the capture stream actually opened. If false (typical
/// in headless environments with null sources, or where the user
/// denied microphone permission), PTT becomes a no-op and
/// `frames_sent` stays at 0.
capture_active: bool,
}
// cpal::Stream is not Send. We keep the engine pinned to the thread
// it was constructed on — `chanora_core` spawns it inside a
// `tokio::task::spawn_blocking` so the streams stay on that worker.
// This `unsafe impl Send` is necessary because the outer Arc<AudioEngine>
// is stored in core's session and must move into a task. The streams
// themselves are only mutated through the Mutex and are dropped on
// the same thread that owns them.
//
// SAFETY: cpal's Stream is not Send because the underlying audio API
// callback thread may not be transferable. We never invoke methods on
// the streams from any thread but the owning one; we only ever *drop*
// them, which cpal documents as safe from any thread for ALSA and
// PipeWire (Linux backend used here). For Windows/macOS the contract
// may differ; production Beta+ work must revisit per-platform.
unsafe impl Send for AudioEngine {}
unsafe impl Sync for AudioEngine {}
impl AudioEngine {
/// Start the engine: open capture + playback streams, spawn the
/// inbound-voice forwarder, return a handle.
pub fn start(
cfg: AudioEngineConfig,
voice_out_tx: mpsc::Sender<OutPacket>,
mut voice_in_rx: mpsc::Receiver<InboundVoice>,
) -> Result<Self, AudioError> {
let host = cpal::default_host();
let in_dev = host
.default_input_device()
.ok_or(AudioError::NoInputDevice)?;
let out_dev = host
.default_output_device()
.ok_or(AudioError::NoOutputDevice)?;
info!(
target: "chanora_audio",
in_device = %in_dev.name().unwrap_or_default(),
out_device = %out_dev.name().unwrap_or_default(),
"starting audio engine"
);
let ptt = Arc::new(AtomicBool::new(cfg.ptt_initial));
let frames_sent = Arc::new(AtomicU32::new(0));
let frames_received = Arc::new(AtomicU32::new(0));
// ---------- Capture ----------
// Capture is best-effort. If the platform default input
// device refuses any supported config (typical for
// headless null sources or for users who deny the mic
// permission) we log and continue — playback alone is
// still useful. PTT becomes a no-op in that case.
let capture_result = try_open_capture(
&in_dev,
voice_out_tx,
ptt.clone(),
frames_sent.clone(),
cfg.mic_gain,
);
let (input_stream, capture_active) = match capture_result {
Ok(s) => (Some(s), true),
Err(e) => {
warn!(
target: "chanora_audio",
error = %e,
"capture stream unavailable; continuing with playback only"
);
(None, false)
}
};
if let Some(s) = &input_stream {
s.play()
.map_err(|e| AudioError::Backend(format!("input play: {e}")))?;
}
// ---------- Playback ----------
let audio_handler: Arc<Mutex<AudioHandler<SessionAudioId>>> =
Arc::new(Mutex::new(AudioHandler::new()));
let out_cfg = out_dev
.default_output_config()
.map_err(|e| AudioError::StreamConfig(format!("output default: {e}")))?;
let out_format = out_cfg.sample_format();
// AudioHandler::fill_buffer expects 48 kHz stereo f32.
let out_stream_cfg = cpal::StreamConfig {
channels: 2,
sample_rate: cpal::SampleRate(SAMPLE_RATE),
buffer_size: cpal::BufferSize::Default,
};
let output_stream = match out_format {
SampleFormat::F32 => build_output_stream::<f32>(
&out_dev,
&out_stream_cfg,
audio_handler.clone(),
)?,
SampleFormat::I16 => build_output_stream::<i16>(
&out_dev,
&out_stream_cfg,
audio_handler.clone(),
)?,
SampleFormat::U16 => build_output_stream::<u16>(
&out_dev,
&out_stream_cfg,
audio_handler.clone(),
)?,
other => {
return Err(AudioError::StreamConfig(format!(
"unsupported output format: {other:?}"
)))
}
};
output_stream
.play()
.map_err(|e| AudioError::Backend(format!("output play: {e}")))?;
// ---------- Inbound forwarder ----------
let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
let handler_for_task = audio_handler.clone();
let frames_received_for_task = frames_received.clone();
tokio::spawn(async move {
loop {
tokio::select! {
_ = &mut shutdown_rx => {
debug!(target: "chanora_audio", "inbound forwarder shutting down");
break;
}
item = voice_in_rx.recv() => {
match item {
Some(v) => {
let id = SessionAudioId(v.from_client);
let mut h = handler_for_task.lock().unwrap();
if let Err(e) = h.handle_packet(id, v.packet) {
debug!(target: "chanora_audio", error = %e, "decode failed");
} else {
frames_received_for_task.fetch_add(1, Ordering::Relaxed);
}
}
None => break,
}
}
}
}
});
Ok(Self {
ptt,
frames_sent,
frames_received,
_input_stream: Mutex::new(input_stream),
_output_stream: Mutex::new(Some(output_stream)),
shutdown_tx: Some(shutdown_tx),
capture_active,
})
}
/// Stop the engine. Idempotent.
pub fn stop(&mut self) {
if let Some(tx) = self.shutdown_tx.take() {
let _ = tx.send(());
}
// Drop the streams, which stops their callback threads.
let _ = self._input_stream.lock().unwrap().take();
let _ = self._output_stream.lock().unwrap().take();
info!(target: "chanora_audio", "audio engine stopped");
}
/// Set the push-to-talk active state. When false, captured audio
/// is discarded before encoding. No-op if capture is inactive.
pub fn set_ptt(&self, active: bool) {
self.ptt.store(active, Ordering::Relaxed);
}
/// Current PTT state.
pub fn ptt(&self) -> bool {
self.ptt.load(Ordering::Relaxed)
}
/// True if the capture stream opened. When false, the engine
/// runs in playback-only mode and PTT is a no-op.
pub fn capture_active(&self) -> bool {
self.capture_active
}
/// Number of Opus frames sent since the engine started.
pub fn frames_sent(&self) -> u32 {
self.frames_sent.load(Ordering::Relaxed)
}
/// Number of inbound voice packets received and decoded.
pub fn frames_received(&self) -> u32 {
self.frames_received.load(Ordering::Relaxed)
}
}
impl Drop for AudioEngine {
fn drop(&mut self) {
self.stop();
}
}
// ---------- Capture pipeline ----------
fn try_open_capture(
in_dev: &cpal::Device,
voice_out_tx: mpsc::Sender<OutPacket>,
ptt: Arc<AtomicBool>,
frames_sent: Arc<AtomicU32>,
mic_gain: f32,
) -> Result<cpal::Stream, AudioError> {
let in_cfg = in_dev
.default_input_config()
.map_err(|e| AudioError::StreamConfig(format!("input default: {e}")))?;
let in_sample_rate = in_cfg.sample_rate().0;
let in_channels = in_cfg.channels() as usize;
let in_format = in_cfg.sample_format();
let in_stream_cfg: cpal::StreamConfig = in_cfg.into();
let opus_enc = OpusEncoder::new(
OpusSampleRate::Hz48000,
OpusChannels::Mono,
OpusApp::Voip,
)
.map_err(|e| AudioError::Opus(format!("encoder new: {e}")))?;
let capture_state = Arc::new(Mutex::new(CaptureState::new(
opus_enc,
in_sample_rate,
in_channels,
mic_gain,
voice_out_tx,
ptt,
frames_sent,
)));
let stream = match in_format {
SampleFormat::F32 => build_input_stream::<f32>(in_dev, &in_stream_cfg, capture_state)?,
SampleFormat::I16 => build_input_stream::<i16>(in_dev, &in_stream_cfg, capture_state)?,
SampleFormat::U16 => build_input_stream::<u16>(in_dev, &in_stream_cfg, capture_state)?,
other => {
return Err(AudioError::StreamConfig(format!(
"unsupported input format: {other:?}"
)))
}
};
Ok(stream)
}
struct CaptureState {
encoder: OpusEncoder,
in_sample_rate: u32,
in_channels: usize,
mic_gain: f32,
/// 48 kHz mono buffer accumulated to FRAME_SAMPLES before each encode.
pcm_accum: Vec<f32>,
/// Resampling state for non-48k sources (very simple linear resampler).
resample_pos: f64,
opus_out: [u8; MAX_OPUS_FRAME],
voice_out_tx: mpsc::Sender<OutPacket>,
ptt: Arc<AtomicBool>,
frames_sent: Arc<AtomicU32>,
}
impl CaptureState {
fn new(
encoder: OpusEncoder,
in_sample_rate: u32,
in_channels: usize,
mic_gain: f32,
voice_out_tx: mpsc::Sender<OutPacket>,
ptt: Arc<AtomicBool>,
frames_sent: Arc<AtomicU32>,
) -> Self {
Self {
encoder,
in_sample_rate,
in_channels,
mic_gain,
pcm_accum: Vec::with_capacity(FRAME_SAMPLES * 2),
resample_pos: 0.0,
opus_out: [0u8; MAX_OPUS_FRAME],
voice_out_tx,
ptt,
frames_sent,
}
}
/// Consume an arbitrary-rate, multichannel cpal buffer; produce
/// 48 kHz mono frames; encode and send on PTT.
fn ingest<T: ToF32 + Copy>(&mut self, buf: &[T]) {
if !self.ptt.load(Ordering::Relaxed) {
// Drain accumulator while muted so we don't pop on PTT release.
self.pcm_accum.clear();
return;
}
// 1. Down-mix to mono + gain.
let mono: Vec<f32> = buf
.chunks(self.in_channels)
.map(|frame| {
let sum: f32 = frame.iter().map(|s| s.to_f32_sample()).sum();
(sum / frame.len() as f32) * self.mic_gain
})
.collect();
// 2. Resample to 48 kHz if needed.
if self.in_sample_rate == SAMPLE_RATE {
self.pcm_accum.extend_from_slice(&mono);
} else {
self.resample_into_accum(&mono);
}
// 3. Encode any complete frames.
while self.pcm_accum.len() >= FRAME_SAMPLES {
let frame: Vec<f32> = self.pcm_accum.drain(..FRAME_SAMPLES).collect();
match self.encoder.encode_float(&frame, &mut self.opus_out[..]) {
Ok(len) => {
let packet = OutAudio::new(&AudioData::C2S {
id: 0,
codec: CodecType::OpusVoice,
data: &self.opus_out[..len],
});
match self.voice_out_tx.try_send(packet) {
Ok(()) => {
self.frames_sent.fetch_add(1, Ordering::Relaxed);
}
Err(mpsc::error::TrySendError::Full(_)) => {
warn!(target: "chanora_audio", "voice_out queue full; dropping frame");
}
Err(mpsc::error::TrySendError::Closed(_)) => {
warn!(target: "chanora_audio", "voice_out closed; stopping send");
}
}
}
Err(e) => {
error!(target: "chanora_audio", error = %e, "opus encode failed");
}
}
}
}
/// Simple linear resampler for `in_sample_rate → 48000`.
/// Production quality work belongs in a Beta+ DSP module.
fn resample_into_accum(&mut self, mono: &[f32]) {
let ratio = self.in_sample_rate as f64 / SAMPLE_RATE as f64;
let mut pos = self.resample_pos;
while pos < mono.len() as f64 {
let i = pos as usize;
let frac = pos - i as f64;
let a = mono[i];
let b = if i + 1 < mono.len() { mono[i + 1] } else { a };
self.pcm_accum
.push((a as f64 + frac * (b - a) as f64) as f32);
pos += ratio;
}
// Keep the leftover sub-sample offset for the next buffer.
self.resample_pos = pos - mono.len() as f64;
}
}
/// Per-sample format conversion to f32 in the range [-1.0, 1.0].
trait ToF32 {
fn to_f32_sample(self) -> f32;
}
impl ToF32 for f32 {
fn to_f32_sample(self) -> f32 {
self
}
}
impl ToF32 for i16 {
fn to_f32_sample(self) -> f32 {
f32::from(self) / f32::from(i16::MAX)
}
}
impl ToF32 for u16 {
fn to_f32_sample(self) -> f32 {
(f32::from(self) - f32::from(i16::MAX) - 1.0) / f32::from(i16::MAX)
}
}
fn build_input_stream<T>(
device: &cpal::Device,
config: &cpal::StreamConfig,
state: Arc<Mutex<CaptureState>>,
) -> Result<cpal::Stream, AudioError>
where
T: SizedSample + ToF32 + Send + 'static,
{
let stream = device
.build_input_stream(
config,
move |data: &[T], _| {
let mut s = state.lock().unwrap();
s.ingest(data);
},
move |e| {
error!(target: "chanora_audio", error = %e, "input stream error");
},
None,
)
.map_err(|e| AudioError::Backend(format!("build_input_stream: {e}")))?;
Ok(stream)
}
// ---------- Playback pipeline ----------
fn build_output_stream<T>(
device: &cpal::Device,
config: &cpal::StreamConfig,
handler: Arc<Mutex<AudioHandler<SessionAudioId>>>,
) -> Result<cpal::Stream, AudioError>
where
T: SizedSample + FromF32 + Send + 'static,
{
// Reusable f32 scratch buffer. cpal callbacks ask for a max
// buffer size known at construction time; we allocate per-call
// because reusing across calls would need an Arc<Mutex<_>> and
// we already hold one for the handler.
let stream = device
.build_output_stream(
config,
move |out: &mut [T], _| {
let mut scratch = vec![0.0f32; out.len()];
{
let mut h = handler.lock().unwrap();
h.fill_buffer(&mut scratch);
}
for (dst, src) in out.iter_mut().zip(scratch.into_iter()) {
*dst = T::from_f32_sample(src);
}
},
move |e| {
error!(target: "chanora_audio", error = %e, "output stream error");
},
None,
)
.map_err(|e| AudioError::Backend(format!("build_output_stream: {e}")))?;
Ok(stream)
}
trait FromF32 {
fn from_f32_sample(v: f32) -> Self;
}
impl FromF32 for f32 {
fn from_f32_sample(v: f32) -> Self {
v
}
}
impl FromF32 for i16 {
fn from_f32_sample(v: f32) -> Self {
(v.clamp(-1.0, 1.0) * f32::from(i16::MAX)) as i16
}
}
impl FromF32 for u16 {
fn from_f32_sample(v: f32) -> Self {
let s = (v.clamp(-1.0, 1.0) * f32::from(i16::MAX)) as i32;
(s + i32::from(i16::MAX) + 1) as u16
}
}
crates/chanora_audio/src/lib.rs
+32 -20
View File
@@ -1,30 +1,37 @@
//! # `chanora_audio`
//!
//! Audio subsystem. Per SAD §7.2:
//! Audio subsystem promoted from `poc/audio-capture-playback-spike`
//! and wired against `chanora_protocol`'s voice channels.
//!
//! * capture from the platform default input device
//! * DSP chain: high-pass filter → noise suppression → echo
//! cancellation → automatic gain control → gate
//! (per-effect defaults: DEC-007..010)
//! * Opus encode/decode
//! * jitter buffer, mixer
//! * playback to the platform default output device
//! ## What's wired in this Beta
//!
//! Crate selection per DEC-011.1: `cpal` on desktop and Android.
//! iOS audio crate remains deferred.
//! * Default-input capture via `cpal` (DEC-011.1)
//! * Frame-aligned 20 ms / 48 kHz mono Opus encoding via `audiopus`
//! * Forward encoded frames to the protocol crate as `OutPacket`s
//! * Inbound voice packets fed to `tsclientlib::audio::AudioHandler`
//! which owns Opus decode + per-client jitter buffer + mix
//! * Mixed f32 PCM pulled by the cpal output callback at 48 kHz stereo
//! * Push-to-talk: capture stream is permanently open; encoding is
//! gated by an atomic `ptt_active` flag
//!
//! Empirical evidence from PoC:
//! `poc/audio-capture-playback-spike` (Linux/PipeWire) +
//! `poc/audio-capture-playback-android-spike` (Android 14 arm64-v8a,
//! physical device).
//! ## What's NOT wired in this Beta
//!
//! ## Status
//!
//! Scaffold only. PoC code is not promoted here yet.
//! * AEC / AGC / NS / HPF DSP chain (DEC-007/008/009/010 — Beta+
//! work; the toggles in `AudioEffects` are honoured by *naming*
//! but the filters are no-ops)
//! * Mobile audio paths (DEC-011.1 desktop + Android proven; this
//! integration is desktop-only for v0.2.0-beta.1)
//! * Hot-plug device-change handling
//! * Sample-rate adaptation if the device cannot do 48 kHz / mono in
//! the format we request (returns `AudioError::StreamConfig`)
//! * Multi-channel speaker layouts beyond stereo
#![forbid(unsafe_code)]
#![warn(missing_docs)]
mod engine;
pub use engine::{AudioEngine, AudioEngineConfig};
use thiserror::Error;
/// Errors raised by the audio subsystem.
@@ -39,15 +46,20 @@ pub enum AudioError {
/// The audio backend rejected a stream configuration.
#[error("stream config rejected: {0}")]
StreamConfig(String),
/// Opus codec init/encode/decode failure.
#[error("opus: {0}")]
Opus(String),
/// A backend-specific failure surfaced without a typed mapping.
/// Production code must narrow this further as failure modes
/// are catalogued.
#[error("audio backend: {0}")]
Backend(String),
}
/// Audio-effect toggles. Defaults match DEC-007 (AEC),
/// DEC-008 (AGC), DEC-009 (NS), DEC-010 (HPF) — all enabled.
///
/// Note: in Beta v0.2.0-beta.1 the actual DSP filters are not yet
/// implemented; the struct is kept here as the public API surface so
/// later work can flip an internal flag without breaking callers.
#[derive(Debug, Clone, Copy)]
pub struct AudioEffects {
/// Acoustic echo cancellation (DEC-007).
crates/chanora_bridge/src/api.rs
+49
View File
@@ -175,3 +175,52 @@ pub async fn is_connected() -> bool {
.await
.unwrap_or(false)
}
// ---------- Audio commands (Beta) ----------
/// Start the audio engine on the active connection. Requires a
/// connection; idempotent (will replace any previous engine).
pub async fn start_audio() -> Result<(), BridgeError> {
runtime()
.spawn(async {
session()
.start_audio(chanora_core::AudioEngineConfig::default())
.await
})
.await
.map_err(|e| BridgeError::Unmapped(format!("join: {e}")))??;
Ok(())
}
/// Set the push-to-talk state.
pub async fn set_ptt(active: bool) -> Result<(), BridgeError> {
runtime()
.spawn(async move { session().set_ptt(active).await })
.await
.map_err(|e| BridgeError::Unmapped(format!("join: {e}")))??;
Ok(())
}
/// Statistics from the audio engine.
#[derive(Debug, Clone)]
pub struct BridgeAudioStats {
/// Number of Opus frames sent since audio started.
pub frames_sent: u32,
/// Number of inbound voice packets decoded.
pub frames_received: u32,
/// Current push-to-talk state.
pub ptt_active: bool,
}
/// Read audio statistics. Errors if no connection or audio not started.
pub async fn audio_stats() -> Result<BridgeAudioStats, BridgeError> {
let (s, r, p) = runtime()
.spawn(async { session().audio_stats().await })
.await
.map_err(|e| BridgeError::Unmapped(format!("join: {e}")))??;
Ok(BridgeAudioStats {
frames_sent: s,
frames_received: r,
ptt_active: p,
})
}
crates/chanora_bridge/src/frb_generated.rs
+171 -6
View File
@@ -38,7 +38,7 @@ flutter_rust_bridge::frb_generated_boilerplate!(
default_rust_auto_opaque = RustAutoOpaqueMoi,
);
pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_VERSION: &str = "2.12.0";
pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_CONTENT_HASH: i32 = 978717843;
pub(crate) const FLUTTER_RUST_BRIDGE_CODEGEN_CONTENT_HASH: i32 = 1944264248;
// Section: executor
@@ -46,6 +46,41 @@ flutter_rust_bridge::frb_generated_default_handler!();
// Section: wire_funcs
fn wire__crate__api__audio_stats_impl(
port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
rust_vec_len_: i32,
data_len_: i32,
) {
FLUTTER_RUST_BRIDGE_HANDLER.wrap_async::<flutter_rust_bridge::for_generated::SseCodec, _, _, _>(
flutter_rust_bridge::for_generated::TaskInfo {
debug_name: "audio_stats",
port: Some(port_),
mode: flutter_rust_bridge::for_generated::FfiCallMode::Normal,
},
move || {
let message = unsafe {
flutter_rust_bridge::for_generated::Dart2RustMessageSse::from_wire(
ptr_,
rust_vec_len_,
data_len_,
)
};
let mut deserializer =
flutter_rust_bridge::for_generated::SseDeserializer::new(message);
deserializer.end();
move |context| async move {
transform_result_sse::<_, crate::BridgeError>(
(move || async move {
let output_ok = crate::api::audio_stats().await?;
Ok(output_ok)
})()
.await,
)
}
},
)
}
fn wire__crate__api__bridge_init_impl(
port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
@@ -187,6 +222,42 @@ fn wire__crate__api__is_connected_impl(
},
)
}
fn wire__crate__api__set_ptt_impl(
port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
rust_vec_len_: i32,
data_len_: i32,
) {
FLUTTER_RUST_BRIDGE_HANDLER.wrap_async::<flutter_rust_bridge::for_generated::SseCodec, _, _, _>(
flutter_rust_bridge::for_generated::TaskInfo {
debug_name: "set_ptt",
port: Some(port_),
mode: flutter_rust_bridge::for_generated::FfiCallMode::Normal,
},
move || {
let message = unsafe {
flutter_rust_bridge::for_generated::Dart2RustMessageSse::from_wire(
ptr_,
rust_vec_len_,
data_len_,
)
};
let mut deserializer =
flutter_rust_bridge::for_generated::SseDeserializer::new(message);
let api_active = <bool>::sse_decode(&mut deserializer);
deserializer.end();
move |context| async move {
transform_result_sse::<_, crate::BridgeError>(
(move || async move {
let output_ok = crate::api::set_ptt(api_active).await?;
Ok(output_ok)
})()
.await,
)
}
},
)
}
fn wire__crate__api__snapshot_impl(
port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
@@ -222,6 +293,41 @@ fn wire__crate__api__snapshot_impl(
},
)
}
fn wire__crate__api__start_audio_impl(
port_: flutter_rust_bridge::for_generated::MessagePort,
ptr_: flutter_rust_bridge::for_generated::PlatformGeneralizedUint8ListPtr,
rust_vec_len_: i32,
data_len_: i32,
) {
FLUTTER_RUST_BRIDGE_HANDLER.wrap_async::<flutter_rust_bridge::for_generated::SseCodec, _, _, _>(
flutter_rust_bridge::for_generated::TaskInfo {
debug_name: "start_audio",
port: Some(port_),
mode: flutter_rust_bridge::for_generated::FfiCallMode::Normal,
},
move || {
let message = unsafe {
flutter_rust_bridge::for_generated::Dart2RustMessageSse::from_wire(
ptr_,
rust_vec_len_,
data_len_,
)
};
let mut deserializer =
flutter_rust_bridge::for_generated::SseDeserializer::new(message);
deserializer.end();
move |context| async move {
transform_result_sse::<_, crate::BridgeError>(
(move || async move {
let output_ok = crate::api::start_audio().await?;
Ok(output_ok)
})()
.await,
)
}
},
)
}
// Section: dart2rust
@@ -240,6 +346,20 @@ impl SseDecode for bool {
}
}
impl SseDecode for crate::api::BridgeAudioStats {
// Codec=Sse (Serialization based), see doc to use other codecs
fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self {
let mut var_framesSent = <u32>::sse_decode(deserializer);
let mut var_framesReceived = <u32>::sse_decode(deserializer);
let mut var_pttActive = <bool>::sse_decode(deserializer);
return crate::api::BridgeAudioStats {
frames_sent: var_framesSent,
frames_received: var_framesReceived,
ptt_active: var_pttActive,
};
}
}
impl SseDecode for crate::api::BridgeChannel {
// Codec=Sse (Serialization based), see doc to use other codecs
fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self {
@@ -363,6 +483,13 @@ impl SseDecode for Vec<u8> {
}
}
impl SseDecode for u32 {
// Codec=Sse (Serialization based), see doc to use other codecs
fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self {
deserializer.cursor.read_u32::<NativeEndian>().unwrap()
}
}
impl SseDecode for u64 {
// Codec=Sse (Serialization based), see doc to use other codecs
fn sse_decode(deserializer: &mut flutter_rust_bridge::for_generated::SseDeserializer) -> Self {
@@ -398,11 +525,14 @@ fn pde_ffi_dispatcher_primary_impl(
) {
// Codec=Pde (Serialization + dispatch), see doc to use other codecs
match func_id {
1 => wire__crate__api__bridge_init_impl(port, ptr, rust_vec_len, data_len),
2 => wire__crate__api__connect_impl(port, ptr, rust_vec_len, data_len),
3 => wire__crate__api__disconnect_impl(port, ptr, rust_vec_len, data_len),
4 => wire__crate__api__is_connected_impl(port, ptr, rust_vec_len, data_len),
5 => wire__crate__api__snapshot_impl(port, ptr, rust_vec_len, data_len),
1 => wire__crate__api__audio_stats_impl(port, ptr, rust_vec_len, data_len),
2 => wire__crate__api__bridge_init_impl(port, ptr, rust_vec_len, data_len),
3 => wire__crate__api__connect_impl(port, ptr, rust_vec_len, data_len),
4 => wire__crate__api__disconnect_impl(port, ptr, rust_vec_len, data_len),
5 => wire__crate__api__is_connected_impl(port, ptr, rust_vec_len, data_len),
6 => wire__crate__api__set_ptt_impl(port, ptr, rust_vec_len, data_len),
7 => wire__crate__api__snapshot_impl(port, ptr, rust_vec_len, data_len),
8 => wire__crate__api__start_audio_impl(port, ptr, rust_vec_len, data_len),
_ => unreachable!(),
}
}
@@ -421,6 +551,25 @@ fn pde_ffi_dispatcher_sync_impl(
// Section: rust2dart
// Codec=Dco (DartCObject based), see doc to use other codecs
impl flutter_rust_bridge::IntoDart for crate::api::BridgeAudioStats {
fn into_dart(self) -> flutter_rust_bridge::for_generated::DartAbi {
[
self.frames_sent.into_into_dart().into_dart(),
self.frames_received.into_into_dart().into_dart(),
self.ptt_active.into_into_dart().into_dart(),
]
.into_dart()
}
}
impl flutter_rust_bridge::for_generated::IntoDartExceptPrimitive for crate::api::BridgeAudioStats {}
impl flutter_rust_bridge::IntoIntoDart<crate::api::BridgeAudioStats>
for crate::api::BridgeAudioStats
{
fn into_into_dart(self) -> crate::api::BridgeAudioStats {
self
}
}
// Codec=Dco (DartCObject based), see doc to use other codecs
impl flutter_rust_bridge::IntoDart for crate::api::BridgeChannel {
fn into_dart(self) -> flutter_rust_bridge::for_generated::DartAbi {
@@ -518,6 +667,15 @@ impl SseEncode for bool {
}
}
impl SseEncode for crate::api::BridgeAudioStats {
// Codec=Sse (Serialization based), see doc to use other codecs
fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) {
<u32>::sse_encode(self.frames_sent, serializer);
<u32>::sse_encode(self.frames_received, serializer);
<bool>::sse_encode(self.ptt_active, serializer);
}
}
impl SseEncode for crate::api::BridgeChannel {
// Codec=Sse (Serialization based), see doc to use other codecs
fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) {
@@ -615,6 +773,13 @@ impl SseEncode for Vec<u8> {
}
}
impl SseEncode for u32 {
// Codec=Sse (Serialization based), see doc to use other codecs
fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) {
serializer.cursor.write_u32::<NativeEndian>(self).unwrap();
}
}
impl SseEncode for u64 {
// Codec=Sse (Serialization based), see doc to use other codecs
fn sse_encode(self, serializer: &mut flutter_rust_bridge::for_generated::SseSerializer) {
crates/chanora_bridge/src/lib.rs
+4
View File
@@ -66,7 +66,11 @@ impl From<chanora_core::CoreError> for BridgeError {
match e {
chanora_core::CoreError::NotConnected => BridgeError::NotConnected,
chanora_core::CoreError::AlreadyConnected => BridgeError::AlreadyConnected,
chanora_core::CoreError::AudioNotStarted => {
BridgeError::InvalidCommand("audio not started".to_string())
}
chanora_core::CoreError::Protocol(p) => BridgeError::Connection(format!("{p}")),
chanora_core::CoreError::Audio(a) => BridgeError::Connection(format!("audio: {a}")),
other => BridgeError::Unmapped(format!("{other}")),
}
}
crates/chanora_protocol/Cargo.toml
+9 -4
View File
@@ -14,10 +14,15 @@ thiserror.workspace = true
tracing.workspace = true
serde.workspace = true
# tsclientlib is git-only and not on crates.io. Audio feature disabled
# because chanora_audio owns audio paths; the protocol crate only
# handles connection lifecycle + state book events.
tsclientlib = { git = "https://github.com/ReSpeak/tsclientlib.git", rev = "04aa2491", default-features = false, features = ["default-tls"] }
# tsclientlib is git-only and not on crates.io. The "audio" feature
# pulls in `audiopus` only — `sdl2` is a dev-dep used by upstream
# examples; the library itself does not link SDL.
tsclientlib = { git = "https://github.com/ReSpeak/tsclientlib.git", rev = "04aa2491", default-features = false, features = ["default-tls", "audio"] }
# tsproto_packets exposes OutAudio / InAudioBuf / AudioData /
# CodecType / Direction. Pinning to the same git rev as tsclientlib
# avoids any version-skew confusion.
tsproto-packets = { git = "https://github.com/ReSpeak/tsclientlib.git", rev = "04aa2491" }
# Async runtime utilities used by the connection task.
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "sync"] }
crates/chanora_protocol/src/adapter.rs
+87 -16
View File
@@ -11,6 +11,9 @@
//! initial state snapshot is ready.
//! * Snapshot reads are served by sending a request over an
//! `mpsc::channel`; the task replies on a `oneshot` per request.
//! * Outbound voice packets are submitted via a separate mpsc;
//! inbound voice packets are forwarded out via a broadcast channel
//! so multiple sinks (recorder, audio mixer, …) can subscribe.
//! * Disconnect is requested via a `oneshot`; the task drains
//! `tsclientlib`'s outbound events and exits.
@@ -24,6 +27,7 @@ use tsclientlib::data::{self, Channel, Client};
use tsclientlib::{
ChannelId as TsChannelId, Connection, DisconnectOptions, Identity, OutCommandExt, StreamItem,
};
use tsproto_packets::packets::{InAudioBuf, OutPacket};
use crate::dto::{ChannelId, ChannelInfo, ClientId, ClientInfo, ServerSnapshot};
use crate::ProtocolError;
@@ -67,6 +71,21 @@ enum Request {
/// Async handle owning a live protocol connection. Drop = disconnect.
pub struct ProtocolClient {
tx: mpsc::Sender<Request>,
/// Submit outbound voice packets here. Built by `chanora_audio`
/// via [`Self::voice_out`].
voice_out_tx: mpsc::Sender<OutPacket>,
/// Inbound voice packets land here. Consumed by `chanora_audio`.
/// Wrapped in a `Mutex<Option<_>>` so the consumer can take it
/// exactly once.
voice_in_rx: std::sync::Mutex<Option<mpsc::Receiver<InboundVoice>>>,
}
/// One inbound voice packet from a remote client.
pub struct InboundVoice {
/// The remote client this audio came from.
pub from_client: u64,
/// Raw packet bytes for `AudioHandler::handle_packet`.
pub packet: InAudioBuf,
}
impl ProtocolClient {
@@ -82,12 +101,24 @@ impl ProtocolClient {
}
let (tx, rx) = mpsc::channel::<Request>(8);
let (voice_out_tx, voice_out_rx) = mpsc::channel::<OutPacket>(64);
let (voice_in_tx, voice_in_rx) = mpsc::channel::<InboundVoice>(64);
let (ready_tx, ready_rx) = oneshot::channel::<Result<(), ProtocolError>>();
tokio::spawn(connection_task(cfg.clone(), rx, ready_tx));
tokio::spawn(connection_task(
cfg.clone(),
rx,
voice_out_rx,
voice_in_tx,
ready_tx,
));
match tokio::time::timeout(cfg.ready_timeout, ready_rx).await {
Ok(Ok(Ok(()))) => Ok(Self { tx }),
Ok(Ok(Ok(()))) => Ok(Self {
tx,
voice_out_tx,
voice_in_rx: std::sync::Mutex::new(Some(voice_in_rx)),
}),
Ok(Ok(Err(e))) => Err(e),
Ok(Err(_)) => Err(ProtocolError::Backend(
"connection task exited before signalling ready".to_string(),
@@ -114,11 +145,24 @@ impl ProtocolClient {
let _ = rx.await;
}
}
/// Sender for outbound voice packets. Clone freely.
pub fn voice_out(&self) -> mpsc::Sender<OutPacket> {
self.voice_out_tx.clone()
}
/// Take the inbound-voice receiver. Returns `None` if it has
/// already been taken; only one consumer is allowed.
pub fn take_voice_in(&self) -> Option<mpsc::Receiver<InboundVoice>> {
self.voice_in_rx.lock().ok().and_then(|mut g| g.take())
}
}
async fn connection_task(
cfg: ConnectConfig,
mut rx: mpsc::Receiver<Request>,
mut voice_out_rx: mpsc::Receiver<OutPacket>,
voice_in_tx: mpsc::Sender<InboundVoice>,
ready_tx: oneshot::Sender<Result<(), ProtocolError>>,
) {
let mut builder = Connection::build(cfg.address.clone()).name(cfg.nickname.clone());
@@ -200,19 +244,38 @@ async fn connection_task(
let _ = ready_tx.send(Ok(()));
// Request loop with a continuously-pumped event stream. We pump
// one event at a time, then check for one pending request, then
// repeat. This avoids holding a borrow on `con` across an await
// boundary in `tokio::select!`.
// Main loop: pump events, service requests, forward voice.
loop {
// Try to advance the event stream by one event with a small
// timeout. Errors are logged; stream end is fatal.
// 1. Drain any outbound voice packets first — they're time-sensitive.
while let Ok(pkt) = voice_out_rx.try_recv() {
if let Err(e) = con.send_audio(pkt) {
warn!(target: "chanora_protocol", error = %e, "send_audio failed");
}
}
// 2. Advance event stream by at most one event with a small timeout.
let pump = async {
let mut ev_stream = con.events();
tokio::time::timeout(Duration::from_millis(50), ev_stream.next()).await
tokio::time::timeout(Duration::from_millis(20), ev_stream.next()).await
};
match pump.await {
Ok(Some(Ok(_))) => { /* event consumed */ }
Ok(Some(Ok(item))) => {
if let StreamItem::Audio(buf) = item {
// Extract `from` client id then forward.
let from = packet_sender_id(&buf);
if let Some(from) = from {
if voice_in_tx
.try_send(InboundVoice {
from_client: from,
packet: buf,
})
.is_err()
{
// Subscriber is too slow or absent; drop.
}
}
}
}
Ok(Some(Err(e))) => {
warn!(target: "chanora_protocol", error = %e, "event error");
}
@@ -220,11 +283,10 @@ async fn connection_task(
warn!(target: "chanora_protocol", "event stream ended");
return;
}
Err(_) => { /* no event in 50 ms — service requests */ }
Err(_) => { /* no event in 20 ms */ }
}
// Service at most one request (non-blocking) so we keep
// pumping events too.
// 3. Service at most one control request (non-blocking).
match rx.try_recv() {
Ok(Request::Snapshot(reply)) => {
let snap = build_snapshot(&con);
@@ -237,7 +299,7 @@ async fn connection_task(
info!(target: "chanora_protocol", "clean disconnect");
return;
}
Err(mpsc::error::TryRecvError::Empty) => { /* nothing to do */ }
Err(mpsc::error::TryRecvError::Empty) => {}
Err(mpsc::error::TryRecvError::Disconnected) => {
let _ = con.disconnect(DisconnectOptions::new());
con.events().for_each(|_| future::ready(())).await;
@@ -248,6 +310,16 @@ async fn connection_task(
}
}
/// Extract the originating `client_id` from an inbound voice packet.
fn packet_sender_id(buf: &InAudioBuf) -> Option<u64> {
use tsproto_packets::packets::AudioData;
match buf.data().data() {
AudioData::S2C { from, .. } => Some(*from as u64),
AudioData::S2CWhisper { from, .. } => Some(*from as u64),
_ => None,
}
}
fn build_snapshot(con: &Connection) -> Result<ServerSnapshot, ProtocolError> {
let state: &data::Connection = con
.get_state()
@@ -299,7 +371,6 @@ fn sanitize(s: &str) -> String {
#[allow(dead_code)]
const _ROOT_MATCHES_UPSTREAM: () = {
// Compile-time assertion that ChannelId(0) maps to what tsclientlib
// also considers the root. If upstream ever changes, this stops
// compiling and forces an audit.
// also considers the root.
let _ = TsChannelId(0);
};
crates/chanora_protocol/src/lib.rs
+10 -1
View File
@@ -27,9 +27,18 @@
mod adapter;
mod dto;
pub use adapter::{ConnectConfig, ProtocolClient};
pub use adapter::{ConnectConfig, InboundVoice, ProtocolClient};
pub use dto::{ChannelInfo, ClientInfo, ServerSnapshot};
// Re-export the upstream voice types so chanora_audio can build outbound
// voice packets without taking a direct dependency on tsclientlib /
// tsproto_packets. Per SAD-067 this is the *one* deliberate
// re-export: the audio path is performance-sensitive and a parallel
// type hierarchy would force copies for every 20 ms frame.
pub use tsproto_packets::packets::{
AudioData, CodecType, Direction, InAudioBuf, OutAudio, OutPacket,
};
use thiserror::Error;
/// Errors surfaced by the protocol adapter. None of these expose