2381 lines
85 KiB
Rust
2381 lines
85 KiB
Rust
//! The adapter that drives `tsclientlib` on a background task and
|
|
//! exposes a typed channel-and-future API to the rest of Chanora.
|
|
//!
|
|
//! Threading model:
|
|
//!
|
|
//! * `ProtocolClient::connect` spawns a tokio task that owns the
|
|
//! `tsclientlib::Connection` (which is not `Send`-safe to move
|
|
//! across awaits in some shapes — keeping it inside a single task
|
|
//! sidesteps the problem entirely).
|
|
//! * The task exposes its life via a `oneshot` that fires when the
|
|
//! 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.
|
|
|
|
use std::net::SocketAddr;
|
|
use std::time::{Duration, Instant};
|
|
|
|
use base64::prelude::*;
|
|
use chanora_resolver::ChanoraResolver;
|
|
use futures::prelude::*;
|
|
use std::collections::HashMap;
|
|
use tokio::sync::{mpsc, oneshot};
|
|
use tracing::{info, warn};
|
|
|
|
use tsclientlib::data::{self, Channel, Client};
|
|
use tsclientlib::messages::s2c::{InClientDbInfoPart, InMessage};
|
|
use tsclientlib::prelude::*;
|
|
use tsclientlib::{
|
|
ChannelId as TsChannelId, ClientId as TsClientId, Connection, ConnectionStats,
|
|
DisconnectOptions, Identity, MessageHandle, OutCommandExt, StreamItem, Version,
|
|
};
|
|
use tsproto_packets::packets::{Direction, Flags, InAudioBuf, OutCommand, OutPacket, PacketType};
|
|
use tsproto_types::ClientType;
|
|
|
|
use crate::dto::{
|
|
ChannelId, ChannelInfo, ChatMessage, ClientId, ClientInfo, ClientProfile, MessageTarget,
|
|
ProtocolDelta, ServerActivity, ServerSnapshot,
|
|
};
|
|
use crate::poke_limiter::PokeLimiter;
|
|
use crate::ProtocolError;
|
|
|
|
const SPEAKING_ACTIVITY_WINDOW: Duration = Duration::from_millis(750);
|
|
const INBOUND_VOICE_SEND_TIMEOUT: Duration = Duration::from_millis(40);
|
|
const PROFILE_REFRESH_RESULT_TIMEOUT: Duration = Duration::from_secs(3);
|
|
const OUTBOUND_VOICE_PACKETS_PER_TICK: usize = 8;
|
|
const DISCONNECT_REPLY_TIMEOUT: Duration = Duration::from_secs(1);
|
|
const DISCONNECT_EVENT_DRAIN_TIMEOUT: Duration = Duration::from_millis(500);
|
|
|
|
type PendingMoves = HashMap<
|
|
MessageHandle,
|
|
(
|
|
u64,
|
|
Option<oneshot::Sender<Result<(), ProtocolError>>>,
|
|
std::time::Instant,
|
|
),
|
|
>;
|
|
|
|
struct EventChannels {
|
|
voice_in: mpsc::Sender<InboundVoice>,
|
|
chat: mpsc::Sender<ChatMessage>,
|
|
activity: mpsc::Sender<ServerActivity>,
|
|
delta: mpsc::Sender<ProtocolDelta>,
|
|
}
|
|
|
|
#[derive(Debug, PartialEq, Eq)]
|
|
enum SendTimeoutError<T> {
|
|
Timeout(T),
|
|
Closed(T),
|
|
}
|
|
|
|
async fn send_with_timeout<T: Send>(
|
|
tx: &mpsc::Sender<T>,
|
|
value: T,
|
|
timeout_duration: Duration,
|
|
) -> Result<(), SendTimeoutError<T>> {
|
|
match tokio::time::timeout(timeout_duration, tx.reserve()).await {
|
|
Ok(Ok(permit)) => {
|
|
permit.send(value);
|
|
Ok(())
|
|
}
|
|
Ok(Err(_)) => Err(SendTimeoutError::Closed(value)),
|
|
Err(_) => Err(SendTimeoutError::Timeout(value)),
|
|
}
|
|
}
|
|
|
|
fn drain_voice_packets_for_tick<T, E>(
|
|
voice_out_rx: &mut mpsc::Receiver<T>,
|
|
max_packets: usize,
|
|
mut send: impl FnMut(T) -> Result<(), E>,
|
|
) -> usize {
|
|
let mut drained = 0;
|
|
for _ in 0..max_packets {
|
|
let packet = match voice_out_rx.try_recv() {
|
|
Ok(packet) => packet,
|
|
Err(_) => break,
|
|
};
|
|
let _ = send(packet);
|
|
drained += 1;
|
|
}
|
|
drained
|
|
}
|
|
|
|
async fn bounded_drain_stream<S>(stream: S, timeout_duration: Duration)
|
|
where
|
|
S: futures::Stream,
|
|
{
|
|
let _ = tokio::time::timeout(timeout_duration, stream.for_each(|_| future::ready(()))).await;
|
|
}
|
|
|
|
/// Pick the TeamSpeak `client_version`/platform/signature triple
|
|
/// (sourced from `ReSpeak/tsdeclarations/Versions.csv`, baked into
|
|
/// `tsproto-types` at vendor-time) that best matches the *runtime*
|
|
/// platform Chanora is executing on.
|
|
///
|
|
/// Per-platform selection (set by the project owner):
|
|
///
|
|
/// * Windows / Linux / macOS desktops announce as TeamSpeak **5
|
|
/// beta51** — the latest TS5 desktop signature in the vendored
|
|
/// `tsproto-types` enum. TS5 servers expect this shape; TS3
|
|
/// servers are happy to accept any well-formed signed client
|
|
/// descriptor and have no codec-version coupling.
|
|
/// * Android announces as **3.5.0__7** (the latest Android
|
|
/// signature in the vendored enum).
|
|
/// * iOS announces as **3.5.6** (latest iOS signature in the
|
|
/// vendored enum).
|
|
///
|
|
/// All five variants are guaranteed to exist in the generated
|
|
/// `Version` enum at compile time; if upstream rotates the CSV the
|
|
/// build will fail loudly here rather than silently fall back.
|
|
fn pick_client_version() -> Version {
|
|
Version::Windows_3_X_X__1
|
|
}
|
|
|
|
/// Typed configuration for a connection attempt.
|
|
#[derive(Debug, Clone)]
|
|
pub struct ConnectConfig {
|
|
/// Server address: `hostname[:port]` or TSDNS name.
|
|
pub address: String,
|
|
/// Nickname to use on the server.
|
|
pub nickname: String,
|
|
/// Optional server password.
|
|
pub password: Option<String>,
|
|
/// Optional pre-existing identity (base64 string accepted by
|
|
/// `tsclientlib::Identity::new_from_str`). If `None`, a fresh
|
|
/// identity is generated and **not persisted** — production callers
|
|
/// should provide one from secure identity storage.
|
|
pub identity: Option<String>,
|
|
/// How long to wait for the initial state snapshot before
|
|
/// returning `ProtocolError::Timeout`.
|
|
pub ready_timeout: Duration,
|
|
/// Optional already-resolved socket address from core's invisible
|
|
/// prefetch cache. When present, the protocol layer skips address
|
|
/// resolution but still opens a normal TS3 connection only after
|
|
/// the user requested Connect.
|
|
pub resolved_address: Option<std::net::SocketAddr>,
|
|
}
|
|
|
|
impl Default for ConnectConfig {
|
|
fn default() -> Self {
|
|
Self {
|
|
address: String::new(),
|
|
nickname: "Chanora".to_string(),
|
|
password: None,
|
|
identity: None,
|
|
ready_timeout: Duration::from_secs(10),
|
|
resolved_address: None,
|
|
}
|
|
}
|
|
}
|
|
|
|
enum Request {
|
|
Snapshot(oneshot::Sender<Result<ServerSnapshot, ProtocolError>>),
|
|
Disconnect(oneshot::Sender<()>),
|
|
/// Move self to a channel. Optional channel password.
|
|
MoveToChannel {
|
|
channel_id: u64,
|
|
password: Option<String>,
|
|
reply: oneshot::Sender<Result<(), ProtocolError>>,
|
|
},
|
|
/// Move self to a channel without waiting for the reply.
|
|
MoveToChannelNoWait {
|
|
channel_id: u64,
|
|
password: Option<String>,
|
|
},
|
|
/// Update own client mute state (input and/or output).
|
|
SetMuted {
|
|
input: Option<bool>,
|
|
output: Option<bool>,
|
|
reply: oneshot::Sender<Result<(), ProtocolError>>,
|
|
},
|
|
/// Send a text message to a target.
|
|
SendTextMessage {
|
|
/// Message content.
|
|
message: String,
|
|
/// Target scope.
|
|
target: MessageTarget,
|
|
/// Reply channel for outcome.
|
|
reply: oneshot::Sender<Result<(), ProtocolError>>,
|
|
},
|
|
/// Fetch richer profile/connection details for an online client.
|
|
FetchClientProfile {
|
|
client_id: u64,
|
|
reply: oneshot::Sender<Result<ClientProfile, ProtocolError>>,
|
|
},
|
|
}
|
|
|
|
/// Why a [`ProtocolClient`] task ended. Distinguishes a user-driven
|
|
/// disconnect (the supervisor must NOT retry) from a network-driven
|
|
/// loss (the supervisor should consider retrying).
|
|
#[derive(Debug, Clone)]
|
|
pub enum DisconnectReason {
|
|
/// The caller explicitly called [`ProtocolClient::disconnect`]
|
|
/// or dropped the handle.
|
|
UserRequested,
|
|
/// The underlying tsclientlib event stream ended.
|
|
StreamEnded,
|
|
/// A protocol-layer error caused the task to abort.
|
|
Error(String),
|
|
}
|
|
|
|
/// 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>>>,
|
|
/// Fires exactly once when the connection task exits, with the
|
|
/// reason. Used by the supervisor in `chanora_core` to drive
|
|
/// auto-reconnect. Wrapped in a Mutex<Option<_>> so it can be
|
|
/// taken once by the supervisor and never resurfaced.
|
|
lost_rx: std::sync::Mutex<Option<oneshot::Receiver<DisconnectReason>>>,
|
|
/// Inbound chat message stream from the connection task. The
|
|
/// receiver is taken by the supervisor and forwarded to UI.
|
|
chat_rx: std::sync::Mutex<Option<mpsc::Receiver<ChatMessage>>>,
|
|
/// Inbound server-activity stream from the connection task.
|
|
activity_rx: std::sync::Mutex<Option<mpsc::Receiver<ServerActivity>>>,
|
|
/// Inbound state-delta stream from the connection task.
|
|
delta_rx: std::sync::Mutex<Option<mpsc::Receiver<ProtocolDelta>>>,
|
|
}
|
|
|
|
/// 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,
|
|
}
|
|
|
|
/// A cheap, clone-free probe handle for the watchdog. Owns its own
|
|
/// clone of the connection task's request channel.
|
|
#[derive(Clone)]
|
|
pub struct SnapshotProbe {
|
|
tx: mpsc::Sender<Request>,
|
|
}
|
|
|
|
impl SnapshotProbe {
|
|
/// Issue a single snapshot RPC. Returns the same error shape as
|
|
/// [`ProtocolClient::snapshot`]. Suitable for use under a
|
|
/// `tokio::time::timeout`.
|
|
pub async fn probe(&self) -> Result<ServerSnapshot, ProtocolError> {
|
|
let (tx, rx) = oneshot::channel();
|
|
self.tx
|
|
.send(Request::Snapshot(tx))
|
|
.await
|
|
.map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?;
|
|
rx.await
|
|
.map_err(|_| ProtocolError::Lost("snapshot reply dropped".to_string()))?
|
|
}
|
|
}
|
|
|
|
impl ProtocolClient {
|
|
/// Generate a fresh, persistable TS3 identity string. The
|
|
/// returned value is the canonical `counter`V`base64key` form
|
|
/// accepted by [`ConnectConfig::identity`] and by tsclientlib's
|
|
/// `Identity::new_from_str`. Callers should persist it via
|
|
/// `chanora_storage` so subsequent connects reuse the same
|
|
/// identity and the server sees the same client UID.
|
|
pub fn generate_identity() -> String {
|
|
let id = Identity::create();
|
|
format!("{}V{}", id.counter(), id.key().to_ts())
|
|
}
|
|
|
|
/// Dial the server and wait for the initial state snapshot. The
|
|
/// returned client is ready for [`Self::snapshot`] and
|
|
/// [`Self::disconnect`] calls.
|
|
pub async fn connect(cfg: ConnectConfig) -> Result<Self, ProtocolError> {
|
|
if cfg.address.trim().is_empty() {
|
|
return Err(ProtocolError::Invalid("address is empty".to_string()));
|
|
}
|
|
if cfg.nickname.trim().is_empty() {
|
|
return Err(ProtocolError::Invalid("nickname is empty".to_string()));
|
|
}
|
|
|
|
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 (chat_tx, chat_rx) = mpsc::channel::<ChatMessage>(64);
|
|
let (activity_tx, activity_rx) = mpsc::channel::<ServerActivity>(128);
|
|
let (delta_tx, delta_rx) = mpsc::channel::<ProtocolDelta>(256);
|
|
let (ready_tx, ready_rx) = oneshot::channel::<Result<(), ProtocolError>>();
|
|
let (lost_tx, lost_rx) = oneshot::channel::<DisconnectReason>();
|
|
|
|
tokio::spawn(connection_task(
|
|
cfg.clone(),
|
|
rx,
|
|
voice_out_rx,
|
|
EventChannels {
|
|
voice_in: voice_in_tx,
|
|
chat: chat_tx,
|
|
activity: activity_tx,
|
|
delta: delta_tx,
|
|
},
|
|
ready_tx,
|
|
lost_tx,
|
|
));
|
|
|
|
match tokio::time::timeout(cfg.ready_timeout, ready_rx).await {
|
|
Ok(Ok(Ok(()))) => Ok(Self {
|
|
tx,
|
|
voice_out_tx,
|
|
voice_in_rx: std::sync::Mutex::new(Some(voice_in_rx)),
|
|
lost_rx: std::sync::Mutex::new(Some(lost_rx)),
|
|
chat_rx: std::sync::Mutex::new(Some(chat_rx)),
|
|
activity_rx: std::sync::Mutex::new(Some(activity_rx)),
|
|
delta_rx: std::sync::Mutex::new(Some(delta_rx)),
|
|
}),
|
|
Ok(Ok(Err(e))) => Err(e),
|
|
Ok(Err(_)) => Err(ProtocolError::Backend(
|
|
"connection task exited before signalling ready".to_string(),
|
|
)),
|
|
Err(_) => Err(ProtocolError::Timeout),
|
|
}
|
|
}
|
|
|
|
/// Read a typed snapshot of the current server state.
|
|
pub async fn snapshot(&self) -> Result<ServerSnapshot, ProtocolError> {
|
|
let (tx, rx) = oneshot::channel();
|
|
self.tx
|
|
.send(Request::Snapshot(tx))
|
|
.await
|
|
.map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?;
|
|
rx.await
|
|
.map_err(|_| ProtocolError::Lost("snapshot reply dropped".to_string()))?
|
|
}
|
|
|
|
/// Fetch richer profile and live connection details for one online client.
|
|
pub async fn client_profile(&self, client_id: u64) -> Result<ClientProfile, ProtocolError> {
|
|
let (tx, rx) = oneshot::channel();
|
|
self.tx
|
|
.send(Request::FetchClientProfile {
|
|
client_id,
|
|
reply: tx,
|
|
})
|
|
.await
|
|
.map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?;
|
|
rx.await
|
|
.map_err(|_| ProtocolError::Lost("client_profile reply dropped".to_string()))?
|
|
}
|
|
|
|
/// Disconnect cleanly. Blocks until the task exits.
|
|
pub async fn disconnect(self) {
|
|
let (tx, rx) = oneshot::channel();
|
|
let request_path = async {
|
|
if self.tx.send(Request::Disconnect(tx)).await.is_ok() {
|
|
let _ = rx.await;
|
|
}
|
|
};
|
|
|
|
if tokio::time::timeout(DISCONNECT_REPLY_TIMEOUT, request_path)
|
|
.await
|
|
.is_err()
|
|
{
|
|
warn!(
|
|
target: "chanora_protocol",
|
|
timeout_ms = DISCONNECT_REPLY_TIMEOUT.as_millis() as u64,
|
|
"disconnect request did not complete before timeout"
|
|
);
|
|
}
|
|
}
|
|
|
|
/// Move our own client into a channel. `password` is optional
|
|
/// for password-protected channels.
|
|
pub async fn move_to_channel(
|
|
&self,
|
|
channel_id: u64,
|
|
password: Option<String>,
|
|
) -> Result<(), ProtocolError> {
|
|
let (tx, rx) = oneshot::channel();
|
|
self.tx
|
|
.send(Request::MoveToChannel {
|
|
channel_id,
|
|
password,
|
|
reply: tx,
|
|
})
|
|
.await
|
|
.map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?;
|
|
rx.await
|
|
.map_err(|_| ProtocolError::Lost("move_to_channel reply dropped".to_string()))?
|
|
}
|
|
|
|
/// Queue a move command and return once it has been accepted by
|
|
/// the protocol task. The server-side result is still observed
|
|
/// asynchronously by the task for logging and reconciliation.
|
|
pub async fn queue_move_to_channel(
|
|
&self,
|
|
channel_id: u64,
|
|
password: Option<String>,
|
|
) -> Result<(), ProtocolError> {
|
|
self.tx
|
|
.send(Request::MoveToChannelNoWait {
|
|
channel_id,
|
|
password,
|
|
})
|
|
.await
|
|
.map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))
|
|
}
|
|
|
|
/// Update mute state on our own client. Pass `Some(_)` for the
|
|
/// fields you want to change, `None` to leave a field as-is.
|
|
pub async fn set_muted(
|
|
&self,
|
|
input: Option<bool>,
|
|
output: Option<bool>,
|
|
) -> Result<(), ProtocolError> {
|
|
let (tx, rx) = oneshot::channel();
|
|
self.tx
|
|
.send(Request::SetMuted {
|
|
input,
|
|
output,
|
|
reply: tx,
|
|
})
|
|
.await
|
|
.map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?;
|
|
rx.await
|
|
.map_err(|_| ProtocolError::Lost("set_muted reply dropped".to_string()))?
|
|
}
|
|
|
|
/// Sender for outbound voice packets. Clone freely.
|
|
pub fn voice_out(&self) -> mpsc::Sender<OutPacket> {
|
|
self.voice_out_tx.clone()
|
|
}
|
|
|
|
/// Clone the request channel so a watchdog can issue probes
|
|
/// without holding a `&self` reference across the await. The
|
|
/// returned [`SnapshotProbe`] is `Send + 'static` and dispatches
|
|
/// a single snapshot RPC against this protocol task.
|
|
pub fn snapshot_probe(&self) -> SnapshotProbe {
|
|
SnapshotProbe {
|
|
tx: self.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())
|
|
}
|
|
|
|
/// Put a previously-taken voice_in receiver back so a
|
|
/// follow-up `take_voice_in()` succeeds. Used by the core's
|
|
/// `start_audio` to recover from a failed
|
|
/// `AudioEngine::start_with_gate` — without this a single
|
|
/// engine-construction failure would permanently poison the
|
|
/// voice channel and force a reconnect to fix.
|
|
pub fn put_voice_in(&self, rx: mpsc::Receiver<InboundVoice>) {
|
|
if let Ok(mut g) = self.voice_in_rx.lock() {
|
|
// If a consumer is already in possession we drop the
|
|
// duplicate rather than overwriting; this branch
|
|
// should not be reachable in practice because the only
|
|
// caller (start_audio) takes-then-puts inside the same
|
|
// critical section.
|
|
if g.is_none() {
|
|
*g = Some(rx);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Take the loss-notifier. Returns `None` if it has already been
|
|
/// taken. The supervisor in `chanora_core` consumes this to
|
|
/// drive auto-reconnect; nothing else should call it.
|
|
pub fn take_loss_notifier(&self) -> Option<oneshot::Receiver<DisconnectReason>> {
|
|
self.lost_rx.lock().ok().and_then(|mut g| g.take())
|
|
}
|
|
|
|
/// Take the inbound-chat receiver. Returns `None` if it has
|
|
/// already been taken; only one consumer is allowed.
|
|
pub fn take_chat_rx(&self) -> Option<mpsc::Receiver<ChatMessage>> {
|
|
self.chat_rx.lock().ok().and_then(|mut g| g.take())
|
|
}
|
|
|
|
/// Put a previously-taken chat_rx receiver back.
|
|
pub fn put_chat_rx(&self, rx: mpsc::Receiver<ChatMessage>) {
|
|
if let Ok(mut g) = self.chat_rx.lock() {
|
|
if g.is_none() {
|
|
*g = Some(rx);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Take the inbound server-activity receiver. Returns `None` if it has
|
|
/// already been taken; only one consumer is allowed.
|
|
pub fn take_activity_rx(&self) -> Option<mpsc::Receiver<ServerActivity>> {
|
|
self.activity_rx.lock().ok().and_then(|mut g| g.take())
|
|
}
|
|
|
|
/// Put a previously-taken activity receiver back.
|
|
pub fn put_activity_rx(&self, rx: mpsc::Receiver<ServerActivity>) {
|
|
if let Ok(mut g) = self.activity_rx.lock() {
|
|
if g.is_none() {
|
|
*g = Some(rx);
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Takes ownership of the delta receiver channel. Returns `None` if already
|
|
/// taken. Must be called exactly once during initialization to subscribe to
|
|
/// incremental state changes.
|
|
pub fn take_delta_rx(&self) -> Option<mpsc::Receiver<ProtocolDelta>> {
|
|
self.delta_rx.lock().ok().and_then(|mut g| g.take())
|
|
}
|
|
|
|
/// Send a text message to the specified target.
|
|
pub async fn send_text_message(
|
|
&self,
|
|
message: String,
|
|
target: MessageTarget,
|
|
) -> Result<(), ProtocolError> {
|
|
let (tx, rx) = oneshot::channel();
|
|
self.tx
|
|
.send(Request::SendTextMessage {
|
|
message,
|
|
target,
|
|
reply: tx,
|
|
})
|
|
.await
|
|
.map_err(|_| ProtocolError::Lost("connection task is gone".to_string()))?;
|
|
rx.await
|
|
.map_err(|_| ProtocolError::Lost("send_text_message reply dropped".to_string()))?
|
|
}
|
|
}
|
|
|
|
async fn connection_task(
|
|
cfg: ConnectConfig,
|
|
mut rx: mpsc::Receiver<Request>,
|
|
mut voice_out_rx: mpsc::Receiver<OutPacket>,
|
|
channels: EventChannels,
|
|
ready_tx: oneshot::Sender<Result<(), ProtocolError>>,
|
|
lost_tx: oneshot::Sender<DisconnectReason>,
|
|
) {
|
|
// Box the lost_tx so each exit branch can move it.
|
|
let mut lost_tx = Some(lost_tx);
|
|
// Macro: report the disconnect reason and return from the task.
|
|
macro_rules! exit {
|
|
($reason:expr) => {{
|
|
if let Some(tx) = lost_tx.take() {
|
|
let _ = tx.send($reason);
|
|
}
|
|
return;
|
|
}};
|
|
}
|
|
macro_rules! fail_ready {
|
|
($err:expr) => {{
|
|
let err = $err;
|
|
let reason = DisconnectReason::Error(format!("{err}"));
|
|
let _ = ready_tx.send(Err(err));
|
|
exit!(reason);
|
|
}};
|
|
($err:expr, $reason:expr) => {{
|
|
let _ = ready_tx.send(Err($err));
|
|
exit!($reason);
|
|
}};
|
|
}
|
|
|
|
let resolved = match server_socket_from_config(&cfg) {
|
|
Some(addr) => {
|
|
info!(
|
|
target: "chanora_protocol",
|
|
input = %cfg.address,
|
|
resolved = %addr,
|
|
"using prefetched server address"
|
|
);
|
|
addr
|
|
}
|
|
None => {
|
|
let addr = match resolve_server_socket(&cfg.address).await {
|
|
Ok(addr) => addr,
|
|
Err(err) => fail_ready!(err),
|
|
};
|
|
info!(
|
|
target: "chanora_protocol",
|
|
input = %cfg.address,
|
|
resolved = %addr,
|
|
"server address resolved"
|
|
);
|
|
addr
|
|
}
|
|
};
|
|
|
|
// Pass the resolved SocketAddr directly to tsclientlib so it
|
|
// skips its own resolver entirely (tsclientlib accepts
|
|
// SocketAddr via the From<SocketAddr> for ServerAddress impl).
|
|
let client_version = pick_client_version();
|
|
info!(
|
|
target: "chanora_protocol",
|
|
client_version = %client_version,
|
|
"selected TS3 client_version for this platform"
|
|
);
|
|
let mut builder = Connection::build(resolved)
|
|
.name(cfg.nickname.clone())
|
|
.version(client_version);
|
|
|
|
let identity = match cfg.identity.as_deref() {
|
|
Some(value) => match Identity::new_from_str(value) {
|
|
Ok(identity) => identity,
|
|
Err(err) => {
|
|
let msg = err.to_string();
|
|
fail_ready!(
|
|
ProtocolError::Identity(msg.clone()),
|
|
DisconnectReason::Error(format!("identity: {msg}"))
|
|
);
|
|
}
|
|
},
|
|
None => Identity::create(),
|
|
};
|
|
builder = builder.identity(identity);
|
|
|
|
if let Some(pw) = &cfg.password {
|
|
builder = builder.password(pw.clone());
|
|
}
|
|
|
|
let mut con = match builder.connect() {
|
|
Ok(c) => c,
|
|
Err(e) => {
|
|
let msg = format!("{e}");
|
|
fail_ready!(
|
|
ProtocolError::Connect(msg.clone()),
|
|
DisconnectReason::Error(format!("connect: {msg}"))
|
|
);
|
|
}
|
|
};
|
|
|
|
// Wait for the first BookEvents indicating the state snapshot is ready.
|
|
let first = con
|
|
.events()
|
|
.try_filter(|e| future::ready(matches!(e, StreamItem::BookEvents(_))))
|
|
.next()
|
|
.await;
|
|
match first {
|
|
Some(Ok(_)) => {
|
|
info!(target: "chanora_protocol", "initial state snapshot received");
|
|
}
|
|
Some(Err(e)) => {
|
|
let msg = format!("{e}");
|
|
fail_ready!(
|
|
ProtocolError::DisconnectedEarly(msg.clone()),
|
|
DisconnectReason::Error(format!("disconnected early: {msg}"))
|
|
);
|
|
}
|
|
None => {
|
|
let msg = "event stream ended before snapshot".to_string();
|
|
fail_ready!(
|
|
ProtocolError::DisconnectedEarly(msg.clone()),
|
|
DisconnectReason::Error(msg)
|
|
);
|
|
}
|
|
}
|
|
|
|
// Subscribe to the full server tree so snapshot() returns more than just our channel.
|
|
if let Ok(state) = con.get_state() {
|
|
if let Err(e) = state.server.set_subscribed(true).send(&mut con) {
|
|
warn!(target: "chanora_protocol", error = %e, "could not subscribe to server tree");
|
|
}
|
|
}
|
|
|
|
// Settle: pump events for ~2 s so the subscribed tree arrives
|
|
// before the first snapshot. The upstream packet codec emits
|
|
// out-of-order command-packet warnings here; they are harmless
|
|
// and the final state still converges.
|
|
let settle_until = std::time::Instant::now() + Duration::from_secs(2);
|
|
while std::time::Instant::now() < settle_until {
|
|
let ev = tokio::time::timeout(Duration::from_millis(100), con.events().next()).await;
|
|
match ev {
|
|
Ok(Some(Ok(_))) => continue,
|
|
Ok(Some(Err(e))) => {
|
|
warn!(target: "chanora_protocol", error = %e, "event error during settle");
|
|
}
|
|
Ok(None) => {
|
|
let msg = "stream closed during settle".to_string();
|
|
fail_ready!(
|
|
ProtocolError::DisconnectedEarly(msg),
|
|
DisconnectReason::StreamEnded
|
|
);
|
|
}
|
|
Err(_) => { /* no event available right now; keep waiting */ }
|
|
}
|
|
}
|
|
|
|
let _ = ready_tx.send(Ok(()));
|
|
|
|
// Pending `client_move` requests: each one is keyed by the
|
|
// `MessageHandle` tsclientlib returns from `send_with_result`.
|
|
// When the corresponding `StreamItem::MessageResult` arrives we
|
|
// resolve the oneshot back to the caller. Entries also carry a
|
|
// deadline so a server that never replies doesn't leak the
|
|
// reply channel — at most 3 s of pending state per move.
|
|
let mut pending_moves: PendingMoves = HashMap::new();
|
|
let mut voice_activity: HashMap<u64, Instant> = HashMap::new();
|
|
let mut poke_limiter = PokeLimiter::new();
|
|
|
|
// Main loop: pump events, service requests, forward voice.
|
|
loop {
|
|
// 1. Send a bounded batch of outbound voice packets first — they're
|
|
// time-sensitive, but control requests must still make progress.
|
|
drain_voice_packets_for_tick(&mut voice_out_rx, OUTBOUND_VOICE_PACKETS_PER_TICK, |pkt| {
|
|
if let Err(e) = con.send_audio(pkt) {
|
|
warn!(target: "chanora_protocol", error = %e, "send_audio failed");
|
|
}
|
|
Ok::<(), ()>(())
|
|
});
|
|
|
|
// 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(20), ev_stream.next()).await
|
|
};
|
|
match pump.await {
|
|
Ok(Some(Ok(item))) => match item {
|
|
StreamItem::Audio(buf) => {
|
|
handle_audio_stream_item(&channels.voice_in, &mut voice_activity, buf).await;
|
|
}
|
|
other => handle_non_audio_stream_item(
|
|
&con,
|
|
other,
|
|
&channels.chat,
|
|
&channels.activity,
|
|
&channels.delta,
|
|
&mut pending_moves,
|
|
&mut poke_limiter,
|
|
),
|
|
},
|
|
Ok(Some(Err(e))) => {
|
|
warn!(target: "chanora_protocol", error = %e, "event error");
|
|
// Some errors are transient; treat persistent ones
|
|
// as a loss after the next iteration.
|
|
}
|
|
Ok(None) => {
|
|
warn!(target: "chanora_protocol", "event stream ended");
|
|
exit!(DisconnectReason::StreamEnded);
|
|
}
|
|
Err(_) => { /* no event in 20 ms */ }
|
|
}
|
|
|
|
// 2b. Sweep stale pending_moves whose deadline has passed.
|
|
// The server should always reply within ~1 s; 3 s is a
|
|
// generous ceiling. Expired entries fall back to Ok() so
|
|
// the caller's snapshot-confirmation polling still has a
|
|
// chance to detect success — better than a fake
|
|
// ServerRejected for legacy servers that never reply to
|
|
// move requests.
|
|
if !pending_moves.is_empty() {
|
|
let now = std::time::Instant::now();
|
|
let expired: Vec<MessageHandle> = pending_moves
|
|
.iter()
|
|
.filter_map(|(handle, (_, _, deadline))| {
|
|
if now >= *deadline {
|
|
Some(*handle)
|
|
} else {
|
|
None
|
|
}
|
|
})
|
|
.collect();
|
|
for handle in expired {
|
|
if let Some((_target_channel, Some(reply), _)) = pending_moves.remove(&handle) {
|
|
let _ = reply.send(Ok(()));
|
|
}
|
|
}
|
|
}
|
|
|
|
// 3. Service at most one control request (non-blocking).
|
|
match rx.try_recv() {
|
|
Ok(Request::Snapshot(reply)) => {
|
|
let snap = build_snapshot(&con, &voice_activity);
|
|
let _ = reply.send(snap);
|
|
}
|
|
Ok(Request::MoveToChannel {
|
|
channel_id,
|
|
password,
|
|
reply,
|
|
}) => match move_self_to(&mut con, channel_id, password.as_deref()) {
|
|
Ok(handle) => {
|
|
let deadline = std::time::Instant::now() + Duration::from_secs(3);
|
|
pending_moves.insert(handle, (channel_id, Some(reply), deadline));
|
|
}
|
|
Err(e) => {
|
|
let _ = reply.send(Err(e));
|
|
}
|
|
},
|
|
Ok(Request::MoveToChannelNoWait {
|
|
channel_id,
|
|
password,
|
|
}) => match move_self_to(&mut con, channel_id, password.as_deref()) {
|
|
Ok(handle) => {
|
|
let deadline = std::time::Instant::now() + Duration::from_secs(3);
|
|
pending_moves.insert(handle, (channel_id, None, deadline));
|
|
}
|
|
Err(e) => {
|
|
warn!(
|
|
target: "chanora_protocol",
|
|
error = %e,
|
|
channel_id,
|
|
"fire-and-forget client_move could not be queued"
|
|
);
|
|
}
|
|
},
|
|
Ok(Request::SetMuted {
|
|
input,
|
|
output,
|
|
reply,
|
|
}) => {
|
|
let r = set_self_muted(&mut con, input, output);
|
|
let _ = reply.send(r);
|
|
}
|
|
Ok(Request::SendTextMessage {
|
|
message,
|
|
target,
|
|
reply,
|
|
}) => {
|
|
let r = send_text_message(&mut con, &message, target);
|
|
let _ = reply.send(r);
|
|
}
|
|
Ok(Request::FetchClientProfile { client_id, reply }) => {
|
|
let r = fetch_client_profile(
|
|
&mut con,
|
|
client_id,
|
|
&channels,
|
|
&mut pending_moves,
|
|
&mut voice_activity,
|
|
&mut poke_limiter,
|
|
)
|
|
.await;
|
|
let _ = reply.send(r);
|
|
}
|
|
Ok(Request::Disconnect(reply)) => {
|
|
let _ = con.disconnect(DisconnectOptions::new());
|
|
bounded_drain_stream(con.events(), DISCONNECT_EVENT_DRAIN_TIMEOUT).await;
|
|
let _ = reply.send(());
|
|
info!(target: "chanora_protocol", "clean disconnect");
|
|
exit!(DisconnectReason::UserRequested);
|
|
}
|
|
Err(mpsc::error::TryRecvError::Empty) => {}
|
|
Err(mpsc::error::TryRecvError::Disconnected) => {
|
|
let _ = con.disconnect(DisconnectOptions::new());
|
|
bounded_drain_stream(con.events(), DISCONNECT_EVENT_DRAIN_TIMEOUT).await;
|
|
info!(target: "chanora_protocol", "handle dropped; implicit disconnect");
|
|
exit!(DisconnectReason::UserRequested);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn handle_audio_stream_item(
|
|
voice_in_tx: &mpsc::Sender<InboundVoice>,
|
|
voice_activity: &mut HashMap<u64, Instant>,
|
|
buf: InAudioBuf,
|
|
) {
|
|
let from = packet_sender_id(&buf);
|
|
if let Some(from) = from {
|
|
voice_activity.insert(from, Instant::now());
|
|
let inbound = InboundVoice {
|
|
from_client: from,
|
|
packet: buf,
|
|
};
|
|
match send_with_timeout(voice_in_tx, inbound, INBOUND_VOICE_SEND_TIMEOUT).await {
|
|
Ok(()) => {}
|
|
Err(SendTimeoutError::Timeout(_)) => {
|
|
warn!(
|
|
target: "chanora_protocol",
|
|
from_client = from,
|
|
timeout_ms = INBOUND_VOICE_SEND_TIMEOUT.as_millis() as u64,
|
|
"inbound voice queue stayed full; dropping packet"
|
|
);
|
|
}
|
|
Err(SendTimeoutError::Closed(_)) => {
|
|
warn!(
|
|
target: "chanora_protocol",
|
|
from_client = from,
|
|
"inbound voice consumer closed; dropping packet"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
fn handle_non_audio_stream_item(
|
|
con: &Connection,
|
|
item: StreamItem,
|
|
chat_tx: &mpsc::Sender<ChatMessage>,
|
|
activity_tx: &mpsc::Sender<ServerActivity>,
|
|
delta_tx: &mpsc::Sender<ProtocolDelta>,
|
|
pending_moves: &mut PendingMoves,
|
|
poke_limiter: &mut PokeLimiter,
|
|
) {
|
|
match item {
|
|
StreamItem::BookEvents(events) => {
|
|
for ev in events {
|
|
if let tsclientlib::events::Event::PropertyChanged {
|
|
id: ts_bookkeeping::events::PropertyId::ClientChannel(client_id),
|
|
..
|
|
} = &ev
|
|
{
|
|
let own_client = con.get_state().ok().map(|state| state.own_client);
|
|
if let Ok(state) = con.get_state() {
|
|
if let Some(client) = state.clients.get(client_id) {
|
|
let _ = delta_tx.try_send(ProtocolDelta::ClientMoved {
|
|
client_id: client_id.0 as u64,
|
|
new_channel_id: client.channel.0,
|
|
});
|
|
}
|
|
}
|
|
if own_client == Some(*client_id) {
|
|
let current_channel = con.get_state().ok().and_then(|state| {
|
|
state.clients.get(client_id).map(|client| client.channel.0)
|
|
});
|
|
if let Some(current_channel) = current_channel {
|
|
let matched: Vec<MessageHandle> = pending_moves
|
|
.iter()
|
|
.filter_map(|(handle, (target_channel, _, _))| {
|
|
if *target_channel == current_channel {
|
|
Some(*handle)
|
|
} else {
|
|
None
|
|
}
|
|
})
|
|
.collect();
|
|
for handle in matched {
|
|
if let Some((_, reply, _)) = pending_moves.remove(&handle) {
|
|
info!(
|
|
target: "chanora_protocol",
|
|
channel_id = current_channel,
|
|
"client_move resolved by authoritative self channel change"
|
|
);
|
|
if let Some(reply) = reply {
|
|
let _ = reply.send(Ok(()));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
if let Some(activity) = format_server_activity(con, &ev) {
|
|
let _ = activity_tx.try_send(ServerActivity { message: activity });
|
|
}
|
|
|
|
forward_delta(con, &ev, delta_tx);
|
|
|
|
if let tsclientlib::events::Event::Message {
|
|
target,
|
|
invoker,
|
|
message,
|
|
} = ev
|
|
{
|
|
let (mapped, poke_strength) = match target {
|
|
tsclientlib::MessageTarget::Server => (MessageTarget::Server, None),
|
|
tsclientlib::MessageTarget::Channel => (MessageTarget::Channel, None),
|
|
tsclientlib::MessageTarget::Client(id) => {
|
|
(MessageTarget::Client(id.0 as u64), None)
|
|
}
|
|
tsclientlib::MessageTarget::Poke(id) => {
|
|
let own_client_id =
|
|
con.get_state().ok().map(|state| state.own_client.0 as u64);
|
|
let strength = poke_limiter.record(invoker.id.0 as u64, own_client_id);
|
|
(MessageTarget::Poke(id.0 as u64), Some(strength))
|
|
}
|
|
};
|
|
let _ = chat_tx.try_send(ChatMessage {
|
|
sender_id: ClientId(invoker.id.0 as u64),
|
|
sender_name: sanitize(&invoker.name),
|
|
message: sanitize(&message),
|
|
target: mapped,
|
|
poke_strength,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
StreamItem::MessageResult(handle, result) => {
|
|
if let Some((_target_channel, reply, _deadline)) = pending_moves.remove(&handle) {
|
|
let mapped = match result {
|
|
Ok(()) => Ok(()),
|
|
Err(cmd_err) => {
|
|
let code = cmd_err.error as u32;
|
|
let message = cmd_err.error.to_string();
|
|
info!(
|
|
target: "chanora_protocol",
|
|
code,
|
|
message = %message,
|
|
"server rejected client_move"
|
|
);
|
|
Err(ProtocolError::ServerRejected { code, message })
|
|
}
|
|
};
|
|
if let Some(reply) = reply {
|
|
let _ = reply.send(mapped);
|
|
} else if let Err(err) = mapped {
|
|
info!(
|
|
target: "chanora_protocol",
|
|
error = %err,
|
|
"client_move completed in background with error"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
StreamItem::Audio(_) => unreachable!("audio handled separately"),
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
async fn resolve_server_socket(address: &str) -> Result<SocketAddr, ProtocolError> {
|
|
let resolver = ChanoraResolver::new().map_err(|err| ProtocolError::DnsFailed {
|
|
host: address.to_string(),
|
|
reason: format!("resolver initialization failed: {err}"),
|
|
})?;
|
|
let resolved_address = resolver
|
|
.resolve_client_address(address)
|
|
.await
|
|
.map_err(|err| ProtocolError::DnsFailed {
|
|
host: address.to_string(),
|
|
reason: err.to_string(),
|
|
})?;
|
|
resolved_address
|
|
.parse::<SocketAddr>()
|
|
.map_err(|err| ProtocolError::DnsFailed {
|
|
host: address.to_string(),
|
|
reason: format!("resolver returned invalid socket address '{resolved_address}': {err}"),
|
|
})
|
|
}
|
|
|
|
fn server_socket_from_config(cfg: &ConnectConfig) -> Option<SocketAddr> {
|
|
cfg.resolved_address
|
|
}
|
|
|
|
/// Move our own client into `channel_id` with an optional password.
|
|
/// Looks up our `own_client` in the current state and dispatches the
|
|
/// generated `client_move` command via `send_with_result`. The
|
|
/// returned `MessageHandle` is correlated by the connection loop
|
|
/// against the next `StreamItem::MessageResult` so we can surface
|
|
/// typed `ServerRejected` errors (no permission, wrong password,
|
|
/// channel full, etc.) per the TS3 error catalogue.
|
|
fn move_self_to(
|
|
con: &mut Connection,
|
|
channel_id: u64,
|
|
password: Option<&str>,
|
|
) -> Result<MessageHandle, ProtocolError> {
|
|
let state = con
|
|
.get_state()
|
|
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
|
|
let own_id = state.own_client;
|
|
let own_client = state
|
|
.clients
|
|
.get(&own_id)
|
|
.ok_or_else(|| ProtocolError::Backend("own_client not in state".to_string()))?;
|
|
let target = TsChannelId(channel_id);
|
|
let mut part = own_client.client_move(target);
|
|
if let Some(pw) = password {
|
|
part = part.set_password(pw);
|
|
}
|
|
let handle = part
|
|
.send_with_result(con)
|
|
.map_err(|e| ProtocolError::Backend(format!("client_move send: {e}")))?;
|
|
info!(target: "chanora_protocol", channel_id, "client_move sent");
|
|
Ok(handle)
|
|
}
|
|
|
|
/// Send a `clientupdate` with the requested mute fields set. `None`
|
|
/// fields are omitted so callers can toggle just one flag.
|
|
fn set_self_muted(
|
|
con: &mut Connection,
|
|
input: Option<bool>,
|
|
output: Option<bool>,
|
|
) -> Result<(), ProtocolError> {
|
|
if input.is_none() && output.is_none() {
|
|
return Ok(());
|
|
}
|
|
let part = {
|
|
let state = con
|
|
.get_state()
|
|
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
|
|
let mut p = state.client_update();
|
|
if let Some(v) = input {
|
|
p = p.set_input_muted(v);
|
|
}
|
|
if let Some(v) = output {
|
|
p = p.set_output_muted(v);
|
|
}
|
|
p
|
|
};
|
|
part.send(con)
|
|
.map_err(|e| ProtocolError::Backend(format!("client_update send: {e}")))?;
|
|
info!(target: "chanora_protocol", ?input, ?output, "client_update sent");
|
|
Ok(())
|
|
}
|
|
|
|
fn send_text_message(
|
|
con: &mut Connection,
|
|
message: &str,
|
|
target: MessageTarget,
|
|
) -> Result<(), ProtocolError> {
|
|
use tsproto_types::TextMessageTargetMode;
|
|
match target {
|
|
MessageTarget::Server => {
|
|
send_text_to_mode(con, message, TextMessageTargetMode::Server, "server")?;
|
|
}
|
|
MessageTarget::Channel => {
|
|
// Fix: previously channel messages were sent via
|
|
// state.server.send_textmessage() which always uses
|
|
// TextMessageTargetMode::Server. Now correctly uses
|
|
// TextMessageTargetMode::Channel so the message is
|
|
// scoped to the current channel, not server-wide.
|
|
send_text_to_mode(con, message, TextMessageTargetMode::Channel, "channel")?;
|
|
}
|
|
MessageTarget::Client(client_id) => {
|
|
let state = con
|
|
.get_state()
|
|
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
|
|
let client = find_client_by_id(state.clients.values(), client_id)?;
|
|
client
|
|
.send_textmessage(message)
|
|
.send(con)
|
|
.map_err(|e| ProtocolError::Backend(format!("send_textmessage(client): {e}")))?;
|
|
}
|
|
MessageTarget::Poke(client_id) => {
|
|
let state = con
|
|
.get_state()
|
|
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
|
|
let client = find_client_by_id(state.clients.values(), client_id)?;
|
|
client
|
|
.poke(message)
|
|
.send(con)
|
|
.map_err(|e| ProtocolError::Backend(format!("poke: {e}")))?;
|
|
}
|
|
}
|
|
info!(target: "chanora_protocol", len = message.len(), ?target, "text message sent");
|
|
Ok(())
|
|
}
|
|
|
|
fn send_text_to_mode(
|
|
con: &mut Connection,
|
|
message: &str,
|
|
target: tsproto_types::TextMessageTargetMode,
|
|
label: &str,
|
|
) -> Result<(), ProtocolError> {
|
|
use ts_bookkeeping::messages::c2s;
|
|
|
|
c2s::OutSendTextMessageMessage::new(&mut std::iter::once(c2s::OutSendTextMessagePart {
|
|
target,
|
|
target_client_id: None,
|
|
message: message.into(),
|
|
}))
|
|
.send(con)
|
|
.map_err(|e| ProtocolError::Backend(format!("send_textmessage({label}): {e}")))
|
|
}
|
|
|
|
async fn fetch_client_profile(
|
|
con: &mut Connection,
|
|
client_id: u64,
|
|
channels: &EventChannels,
|
|
pending_moves: &mut PendingMoves,
|
|
voice_activity: &mut HashMap<u64, Instant>,
|
|
poke_limiter: &mut PokeLimiter,
|
|
) -> Result<ClientProfile, ProtocolError> {
|
|
let target_id = TsClientId(client_id as u16);
|
|
|
|
let (
|
|
database_id,
|
|
uid_b64,
|
|
has_optional,
|
|
has_connection,
|
|
is_own,
|
|
needs_server_groups,
|
|
needs_channel_groups,
|
|
) = {
|
|
let state = con
|
|
.get_state()
|
|
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
|
|
let client = state
|
|
.clients
|
|
.get(&target_id)
|
|
.ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))?;
|
|
(
|
|
client.database_id,
|
|
client.uid.as_ref().map(|uid| uid_to_b64(uid.as_ref())),
|
|
client.optional_data.is_some(),
|
|
client.connection_data.is_some(),
|
|
state.own_client == target_id,
|
|
state.server_groups.is_empty(),
|
|
state.channel_groups.is_empty(),
|
|
)
|
|
};
|
|
|
|
let refresh_plan = client_profile_refresh_plan(
|
|
is_own,
|
|
has_optional,
|
|
has_connection,
|
|
needs_server_groups,
|
|
needs_channel_groups,
|
|
);
|
|
|
|
if refresh_plan.needs_server_groups {
|
|
let _ = request_messages(
|
|
con,
|
|
build_command("servergrouplist", &[], &[]),
|
|
channels,
|
|
pending_moves,
|
|
voice_activity,
|
|
poke_limiter,
|
|
)
|
|
.await;
|
|
}
|
|
if refresh_plan.needs_channel_groups {
|
|
let _ = request_messages(
|
|
con,
|
|
build_command("channelgrouplist", &[], &[]),
|
|
channels,
|
|
pending_moves,
|
|
voice_activity,
|
|
poke_limiter,
|
|
)
|
|
.await;
|
|
}
|
|
if refresh_plan.needs_client_variables {
|
|
if let Err(e) = request_messages(
|
|
con,
|
|
build_command(
|
|
"clientgetvariables",
|
|
&[("clid", client_id.to_string())],
|
|
&[],
|
|
),
|
|
channels,
|
|
pending_moves,
|
|
voice_activity,
|
|
poke_limiter,
|
|
)
|
|
.await
|
|
{
|
|
warn!(
|
|
target: "chanora_protocol",
|
|
client_id,
|
|
error = %e,
|
|
"clientgetvariables failed"
|
|
);
|
|
}
|
|
}
|
|
if refresh_plan.needs_connection_info {
|
|
if let Err(e) = request_messages(
|
|
con,
|
|
build_command("getconnectioninfo", &[("clid", client_id.to_string())], &[]),
|
|
channels,
|
|
pending_moves,
|
|
voice_activity,
|
|
poke_limiter,
|
|
)
|
|
.await
|
|
{
|
|
warn!(
|
|
target: "chanora_protocol",
|
|
client_id,
|
|
error = %e,
|
|
"getconnectioninfo failed (ping/packet_loss will be unavailable)"
|
|
);
|
|
}
|
|
}
|
|
|
|
let db_info = if refresh_plan.needs_client_db_info {
|
|
request_client_db_info(
|
|
con,
|
|
database_id,
|
|
channels,
|
|
pending_moves,
|
|
voice_activity,
|
|
poke_limiter,
|
|
)
|
|
.await
|
|
.ok()
|
|
} else {
|
|
None
|
|
};
|
|
|
|
let state = con
|
|
.get_state()
|
|
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
|
|
let client = state
|
|
.clients
|
|
.get(&target_id)
|
|
.ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))?;
|
|
|
|
let optional = client.optional_data.as_ref();
|
|
let connection = client.connection_data.as_ref();
|
|
let net_stats: Option<&ConnectionStats> = if is_own {
|
|
con.get_network_stats().ok()
|
|
} else {
|
|
None
|
|
};
|
|
let server_group_names: HashMap<_, _> = state
|
|
.server_groups
|
|
.iter()
|
|
.map(|(id, group)| (*id, group.name.clone()))
|
|
.collect();
|
|
let channel_group_names: HashMap<_, _> = state
|
|
.channel_groups
|
|
.iter()
|
|
.map(|(id, group)| (*id, group.name.clone()))
|
|
.collect();
|
|
|
|
let unique_id = db_info
|
|
.as_ref()
|
|
.map(|info| uid_to_b64(info.uid.as_ref()))
|
|
.or(uid_b64)
|
|
.unwrap_or_default();
|
|
let avatar_path = if client.avatar_hash.is_empty() || unique_id.is_empty() {
|
|
String::new()
|
|
} else {
|
|
format!("/avatar_{}", uid_to_avatar_path(&unique_id))
|
|
};
|
|
|
|
Ok(ClientProfile {
|
|
id: ClientId(client.id.0 as u64),
|
|
channel: ChannelId(client.channel.0),
|
|
name: db_info
|
|
.as_ref()
|
|
.map(|info| sanitize(&info.name))
|
|
.unwrap_or_else(|| sanitize(&client.name)),
|
|
unique_id,
|
|
database_id: Some(client.database_id.0),
|
|
country_code: client.country_code.clone(),
|
|
description: db_info
|
|
.as_ref()
|
|
.map(|info| sanitize(&info.description))
|
|
.unwrap_or_else(|| sanitize(&client.description)),
|
|
version: optional
|
|
.map(|info| info.version.clone())
|
|
.unwrap_or_default(),
|
|
platform: optional
|
|
.map(|info| info.platform.clone())
|
|
.unwrap_or_default(),
|
|
created_unix_seconds: optional
|
|
.map(|info| info.created.unix_timestamp())
|
|
.or_else(|| db_info.as_ref().map(|info| info.created.unix_timestamp())),
|
|
last_connected_unix_seconds: optional
|
|
.map(|info| info.last_connected.unix_timestamp())
|
|
.or_else(|| {
|
|
db_info
|
|
.as_ref()
|
|
.map(|info| info.last_connected.unix_timestamp())
|
|
}),
|
|
connections_total: optional
|
|
.map(|info| u64::from(info.connections_total))
|
|
.or_else(|| {
|
|
db_info
|
|
.as_ref()
|
|
.map(|info| u64::from(info.connections_total))
|
|
}),
|
|
online_seconds: connection
|
|
.and_then(|info| info.connected_time.map(|duration| duration.whole_seconds())),
|
|
idle_milliseconds: connection.map(|info| duration_millis(info.idle_time)),
|
|
ping_milliseconds: net_stats
|
|
.and_then(|s| std_duration_millis(s.rtt))
|
|
.or_else(|| connection.and_then(|info| info.ping.map(duration_millis))),
|
|
ping_deviation_milliseconds: net_stats
|
|
.and_then(|s| std_duration_millis(s.rtt_dev))
|
|
.or_else(|| connection.and_then(|info| info.ping_deviation.map(duration_millis))),
|
|
client_address: connection
|
|
.and_then(|info| info.client_address.map(|address| address.to_string()))
|
|
.unwrap_or_default(),
|
|
server_groups: format_server_groups(client.server_groups.iter(), &server_group_names),
|
|
channel_group: channel_group_names
|
|
.get(&client.channel_group)
|
|
.cloned()
|
|
.unwrap_or_else(|| format!("Unknown ({})", client.channel_group.0)),
|
|
avatar_path,
|
|
bytes_downloaded_month: optional
|
|
.map(|info| info.bytes_downloaded_month)
|
|
.or_else(|| db_info.as_ref().map(|info| info.bytes_downloaded_month)),
|
|
bytes_uploaded_month: optional
|
|
.map(|info| info.bytes_uploaded_month)
|
|
.or_else(|| db_info.as_ref().map(|info| info.bytes_uploaded_month)),
|
|
bytes_downloaded_total: optional
|
|
.map(|info| info.bytes_downloaded_total)
|
|
.or_else(|| db_info.as_ref().map(|info| info.bytes_downloaded_total)),
|
|
bytes_uploaded_total: optional
|
|
.map(|info| info.bytes_uploaded_total)
|
|
.or_else(|| db_info.as_ref().map(|info| info.bytes_uploaded_total)),
|
|
packet_loss_client_to_server_total: net_stats
|
|
.map(|s| s.get_packetloss())
|
|
.or_else(|| connection.map(|info| info.client_to_server_packetloss_total)),
|
|
packet_loss_server_to_client_total: net_stats
|
|
.map(|s| s.get_packetloss_s2c_total())
|
|
.or_else(|| connection.and_then(|info| info.server_to_client_packetloss_total)),
|
|
})
|
|
}
|
|
|
|
fn build_command(name: &str, args: &[(&str, String)], flags: &[&str]) -> OutCommand {
|
|
let mut command = OutCommand::new(Direction::C2S, Flags::empty(), PacketType::Command, name);
|
|
for flag in flags {
|
|
command.write_arg(flag, &"");
|
|
}
|
|
for (key, value) in args {
|
|
command.write_arg(key, value);
|
|
}
|
|
command
|
|
}
|
|
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
struct ClientProfileRefreshPlan {
|
|
needs_server_groups: bool,
|
|
needs_channel_groups: bool,
|
|
needs_client_variables: bool,
|
|
needs_connection_info: bool,
|
|
needs_client_db_info: bool,
|
|
}
|
|
|
|
fn client_profile_refresh_plan(
|
|
is_own: bool,
|
|
has_optional: bool,
|
|
has_connection: bool,
|
|
needs_server_groups: bool,
|
|
needs_channel_groups: bool,
|
|
) -> ClientProfileRefreshPlan {
|
|
ClientProfileRefreshPlan {
|
|
needs_server_groups,
|
|
needs_channel_groups,
|
|
needs_client_variables: !has_optional || !is_own,
|
|
needs_connection_info: !has_connection || !is_own,
|
|
needs_client_db_info: !has_optional || !is_own,
|
|
}
|
|
}
|
|
|
|
async fn request_messages(
|
|
con: &mut Connection,
|
|
command: OutCommand,
|
|
channels: &EventChannels,
|
|
pending_moves: &mut PendingMoves,
|
|
voice_activity: &mut HashMap<u64, Instant>,
|
|
poke_limiter: &mut PokeLimiter,
|
|
) -> Result<Vec<InMessage>, ProtocolError> {
|
|
let handle = command
|
|
.send_with_result(con)
|
|
.map_err(|e| ProtocolError::Backend(format!("send command: {e}")))?;
|
|
let mut messages = Vec::new();
|
|
let deadline = Instant::now() + PROFILE_REFRESH_RESULT_TIMEOUT;
|
|
loop {
|
|
let remaining = deadline.saturating_duration_since(Instant::now());
|
|
if remaining.is_zero() {
|
|
return Err(ProtocolError::Backend(
|
|
"profile refresh command timed out".to_string(),
|
|
));
|
|
}
|
|
let item = {
|
|
let mut stream = con.events();
|
|
tokio::time::timeout(remaining, stream.next())
|
|
.await
|
|
.map_err(|_| {
|
|
ProtocolError::Backend("profile refresh command timed out".to_string())
|
|
})?
|
|
.ok_or_else(|| ProtocolError::Lost("event stream ended".to_string()))?
|
|
.map_err(|e| ProtocolError::Lost(e.to_string()))?
|
|
};
|
|
match item {
|
|
StreamItem::MessageEvent(message) => messages.push(message),
|
|
StreamItem::MessageResult(reply, status) if reply == handle => {
|
|
status.map_err(|error| ProtocolError::ServerRejected {
|
|
code: error.error as u32,
|
|
message: error.error.to_string(),
|
|
})?;
|
|
return Ok(messages);
|
|
}
|
|
StreamItem::Audio(buf) => {
|
|
handle_audio_stream_item(&channels.voice_in, voice_activity, buf).await;
|
|
}
|
|
other => handle_non_audio_stream_item(
|
|
con,
|
|
other,
|
|
&channels.chat,
|
|
&channels.activity,
|
|
&channels.delta,
|
|
pending_moves,
|
|
poke_limiter,
|
|
),
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn request_client_db_info(
|
|
con: &mut Connection,
|
|
dbid: tsclientlib::ClientDbId,
|
|
channels: &EventChannels,
|
|
pending_moves: &mut PendingMoves,
|
|
voice_activity: &mut HashMap<u64, Instant>,
|
|
poke_limiter: &mut PokeLimiter,
|
|
) -> Result<InClientDbInfoPart, ProtocolError> {
|
|
let messages = request_messages(
|
|
con,
|
|
build_command("clientdbinfo", &[("cldbid", dbid.0.to_string())], &[]),
|
|
channels,
|
|
pending_moves,
|
|
voice_activity,
|
|
poke_limiter,
|
|
)
|
|
.await?;
|
|
for message in messages {
|
|
if let InMessage::ClientDbInfo(info) = message {
|
|
if let Some(row) = info.iter().next() {
|
|
return Ok(row.clone());
|
|
}
|
|
}
|
|
}
|
|
Err(ProtocolError::Backend(
|
|
"clientdbinfo returned no row".to_string(),
|
|
))
|
|
}
|
|
|
|
fn format_server_groups<'a>(
|
|
groups: impl IntoIterator<Item = &'a tsclientlib::ServerGroupId>,
|
|
names: &HashMap<tsclientlib::ServerGroupId, String>,
|
|
) -> Vec<String> {
|
|
let mut rendered = groups
|
|
.into_iter()
|
|
.map(|group| {
|
|
names
|
|
.get(group)
|
|
.cloned()
|
|
.unwrap_or_else(|| format!("Unknown ({})", group.0))
|
|
})
|
|
.collect::<Vec<_>>();
|
|
rendered.sort();
|
|
rendered
|
|
}
|
|
|
|
fn duration_millis(duration: time::Duration) -> i64 {
|
|
duration
|
|
.whole_milliseconds()
|
|
.clamp(i64::MIN as i128, i64::MAX as i128) as i64
|
|
}
|
|
|
|
fn std_duration_millis(duration: std::time::Duration) -> Option<i64> {
|
|
duration.as_millis().try_into().ok()
|
|
}
|
|
|
|
fn uid_to_b64(uid: &tsclientlib::Uid) -> String {
|
|
BASE64_STANDARD.encode(&uid.0)
|
|
}
|
|
|
|
fn uid_to_avatar_path(uid_b64: &str) -> String {
|
|
let decoded = BASE64_STANDARD.decode(uid_b64).unwrap_or_default();
|
|
let mut rendered = String::with_capacity(decoded.len() * 2);
|
|
for byte in decoded {
|
|
rendered.push((b'a' + (byte >> 4)) as char);
|
|
rendered.push((b'a' + (byte & 0x0f)) as char);
|
|
}
|
|
rendered
|
|
}
|
|
|
|
fn find_client_by_id<'a>(
|
|
clients: impl IntoIterator<Item = &'a Client>,
|
|
client_id: u64,
|
|
) -> Result<&'a Client, ProtocolError> {
|
|
clients
|
|
.into_iter()
|
|
.find(|client| client.id.0 as u64 == client_id)
|
|
.ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))
|
|
}
|
|
|
|
/// 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,
|
|
}
|
|
}
|
|
|
|
/// Sort TeamSpeak channels by the linked-list ordering carried in
|
|
/// each channel's `order` field (predecessor pointer), producing a
|
|
/// root-first depth-first list suitable for direct UI rendering.
|
|
///
|
|
/// Generic over `T` + an `extract` closure so unit tests can supply
|
|
/// a lightweight fixture struct without constructing a live
|
|
/// `tsclientlib::data::Channel`.
|
|
fn sort_channels_tree<'a>(channels: &'a [&'a Channel]) -> Vec<&'a Channel> {
|
|
sort_channels_tree_by(channels, |c| (c.id.0, c.parent.0, c.order.0))
|
|
}
|
|
|
|
/// Inner implementation that operates on any slice of items via an
|
|
/// extractor returning `(id, parent, order)` u64 triples. Pure;
|
|
/// tested directly by the `#[cfg(test)]` block below without
|
|
/// touching `tsclientlib`.
|
|
fn sort_channels_tree_by<'a, T>(
|
|
items: &'a [&'a T],
|
|
extract: impl Fn(&T) -> (u64, u64, u64),
|
|
) -> Vec<&'a T> {
|
|
// Bucket by parent and build (per parent) the predecessor->item
|
|
// lookup so we can walk the chain in O(n).
|
|
let mut by_parent: HashMap<u64, Vec<&'a T>> = HashMap::new();
|
|
for &item in items {
|
|
let (_id, parent, _order) = extract(item);
|
|
by_parent.entry(parent).or_default().push(item);
|
|
}
|
|
let mut ordered_per_parent: HashMap<u64, Vec<&'a T>> = HashMap::new();
|
|
for (parent, siblings) in by_parent.into_iter() {
|
|
let mut successor: HashMap<u64, &'a T> = HashMap::with_capacity(siblings.len());
|
|
for &c in &siblings {
|
|
let (_id, _parent, order) = extract(c);
|
|
// First-write wins: if the server emits two channels
|
|
// with the same predecessor (corrupted state), keep
|
|
// the first and fall through the leftover path for
|
|
// the duplicates.
|
|
successor.entry(order).or_insert(c);
|
|
}
|
|
let mut ordered: Vec<&'a T> = Vec::with_capacity(siblings.len());
|
|
let mut cursor: u64 = 0;
|
|
let mut visited: std::collections::HashSet<u64> = std::collections::HashSet::new();
|
|
while let Some(next) = successor.get(&cursor).copied() {
|
|
let (id, _parent, _order) = extract(next);
|
|
if !visited.insert(id) {
|
|
// Cycle guard. Should not happen on a well-formed
|
|
// server snapshot but cheap to defend against.
|
|
break;
|
|
}
|
|
ordered.push(next);
|
|
cursor = id;
|
|
}
|
|
// Anything we did not reach (broken predecessor pointer or
|
|
// duplicate predecessor) gets appended sorted by id so the
|
|
// UI does not silently drop channels.
|
|
let reached: std::collections::HashSet<u64> =
|
|
ordered.iter().map(|c| extract(c).0).collect();
|
|
let mut leftover: Vec<&'a T> = siblings
|
|
.into_iter()
|
|
.filter(|c| !reached.contains(&extract(c).0))
|
|
.collect();
|
|
leftover.sort_by_key(|c| extract(c).0);
|
|
ordered.extend(leftover);
|
|
ordered_per_parent.insert(parent, ordered);
|
|
}
|
|
|
|
// Emit root list first then each subtree depth-first.
|
|
let mut out: Vec<&'a T> = Vec::with_capacity(items.len());
|
|
emit_subtree(&ordered_per_parent, 0, &mut out, &extract);
|
|
// Defensive: if a channel's `parent` does not appear anywhere
|
|
// in the emitted tree (orphaned subtree) append it so it isn't
|
|
// lost. We track emitted ids and dump anything else.
|
|
let emitted: std::collections::HashSet<u64> = out.iter().map(|c| extract(c).0).collect();
|
|
let mut orphans: Vec<&'a T> = items
|
|
.iter()
|
|
.copied()
|
|
.filter(|c| !emitted.contains(&extract(c).0))
|
|
.collect();
|
|
orphans.sort_by_key(|c| extract(c).0);
|
|
out.extend(orphans);
|
|
out
|
|
}
|
|
|
|
fn emit_subtree<'a, T>(
|
|
by_parent: &HashMap<u64, Vec<&'a T>>,
|
|
root_id: u64,
|
|
out: &mut Vec<&'a T>,
|
|
extract: &impl Fn(&T) -> (u64, u64, u64),
|
|
) {
|
|
let Some(children) = by_parent.get(&root_id) else {
|
|
return;
|
|
};
|
|
for &ch in children {
|
|
out.push(ch);
|
|
let (child_id, _parent, _order) = extract(ch);
|
|
emit_subtree(by_parent, child_id, out, extract);
|
|
}
|
|
}
|
|
|
|
fn build_snapshot(
|
|
con: &Connection,
|
|
voice_activity: &HashMap<u64, Instant>,
|
|
) -> Result<ServerSnapshot, ProtocolError> {
|
|
let state: &data::Connection = con
|
|
.get_state()
|
|
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
|
|
|
|
// TeamSpeak channel ordering: the `order` field on a channel is
|
|
// NOT a numeric rank but the id of the channel that should
|
|
// appear immediately before this one within the same parent.
|
|
// `order == ChannelId(0)` marks the head of a parent's child
|
|
// list. The previous implementation sorted by `order.0`
|
|
// numerically, which produced a stable-but-arbitrary order
|
|
// that did not match the TS3 client display order and was
|
|
// reported by users as "channel sort in not correct".
|
|
//
|
|
// Correct algorithm:
|
|
// 1. Bucket channels by parent.
|
|
// 2. Within each bucket, walk the linked list starting from
|
|
// the entry whose `order == ChannelId(0)` and following
|
|
// each successive channel via its successor map until the
|
|
// chain terminates.
|
|
// 3. Emit channels root-first depth-first, so callers see a
|
|
// pre-ordered tree without needing to re-sort.
|
|
//
|
|
// Defensive fallback: any siblings the linked-list walk
|
|
// cannot reach (e.g. the server sent a cycle or a dangling
|
|
// predecessor) are appended at the end of the bucket sorted
|
|
// by id so the UI doesn't lose channels.
|
|
let all_channels: Vec<&Channel> = state.channels.values().collect();
|
|
let channels: Vec<&Channel> = sort_channels_tree(&all_channels);
|
|
let clients: Vec<&Client> = state.clients.values().collect();
|
|
|
|
let channels_dto: Vec<ChannelInfo> = channels
|
|
.iter()
|
|
.map(|c| ChannelInfo {
|
|
id: ChannelId(c.id.0),
|
|
parent: ChannelId(c.parent.0),
|
|
name: sanitize(&c.name),
|
|
order: c.order.0 as i64,
|
|
has_password: c.has_password.unwrap_or(false),
|
|
needed_talk_power: c.needed_talk_power,
|
|
})
|
|
.collect();
|
|
|
|
let clients_dto: Vec<ClientInfo> = clients
|
|
.iter()
|
|
.map(|c| ClientInfo {
|
|
id: ClientId(c.id.0 as u64),
|
|
channel: ChannelId(c.channel.0),
|
|
name: sanitize(&c.name),
|
|
input_muted: c.input_muted,
|
|
output_muted: c.output_muted || c.output_only_muted,
|
|
is_speaking: voice_activity
|
|
.get(&(c.id.0 as u64))
|
|
.is_some_and(|last| last.elapsed() <= SPEAKING_ACTIVITY_WINDOW),
|
|
is_server_query: is_server_query_client_type(&c.client_type),
|
|
talk_power: c.talk_power,
|
|
talk_power_granted: c.talk_power_granted,
|
|
})
|
|
.collect();
|
|
|
|
Ok(ServerSnapshot {
|
|
server_name: sanitize(&state.server.name),
|
|
welcome_message: sanitize(&state.server.welcome_message),
|
|
platform: sanitize(&state.server.platform),
|
|
version: sanitize(&state.server.version),
|
|
channels: channels_dto,
|
|
clients: clients_dto,
|
|
own_client_id: state.own_client.0 as u64,
|
|
})
|
|
}
|
|
|
|
fn is_server_query_client_type(client_type: &ClientType) -> bool {
|
|
matches!(client_type, ClientType::Query { .. })
|
|
}
|
|
|
|
/// Light sanitisation of strings before they cross the protocol
|
|
/// boundary. The redaction policy proper lives in
|
|
/// `chanora_diagnostics`; this filter only strips control characters
|
|
/// that would break terminal output or Flutter rendering.
|
|
fn sanitize(s: &str) -> String {
|
|
s.chars()
|
|
.filter(|c| !c.is_control() || *c == '\t' || *c == '\n')
|
|
.collect()
|
|
}
|
|
|
|
fn quoted(s: &str) -> String {
|
|
format!("\"{}\"", sanitize(s))
|
|
}
|
|
|
|
fn activity_channel_name(con: &Connection, id: tsclientlib::ChannelId) -> Option<String> {
|
|
con.get_state()
|
|
.ok()
|
|
.and_then(|state| state.channels.get(&id))
|
|
.map(|channel| quoted(&channel.name))
|
|
}
|
|
|
|
fn activity_client_name(con: &Connection, id: tsclientlib::ClientId) -> Option<String> {
|
|
con.get_state()
|
|
.ok()
|
|
.and_then(|state| state.clients.get(&id))
|
|
.filter(|client| !is_server_query_client_type(&client.client_type))
|
|
.map(|client| quoted(&client.name))
|
|
}
|
|
|
|
fn activity_client(con: &Connection, id: tsclientlib::ClientId) -> Option<Client> {
|
|
con.get_state()
|
|
.ok()
|
|
.and_then(|state| state.clients.get(&id))
|
|
.filter(|client| !is_server_query_client_type(&client.client_type))
|
|
.cloned()
|
|
}
|
|
|
|
fn activity_channel_group_name(
|
|
con: &Connection,
|
|
id: tsclientlib::ChannelGroupId,
|
|
) -> Option<String> {
|
|
con.get_state()
|
|
.ok()
|
|
.and_then(|state| state.channel_groups.get(&id))
|
|
.map(|group| quoted(&group.name))
|
|
}
|
|
|
|
fn activity_server_group_name(con: &Connection, id: tsclientlib::ServerGroupId) -> Option<String> {
|
|
con.get_state()
|
|
.ok()
|
|
.and_then(|state| state.server_groups.get(&id))
|
|
.map(|group| quoted(&group.name))
|
|
}
|
|
|
|
fn activity_invoker_name(invoker: Option<&tsclientlib::Invoker>) -> String {
|
|
invoker
|
|
.map(|invoker| quoted(&invoker.name))
|
|
.unwrap_or_else(|| "\"Server\"".to_string())
|
|
}
|
|
|
|
fn format_server_activity(con: &Connection, ev: &tsclientlib::events::Event) -> Option<String> {
|
|
use ts_bookkeeping::events::{Event, PropertyId, PropertyValue};
|
|
use tsproto_types::Reason;
|
|
|
|
match ev {
|
|
Event::PropertyAdded {
|
|
id: PropertyId::Client(client_id),
|
|
extra,
|
|
..
|
|
} => {
|
|
extra.reason?;
|
|
let client = activity_client(con, *client_id)?;
|
|
let channel = activity_channel_name(con, client.channel)?;
|
|
Some(format!(
|
|
"{} connected to channel {}",
|
|
quoted(&client.name),
|
|
channel
|
|
))
|
|
}
|
|
Event::PropertyRemoved {
|
|
id: PropertyId::Client(_),
|
|
old,
|
|
extra,
|
|
..
|
|
} => {
|
|
let PropertyValue::Client(client) = old else {
|
|
return None;
|
|
};
|
|
if is_server_query_client_type(&client.client_type) {
|
|
return None;
|
|
}
|
|
match extra.reason {
|
|
Some(Reason::Clientdisconnect) => {
|
|
Some(format!("{} disconnected (Leaving)", quoted(&client.name)))
|
|
}
|
|
Some(Reason::ClientdisconnectServerShutdown) | Some(Reason::Serverstop) => Some(
|
|
format!("{} disconnected (server shutdown)", quoted(&client.name)),
|
|
),
|
|
_ => Some(format!(
|
|
"{} dropped (connection lost)",
|
|
quoted(&client.name)
|
|
)),
|
|
}
|
|
}
|
|
Event::PropertyChanged {
|
|
id: PropertyId::ClientChannel(client_id),
|
|
old,
|
|
invoker,
|
|
..
|
|
} => {
|
|
let PropertyValue::ChannelId(from_channel_id) = old else {
|
|
return None;
|
|
};
|
|
let client = activity_client(con, *client_id)?;
|
|
let from = activity_channel_name(con, *from_channel_id)?;
|
|
let to = activity_channel_name(con, client.channel)?;
|
|
if invoker.as_ref().map(|invoker| invoker.id) == Some(*client_id) || invoker.is_none() {
|
|
Some(format!(
|
|
"{} switched from channel {} to {}",
|
|
quoted(&client.name),
|
|
from,
|
|
to
|
|
))
|
|
} else {
|
|
Some(format!(
|
|
"{} was moved from channel {} to {} by {}",
|
|
quoted(&client.name),
|
|
from,
|
|
to,
|
|
activity_invoker_name(invoker.as_ref())
|
|
))
|
|
}
|
|
}
|
|
Event::PropertyChanged {
|
|
id: PropertyId::ClientChannelGroup(client_id),
|
|
invoker,
|
|
..
|
|
} => {
|
|
let client = activity_client(con, *client_id)?;
|
|
let group = activity_channel_group_name(con, client.channel_group)?;
|
|
Some(format!(
|
|
"Channel group {} was assigned to {} by {}.",
|
|
group,
|
|
quoted(&client.name),
|
|
activity_invoker_name(invoker.as_ref())
|
|
))
|
|
}
|
|
Event::PropertyAdded {
|
|
id: PropertyId::ClientServerGroup(client_id, group_id),
|
|
invoker,
|
|
..
|
|
} => {
|
|
let client = activity_client_name(con, *client_id)?;
|
|
let group = activity_server_group_name(con, *group_id)?;
|
|
Some(format!(
|
|
"Server group {} was assigned to {} by {}.",
|
|
group,
|
|
client,
|
|
activity_invoker_name(invoker.as_ref())
|
|
))
|
|
}
|
|
Event::PropertyRemoved {
|
|
id: PropertyId::ClientServerGroup(client_id, group_id),
|
|
invoker,
|
|
..
|
|
} => {
|
|
let client = activity_client_name(con, *client_id)?;
|
|
let group = activity_server_group_name(con, *group_id)?;
|
|
Some(format!(
|
|
"Server group {} was removed from {} by {}.",
|
|
group,
|
|
client,
|
|
activity_invoker_name(invoker.as_ref())
|
|
))
|
|
}
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
const _: () = {
|
|
// Compile-time assertion that ChannelId(0) maps to what tsclientlib
|
|
// also considers the root.
|
|
let _ = TsChannelId(0);
|
|
};
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{
|
|
bounded_drain_stream, client_profile_refresh_plan, drain_voice_packets_for_tick,
|
|
is_server_query_client_type, send_with_timeout, server_socket_from_config,
|
|
sort_channels_tree_by, std_duration_millis, ConnectConfig, ProtocolClient, Request,
|
|
SendTimeoutError, DISCONNECT_REPLY_TIMEOUT,
|
|
};
|
|
use futures::stream;
|
|
use std::time::Duration;
|
|
use tokio::sync::mpsc;
|
|
use tsproto_types::ClientType;
|
|
|
|
/// Lightweight fixture mirroring just the (id, parent, order)
|
|
/// triple that the linked-list sort needs. Avoids constructing
|
|
/// a real `tsclientlib::data::Channel` (which requires a live
|
|
/// connection) in unit tests.
|
|
#[derive(Debug, PartialEq)]
|
|
struct FakeChannel {
|
|
id: u64,
|
|
parent: u64,
|
|
order: u64,
|
|
}
|
|
|
|
fn extract(c: &FakeChannel) -> (u64, u64, u64) {
|
|
(c.id, c.parent, c.order)
|
|
}
|
|
|
|
#[test]
|
|
fn query_client_type_maps_to_server_query_flag() {
|
|
assert!(!is_server_query_client_type(&ClientType::Normal));
|
|
assert!(is_server_query_client_type(&ClientType::Query {
|
|
admin: false
|
|
}));
|
|
assert!(is_server_query_client_type(&ClientType::Query {
|
|
admin: true
|
|
}));
|
|
}
|
|
|
|
#[test]
|
|
fn connect_config_can_carry_prefetched_socket_address() {
|
|
let addr: std::net::SocketAddr = "127.0.0.1:9987".parse().unwrap();
|
|
let cfg = ConnectConfig {
|
|
address: "example.com".to_string(),
|
|
nickname: "Tester".to_string(),
|
|
password: None,
|
|
identity: None,
|
|
ready_timeout: Duration::from_secs(1),
|
|
resolved_address: Some(addr),
|
|
};
|
|
|
|
assert_eq!(cfg.resolved_address, Some(addr));
|
|
}
|
|
|
|
#[test]
|
|
fn server_socket_from_config_prefers_prefetched_address() {
|
|
let addr: std::net::SocketAddr = "127.0.0.1:9987".parse().unwrap();
|
|
let cfg = ConnectConfig {
|
|
address: "example.com".to_string(),
|
|
resolved_address: Some(addr),
|
|
nickname: "Tester".to_string(),
|
|
password: None,
|
|
identity: None,
|
|
ready_timeout: Duration::from_secs(1),
|
|
};
|
|
|
|
assert_eq!(server_socket_from_config(&cfg), Some(addr));
|
|
}
|
|
|
|
#[test]
|
|
fn std_duration_millis_returns_none_when_duration_exceeds_i64() {
|
|
let overflow = Duration::from_millis(i64::MAX as u64) + Duration::from_millis(1);
|
|
|
|
assert_eq!(std_duration_millis(overflow), None);
|
|
}
|
|
|
|
#[test]
|
|
fn non_self_profile_refresh_requests_variables_connection_and_db_info() {
|
|
let plan = client_profile_refresh_plan(false, true, true, false, false);
|
|
|
|
assert!(plan.needs_client_variables);
|
|
assert!(plan.needs_connection_info);
|
|
assert!(plan.needs_client_db_info);
|
|
}
|
|
|
|
#[test]
|
|
fn own_profile_refresh_skips_complete_optional_and_connection_requests() {
|
|
let plan = client_profile_refresh_plan(true, true, true, false, false);
|
|
|
|
assert!(!plan.needs_server_groups);
|
|
assert!(!plan.needs_channel_groups);
|
|
assert!(!plan.needs_client_variables);
|
|
assert!(!plan.needs_connection_info);
|
|
assert!(!plan.needs_client_db_info);
|
|
}
|
|
|
|
#[test]
|
|
fn own_profile_refresh_requests_missing_group_lists() {
|
|
let plan = client_profile_refresh_plan(true, true, true, true, true);
|
|
|
|
assert!(plan.needs_server_groups);
|
|
assert!(plan.needs_channel_groups);
|
|
}
|
|
|
|
#[test]
|
|
fn channel_sort_linked_list_under_one_parent() {
|
|
// Server emits four root-level channels in arbitrary HashMap
|
|
// iteration order. order=0 -> first; order=X means "comes
|
|
// after the channel with id=X". Expected emitted order is
|
|
// the linked-list walk: a -> b -> c -> d.
|
|
let a = FakeChannel {
|
|
id: 100,
|
|
parent: 0,
|
|
order: 0,
|
|
};
|
|
let b = FakeChannel {
|
|
id: 200,
|
|
parent: 0,
|
|
order: 100,
|
|
};
|
|
let c = FakeChannel {
|
|
id: 300,
|
|
parent: 0,
|
|
order: 200,
|
|
};
|
|
let d = FakeChannel {
|
|
id: 400,
|
|
parent: 0,
|
|
order: 300,
|
|
};
|
|
// Deliberately shuffled inputs.
|
|
let inputs: Vec<&FakeChannel> = vec![&c, &a, &d, &b];
|
|
let sorted = sort_channels_tree_by(&inputs, extract);
|
|
let ids: Vec<u64> = sorted.iter().map(|c| c.id).collect();
|
|
assert_eq!(ids, vec![100, 200, 300, 400]);
|
|
}
|
|
|
|
#[test]
|
|
fn channel_sort_disconnected_predecessor_falls_back_by_id() {
|
|
// a is the head. b correctly chains. c claims predecessor
|
|
// = 999 which does not exist among the siblings. c must
|
|
// not be dropped — it falls back to the leftover bucket
|
|
// appended sorted by id at the end.
|
|
let a = FakeChannel {
|
|
id: 100,
|
|
parent: 0,
|
|
order: 0,
|
|
};
|
|
let b = FakeChannel {
|
|
id: 200,
|
|
parent: 0,
|
|
order: 100,
|
|
};
|
|
let c = FakeChannel {
|
|
id: 300,
|
|
parent: 0,
|
|
order: 999,
|
|
};
|
|
let inputs: Vec<&FakeChannel> = vec![&c, &a, &b];
|
|
let sorted = sort_channels_tree_by(&inputs, extract);
|
|
let ids: Vec<u64> = sorted.iter().map(|c| c.id).collect();
|
|
assert_eq!(ids, vec![100, 200, 300]);
|
|
}
|
|
|
|
#[test]
|
|
fn channel_sort_emits_subtree_depth_first() {
|
|
// Tree:
|
|
// root (id=0, implicit)
|
|
// ├── a (id=10, order=0)
|
|
// │ ├── a1 (id=11, parent=10, order=0)
|
|
// │ └── a2 (id=12, parent=10, order=11)
|
|
// └── b (id=20, order=10)
|
|
// Expected emission: a, a1, a2, b
|
|
let a = FakeChannel {
|
|
id: 10,
|
|
parent: 0,
|
|
order: 0,
|
|
};
|
|
let a1 = FakeChannel {
|
|
id: 11,
|
|
parent: 10,
|
|
order: 0,
|
|
};
|
|
let a2 = FakeChannel {
|
|
id: 12,
|
|
parent: 10,
|
|
order: 11,
|
|
};
|
|
let b = FakeChannel {
|
|
id: 20,
|
|
parent: 0,
|
|
order: 10,
|
|
};
|
|
let inputs: Vec<&FakeChannel> = vec![&b, &a2, &a, &a1];
|
|
let sorted = sort_channels_tree_by(&inputs, extract);
|
|
let ids: Vec<u64> = sorted.iter().map(|c| c.id).collect();
|
|
assert_eq!(ids, vec![10, 11, 12, 20]);
|
|
}
|
|
|
|
#[test]
|
|
fn channel_sort_does_not_loop_on_cycle() {
|
|
// a says "comes after b"; b says "comes after a". The
|
|
// walk must terminate (cycle guard) and both channels
|
|
// must still appear in the output via the leftover path.
|
|
let a = FakeChannel {
|
|
id: 1,
|
|
parent: 0,
|
|
order: 2,
|
|
};
|
|
let b = FakeChannel {
|
|
id: 2,
|
|
parent: 0,
|
|
order: 1,
|
|
};
|
|
let inputs: Vec<&FakeChannel> = vec![&a, &b];
|
|
let sorted = sort_channels_tree_by(&inputs, extract);
|
|
// Both reachable in some deterministic order (id-sorted
|
|
// in the leftover bucket since cursor=0 finds nothing).
|
|
let ids: Vec<u64> = sorted.iter().map(|c| c.id).collect();
|
|
assert_eq!(ids.len(), 2);
|
|
assert!(ids.contains(&1));
|
|
assert!(ids.contains(&2));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn send_with_timeout_waits_for_capacity_instead_of_dropping() {
|
|
let (tx, mut rx) = mpsc::channel(1);
|
|
tx.send(1_u8).await.expect("seed first item");
|
|
|
|
let drain = tokio::spawn(async move {
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
assert_eq!(rx.recv().await, Some(1));
|
|
rx.recv().await
|
|
});
|
|
|
|
send_with_timeout(&tx, 2_u8, Duration::from_millis(100))
|
|
.await
|
|
.expect("second item should enqueue once capacity frees");
|
|
|
|
assert_eq!(drain.await.expect("drain task should succeed"), Some(2));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn send_with_timeout_times_out_when_capacity_stays_full() {
|
|
let (tx, _rx) = mpsc::channel(1);
|
|
tx.send(1_u8).await.expect("seed first item");
|
|
|
|
let result = send_with_timeout(&tx, 2_u8, Duration::from_millis(10)).await;
|
|
|
|
assert_eq!(result, Err(SendTimeoutError::Timeout(2)));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn disconnect_request_send_is_bounded_when_request_channel_is_full() {
|
|
let (tx, _rx) = mpsc::channel(1);
|
|
let (reply_tx, _reply_rx) = tokio::sync::oneshot::channel();
|
|
tx.send(Request::Snapshot(reply_tx))
|
|
.await
|
|
.expect("seed first request");
|
|
|
|
let (disconnect_tx, _disconnect_rx) = tokio::sync::oneshot::channel();
|
|
let result = send_with_timeout(
|
|
&tx,
|
|
Request::Disconnect(disconnect_tx),
|
|
Duration::from_millis(10),
|
|
)
|
|
.await;
|
|
|
|
assert!(matches!(result, Err(SendTimeoutError::Timeout(_))));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn protocol_client_disconnect_returns_when_request_channel_is_full() {
|
|
let (tx, _rx) = mpsc::channel(1);
|
|
let (snapshot_tx, _snapshot_rx) = tokio::sync::oneshot::channel();
|
|
tx.send(Request::Snapshot(snapshot_tx))
|
|
.await
|
|
.expect("seed first request");
|
|
let (voice_out_tx, _voice_out_rx) = mpsc::channel(1);
|
|
let (_voice_in_tx, voice_in_rx) = mpsc::channel(1);
|
|
let (_lost_tx, lost_rx) = tokio::sync::oneshot::channel();
|
|
let (_chat_tx, chat_rx) = mpsc::channel(1);
|
|
let (_activity_tx, activity_rx) = mpsc::channel(1);
|
|
let (_delta_tx, delta_rx) = mpsc::channel(1);
|
|
let client = ProtocolClient {
|
|
tx,
|
|
voice_out_tx,
|
|
voice_in_rx: std::sync::Mutex::new(Some(voice_in_rx)),
|
|
lost_rx: std::sync::Mutex::new(Some(lost_rx)),
|
|
chat_rx: std::sync::Mutex::new(Some(chat_rx)),
|
|
activity_rx: std::sync::Mutex::new(Some(activity_rx)),
|
|
delta_rx: std::sync::Mutex::new(Some(delta_rx)),
|
|
};
|
|
|
|
tokio::time::timeout(
|
|
DISCONNECT_REPLY_TIMEOUT + Duration::from_millis(100),
|
|
client.disconnect(),
|
|
)
|
|
.await
|
|
.expect("disconnect should not wait indefinitely for request channel capacity");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn voice_drain_stops_at_per_tick_budget() {
|
|
let (tx, mut rx) = mpsc::channel(8);
|
|
for value in 0_u8..5 {
|
|
tx.send(value).await.expect("seed voice packet");
|
|
}
|
|
|
|
let mut sent = Vec::new();
|
|
let drained = drain_voice_packets_for_tick(&mut rx, 2, |value| {
|
|
sent.push(value);
|
|
Ok::<(), ()>(())
|
|
});
|
|
|
|
assert_eq!(drained, 2);
|
|
assert_eq!(sent, vec![0, 1]);
|
|
assert_eq!(rx.len(), 3);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn disconnect_stream_drain_returns_after_timeout() {
|
|
let start = tokio::time::Instant::now();
|
|
|
|
bounded_drain_stream(stream::pending::<()>(), Duration::from_millis(10)).await;
|
|
|
|
assert!(start.elapsed() < Duration::from_millis(100));
|
|
}
|
|
}
|
|
|
|
fn forward_delta(
|
|
con: &Connection,
|
|
ev: &tsclientlib::events::Event,
|
|
delta_tx: &mpsc::Sender<ProtocolDelta>,
|
|
) {
|
|
use ts_bookkeeping::events::{Event, PropertyId, PropertyValue};
|
|
|
|
match ev {
|
|
Event::PropertyAdded {
|
|
id: PropertyId::Client(client_id),
|
|
..
|
|
} => {
|
|
if let Ok(state) = con.get_state() {
|
|
if let Some(client) = state.clients.get(client_id) {
|
|
let _ = delta_tx.try_send(ProtocolDelta::ClientJoined {
|
|
client_id: client_id.0 as u64,
|
|
channel_id: client.channel.0,
|
|
name: client.name.clone(),
|
|
input_muted: client.input_muted,
|
|
output_muted: client.output_muted || client.output_only_muted,
|
|
is_server_query: is_server_query_client_type(&client.client_type),
|
|
talk_power: client.talk_power,
|
|
talk_power_granted: client.talk_power_granted,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
Event::PropertyRemoved {
|
|
id: PropertyId::Client(_),
|
|
old: PropertyValue::Client(client),
|
|
..
|
|
} => {
|
|
let _ = delta_tx.try_send(ProtocolDelta::ClientLeft {
|
|
client_id: client.id.0 as u64,
|
|
name: client.name.clone(),
|
|
});
|
|
}
|
|
Event::PropertyChanged {
|
|
id: PropertyId::ClientChannel(_),
|
|
..
|
|
} => {}
|
|
Event::PropertyChanged {
|
|
id: PropertyId::Client(client_id),
|
|
..
|
|
} => {
|
|
if let Ok(state) = con.get_state() {
|
|
if let Some(client) = state.clients.get(client_id) {
|
|
let _ = delta_tx.try_send(ProtocolDelta::ClientUpdated {
|
|
client_id: client_id.0 as u64,
|
|
input_muted: client.input_muted,
|
|
output_muted: client.output_muted || client.output_only_muted,
|
|
is_server_query: is_server_query_client_type(&client.client_type),
|
|
talk_power: client.talk_power,
|
|
talk_power_granted: client.talk_power_granted,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
Event::PropertyAdded {
|
|
id: PropertyId::Channel(channel_id),
|
|
..
|
|
} => {
|
|
if let Ok(state) = con.get_state() {
|
|
if let Some(channel) = state.channels.get(channel_id) {
|
|
let _ = delta_tx.try_send(ProtocolDelta::ChannelAdded {
|
|
id: channel_id.0,
|
|
parent: channel.parent.0,
|
|
name: channel.name.clone(),
|
|
order: channel.order.0 as i64,
|
|
has_password: channel.has_password.unwrap_or(false),
|
|
needed_talk_power: channel.needed_talk_power,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
Event::PropertyRemoved {
|
|
id: PropertyId::Channel(_),
|
|
old: PropertyValue::Channel(channel),
|
|
..
|
|
} => {
|
|
let _ = delta_tx.try_send(ProtocolDelta::ChannelRemoved { id: channel.id.0 });
|
|
}
|
|
Event::PropertyChanged {
|
|
id: PropertyId::Channel(channel_id),
|
|
..
|
|
} => {
|
|
if let Ok(state) = con.get_state() {
|
|
if let Some(channel) = state.channels.get(channel_id) {
|
|
let _ = delta_tx.try_send(ProtocolDelta::ChannelUpdated {
|
|
id: channel_id.0,
|
|
name: channel.name.clone(),
|
|
has_password: channel.has_password.unwrap_or(false),
|
|
needed_talk_power: channel.needed_talk_power,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|