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

feat: add TeamSpeak address resolver

Browse Source
This commit is contained in:
Edison Jwa
2026-05-25 01:12:50 +09:00
parent d7556cd39f
commit eb9014cd81
15 changed files with 2360 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/chanora_protocol/src/adapter.rs
+391 -71
View File
@@ -17,8 +17,10 @@
//! * 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 chanora_resolver::ChanoraResolver;
use futures::prelude::*;
use std::collections::HashMap;
use tokio::sync::{mpsc, oneshot};
@@ -34,7 +36,8 @@ use tsproto_packets::packets::{InAudioBuf, OutPacket};
use tsproto_types::ClientType;
use crate::dto::{
ChannelId, ChannelInfo, ChatMessage, ClientId, ClientInfo, MessageTarget, ServerSnapshot,
ChannelId, ChannelInfo, ChatMessage, ClientId, ClientInfo, MessageTarget, ServerActivity,
ServerSnapshot,
};
use crate::ProtocolError;
@@ -135,6 +138,11 @@ enum Request {
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>,
@@ -184,6 +192,8 @@ pub struct ProtocolClient {
/// 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>>>,
}
/// One inbound voice packet from a remote client.
@@ -243,6 +253,7 @@ impl ProtocolClient {
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 (ready_tx, ready_rx) = oneshot::channel::<Result<(), ProtocolError>>();
let (lost_tx, lost_rx) = oneshot::channel::<DisconnectReason>();
@@ -252,6 +263,7 @@ impl ProtocolClient {
voice_out_rx,
voice_in_tx,
chat_tx,
activity_tx,
ready_tx,
lost_tx,
));
@@ -263,6 +275,7 @@ impl ProtocolClient {
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)),
}),
Ok(Ok(Err(e))) => Err(e),
Ok(Err(_)) => Err(ProtocolError::Backend(
@@ -311,6 +324,23 @@ impl ProtocolClient {
.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(
@@ -393,6 +423,21 @@ impl ProtocolClient {
}
}
/// 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);
}
}
}
/// Send a text message to the specified target.
pub async fn send_text_message(
&self,
@@ -419,6 +464,7 @@ async fn connection_task(
mut voice_out_rx: mpsc::Receiver<OutPacket>,
voice_in_tx: mpsc::Sender<InboundVoice>,
chat_tx: mpsc::Sender<ChatMessage>,
activity_tx: mpsc::Sender<ServerActivity>,
ready_tx: oneshot::Sender<Result<(), ProtocolError>>,
lost_tx: oneshot::Sender<DisconnectReason>,
) {
@@ -433,28 +479,28 @@ async fn connection_task(
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);
}};
}
// 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));
}
let resolved = match resolve_server_socket(&cfg.address).await {
Ok(addr) => addr,
Err(err) => fail_ready!(err),
};
// 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"
"server address resolved"
);
// Pass the resolved SocketAddr directly to tsclientlib so it
@@ -471,12 +517,14 @@ async fn connection_task(
.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}")));
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(),
@@ -491,8 +539,10 @@ async fn connection_task(
Ok(c) => c,
Err(e) => {
let msg = format!("{e}");
let _ = ready_tx.send(Err(ProtocolError::Connect(msg.clone())));
exit!(DisconnectReason::Error(format!("connect: {msg}")));
fail_ready!(
ProtocolError::Connect(msg.clone()),
DisconnectReason::Error(format!("connect: {msg}"))
);
}
};
@@ -508,15 +558,17 @@ async fn connection_task(
}
Some(Err(e)) => {
let msg = format!("{e}");
let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone())));
exit!(DisconnectReason::Error(format!(
"disconnected early: {msg}"
)));
fail_ready!(
ProtocolError::DisconnectedEarly(msg.clone()),
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));
fail_ready!(
ProtocolError::DisconnectedEarly(msg.clone()),
DisconnectReason::Error(msg)
);
}
}
@@ -541,8 +593,10 @@ async fn connection_task(
}
Ok(None) => {
let msg = "stream closed during settle".to_string();
let _ = ready_tx.send(Err(ProtocolError::DisconnectedEarly(msg.clone())));
exit!(DisconnectReason::StreamEnded);
fail_ready!(
ProtocolError::DisconnectedEarly(msg),
DisconnectReason::StreamEnded
);
}
Err(_) => { /* no event available right now; keep waiting */ }
}
@@ -559,7 +613,8 @@ async fn connection_task(
let mut pending_moves: HashMap<
MessageHandle,
(
oneshot::Sender<Result<(), ProtocolError>>,
u64,
Option<oneshot::Sender<Result<(), ProtocolError>>>,
std::time::Instant,
),
> = HashMap::new();
@@ -599,6 +654,53 @@ async fn connection_task(
}
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 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 });
}
if let tsclientlib::events::Event::Message {
target,
invoker,
@@ -625,7 +727,7 @@ async fn connection_task(
}
}
StreamItem::MessageResult(handle, result) => {
if let Some((reply, _deadline)) = pending_moves.remove(&handle) {
if let Some((_target_channel, reply, _deadline)) = pending_moves.remove(&handle) {
let mapped = match result {
Ok(()) => Ok(()),
Err(cmd_err) => {
@@ -648,7 +750,15 @@ async fn connection_task(
Err(ProtocolError::ServerRejected { code, message })
}
};
let _ = reply.send(mapped);
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"
);
}
}
}
_ => { /* book / message / other events: ignore */ }
@@ -677,7 +787,7 @@ async fn connection_task(
let now = std::time::Instant::now();
let expired: Vec<MessageHandle> = pending_moves
.iter()
.filter_map(|(handle, (_, deadline))| {
.filter_map(|(handle, (_, _, deadline))| {
if now >= *deadline {
Some(*handle)
} else {
@@ -686,8 +796,10 @@ async fn connection_task(
})
.collect();
for handle in expired {
if let Some((reply, _)) = pending_moves.remove(&handle) {
let _ = reply.send(Ok(()));
if let Some((_target_channel, reply, _)) = pending_moves.remove(&handle) {
if let Some(reply) = reply {
let _ = reply.send(Ok(()));
}
}
}
}
@@ -706,15 +818,32 @@ async fn connection_task(
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));
pending_moves.insert(handle, (channel_id, Some(reply), deadline));
}
Err(e) => {
// Couldn't even send the command; report
// immediately.
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,
@@ -749,6 +878,26 @@ async fn connection_task(
}
}
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}"),
})
}
/// 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
@@ -815,19 +964,10 @@ fn send_text_message(
message: &str,
target: MessageTarget,
) -> Result<(), ProtocolError> {
use ts_bookkeeping::messages::c2s;
use tsproto_types::TextMessageTargetMode;
match target {
MessageTarget::Server => {
c2s::OutSendTextMessageMessage::new(&mut std::iter::once(
c2s::OutSendTextMessagePart {
target: TextMessageTargetMode::Server,
target_client_id: None,
message: message.into(),
},
))
.send(con)
.map_err(|e| ProtocolError::Backend(format!("send_textmessage(server): {e}")))?;
send_text_to_mode(con, message, TextMessageTargetMode::Server, "server")?;
}
MessageTarget::Channel => {
// Fix: previously channel messages were sent via
@@ -835,25 +975,13 @@ fn send_text_message(
// TextMessageTargetMode::Server. Now correctly uses
// TextMessageTargetMode::Channel so the message is
// scoped to the current channel, not server-wide.
c2s::OutSendTextMessageMessage::new(&mut std::iter::once(
c2s::OutSendTextMessagePart {
target: TextMessageTargetMode::Channel,
target_client_id: None,
message: message.into(),
},
))
.send(con)
.map_err(|e| ProtocolError::Backend(format!("send_textmessage(channel): {e}")))?;
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 = state
.clients
.values()
.find(|c| c.id.0 as u64 == client_id)
.ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))?;
let client = find_client_by_id(state.clients.values(), client_id)?;
client
.send_textmessage(message)
.send(con)
@@ -863,11 +991,7 @@ fn send_text_message(
let state = con
.get_state()
.map_err(|e| ProtocolError::Backend(format!("get_state: {e}")))?;
let client = state
.clients
.values()
.find(|c| c.id.0 as u64 == client_id)
.ok_or_else(|| ProtocolError::Backend(format!("client {client_id} not found")))?;
let client = find_client_by_id(state.clients.values(), client_id)?;
client
.poke(message)
.send(con)
@@ -878,6 +1002,33 @@ fn send_text_message(
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}")))
}
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;
@@ -1073,6 +1224,175 @@ fn sanitize(s: &str) -> String {
.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, .. } => {
if extra.reason.is_none() {
return None;
}
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.