feat: direct 1:1 calling via relay signaling (Phase 1)
New feature: call someone directly by fingerprint through the relay.
- Client connects with SNI "_signal" for persistent signaling
- RegisterPresence/RegisterPresenceAck for relay registration
- DirectCallOffer routed to target by fingerprint
- DirectCallAnswer with AcceptGeneric/AcceptTrusted/Reject modes
- Relay creates private room (call-{id}), sends CallSetup to both
- Both clients connect to private room for media (existing SFU path)
- Hangup forwarding + cleanup on disconnect
- Desktop CLI: --signal + --call <fingerprint> for testing
- CallRegistry tracks call state (Pending/Ringing/Active/Ended)
- SignalHub manages persistent signaling connections
Tested: Alice calls Bob by fingerprint, relay routes offer, Bob
auto-accepts, both join private room, media flows bidirectionally.
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -48,6 +48,10 @@ struct CliArgs {
|
|||||||
token: Option<String>,
|
token: Option<String>,
|
||||||
_metrics_file: Option<String>,
|
_metrics_file: Option<String>,
|
||||||
version_check: bool,
|
version_check: bool,
|
||||||
|
/// Connect to relay for persistent signaling (direct calls).
|
||||||
|
signal: bool,
|
||||||
|
/// Place a direct call to a fingerprint (requires --signal).
|
||||||
|
call_target: Option<String>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl CliArgs {
|
impl CliArgs {
|
||||||
@@ -91,11 +95,18 @@ fn parse_args() -> CliArgs {
|
|||||||
let mut metrics_file = None;
|
let mut metrics_file = None;
|
||||||
let mut version_check = false;
|
let mut version_check = false;
|
||||||
let mut relay_str = None;
|
let mut relay_str = None;
|
||||||
|
let mut signal = false;
|
||||||
|
let mut call_target = None;
|
||||||
|
|
||||||
let mut i = 1;
|
let mut i = 1;
|
||||||
while i < args.len() {
|
while i < args.len() {
|
||||||
match args[i].as_str() {
|
match args[i].as_str() {
|
||||||
"--live" => live = true,
|
"--live" => live = true,
|
||||||
|
"--signal" => signal = true,
|
||||||
|
"--call" => {
|
||||||
|
i += 1;
|
||||||
|
call_target = Some(args.get(i).expect("--call requires a fingerprint").to_string());
|
||||||
|
}
|
||||||
"--send-tone" => {
|
"--send-tone" => {
|
||||||
i += 1;
|
i += 1;
|
||||||
send_tone_secs = Some(
|
send_tone_secs = Some(
|
||||||
@@ -225,6 +236,8 @@ fn parse_args() -> CliArgs {
|
|||||||
token,
|
token,
|
||||||
_metrics_file: metrics_file,
|
_metrics_file: metrics_file,
|
||||||
version_check,
|
version_check,
|
||||||
|
signal,
|
||||||
|
call_target,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -263,6 +276,12 @@ async fn main() -> anyhow::Result<()> {
|
|||||||
return Ok(());
|
return Ok(());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// --signal mode: persistent signaling for direct calls
|
||||||
|
if cli.signal {
|
||||||
|
let seed = cli.resolve_seed();
|
||||||
|
return run_signal_mode(cli.relay_addr, seed, cli.token, cli.call_target).await;
|
||||||
|
}
|
||||||
|
|
||||||
let seed = cli.resolve_seed();
|
let seed = cli.resolve_seed();
|
||||||
|
|
||||||
info!(
|
info!(
|
||||||
@@ -667,3 +686,195 @@ async fn run_live(transport: Arc<wzp_transport::QuinnTransport>) -> anyhow::Resu
|
|||||||
info!("done");
|
info!("done");
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Persistent signaling mode for direct 1:1 calls.
|
||||||
|
async fn run_signal_mode(
|
||||||
|
relay_addr: SocketAddr,
|
||||||
|
seed: wzp_crypto::Seed,
|
||||||
|
token: Option<String>,
|
||||||
|
call_target: Option<String>,
|
||||||
|
) -> anyhow::Result<()> {
|
||||||
|
use wzp_proto::SignalMessage;
|
||||||
|
|
||||||
|
let identity = seed.derive_identity();
|
||||||
|
let pub_id = identity.public_identity();
|
||||||
|
let fp = pub_id.fingerprint.to_string();
|
||||||
|
let identity_pub = *pub_id.signing.as_bytes();
|
||||||
|
info!(fingerprint = %fp, "signal mode");
|
||||||
|
|
||||||
|
// Connect to relay with SNI "_signal"
|
||||||
|
let client_config = wzp_transport::client_config();
|
||||||
|
let bind_addr: SocketAddr = if relay_addr.is_ipv6() {
|
||||||
|
"[::]:0".parse()?
|
||||||
|
} else {
|
||||||
|
"0.0.0.0:0".parse()?
|
||||||
|
};
|
||||||
|
let endpoint = wzp_transport::create_endpoint(bind_addr, None)?;
|
||||||
|
let conn = wzp_transport::connect(&endpoint, relay_addr, "_signal", client_config).await?;
|
||||||
|
let transport = Arc::new(wzp_transport::QuinnTransport::new(conn));
|
||||||
|
info!("connected to relay (signal channel)");
|
||||||
|
|
||||||
|
// Auth if token provided
|
||||||
|
if let Some(ref tok) = token {
|
||||||
|
transport.send_signal(&SignalMessage::AuthToken { token: tok.clone() }).await?;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Register presence (signature not verified in Phase 1)
|
||||||
|
transport.send_signal(&SignalMessage::RegisterPresence {
|
||||||
|
identity_pub,
|
||||||
|
signature: vec![], // Phase 1: not verified
|
||||||
|
alias: None,
|
||||||
|
}).await?;
|
||||||
|
|
||||||
|
// Wait for ack
|
||||||
|
match transport.recv_signal().await? {
|
||||||
|
Some(SignalMessage::RegisterPresenceAck { success: true, .. }) => {
|
||||||
|
info!(fingerprint = %fp, "registered on relay — waiting for calls");
|
||||||
|
}
|
||||||
|
Some(SignalMessage::RegisterPresenceAck { success: false, error }) => {
|
||||||
|
anyhow::bail!("registration failed: {}", error.unwrap_or_default());
|
||||||
|
}
|
||||||
|
other => {
|
||||||
|
anyhow::bail!("unexpected response: {other:?}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// If --call specified, place the call
|
||||||
|
if let Some(ref target) = call_target {
|
||||||
|
info!(target = %target, "placing direct call...");
|
||||||
|
let call_id = format!("{:016x}", std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH).unwrap().as_nanos());
|
||||||
|
|
||||||
|
transport.send_signal(&SignalMessage::DirectCallOffer {
|
||||||
|
caller_fingerprint: fp.clone(),
|
||||||
|
caller_alias: None,
|
||||||
|
target_fingerprint: target.clone(),
|
||||||
|
call_id: call_id.clone(),
|
||||||
|
identity_pub,
|
||||||
|
ephemeral_pub: [0u8; 32], // Phase 1: not used for key exchange
|
||||||
|
signature: vec![],
|
||||||
|
supported_profiles: vec![wzp_proto::QualityProfile::GOOD],
|
||||||
|
}).await?;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Signal recv loop — handle incoming signals
|
||||||
|
let signal_transport = transport.clone();
|
||||||
|
let relay = relay_addr;
|
||||||
|
let my_fp = fp.clone();
|
||||||
|
let my_seed = seed.0;
|
||||||
|
|
||||||
|
loop {
|
||||||
|
match signal_transport.recv_signal().await {
|
||||||
|
Ok(Some(msg)) => match msg {
|
||||||
|
SignalMessage::CallRinging { call_id } => {
|
||||||
|
info!(call_id = %call_id, "ringing...");
|
||||||
|
}
|
||||||
|
SignalMessage::DirectCallOffer { caller_fingerprint, caller_alias, call_id, .. } => {
|
||||||
|
info!(
|
||||||
|
from = %caller_fingerprint,
|
||||||
|
alias = ?caller_alias,
|
||||||
|
call_id = %call_id,
|
||||||
|
"incoming call — auto-accepting (generic)"
|
||||||
|
);
|
||||||
|
// Auto-accept for CLI testing
|
||||||
|
let _ = signal_transport.send_signal(&SignalMessage::DirectCallAnswer {
|
||||||
|
call_id,
|
||||||
|
accept_mode: wzp_proto::CallAcceptMode::AcceptGeneric,
|
||||||
|
identity_pub: Some(identity_pub),
|
||||||
|
ephemeral_pub: None,
|
||||||
|
signature: None,
|
||||||
|
chosen_profile: Some(wzp_proto::QualityProfile::GOOD),
|
||||||
|
}).await;
|
||||||
|
}
|
||||||
|
SignalMessage::DirectCallAnswer { call_id, accept_mode, .. } => {
|
||||||
|
info!(call_id = %call_id, mode = ?accept_mode, "call answered");
|
||||||
|
}
|
||||||
|
SignalMessage::CallSetup { call_id, room, relay_addr: setup_relay } => {
|
||||||
|
info!(call_id = %call_id, room = %room, relay = %setup_relay, "call setup — connecting to media room");
|
||||||
|
|
||||||
|
// Connect to the media room
|
||||||
|
let media_relay: SocketAddr = setup_relay.parse().unwrap_or(relay);
|
||||||
|
let media_cfg = wzp_transport::client_config();
|
||||||
|
match wzp_transport::connect(&endpoint, media_relay, &room, media_cfg).await {
|
||||||
|
Ok(media_conn) => {
|
||||||
|
let media_transport = Arc::new(wzp_transport::QuinnTransport::new(media_conn));
|
||||||
|
|
||||||
|
// Crypto handshake
|
||||||
|
match wzp_client::handshake::perform_handshake(&*media_transport, &my_seed, None).await {
|
||||||
|
Ok(_session) => {
|
||||||
|
info!("media connected — sending tone (press Ctrl+C to hang up)");
|
||||||
|
|
||||||
|
// Simple tone sender for testing
|
||||||
|
let mt = media_transport.clone();
|
||||||
|
let send_task = tokio::spawn(async move {
|
||||||
|
let config = wzp_client::call::CallConfig::default();
|
||||||
|
let mut encoder = wzp_client::call::CallEncoder::new(&config);
|
||||||
|
let duration = tokio::time::Duration::from_millis(20);
|
||||||
|
loop {
|
||||||
|
let pcm: Vec<i16> = (0..FRAME_SAMPLES)
|
||||||
|
.map(|_| 0i16) // silence — could be tone
|
||||||
|
.collect();
|
||||||
|
if let Ok(pkts) = encoder.encode_frame(&pcm) {
|
||||||
|
for pkt in &pkts {
|
||||||
|
if mt.send_media(pkt).await.is_err() { return; }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
tokio::time::sleep(duration).await;
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
// Wait for hangup or ctrl+c
|
||||||
|
loop {
|
||||||
|
tokio::select! {
|
||||||
|
sig = signal_transport.recv_signal() => {
|
||||||
|
match sig {
|
||||||
|
Ok(Some(SignalMessage::Hangup { .. })) => {
|
||||||
|
info!("remote hung up");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Ok(None) | Err(_) => break,
|
||||||
|
_ => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_ = tokio::signal::ctrl_c() => {
|
||||||
|
info!("hanging up...");
|
||||||
|
let _ = signal_transport.send_signal(&SignalMessage::Hangup {
|
||||||
|
reason: wzp_proto::HangupReason::Normal,
|
||||||
|
}).await;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
send_task.abort();
|
||||||
|
media_transport.close().await.ok();
|
||||||
|
info!("call ended");
|
||||||
|
}
|
||||||
|
Err(e) => error!("media handshake failed: {e}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(e) => error!("media connect failed: {e}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
SignalMessage::Hangup { reason } => {
|
||||||
|
info!(reason = ?reason, "call ended by remote");
|
||||||
|
}
|
||||||
|
SignalMessage::Pong { .. } => {}
|
||||||
|
other => {
|
||||||
|
info!("signal: {:?}", std::mem::discriminant(&other));
|
||||||
|
}
|
||||||
|
},
|
||||||
|
Ok(None) => {
|
||||||
|
info!("signal connection closed");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
error!("signal error: {e}");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
transport.close().await.ok();
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|||||||
@@ -113,6 +113,12 @@ pub fn signal_to_call_type(signal: &SignalMessage) -> CallSignalType {
|
|||||||
SignalMessage::FederationHello { .. }
|
SignalMessage::FederationHello { .. }
|
||||||
| SignalMessage::GlobalRoomActive { .. }
|
| SignalMessage::GlobalRoomActive { .. }
|
||||||
| SignalMessage::GlobalRoomInactive { .. } => CallSignalType::Offer, // relay-only
|
| SignalMessage::GlobalRoomInactive { .. } => CallSignalType::Offer, // relay-only
|
||||||
|
SignalMessage::DirectCallOffer { .. } => CallSignalType::Offer,
|
||||||
|
SignalMessage::DirectCallAnswer { .. } => CallSignalType::Answer,
|
||||||
|
SignalMessage::CallSetup { .. } => CallSignalType::Offer, // relay-only
|
||||||
|
SignalMessage::CallRinging { .. } => CallSignalType::Ringing,
|
||||||
|
SignalMessage::RegisterPresence { .. }
|
||||||
|
| SignalMessage::RegisterPresenceAck { .. } => CallSignalType::Offer, // relay-only
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -25,8 +25,9 @@ pub mod traits;
|
|||||||
pub use codec_id::{CodecId, QualityProfile};
|
pub use codec_id::{CodecId, QualityProfile};
|
||||||
pub use error::*;
|
pub use error::*;
|
||||||
pub use packet::{
|
pub use packet::{
|
||||||
HangupReason, MediaHeader, MediaPacket, MiniFrameContext, MiniHeader, QualityReport,
|
CallAcceptMode, HangupReason, MediaHeader, MediaPacket, MiniFrameContext, MiniHeader,
|
||||||
RoomParticipant, SignalMessage, TrunkEntry, TrunkFrame, FRAME_TYPE_FULL, FRAME_TYPE_MINI,
|
QualityReport, RoomParticipant, SignalMessage, TrunkEntry, TrunkFrame, FRAME_TYPE_FULL,
|
||||||
|
FRAME_TYPE_MINI,
|
||||||
};
|
};
|
||||||
pub use bandwidth::{BandwidthEstimator, CongestionState};
|
pub use bandwidth::{BandwidthEstimator, CongestionState};
|
||||||
pub use quality::{AdaptiveQualityController, NetworkContext, Tier};
|
pub use quality::{AdaptiveQualityController, NetworkContext, Tier};
|
||||||
|
|||||||
@@ -677,6 +677,91 @@ pub enum SignalMessage {
|
|||||||
GlobalRoomInactive {
|
GlobalRoomInactive {
|
||||||
room: String,
|
room: String,
|
||||||
},
|
},
|
||||||
|
|
||||||
|
// ── Direct calling signals (client ↔ relay signaling) ──
|
||||||
|
|
||||||
|
/// Register on relay for direct calls. Sent on `_signal` connections
|
||||||
|
/// after optional AuthToken.
|
||||||
|
RegisterPresence {
|
||||||
|
/// Client's Ed25519 identity public key.
|
||||||
|
identity_pub: [u8; 32],
|
||||||
|
/// Signature over ("register-presence" || identity_pub).
|
||||||
|
signature: Vec<u8>,
|
||||||
|
/// Optional display name.
|
||||||
|
alias: Option<String>,
|
||||||
|
},
|
||||||
|
|
||||||
|
/// Relay confirms presence registration.
|
||||||
|
RegisterPresenceAck {
|
||||||
|
success: bool,
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
error: Option<String>,
|
||||||
|
},
|
||||||
|
|
||||||
|
/// Direct call offer routed through the relay to a specific peer.
|
||||||
|
DirectCallOffer {
|
||||||
|
/// Caller's fingerprint.
|
||||||
|
caller_fingerprint: String,
|
||||||
|
/// Caller's display name.
|
||||||
|
caller_alias: Option<String>,
|
||||||
|
/// Target's fingerprint.
|
||||||
|
target_fingerprint: String,
|
||||||
|
/// Unique call session ID (UUID).
|
||||||
|
call_id: String,
|
||||||
|
/// Caller's Ed25519 identity pub.
|
||||||
|
identity_pub: [u8; 32],
|
||||||
|
/// Caller's ephemeral X25519 pub (for key exchange on media connect).
|
||||||
|
ephemeral_pub: [u8; 32],
|
||||||
|
/// Signature over (ephemeral_pub || target_fingerprint || call_id).
|
||||||
|
signature: Vec<u8>,
|
||||||
|
/// Supported quality profiles.
|
||||||
|
supported_profiles: Vec<crate::QualityProfile>,
|
||||||
|
},
|
||||||
|
|
||||||
|
/// Callee's response to a direct call.
|
||||||
|
DirectCallAnswer {
|
||||||
|
call_id: String,
|
||||||
|
/// How the callee accepts (or rejects).
|
||||||
|
accept_mode: CallAcceptMode,
|
||||||
|
/// Callee's identity pub (present when accepting).
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
identity_pub: Option<[u8; 32]>,
|
||||||
|
/// Callee's ephemeral pub (present when accepting).
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
ephemeral_pub: Option<[u8; 32]>,
|
||||||
|
/// Signature (present when accepting).
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
signature: Option<Vec<u8>>,
|
||||||
|
/// Chosen quality profile (present when accepting).
|
||||||
|
#[serde(skip_serializing_if = "Option::is_none")]
|
||||||
|
chosen_profile: Option<crate::QualityProfile>,
|
||||||
|
},
|
||||||
|
|
||||||
|
/// Relay tells both parties: media room is ready.
|
||||||
|
CallSetup {
|
||||||
|
call_id: String,
|
||||||
|
/// Room name on the relay for the media session (e.g., "_call:a1b2c3d4").
|
||||||
|
room: String,
|
||||||
|
/// Relay address for the QUIC media connection.
|
||||||
|
relay_addr: String,
|
||||||
|
},
|
||||||
|
|
||||||
|
/// Ringing notification (relay → caller, callee received the offer).
|
||||||
|
CallRinging {
|
||||||
|
call_id: String,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
/// How the callee responds to a direct call.
|
||||||
|
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
|
pub enum CallAcceptMode {
|
||||||
|
/// Reject the call.
|
||||||
|
Reject,
|
||||||
|
/// Accept with trust — in Phase 2, this enables P2P (reveals IP).
|
||||||
|
/// In Phase 1, behaves the same as AcceptGeneric.
|
||||||
|
AcceptTrusted,
|
||||||
|
/// Accept with privacy — relay always mediates media.
|
||||||
|
AcceptGeneric,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A participant entry in a RoomUpdate message.
|
/// A participant entry in a RoomUpdate message.
|
||||||
|
|||||||
199
crates/wzp-relay/src/call_registry.rs
Normal file
199
crates/wzp-relay/src/call_registry.rs
Normal file
@@ -0,0 +1,199 @@
|
|||||||
|
//! Direct call state tracking.
|
||||||
|
//!
|
||||||
|
//! Manages the lifecycle of 1:1 direct calls placed via the `_signal` channel.
|
||||||
|
//! Each call goes through: Pending → Ringing → Active → Ended.
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
/// State of a direct call.
|
||||||
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||||
|
pub enum DirectCallState {
|
||||||
|
/// Offer sent to callee, waiting for response.
|
||||||
|
Pending,
|
||||||
|
/// Callee acknowledged, ringing.
|
||||||
|
Ringing,
|
||||||
|
/// Call accepted, media room active.
|
||||||
|
Active,
|
||||||
|
/// Call ended (hangup, reject, timeout, or error).
|
||||||
|
Ended,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A tracked direct call between two users.
|
||||||
|
pub struct DirectCall {
|
||||||
|
pub call_id: String,
|
||||||
|
pub caller_fingerprint: String,
|
||||||
|
pub callee_fingerprint: String,
|
||||||
|
pub state: DirectCallState,
|
||||||
|
pub accept_mode: Option<wzp_proto::CallAcceptMode>,
|
||||||
|
/// Private room name (set when accepted).
|
||||||
|
pub room_name: Option<String>,
|
||||||
|
pub created_at: Instant,
|
||||||
|
pub answered_at: Option<Instant>,
|
||||||
|
pub ended_at: Option<Instant>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Registry of active direct calls.
|
||||||
|
pub struct CallRegistry {
|
||||||
|
calls: HashMap<String, DirectCall>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl CallRegistry {
|
||||||
|
pub fn new() -> Self {
|
||||||
|
Self {
|
||||||
|
calls: HashMap::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Create a new pending call. Returns the call_id.
|
||||||
|
pub fn create_call(&mut self, call_id: String, caller_fp: String, callee_fp: String) -> &DirectCall {
|
||||||
|
let call = DirectCall {
|
||||||
|
call_id: call_id.clone(),
|
||||||
|
caller_fingerprint: caller_fp,
|
||||||
|
callee_fingerprint: callee_fp,
|
||||||
|
state: DirectCallState::Pending,
|
||||||
|
accept_mode: None,
|
||||||
|
room_name: None,
|
||||||
|
created_at: Instant::now(),
|
||||||
|
answered_at: None,
|
||||||
|
ended_at: None,
|
||||||
|
};
|
||||||
|
self.calls.insert(call_id.clone(), call);
|
||||||
|
self.calls.get(&call_id).unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get a call by ID.
|
||||||
|
pub fn get(&self, call_id: &str) -> Option<&DirectCall> {
|
||||||
|
self.calls.get(call_id)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get a mutable call by ID.
|
||||||
|
pub fn get_mut(&mut self, call_id: &str) -> Option<&mut DirectCall> {
|
||||||
|
self.calls.get_mut(call_id)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Transition to Ringing state.
|
||||||
|
pub fn set_ringing(&mut self, call_id: &str) -> bool {
|
||||||
|
if let Some(call) = self.calls.get_mut(call_id) {
|
||||||
|
if call.state == DirectCallState::Pending {
|
||||||
|
call.state = DirectCallState::Ringing;
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
false
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Transition to Active state.
|
||||||
|
pub fn set_active(&mut self, call_id: &str, mode: wzp_proto::CallAcceptMode, room: String) -> bool {
|
||||||
|
if let Some(call) = self.calls.get_mut(call_id) {
|
||||||
|
if call.state == DirectCallState::Pending || call.state == DirectCallState::Ringing {
|
||||||
|
call.state = DirectCallState::Active;
|
||||||
|
call.accept_mode = Some(mode);
|
||||||
|
call.room_name = Some(room);
|
||||||
|
call.answered_at = Some(Instant::now());
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
false
|
||||||
|
}
|
||||||
|
|
||||||
|
/// End a call.
|
||||||
|
pub fn end_call(&mut self, call_id: &str) -> Option<DirectCall> {
|
||||||
|
if let Some(call) = self.calls.get_mut(call_id) {
|
||||||
|
call.state = DirectCallState::Ended;
|
||||||
|
call.ended_at = Some(Instant::now());
|
||||||
|
}
|
||||||
|
self.calls.remove(call_id)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Find active/pending calls involving a fingerprint.
|
||||||
|
pub fn calls_for_fingerprint(&self, fp: &str) -> Vec<&DirectCall> {
|
||||||
|
self.calls.values()
|
||||||
|
.filter(|c| {
|
||||||
|
c.state != DirectCallState::Ended
|
||||||
|
&& (c.caller_fingerprint == fp || c.callee_fingerprint == fp)
|
||||||
|
})
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Find the peer's fingerprint in a call.
|
||||||
|
pub fn peer_fingerprint(&self, call_id: &str, my_fp: &str) -> Option<&str> {
|
||||||
|
self.calls.get(call_id).map(|c| {
|
||||||
|
if c.caller_fingerprint == my_fp {
|
||||||
|
c.callee_fingerprint.as_str()
|
||||||
|
} else {
|
||||||
|
c.caller_fingerprint.as_str()
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Remove calls that have been pending longer than the timeout.
|
||||||
|
/// Returns call IDs of expired calls.
|
||||||
|
pub fn expire_stale(&mut self, timeout: Duration) -> Vec<DirectCall> {
|
||||||
|
let now = Instant::now();
|
||||||
|
let expired: Vec<String> = self.calls.iter()
|
||||||
|
.filter(|(_, c)| {
|
||||||
|
c.state == DirectCallState::Pending
|
||||||
|
&& now.duration_since(c.created_at) > timeout
|
||||||
|
})
|
||||||
|
.map(|(id, _)| id.clone())
|
||||||
|
.collect();
|
||||||
|
|
||||||
|
expired.into_iter()
|
||||||
|
.filter_map(|id| self.calls.remove(&id))
|
||||||
|
.collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Number of active (non-ended) calls.
|
||||||
|
pub fn active_count(&self) -> usize {
|
||||||
|
self.calls.values()
|
||||||
|
.filter(|c| c.state != DirectCallState::Ended)
|
||||||
|
.count()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn call_lifecycle() {
|
||||||
|
let mut reg = CallRegistry::new();
|
||||||
|
reg.create_call("c1".into(), "alice".into(), "bob".into());
|
||||||
|
|
||||||
|
assert_eq!(reg.get("c1").unwrap().state, DirectCallState::Pending);
|
||||||
|
assert!(reg.set_ringing("c1"));
|
||||||
|
assert_eq!(reg.get("c1").unwrap().state, DirectCallState::Ringing);
|
||||||
|
|
||||||
|
assert!(reg.set_active("c1", wzp_proto::CallAcceptMode::AcceptGeneric, "_call:c1".into()));
|
||||||
|
assert_eq!(reg.get("c1").unwrap().state, DirectCallState::Active);
|
||||||
|
assert_eq!(reg.get("c1").unwrap().room_name.as_deref(), Some("_call:c1"));
|
||||||
|
|
||||||
|
let ended = reg.end_call("c1").unwrap();
|
||||||
|
assert_eq!(ended.state, DirectCallState::Ended);
|
||||||
|
assert_eq!(reg.active_count(), 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn expire_stale_calls() {
|
||||||
|
let mut reg = CallRegistry::new();
|
||||||
|
reg.create_call("c1".into(), "alice".into(), "bob".into());
|
||||||
|
|
||||||
|
// Not expired yet
|
||||||
|
let expired = reg.expire_stale(Duration::from_secs(30));
|
||||||
|
assert!(expired.is_empty());
|
||||||
|
|
||||||
|
// Force expiry with 0 timeout
|
||||||
|
let expired = reg.expire_stale(Duration::from_secs(0));
|
||||||
|
assert_eq!(expired.len(), 1);
|
||||||
|
assert_eq!(expired[0].call_id, "c1");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn peer_lookup() {
|
||||||
|
let mut reg = CallRegistry::new();
|
||||||
|
reg.create_call("c1".into(), "alice".into(), "bob".into());
|
||||||
|
assert_eq!(reg.peer_fingerprint("c1", "alice"), Some("bob"));
|
||||||
|
assert_eq!(reg.peer_fingerprint("c1", "bob"), Some("alice"));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -8,9 +8,11 @@
|
|||||||
//! quality transitions.
|
//! quality transitions.
|
||||||
|
|
||||||
pub mod auth;
|
pub mod auth;
|
||||||
|
pub mod call_registry;
|
||||||
pub mod config;
|
pub mod config;
|
||||||
pub mod event_log;
|
pub mod event_log;
|
||||||
pub mod federation;
|
pub mod federation;
|
||||||
|
pub mod signal_hub;
|
||||||
pub mod handshake;
|
pub mod handshake;
|
||||||
pub mod metrics;
|
pub mod metrics;
|
||||||
pub mod pipeline;
|
pub mod pipeline;
|
||||||
|
|||||||
@@ -424,6 +424,10 @@ async fn main() -> anyhow::Result<()> {
|
|||||||
// Session manager — enforces max concurrent sessions
|
// Session manager — enforces max concurrent sessions
|
||||||
let session_mgr = Arc::new(Mutex::new(SessionManager::new(config.max_sessions)));
|
let session_mgr = Arc::new(Mutex::new(SessionManager::new(config.max_sessions)));
|
||||||
|
|
||||||
|
// Signal hub + call registry for direct 1:1 calls
|
||||||
|
let signal_hub = Arc::new(Mutex::new(wzp_relay::signal_hub::SignalHub::new()));
|
||||||
|
let call_registry = Arc::new(Mutex::new(wzp_relay::call_registry::CallRegistry::new()));
|
||||||
|
|
||||||
// Spawn inter-relay health probes via ProbeMesh coordinator
|
// Spawn inter-relay health probes via ProbeMesh coordinator
|
||||||
if !config.probe_targets.is_empty() {
|
if !config.probe_targets.is_empty() {
|
||||||
let mesh = wzp_relay::probe::ProbeMesh::new(
|
let mesh = wzp_relay::probe::ProbeMesh::new(
|
||||||
@@ -487,6 +491,9 @@ async fn main() -> anyhow::Result<()> {
|
|||||||
let presence = presence.clone();
|
let presence = presence.clone();
|
||||||
let route_resolver = route_resolver.clone();
|
let route_resolver = route_resolver.clone();
|
||||||
let federation_mgr = federation_mgr.clone();
|
let federation_mgr = federation_mgr.clone();
|
||||||
|
let signal_hub = signal_hub.clone();
|
||||||
|
let call_registry = call_registry.clone();
|
||||||
|
let listen_addr_str = config.listen_addr.to_string();
|
||||||
|
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
let addr = connection.remote_address();
|
let addr = connection.remote_address();
|
||||||
@@ -641,6 +648,244 @@ async fn main() -> anyhow::Result<()> {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Direct calling: persistent signaling connection
|
||||||
|
if room_name == "_signal" {
|
||||||
|
info!(%addr, "signal connection");
|
||||||
|
|
||||||
|
// Optional auth
|
||||||
|
let auth_fp: Option<String> = if let Some(ref url) = auth_url {
|
||||||
|
match transport.recv_signal().await {
|
||||||
|
Ok(Some(SignalMessage::AuthToken { token })) => {
|
||||||
|
match wzp_relay::auth::validate_token(url, &token).await {
|
||||||
|
Ok(client) => Some(client.fingerprint),
|
||||||
|
Err(e) => {
|
||||||
|
error!(%addr, "signal auth failed: {e}");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
_ => { warn!(%addr, "signal: expected AuthToken"); return; }
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
None
|
||||||
|
};
|
||||||
|
|
||||||
|
// Wait for RegisterPresence
|
||||||
|
let (client_fp, client_alias) = match tokio::time::timeout(
|
||||||
|
std::time::Duration::from_secs(10),
|
||||||
|
transport.recv_signal(),
|
||||||
|
).await {
|
||||||
|
Ok(Ok(Some(SignalMessage::RegisterPresence { identity_pub, signature: _, alias }))) => {
|
||||||
|
// Compute fingerprint: SHA-256(Ed25519 pub key)[:16] as hex pairs with colons
|
||||||
|
let hash = {
|
||||||
|
use sha2::{Sha256, Digest};
|
||||||
|
Sha256::digest(&identity_pub)
|
||||||
|
};
|
||||||
|
let fp = hash[..16].iter()
|
||||||
|
.map(|b| format!("{b:02x}"))
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
.chunks(2)
|
||||||
|
.map(|c| c.join(""))
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
.join(":");
|
||||||
|
let fp = auth_fp.unwrap_or(fp);
|
||||||
|
(fp, alias)
|
||||||
|
}
|
||||||
|
_ => {
|
||||||
|
warn!(%addr, "signal: no RegisterPresence received");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
// Register in signal hub + presence
|
||||||
|
{
|
||||||
|
let mut hub = signal_hub.lock().await;
|
||||||
|
hub.register(client_fp.clone(), transport.clone(), client_alias.clone());
|
||||||
|
}
|
||||||
|
{
|
||||||
|
let mut reg = presence.lock().await;
|
||||||
|
reg.register_local(&client_fp, client_alias.clone(), None);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Send ack
|
||||||
|
let _ = transport.send_signal(&SignalMessage::RegisterPresenceAck {
|
||||||
|
success: true,
|
||||||
|
error: None,
|
||||||
|
}).await;
|
||||||
|
|
||||||
|
info!(%addr, fingerprint = %client_fp, alias = ?client_alias, "signal client registered");
|
||||||
|
|
||||||
|
// Signal recv loop
|
||||||
|
loop {
|
||||||
|
match transport.recv_signal().await {
|
||||||
|
Ok(Some(msg)) => {
|
||||||
|
match msg {
|
||||||
|
SignalMessage::DirectCallOffer { ref target_fingerprint, ref call_id, ref caller_alias, .. } => {
|
||||||
|
let target_fp = target_fingerprint.clone();
|
||||||
|
let call_id = call_id.clone();
|
||||||
|
|
||||||
|
// Check if target is online
|
||||||
|
let online = {
|
||||||
|
let hub = signal_hub.lock().await;
|
||||||
|
hub.is_online(&target_fp)
|
||||||
|
};
|
||||||
|
if !online {
|
||||||
|
info!(%addr, target = %target_fp, "call target not online");
|
||||||
|
let _ = transport.send_signal(&SignalMessage::Hangup {
|
||||||
|
reason: wzp_proto::HangupReason::Normal,
|
||||||
|
}).await;
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Create call in registry
|
||||||
|
{
|
||||||
|
let mut reg = call_registry.lock().await;
|
||||||
|
reg.create_call(call_id.clone(), client_fp.clone(), target_fp.clone());
|
||||||
|
}
|
||||||
|
|
||||||
|
// Forward offer to callee
|
||||||
|
info!(caller = %client_fp, callee = %target_fp, call_id = %call_id, "routing direct call offer");
|
||||||
|
let hub = signal_hub.lock().await;
|
||||||
|
if let Err(e) = hub.send_to(&target_fp, &msg).await {
|
||||||
|
warn!("failed to forward call offer: {e}");
|
||||||
|
}
|
||||||
|
|
||||||
|
// Send ringing to caller
|
||||||
|
drop(hub);
|
||||||
|
let _ = transport.send_signal(&SignalMessage::CallRinging {
|
||||||
|
call_id: call_id.clone(),
|
||||||
|
}).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
SignalMessage::DirectCallAnswer { ref call_id, ref accept_mode, .. } => {
|
||||||
|
let call_id = call_id.clone();
|
||||||
|
let mode = *accept_mode;
|
||||||
|
|
||||||
|
let peer_fp = {
|
||||||
|
let reg = call_registry.lock().await;
|
||||||
|
reg.peer_fingerprint(&call_id, &client_fp).map(|s| s.to_string())
|
||||||
|
};
|
||||||
|
|
||||||
|
let Some(peer_fp) = peer_fp else {
|
||||||
|
warn!(call_id = %call_id, "answer for unknown call");
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
|
||||||
|
if mode == wzp_proto::CallAcceptMode::Reject {
|
||||||
|
info!(call_id = %call_id, "call rejected");
|
||||||
|
let mut reg = call_registry.lock().await;
|
||||||
|
reg.end_call(&call_id);
|
||||||
|
drop(reg);
|
||||||
|
let hub = signal_hub.lock().await;
|
||||||
|
let _ = hub.send_to(&peer_fp, &SignalMessage::Hangup {
|
||||||
|
reason: wzp_proto::HangupReason::Normal,
|
||||||
|
}).await;
|
||||||
|
} else {
|
||||||
|
// Accept — create private room
|
||||||
|
let room = format!("call-{call_id}");
|
||||||
|
{
|
||||||
|
let mut reg = call_registry.lock().await;
|
||||||
|
reg.set_active(&call_id, mode, room.clone());
|
||||||
|
}
|
||||||
|
info!(call_id = %call_id, room = %room, mode = ?mode, "call accepted, creating room");
|
||||||
|
|
||||||
|
// Forward answer to caller
|
||||||
|
{
|
||||||
|
let hub = signal_hub.lock().await;
|
||||||
|
let _ = hub.send_to(&peer_fp, &msg).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Send CallSetup to both parties
|
||||||
|
let setup = SignalMessage::CallSetup {
|
||||||
|
call_id: call_id.clone(),
|
||||||
|
room: room.clone(),
|
||||||
|
relay_addr: listen_addr_str.clone(),
|
||||||
|
};
|
||||||
|
{
|
||||||
|
let hub = signal_hub.lock().await;
|
||||||
|
let _ = hub.send_to(&peer_fp, &setup).await;
|
||||||
|
let _ = hub.send_to(&client_fp, &setup).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
SignalMessage::Hangup { .. } => {
|
||||||
|
// Forward hangup to all active calls for this user
|
||||||
|
let calls = {
|
||||||
|
let reg = call_registry.lock().await;
|
||||||
|
reg.calls_for_fingerprint(&client_fp)
|
||||||
|
.iter()
|
||||||
|
.map(|c| (c.call_id.clone(), if c.caller_fingerprint == client_fp {
|
||||||
|
c.callee_fingerprint.clone()
|
||||||
|
} else {
|
||||||
|
c.caller_fingerprint.clone()
|
||||||
|
}))
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
};
|
||||||
|
for (call_id, peer_fp) in &calls {
|
||||||
|
let hub = signal_hub.lock().await;
|
||||||
|
let _ = hub.send_to(peer_fp, &msg).await;
|
||||||
|
drop(hub);
|
||||||
|
let mut reg = call_registry.lock().await;
|
||||||
|
reg.end_call(call_id);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
SignalMessage::Ping { timestamp_ms } => {
|
||||||
|
let _ = transport.send_signal(&SignalMessage::Pong { timestamp_ms }).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
other => {
|
||||||
|
warn!(%addr, "signal: unexpected message: {:?}", std::mem::discriminant(&other));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(None) => {
|
||||||
|
info!(%addr, "signal connection closed");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
warn!(%addr, "signal recv error: {e}");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Cleanup: unregister + end active calls
|
||||||
|
let active_calls = {
|
||||||
|
let reg = call_registry.lock().await;
|
||||||
|
reg.calls_for_fingerprint(&client_fp)
|
||||||
|
.iter()
|
||||||
|
.map(|c| (c.call_id.clone(), if c.caller_fingerprint == client_fp {
|
||||||
|
c.callee_fingerprint.clone()
|
||||||
|
} else {
|
||||||
|
c.caller_fingerprint.clone()
|
||||||
|
}))
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
};
|
||||||
|
for (call_id, peer_fp) in &active_calls {
|
||||||
|
let hub = signal_hub.lock().await;
|
||||||
|
let _ = hub.send_to(peer_fp, &SignalMessage::Hangup {
|
||||||
|
reason: wzp_proto::HangupReason::Normal,
|
||||||
|
}).await;
|
||||||
|
drop(hub);
|
||||||
|
let mut reg = call_registry.lock().await;
|
||||||
|
reg.end_call(call_id);
|
||||||
|
}
|
||||||
|
|
||||||
|
{
|
||||||
|
let mut hub = signal_hub.lock().await;
|
||||||
|
hub.unregister(&client_fp);
|
||||||
|
}
|
||||||
|
{
|
||||||
|
let mut reg = presence.lock().await;
|
||||||
|
reg.unregister_local(&client_fp);
|
||||||
|
}
|
||||||
|
|
||||||
|
transport.close().await.ok();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
// Auth check: if --auth-url is set, expect first signal message to be a token
|
// Auth check: if --auth-url is set, expect first signal message to be a token
|
||||||
// Auth: if --auth-url is set, expect AuthToken as first signal
|
// Auth: if --auth-url is set, expect AuthToken as first signal
|
||||||
let authenticated_fp: Option<String> = if let Some(ref url) = auth_url {
|
let authenticated_fp: Option<String> = if let Some(ref url) = auth_url {
|
||||||
|
|||||||
105
crates/wzp-relay/src/signal_hub.rs
Normal file
105
crates/wzp-relay/src/signal_hub.rs
Normal file
@@ -0,0 +1,105 @@
|
|||||||
|
//! Persistent signaling connection manager.
|
||||||
|
//!
|
||||||
|
//! Tracks clients connected via `_signal` SNI. Routes call signals
|
||||||
|
//! (DirectCallOffer, DirectCallAnswer, Hangup) between registered users.
|
||||||
|
|
||||||
|
use std::collections::HashMap;
|
||||||
|
use std::sync::Arc;
|
||||||
|
use std::time::Instant;
|
||||||
|
|
||||||
|
use tracing::{info, warn};
|
||||||
|
use wzp_proto::{MediaTransport, SignalMessage};
|
||||||
|
use wzp_transport::QuinnTransport;
|
||||||
|
|
||||||
|
/// A client connected via `_signal` for direct calling.
|
||||||
|
pub struct SignalClient {
|
||||||
|
pub fingerprint: String,
|
||||||
|
pub alias: Option<String>,
|
||||||
|
pub transport: Arc<QuinnTransport>,
|
||||||
|
pub connected_at: Instant,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Manages persistent signaling connections.
|
||||||
|
pub struct SignalHub {
|
||||||
|
clients: HashMap<String, SignalClient>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SignalHub {
|
||||||
|
pub fn new() -> Self {
|
||||||
|
Self {
|
||||||
|
clients: HashMap::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Register a new signaling client.
|
||||||
|
pub fn register(&mut self, fp: String, transport: Arc<QuinnTransport>, alias: Option<String>) {
|
||||||
|
info!(fingerprint = %fp, alias = ?alias, "signal client registered");
|
||||||
|
self.clients.insert(fp.clone(), SignalClient {
|
||||||
|
fingerprint: fp,
|
||||||
|
alias,
|
||||||
|
transport,
|
||||||
|
connected_at: Instant::now(),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Unregister a signaling client. Returns the client if found.
|
||||||
|
pub fn unregister(&mut self, fp: &str) -> Option<SignalClient> {
|
||||||
|
let client = self.clients.remove(fp);
|
||||||
|
if client.is_some() {
|
||||||
|
info!(fingerprint = %fp, "signal client unregistered");
|
||||||
|
}
|
||||||
|
client
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Look up a client by fingerprint.
|
||||||
|
pub fn get(&self, fp: &str) -> Option<&SignalClient> {
|
||||||
|
self.clients.get(fp)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Check if a fingerprint is online.
|
||||||
|
pub fn is_online(&self, fp: &str) -> bool {
|
||||||
|
self.clients.contains_key(fp)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Send a signal message to a client by fingerprint.
|
||||||
|
pub async fn send_to(&self, fp: &str, msg: &SignalMessage) -> Result<(), String> {
|
||||||
|
match self.clients.get(fp) {
|
||||||
|
Some(client) => {
|
||||||
|
client.transport.send_signal(msg).await
|
||||||
|
.map_err(|e| format!("send to {fp}: {e}"))
|
||||||
|
}
|
||||||
|
None => Err(format!("{fp} not online")),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Number of connected signaling clients.
|
||||||
|
pub fn online_count(&self) -> usize {
|
||||||
|
self.clients.len()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// List all online fingerprints.
|
||||||
|
pub fn online_fingerprints(&self) -> Vec<&str> {
|
||||||
|
self.clients.keys().map(|s| s.as_str()).collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get alias for a fingerprint.
|
||||||
|
pub fn alias(&self, fp: &str) -> Option<&str> {
|
||||||
|
self.clients.get(fp).and_then(|c| c.alias.as_deref())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn register_unregister() {
|
||||||
|
let mut hub = SignalHub::new();
|
||||||
|
assert_eq!(hub.online_count(), 0);
|
||||||
|
assert!(!hub.is_online("alice"));
|
||||||
|
|
||||||
|
// Can't easily construct QuinnTransport in a unit test,
|
||||||
|
// so we just test the HashMap logic conceptually.
|
||||||
|
// Integration tests cover the full flow.
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user