1045 lines
39 KiB
Rust
1045 lines
39 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::time::Duration;
|
|
|
|
use futures::prelude::*;
|
|
use std::collections::HashMap;
|
|
use tokio::sync::{mpsc, oneshot};
|
|
use tracing::{info, warn};
|
|
|
|
use tsclientlib::data::{self, Channel, Client};
|
|
use tsclientlib::prelude::*;
|
|
use tsclientlib::{
|
|
ChannelId as TsChannelId, Connection, DisconnectOptions, Identity, MessageHandle,
|
|
OutCommandExt, StreamItem, Version,
|
|
};
|
|
use tsproto_packets::packets::{InAudioBuf, OutPacket};
|
|
|
|
use crate::dto::{ChannelId, ChannelInfo, ClientId, ClientInfo, ServerSnapshot};
|
|
use crate::ProtocolError;
|
|
|
|
/// 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 {
|
|
#[cfg(target_os = "windows")]
|
|
{
|
|
Version::Windows_5_0_0_beta51
|
|
}
|
|
#[cfg(target_os = "macos")]
|
|
{
|
|
Version::macOS_5_0_0_beta51
|
|
}
|
|
#[cfg(target_os = "ios")]
|
|
{
|
|
Version::iOS_3_5_6
|
|
}
|
|
#[cfg(target_os = "android")]
|
|
{
|
|
Version::Android_3_5_0__7
|
|
}
|
|
#[cfg(target_os = "linux")]
|
|
{
|
|
Version::Linux_5_0_0_beta51
|
|
}
|
|
#[cfg(not(any(
|
|
target_os = "windows",
|
|
target_os = "macos",
|
|
target_os = "ios",
|
|
target_os = "android",
|
|
target_os = "linux"
|
|
)))]
|
|
{
|
|
// Last-ditch fallback for unanticipated targets (BSDs,
|
|
// Solaris-likes). Linux signature is the closest analogue.
|
|
Version::Linux_5_0_0_beta51
|
|
}
|
|
}
|
|
|
|
/// 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 must wire this to `chanora_storage::SecretStorageRepository`.
|
|
pub identity: Option<String>,
|
|
/// How long to wait for the initial state snapshot before
|
|
/// returning `ProtocolError::Timeout`.
|
|
pub ready_timeout: Duration,
|
|
}
|
|
|
|
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),
|
|
}
|
|
}
|
|
}
|
|
|
|
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>>,
|
|
},
|
|
/// Update own client mute state (input and/or output).
|
|
SetMuted {
|
|
input: Option<bool>,
|
|
output: Option<bool>,
|
|
reply: oneshot::Sender<Result<(), 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>>>,
|
|
}
|
|
|
|
/// 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 (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,
|
|
voice_in_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)),
|
|
}),
|
|
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()))?
|
|
}
|
|
|
|
/// Disconnect cleanly. Blocks until the task exits.
|
|
pub async fn disconnect(self) {
|
|
let (tx, rx) = oneshot::channel();
|
|
if self.tx.send(Request::Disconnect(tx)).await.is_ok() {
|
|
let _ = rx.await;
|
|
}
|
|
}
|
|
|
|
/// 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()))?
|
|
}
|
|
|
|
/// 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())
|
|
}
|
|
}
|
|
|
|
async fn connection_task(
|
|
cfg: ConnectConfig,
|
|
mut rx: mpsc::Receiver<Request>,
|
|
mut voice_out_rx: mpsc::Receiver<OutPacket>,
|
|
voice_in_tx: mpsc::Sender<InboundVoice>,
|
|
ready_tx: oneshot::Sender<Result<(), ProtocolError>>,
|
|
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;
|
|
}};
|
|
}
|
|
|
|
// Resolve the hostname OURSELVES using the platform resolver.
|
|
// tsclientlib's built-in hickory-resolver reads /etc/resolv.conf,
|
|
// which does not exist on Android or iOS — by side-stepping it
|
|
// here we get hostname connects working on every platform.
|
|
let addrs = match crate::resolver::resolve(&cfg.address).await {
|
|
Ok(a) => a,
|
|
Err(e) => {
|
|
let msg = format!("{e}");
|
|
let _ = ready_tx.send(Err(e));
|
|
exit!(DisconnectReason::Error(msg));
|
|
}
|
|
};
|
|
// Pick the first address (IPv4 preferred by the resolver's
|
|
// ordering). Future retry logic could fall back to subsequent
|
|
// addresses; one is enough for the Beta connect flow.
|
|
let resolved = addrs[0];
|
|
info!(
|
|
target: "chanora_protocol",
|
|
input = %cfg.address,
|
|
resolved = %resolved,
|
|
"dns resolved"
|
|
);
|
|
|
|
// 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(s) => match Identity::new_from_str(s) {
|
|
Ok(id) => id,
|
|
Err(e) => {
|
|
let msg = format!("{e}");
|
|
let _ = ready_tx.send(Err(ProtocolError::Identity(msg.clone())));
|
|
exit!(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}");
|
|
let _ = ready_tx.send(Err(ProtocolError::Connect(msg.clone())));
|
|
exit!(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}");
|
|
let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone())));
|
|
exit!(DisconnectReason::Error(format!(
|
|
"disconnected early: {msg}"
|
|
)));
|
|
}
|
|
None => {
|
|
let msg = "event stream ended before snapshot".to_string();
|
|
let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone())));
|
|
exit!(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();
|
|
let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone())));
|
|
exit!(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: HashMap<
|
|
MessageHandle,
|
|
(
|
|
oneshot::Sender<Result<(), ProtocolError>>,
|
|
std::time::Instant,
|
|
),
|
|
> = HashMap::new();
|
|
|
|
// Main loop: pump events, service requests, forward voice.
|
|
loop {
|
|
// 1. Drain any outbound voice packets first — they're time-sensitive.
|
|
while let Ok(pkt) = voice_out_rx.try_recv() {
|
|
if let Err(e) = con.send_audio(pkt) {
|
|
warn!(target: "chanora_protocol", error = %e, "send_audio failed");
|
|
}
|
|
}
|
|
|
|
// 2. Advance event stream by at most one event with a small timeout.
|
|
let pump = async {
|
|
let mut ev_stream = con.events();
|
|
tokio::time::timeout(Duration::from_millis(20), ev_stream.next()).await
|
|
};
|
|
match pump.await {
|
|
Ok(Some(Ok(item))) => {
|
|
match item {
|
|
StreamItem::Audio(buf) => {
|
|
let from = packet_sender_id(&buf);
|
|
if let Some(from) = from {
|
|
if voice_in_tx
|
|
.try_send(InboundVoice {
|
|
from_client: from,
|
|
packet: buf,
|
|
})
|
|
.is_err()
|
|
{
|
|
// Subscriber is too slow or absent; drop.
|
|
}
|
|
}
|
|
}
|
|
StreamItem::MessageResult(handle, result) => {
|
|
if let Some((reply, _deadline)) = pending_moves.remove(&handle) {
|
|
let mapped = match result {
|
|
Ok(()) => Ok(()),
|
|
Err(cmd_err) => {
|
|
// tsclientlib's CommandError carries a
|
|
// typed `TsError` (the canonical TS3
|
|
// error code) plus an optional missing
|
|
// permission. We convert to our typed
|
|
// ProtocolError::ServerRejected so the
|
|
// upper layers can render a localised
|
|
// explanation by code instead of a
|
|
// generic backend string.
|
|
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 })
|
|
}
|
|
};
|
|
let _ = reply.send(mapped);
|
|
}
|
|
}
|
|
_ => { /* book / message / other events: ignore */ }
|
|
}
|
|
}
|
|
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((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);
|
|
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, (reply, deadline));
|
|
}
|
|
Err(e) => {
|
|
// Couldn't even send the command; report
|
|
// immediately.
|
|
let _ = reply.send(Err(e));
|
|
}
|
|
}
|
|
}
|
|
Ok(Request::SetMuted {
|
|
input,
|
|
output,
|
|
reply,
|
|
}) => {
|
|
let r = set_self_muted(&mut con, input, output);
|
|
let _ = reply.send(r);
|
|
}
|
|
Ok(Request::Disconnect(reply)) => {
|
|
let _ = con.disconnect(DisconnectOptions::new());
|
|
con.events().for_each(|_| future::ready(())).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());
|
|
con.events().for_each(|_| future::ready(())).await;
|
|
info!(target: "chanora_protocol", "handle dropped; implicit disconnect");
|
|
exit!(DisconnectReason::UserRequested);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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(())
|
|
}
|
|
|
|
/// 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) -> 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,
|
|
})
|
|
.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),
|
|
})
|
|
.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,
|
|
})
|
|
}
|
|
|
|
/// 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()
|
|
}
|
|
|
|
#[allow(dead_code)]
|
|
const _ROOT_MATCHES_UPSTREAM: () = {
|
|
// Compile-time assertion that ChannelId(0) maps to what tsclientlib
|
|
// also considers the root.
|
|
let _ = TsChannelId(0);
|
|
};
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::sort_channels_tree_by;
|
|
|
|
/// 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 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));
|
|
}
|
|
}
|