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

refactor: separate SignalManager from WzpEngine for direct calling
Some checks failed
Mirror to GitHub / mirror (push) Failing after 40s
Details
Build Release Binaries / build-amd64 (push) Failing after 3m40s
Details

Browse Source 
SignalManager (NEW):
- Dedicated Rust struct with its own QUIC connection to _signal
- Separate JNI handle (nativeSignalConnect/GetState/PlaceCall/etc)
- Kotlin wrapper polls state every 500ms via getState() JSON
- Lives independently of WzpEngine — survives across calls
- connect() blocks briefly on 8MB thread, then recv loop runs on dedicated thread

WzpEngine (CLEANED):
- Back to pure media-only role (audio, codec, FEC, jitter)
- Removed start_signaling/place_call/answer_call methods
- Removed signal_transport/signal_fingerprint from EngineState

CallViewModel:
- Two separate managers: signalManager (persistent) + engine (per-call)
- Two separate polling loops: signalPollJob + statsJob
- Auto-connect to media room when signal polling detects "setup" state
- hangupDirectCall() ends media but keeps signal alive

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Siavash Sameni
2026-04-09 09:34:36 +04:00
parent 3a6ae61f8d
commit abc96e8887
7 changed files with 542 additions and 326 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -97,10 +97,6 @@ pub(crate) struct EngineState {
/// QUIC transport handle — stored so stop_call() can close it immediately,
/// triggering relay-side leave + RoomUpdate broadcast.
pub quic_transport: Mutex<Option<Arc<wzp_transport::QuinnTransport>>>,
/// Signal transport for direct calling — stored so place_call/answer_call can send.
pub signal_transport: Mutex<Option<Arc<wzp_transport::QuinnTransport>>>,
/// Our fingerprint (set during signaling registration).
pub signal_fingerprint: Mutex<Option<String>>,
}
pub struct WzpEngine {
@@ -122,8 +118,6 @@ impl WzpEngine {
playout_ring: AudioRing::new(),
audio_level_rms: AtomicU32::new(0),
quic_transport: Mutex::new(None),
signal_transport: Mutex::new(None),
signal_fingerprint: Mutex::new(None),
});
Self {
state,
@@ -250,203 +244,7 @@ impl WzpEngine {
}
/// Start persistent signaling connection for direct calls.
/// Spawns a background task that maintains the `_signal` connection.
/// Start persistent signaling for direct calls.
/// Blocks the calling thread (Kotlin provides a Thread with 8MB stack).
/// Same pattern as start_call: tokio block_on on the caller's thread.
pub fn start_signaling(
&mut self,
relay_addr: &str,
seed_hex: &str,
token: Option<&str>,
alias: Option<&str>,
) -> Result<(), anyhow::Error> {
use wzp_proto::{MediaTransport, SignalMessage};
let addr: SocketAddr = relay_addr.parse()?;
let seed = if seed_hex.is_empty() {
wzp_crypto::Seed::generate()
} else {
wzp_crypto::Seed::from_hex(seed_hex).map_err(|e| anyhow::anyhow!(e))?
};
let identity = seed.derive_identity();
let pub_id = identity.public_identity();
let identity_pub = *pub_id.signing.as_bytes();
let fp = pub_id.fingerprint.to_string();
let token = token.map(|s| s.to_string());
let alias = alias.map(|s| s.to_string());
let state = self.state.clone();
info!(fingerprint = %fp, relay = %addr, "starting signaling");
self.state.running.store(true, Ordering::Release);
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()?;
let signal_state = state.clone();
rt.block_on(async move {
let bind: SocketAddr = "0.0.0.0:0".parse().unwrap();
let endpoint = match wzp_transport::create_endpoint(bind, None) {
Ok(e) => e,
Err(e) => { error!("signal endpoint: {e}"); return; }
};
let client_cfg = wzp_transport::client_config();
let conn = match wzp_transport::connect(&endpoint, addr, "_signal", client_cfg).await {
Ok(c) => c,
Err(e) => { error!("signal connect: {e}"); return; }
};
let transport = std::sync::Arc::new(wzp_transport::QuinnTransport::new(conn));
// Auth if token provided
if let Some(ref tok) = token {
let _ = transport.send_signal(&SignalMessage::AuthToken { token: tok.clone() }).await;
}
// Register presence
let _ = transport.send_signal(&SignalMessage::RegisterPresence {
identity_pub,
signature: vec![],
alias: alias.clone(),
}).await;
// Wait for ack
match transport.recv_signal().await {
Ok(Some(SignalMessage::RegisterPresenceAck { success: true, .. })) => {
info!(fingerprint = %fp, "signal: registered");
let mut stats = signal_state.stats.lock().unwrap();
stats.state = crate::stats::CallState::Registered;
drop(stats);
// Store transport + fingerprint so place_call/answer_call can use them
*signal_state.signal_transport.lock().unwrap() = Some(transport.clone());
*signal_state.signal_fingerprint.lock().unwrap() = Some(fp.clone());
}
other => {
error!("signal registration failed: {other:?}");
return;
}
}
// Signal recv loop
loop {
if !signal_state.running.load(Ordering::Relaxed) {
break;
}
match transport.recv_signal().await {
Ok(Some(SignalMessage::CallRinging { call_id })) => {
info!(call_id = %call_id, "signal: ringing");
let mut stats = signal_state.stats.lock().unwrap();
stats.state = crate::stats::CallState::Ringing;
}
Ok(Some(SignalMessage::DirectCallOffer { caller_fingerprint, caller_alias, call_id, .. })) => {
info!(from = %caller_fingerprint, call_id = %call_id, "signal: incoming call");
let mut stats = signal_state.stats.lock().unwrap();
stats.state = crate::stats::CallState::IncomingCall;
stats.incoming_call_id = Some(call_id);
stats.incoming_caller_fp = Some(caller_fingerprint);
stats.incoming_caller_alias = caller_alias;
}
Ok(Some(SignalMessage::DirectCallAnswer { call_id, accept_mode, .. })) => {
info!(call_id = %call_id, mode = ?accept_mode, "signal: call answered");
}
Ok(Some(SignalMessage::CallSetup { call_id, room, relay_addr })) => {
info!(call_id = %call_id, room = %room, relay = %relay_addr, "signal: call setup");
// Connect to media room via the existing start_call mechanism
// Store the room info so Kotlin can call startCall with it
let mut stats = signal_state.stats.lock().unwrap();
stats.state = crate::stats::CallState::Connecting;
// Store call setup info for Kotlin to pick up
stats.incoming_call_id = Some(format!("{relay_addr}|{room}"));
}
Ok(Some(SignalMessage::Hangup { reason })) => {
info!(reason = ?reason, "signal: call ended by remote");
let mut stats = signal_state.stats.lock().unwrap();
stats.state = crate::stats::CallState::Closed;
stats.incoming_call_id = None;
stats.incoming_caller_fp = None;
stats.incoming_caller_alias = None;
}
Ok(Some(_)) => {}
Ok(None) => {
info!("signal: connection closed");
break;
}
Err(e) => {
error!("signal recv error: {e}");
break;
}
}
}
let mut stats = signal_state.stats.lock().unwrap();
stats.state = crate::stats::CallState::Closed;
}); // block_on
Ok(())
}
/// Place a direct call to a target fingerprint via the signal transport.
pub fn place_call(&self, target_fingerprint: &str) -> Result<(), anyhow::Error> {
use wzp_proto::SignalMessage;
let transport = self.state.signal_transport.lock().unwrap().clone()
.ok_or_else(|| anyhow::anyhow!("not registered"))?;
let caller_fp = self.state.signal_fingerprint.lock().unwrap().clone()
.unwrap_or_default();
let target = target_fingerprint.to_string();
let call_id = format!("{:016x}", std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos());
// Send on a separate thread since we can't block the UI thread
std::thread::spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("tokio runtime");
rt.block_on(async {
let _ = transport.send_signal(&SignalMessage::DirectCallOffer {
caller_fingerprint: caller_fp,
caller_alias: None,
target_fingerprint: target,
call_id,
identity_pub: [0u8; 32],
ephemeral_pub: [0u8; 32],
signature: vec![],
supported_profiles: vec![wzp_proto::QualityProfile::GOOD],
}).await;
});
});
Ok(())
}
/// Answer an incoming direct call via the signal transport.
pub fn answer_call(&self, call_id: &str, mode: wzp_proto::CallAcceptMode) -> Result<(), anyhow::Error> {
use wzp_proto::SignalMessage;
let transport = self.state.signal_transport.lock().unwrap().clone()
.ok_or_else(|| anyhow::anyhow!("not registered"))?;
let call_id = call_id.to_string();
std::thread::spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("tokio runtime");
rt.block_on(async {
let _ = transport.send_signal(&SignalMessage::DirectCallAnswer {
call_id,
accept_mode: mode,
identity_pub: None,
ephemeral_pub: None,
signature: None,
chosen_profile: Some(wzp_proto::QualityProfile::GOOD),
}).await;
});
});
Ok(())
}
// Signal methods (start_signaling, place_call, answer_call) moved to signal_mgr.rs
pub fn set_mute(&self, muted: bool) {
self.state.muted.store(muted, Ordering::Relaxed);