TeamSpeak: ServerQuery über SSH statt Beobachter-Client (kein dauerhafter Slot)
This commit is contained in:
1 parent
0574fd7f6d
commit
5bf3b845e4
6 files changed
+364
-192
No files matched your search
@@ -21,8 +21,9 @@ public record BridgeConfig(
|
||||
Discord discord,
|
||||
List<RoomConfig> rooms
|
||||
) {
|
||||
public record Teamspeak(String host, int port, String password, String parentChannel, String nickname,
|
||||
String roomNickname, int securityLevel, ClientVersion clientVersion) {
|
||||
public record Teamspeak(String host, int port, String password, int queryPort, String queryLogin,
|
||||
String queryPassword, String parentChannel, String roomNickname, int securityLevel,
|
||||
ClientVersion clientVersion) {
|
||||
public boolean enabled() { return !host.isBlank(); }
|
||||
}
|
||||
|
||||
@@ -72,8 +73,9 @@ public record BridgeConfig(
|
||||
Path.of(string(root, "dataDir", "data")),
|
||||
integer(root, "lingerSeconds", 15) * 1000L,
|
||||
new Teamspeak(string(ts, "host", ""), integer(ts, "port", 9987), string(ts, "password", ""),
|
||||
string(ts, "parentChannel", "0"), string(ts, "nickname", "Crosstalk"),
|
||||
string(ts, "roomNickname", "%room% (MC/DC)"), integer(ts, "securityLevel", 15), version),
|
||||
integer(ts, "queryPort", 10022), string(ts, "queryLogin", ""), string(ts, "queryPassword", ""),
|
||||
string(ts, "parentChannel", "0"), string(ts, "roomNickname", "%room% (MC/DC)"),
|
||||
integer(ts, "securityLevel", 15), version),
|
||||
new Discord(string(dc, "guild", ""), string(dc, "category", ""),
|
||||
string(dc, "nickname", "%room% (MC/TS)")),
|
||||
List.copyOf(rooms));
|
||||
|
||||
@@ -1,27 +1,12 @@
|
||||
package eu.mhsl.voicebridge.bridge.teamspeak;
|
||||
|
||||
import com.github.manevolent.ts3j.command.SingleCommand;
|
||||
import com.github.manevolent.ts3j.command.parameter.CommandSingleParameter;
|
||||
import com.github.manevolent.ts3j.event.ChannelCreateEvent;
|
||||
import com.github.manevolent.ts3j.event.ChannelDeletedEvent;
|
||||
import com.github.manevolent.ts3j.event.ChannelEditedEvent;
|
||||
import com.github.manevolent.ts3j.event.ChannelListEvent;
|
||||
import com.github.manevolent.ts3j.event.ChannelMovedEvent;
|
||||
import com.github.manevolent.ts3j.event.ClientJoinEvent;
|
||||
import com.github.manevolent.ts3j.event.ClientLeaveEvent;
|
||||
import com.github.manevolent.ts3j.event.ClientMovedEvent;
|
||||
import com.github.manevolent.ts3j.event.DisconnectedEvent;
|
||||
import com.github.manevolent.ts3j.event.TS3Listener;
|
||||
import com.github.manevolent.ts3j.identity.LocalIdentity;
|
||||
import com.github.manevolent.ts3j.protocol.ProtocolRole;
|
||||
import com.github.manevolent.ts3j.protocol.socket.client.LocalTeamspeakClientSocket;
|
||||
import eu.mhsl.voicebridge.bridge.BridgeConfig;
|
||||
import eu.mhsl.voicebridge.bridge.Room;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.security.GeneralSecurityException;
|
||||
@@ -35,13 +20,14 @@ import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* TeamSpeak-Anbindung. Ein Beobachter-Client bleibt dauerhaft verbunden, sieht so alle Clients und
|
||||
* Channels und legt fehlende Raum-Channels an. Pro aktivem Raum verbindet sich zusätzlich ein
|
||||
* {@link TeamspeakRoomClient}, der im Raum-Channel spricht und zuhört.
|
||||
* TeamSpeak-Anbindung. Eine ServerQuery (belegt keinen Slot) sieht alle Clients und Channels und legt
|
||||
* fehlende Raum-Channels an. Pro aktivem Raum verbindet sich ein {@link TeamspeakRoomClient}, der im
|
||||
* Raum-Channel spricht und zuhört; nur diese Clients belegen Slots.
|
||||
*/
|
||||
public final class TeamspeakSide implements TS3Listener {
|
||||
public final class TeamspeakSide {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(TeamspeakSide.class);
|
||||
private static final long RECONNECT_SECONDS = 30;
|
||||
private static final long RECONNECT_SECONDS = 60;
|
||||
private static final long RESYNC_SECONDS = 60;
|
||||
/** client_type 0 = normaler Client, 1 = ServerQuery. */
|
||||
private static final int CLIENT_TYPE_VOICE = 0;
|
||||
private static final long RETRY_MIN_MILLIS = 30_000;
|
||||
@@ -64,7 +50,7 @@ public final class TeamspeakSide implements TS3Listener {
|
||||
private final Map<String, Long> backoff = new ConcurrentHashMap<>();
|
||||
private final Map<String, Long> nextAttempt = new ConcurrentHashMap<>();
|
||||
|
||||
private volatile LocalTeamspeakClientSocket observer;
|
||||
private volatile TsQuery query;
|
||||
private volatile boolean ready;
|
||||
/** Channel, unter dem die Raum-Channels liegen (0 = oberste Ebene). */
|
||||
private volatile int parentChannel;
|
||||
@@ -76,22 +62,8 @@ public final class TeamspeakSide implements TS3Listener {
|
||||
}
|
||||
|
||||
public void start() {
|
||||
executor.execute(this::connectObserver);
|
||||
executor.scheduleWithFixedDelay(this::retryMissingChannels, RECONNECT_SECONDS, RECONNECT_SECONDS, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
/** Fehlende Raum-Channels erneut anlegen, z. B. nachdem jemand dem Beobachter die Rechte gegeben hat. */
|
||||
private void retryMissingChannels() {
|
||||
LocalTeamspeakClientSocket socket = observer;
|
||||
if (!ready || socket == null || rooms.stream().allMatch(room -> roomChannels.containsKey(room.key()))) return;
|
||||
try {
|
||||
resolveChannels(socket);
|
||||
joinParent(socket);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
} catch (Exception e) {
|
||||
LOG.warn("Fehlende TeamSpeak-Channels: {}", e.toString());
|
||||
}
|
||||
executor.execute(this::connectQuery);
|
||||
executor.scheduleWithFixedDelay(this::resync, RESYNC_SECONDS, RESYNC_SECONDS, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
public boolean ready() {
|
||||
@@ -157,88 +129,78 @@ public final class TeamspeakSide implements TS3Listener {
|
||||
|
||||
private Set<Integer> ownClientIds() {
|
||||
Set<Integer> own = new HashSet<>();
|
||||
LocalTeamspeakClientSocket socket = observer;
|
||||
if (socket != null) own.add(socket.getClientId());
|
||||
for (TeamspeakRoomClient client : roomClients.values()) own.add(client.clientId());
|
||||
return own;
|
||||
}
|
||||
|
||||
private void connectObserver() {
|
||||
// --- ServerQuery ---
|
||||
|
||||
private void connectQuery() {
|
||||
try {
|
||||
LocalTeamspeakClientSocket socket = TeamspeakRoomClient.newSocket(config, loadIdentity("ts-observer.ini"),
|
||||
config.nickname());
|
||||
socket.addListener(this);
|
||||
observer = socket;
|
||||
socket.connect(new InetSocketAddress(config.host(), config.port()), emptyToNull(config.password()), 15000L);
|
||||
// Clients in nicht abonnierten Channels meldet der Server nicht
|
||||
socket.subscribeAll();
|
||||
awaitInitialState();
|
||||
LOG.info("TeamSpeak verbunden mit {}:{} ({} Channels, {} Clients sichtbar)",
|
||||
config.host(), config.port(), channels.size(), clients.size());
|
||||
TsQuery q = new TsQuery(config.host(), config.queryPort(), config.queryLogin(), config.queryPassword(),
|
||||
this::onNotification, () -> executor.execute(this::onQueryClosed));
|
||||
query = q;
|
||||
q.send("use", "port", Integer.toString(config.port()));
|
||||
q.send("servernotifyregister", "event", "server");
|
||||
q.send("servernotifyregister", "event", "channel", "id", "0");
|
||||
loadChannels(q);
|
||||
loadClients(q);
|
||||
parentChannel = resolveParent();
|
||||
resolveChannels(socket);
|
||||
joinParent(socket);
|
||||
resolveChannels(q);
|
||||
ready = true;
|
||||
LOG.info("TeamSpeak-Query verbunden mit {}:{} ({} Channels, {} Clients)",
|
||||
config.host(), config.queryPort(), channels.size(), clients.size());
|
||||
} catch (Exception e) {
|
||||
LOG.warn("TeamSpeak-Verbindung fehlgeschlagen ({}), neuer Versuch in {} s", e.toString(), RECONNECT_SECONDS);
|
||||
disconnectObserver();
|
||||
executor.schedule(this::connectObserver, RECONNECT_SECONDS, TimeUnit.SECONDS);
|
||||
LOG.warn("TeamSpeak-Query fehlgeschlagen ({}), neuer Versuch in {} s", e.toString(), RECONNECT_SECONDS);
|
||||
closeQuery();
|
||||
executor.schedule(this::connectQuery, RECONNECT_SECONDS, TimeUnit.SECONDS);
|
||||
}
|
||||
}
|
||||
|
||||
/** ts3j liefert die Channel- und Client-Liste asynchron aus; warten, bis sie sich nicht mehr ändert. */
|
||||
private void awaitInitialState() throws InterruptedException {
|
||||
int last = -1;
|
||||
for (int i = 0; i < 50; i++) {
|
||||
int now = channels.size() + clients.size();
|
||||
if (now == last && now > 0) return;
|
||||
last = now;
|
||||
Thread.sleep(200);
|
||||
private void onQueryClosed() {
|
||||
if (query == null) return; // absichtlich geschlossen
|
||||
LOG.warn("TeamSpeak-Query getrennt, neuer Versuch in {} s", RECONNECT_SECONDS);
|
||||
closeQuery();
|
||||
executor.schedule(this::connectQuery, RECONNECT_SECONDS, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
private void closeQuery() {
|
||||
ready = false;
|
||||
TsQuery q = query;
|
||||
query = null;
|
||||
if (q != null) q.close();
|
||||
clients.clear();
|
||||
channels.clear();
|
||||
}
|
||||
|
||||
/** Hält die Verbindung offen und gleicht verpasste Ereignisse aus. */
|
||||
private void resync() {
|
||||
TsQuery q = query;
|
||||
if (!ready || q == null) return;
|
||||
try {
|
||||
loadClients(q);
|
||||
if (rooms.stream().anyMatch(room -> !roomChannels.containsKey(room.key()))) resolveChannels(q);
|
||||
} catch (IOException e) {
|
||||
LOG.debug("TeamSpeak-Abgleich fehlgeschlagen: {}", e.toString());
|
||||
}
|
||||
}
|
||||
|
||||
private void resolveChannels(LocalTeamspeakClientSocket socket) throws InterruptedException {
|
||||
for (Room room : rooms) {
|
||||
if (roomChannels.containsKey(room.key())) continue;
|
||||
int configured = room.config().teamspeakChannel();
|
||||
if (configured > 0) {
|
||||
roomChannels.put(room.key(), configured);
|
||||
continue;
|
||||
}
|
||||
Integer found = findChannel(room.name());
|
||||
if (found == null) {
|
||||
try {
|
||||
socket.executeCommand(new SingleCommand("channelcreate", ProtocolRole.CLIENT,
|
||||
new CommandSingleParameter("channel_name", room.name()),
|
||||
new CommandSingleParameter("cpid", Integer.toString(parentChannel)),
|
||||
new CommandSingleParameter("channel_flag_permanent", "1"),
|
||||
new CommandSingleParameter("channel_codec", "4"))).get();
|
||||
} catch (InterruptedException e) {
|
||||
throw e;
|
||||
} catch (Exception e) {
|
||||
LOG.error("{}: TeamSpeak-Channel konnte nicht angelegt werden ({}). Fehlen dem Client "
|
||||
+ "\"{}\" die Rechte zum Anlegen? Bis dahin ohne TeamSpeak.", room.name(), e, config.nickname());
|
||||
continue;
|
||||
}
|
||||
// Die Antwort enthält die cid nicht, sie kommt per notifychannelcreated.
|
||||
for (int i = 0; i < 50 && found == null; i++) {
|
||||
Thread.sleep(100);
|
||||
found = findChannel(room.name());
|
||||
}
|
||||
if (found == null) {
|
||||
LOG.error("{}: TeamSpeak-Channel wurde nicht angelegt", room.name());
|
||||
continue;
|
||||
}
|
||||
LOG.info("TeamSpeak-Channel angelegt: {} ({})", room.name(), found);
|
||||
}
|
||||
roomChannels.put(room.key(), found);
|
||||
private void loadChannels(TsQuery q) throws IOException {
|
||||
Map<Integer, TsChannel> fresh = new ConcurrentHashMap<>();
|
||||
for (Map<String, String> c : q.send("channellist")) {
|
||||
fresh.put(parse(c.get("cid"), -1), new TsChannel(parse(c.get("pid"), 0), c.getOrDefault("channel_name", "")));
|
||||
}
|
||||
channels.clear();
|
||||
channels.putAll(fresh);
|
||||
}
|
||||
|
||||
/** Der Beobachter wartet im Eltern-Channel; wer einen Channel anlegt, kann dorthin verschoben werden. */
|
||||
private void joinParent(LocalTeamspeakClientSocket socket) throws Exception {
|
||||
TsClient self = clients.get(socket.getClientId());
|
||||
if (parentChannel > 0 && (self == null || self.channel() != parentChannel)) socket.joinChannel(parentChannel, null);
|
||||
private void loadClients(TsQuery q) throws IOException {
|
||||
Map<Integer, TsClient> fresh = new ConcurrentHashMap<>();
|
||||
for (Map<String, String> c : q.send("clientlist")) {
|
||||
fresh.put(parse(c.get("clid"), -1), new TsClient(parse(c.get("cid"), 0), parse(c.get("client_type"), 0)));
|
||||
}
|
||||
clients.clear();
|
||||
clients.putAll(fresh);
|
||||
}
|
||||
|
||||
/** {@code parentChannel} ist eine Channel-ID oder der Name eines Channels auf oberster Ebene. */
|
||||
@@ -249,12 +211,32 @@ public final class TeamspeakSide implements TS3Listener {
|
||||
if (id >= 0) return id;
|
||||
Integer found = findChannel(0, configured);
|
||||
if (found == null) throw new IOException("TeamSpeak-Channel \"" + configured + "\" gibt es nicht");
|
||||
LOG.info("TeamSpeak-Eltern-Channel {} hat die ID {}", configured, found);
|
||||
return found;
|
||||
}
|
||||
|
||||
private Integer findChannel(String name) {
|
||||
return findChannel(parentChannel, name);
|
||||
private void resolveChannels(TsQuery q) {
|
||||
for (Room room : rooms) {
|
||||
if (roomChannels.containsKey(room.key())) continue;
|
||||
int configured = room.config().teamspeakChannel();
|
||||
if (configured > 0) {
|
||||
roomChannels.put(room.key(), configured);
|
||||
continue;
|
||||
}
|
||||
Integer found = findChannel(parentChannel, room.name());
|
||||
if (found == null) {
|
||||
try {
|
||||
found = parse(q.send("channelcreate", "channel_name", room.name(), "cpid", Integer.toString(parentChannel),
|
||||
"channel_flag_permanent", "1", "channel_codec", "4").getFirst().get("cid"), -1);
|
||||
channels.put(found, new TsChannel(parentChannel, room.name()));
|
||||
LOG.info("TeamSpeak-Channel angelegt: {} ({})", room.name(), found);
|
||||
} catch (IOException e) {
|
||||
LOG.error("{}: TeamSpeak-Channel konnte nicht angelegt werden ({}). Fehlen dem Query-Login die "
|
||||
+ "Rechte zum Anlegen? Bis dahin ohne TeamSpeak.", room.name(), e.getMessage());
|
||||
continue;
|
||||
}
|
||||
}
|
||||
roomChannels.put(room.key(), found);
|
||||
}
|
||||
}
|
||||
|
||||
private Integer findChannel(int parent, String name) {
|
||||
@@ -264,22 +246,28 @@ public final class TeamspeakSide implements TS3Listener {
|
||||
return null;
|
||||
}
|
||||
|
||||
private void disconnectObserver() {
|
||||
ready = false;
|
||||
LocalTeamspeakClientSocket socket = observer;
|
||||
observer = null;
|
||||
if (socket != null) {
|
||||
socket.removeListener(this);
|
||||
try {
|
||||
socket.disconnect();
|
||||
} catch (Exception ignored) {
|
||||
// Verbindung ist ohnehin weg
|
||||
/** Lesethread der Query: nur Karten aktualisieren, keine Befehle schicken. */
|
||||
private void onNotification(TsQuery.Notification n) {
|
||||
Map<String, String> f = n.fields();
|
||||
int clid = parse(f.get("clid"), -1);
|
||||
int cid = parse(f.get("cid"), -1);
|
||||
switch (n.name()) {
|
||||
case "notifycliententerview" -> clients.put(clid, new TsClient(parse(f.get("ctid"), 0), parse(f.get("client_type"), 0)));
|
||||
case "notifyclientleftview" -> clients.remove(clid);
|
||||
case "notifyclientmoved" -> clients.compute(clid, (id, old) ->
|
||||
new TsClient(parse(f.get("ctid"), 0), old == null ? CLIENT_TYPE_VOICE : old.type()));
|
||||
case "notifychannelcreated" -> channels.put(cid, new TsChannel(parse(f.get("cpid"), 0), f.getOrDefault("channel_name", "")));
|
||||
case "notifychanneledited" -> {
|
||||
String name = f.get("channel_name");
|
||||
if (name != null) channels.computeIfPresent(cid, (id, old) -> new TsChannel(old.parent(), name));
|
||||
}
|
||||
case "notifychannelmoved" -> channels.computeIfPresent(cid, (id, old) -> new TsChannel(parse(f.get("cpid"), 0), old.name()));
|
||||
case "notifychanneldeleted" -> {
|
||||
channels.remove(cid);
|
||||
roomChannels.values().removeIf(c -> c == cid);
|
||||
}
|
||||
default -> {}
|
||||
}
|
||||
for (TeamspeakRoomClient client : roomClients.values()) client.disconnect();
|
||||
roomClients.clear();
|
||||
clients.clear();
|
||||
channels.clear();
|
||||
}
|
||||
|
||||
LocalIdentity loadIdentity(String file) throws IOException, GeneralSecurityException {
|
||||
@@ -306,73 +294,9 @@ public final class TeamspeakSide implements TS3Listener {
|
||||
|
||||
public void stop() {
|
||||
executor.shutdownNow();
|
||||
disconnectObserver();
|
||||
}
|
||||
|
||||
// --- Ereignisse des Beobachters ---
|
||||
|
||||
@Override
|
||||
public void onClientJoin(ClientJoinEvent e) {
|
||||
clients.put(e.getClientId(), new TsClient(e.getClientTargetId(), e.getClientType()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onClientMoved(ClientMovedEvent e) {
|
||||
clients.compute(e.getClientId(), (id, old) ->
|
||||
new TsClient(e.getTargetChannelId(), old == null ? CLIENT_TYPE_VOICE : old.type()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onClientLeave(ClientLeaveEvent e) {
|
||||
clients.remove(e.getClientId());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onChannelList(ChannelListEvent e) {
|
||||
putChannel(e.getMap());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onChannelCreate(ChannelCreateEvent e) {
|
||||
putChannel(e.getMap());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onChannelEdit(ChannelEditedEvent e) {
|
||||
Map<String, String> map = e.getMap();
|
||||
String name = map.get("channel_name");
|
||||
int cid = parse(map.get("cid"), -1);
|
||||
if (name != null && cid >= 0) channels.computeIfPresent(cid, (id, old) -> new TsChannel(old.parent(), name));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onChannelMoved(ChannelMovedEvent e) {
|
||||
Map<String, String> map = e.getMap();
|
||||
int cid = parse(map.get("cid"), -1);
|
||||
int parent = parse(map.get("cpid"), 0);
|
||||
channels.computeIfPresent(cid, (id, old) -> new TsChannel(parent, old.name()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onChannelDeleted(ChannelDeletedEvent e) {
|
||||
channels.remove(e.getChannelId());
|
||||
roomChannels.values().removeIf(cid -> cid == e.getChannelId());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onDisconnected(DisconnectedEvent e) {
|
||||
// Beim fehlgeschlagenen Verbinden plant connectObserver den Neuversuch selbst
|
||||
if (!ready) return;
|
||||
ready = false;
|
||||
LOG.warn("TeamSpeak-Beobachter getrennt, neuer Versuch in {} s", RECONNECT_SECONDS);
|
||||
executor.execute(this::disconnectObserver);
|
||||
executor.schedule(this::connectObserver, RECONNECT_SECONDS, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
private void putChannel(Map<String, String> map) {
|
||||
int cid = parse(map.get("cid"), -1);
|
||||
if (cid < 0) return;
|
||||
channels.put(cid, new TsChannel(parse(map.get("cpid"), 0), map.getOrDefault("channel_name", "")));
|
||||
for (TeamspeakRoomClient client : roomClients.values()) client.disconnect();
|
||||
roomClients.clear();
|
||||
closeQuery();
|
||||
}
|
||||
|
||||
private static int parse(String value, int def) {
|
||||
|
||||
@@ -0,0 +1,207 @@
|
||||
package eu.mhsl.voicebridge.bridge.teamspeak;
|
||||
|
||||
import com.jcraft.jsch.ChannelShell;
|
||||
import com.jcraft.jsch.JSch;
|
||||
import com.jcraft.jsch.JSchException;
|
||||
import com.jcraft.jsch.Session;
|
||||
|
||||
import java.io.BufferedReader;
|
||||
import java.io.IOException;
|
||||
import java.io.InputStreamReader;
|
||||
import java.io.OutputStream;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ArrayBlockingQueue;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
/**
|
||||
* ServerQuery über SSH (Port 10022). Query-Verbindungen belegen keinen Client-Slot.
|
||||
* Antworten kommen als Datenzeilen und einer abschließenden {@code error id=…}-Zeile,
|
||||
* Ereignisse (nach {@code servernotifyregister}) als {@code notify…}-Zeilen dazwischen.
|
||||
*/
|
||||
public final class TsQuery implements AutoCloseable {
|
||||
public static final class QueryException extends IOException {
|
||||
public final int id;
|
||||
|
||||
QueryException(int id, String message) {
|
||||
super("Fehler " + id + ": " + message);
|
||||
this.id = id;
|
||||
}
|
||||
}
|
||||
|
||||
/** Ein Ereignis: Name (z. B. {@code notifycliententerview}) und seine Felder. */
|
||||
public record Notification(String name, Map<String, String> fields) {}
|
||||
|
||||
private final Session session;
|
||||
private final ChannelShell channel;
|
||||
private final OutputStream out;
|
||||
private final BlockingQueue<Object> responses = new ArrayBlockingQueue<>(4);
|
||||
private final Consumer<Notification> notifications;
|
||||
private final Runnable onClose;
|
||||
private final Thread reader;
|
||||
private volatile boolean closed;
|
||||
private long lastCommand;
|
||||
/** Mindestabstand zwischen zwei Befehlen (die Query sperrt die IP bei zu vielen Befehlen). */
|
||||
private static final long MIN_INTERVAL_MILLIS = 1000;
|
||||
|
||||
public TsQuery(String host, int port, String user, String password,
|
||||
Consumer<Notification> notifications, Runnable onClose) throws JSchException, IOException {
|
||||
this.notifications = notifications;
|
||||
this.onClose = onClose;
|
||||
JSch jsch = new JSch();
|
||||
session = jsch.getSession(user, host, port);
|
||||
session.setPassword(password);
|
||||
// TeamSpeak erzeugt seinen Host-Key selbst, es gibt nichts zum Abgleichen
|
||||
session.setConfig("StrictHostKeyChecking", "no");
|
||||
session.setServerAliveInterval(60_000);
|
||||
session.connect(15_000);
|
||||
channel = (ChannelShell) session.openChannel("shell");
|
||||
channel.setPty(false);
|
||||
BufferedReader in = new BufferedReader(new InputStreamReader(channel.getInputStream(), StandardCharsets.UTF_8));
|
||||
out = channel.getOutputStream();
|
||||
channel.connect(15_000);
|
||||
reader = Thread.ofPlatform().daemon().name("ts-query").start(() -> readLoop(in));
|
||||
}
|
||||
|
||||
private void readLoop(BufferedReader in) {
|
||||
List<Map<String, String>> data = new ArrayList<>();
|
||||
try {
|
||||
String line;
|
||||
while ((line = in.readLine()) != null) {
|
||||
line = line.strip();
|
||||
if (line.isEmpty() || line.equals("TS3") || line.startsWith("Welcome") || line.startsWith("Please type")) continue;
|
||||
if (line.startsWith("notify")) {
|
||||
int space = line.indexOf(' ');
|
||||
String name = space < 0 ? line : line.substring(0, space);
|
||||
for (Map<String, String> entry : parse(space < 0 ? "" : line.substring(space + 1))) {
|
||||
notifications.accept(new Notification(name, entry));
|
||||
}
|
||||
} else if (line.startsWith("error ")) {
|
||||
Map<String, String> error = parse(line.substring(6)).getFirst();
|
||||
int id = Integer.parseInt(error.getOrDefault("id", "-1"));
|
||||
responses.put(id == 0 ? List.copyOf(data) : new QueryException(id, error.getOrDefault("msg", "")));
|
||||
data.clear();
|
||||
} else {
|
||||
data.addAll(parse(line));
|
||||
}
|
||||
}
|
||||
} catch (IOException | InterruptedException e) {
|
||||
// Verbindung beendet
|
||||
} finally {
|
||||
closed = true;
|
||||
responses.offer(new IOException("Query-Verbindung getrennt"));
|
||||
onClose.run();
|
||||
}
|
||||
}
|
||||
|
||||
/** Befehl schicken und auf die Antwort warten. Flags wie {@code -uid} gehören in {@code command}. */
|
||||
@SuppressWarnings("unchecked")
|
||||
public synchronized List<Map<String, String>> send(String command, String... keyValues) throws IOException {
|
||||
if (closed) throw new IOException("Query-Verbindung getrennt");
|
||||
StringBuilder line = new StringBuilder(command);
|
||||
for (int i = 0; i + 1 < keyValues.length; i += 2) {
|
||||
line.append(' ').append(keyValues[i]).append('=').append(escape(keyValues[i + 1]));
|
||||
}
|
||||
// Flood-Schutz der Query: lieber langsam als eine IP-Sperre
|
||||
long wait = lastCommand + MIN_INTERVAL_MILLIS - System.currentTimeMillis();
|
||||
if (wait > 0) {
|
||||
try {
|
||||
Thread.sleep(wait);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new IOException("unterbrochen", e);
|
||||
}
|
||||
}
|
||||
lastCommand = System.currentTimeMillis();
|
||||
responses.clear();
|
||||
out.write((line + "\n").getBytes(StandardCharsets.UTF_8));
|
||||
out.flush();
|
||||
Object response;
|
||||
try {
|
||||
response = responses.poll(10, TimeUnit.SECONDS);
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
throw new IOException("unterbrochen", e);
|
||||
}
|
||||
if (response == null) throw new IOException("Keine Antwort auf " + command);
|
||||
if (response instanceof IOException e) throw e;
|
||||
return (List<Map<String, String>>) response;
|
||||
}
|
||||
|
||||
public boolean isOpen() {
|
||||
return !closed;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
closed = true;
|
||||
try {
|
||||
out.write("quit\n".getBytes(StandardCharsets.UTF_8));
|
||||
out.flush();
|
||||
} catch (IOException ignored) {
|
||||
// Verbindung ist ohnehin weg
|
||||
}
|
||||
channel.disconnect();
|
||||
session.disconnect();
|
||||
}
|
||||
|
||||
static List<Map<String, String>> parse(String text) {
|
||||
List<Map<String, String>> entries = new ArrayList<>();
|
||||
for (String entry : text.split("\\|")) {
|
||||
Map<String, String> map = new LinkedHashMap<>();
|
||||
for (String token : entry.split(" ")) {
|
||||
if (token.isEmpty()) continue;
|
||||
int eq = token.indexOf('=');
|
||||
if (eq < 0) map.put(token, "");
|
||||
else map.put(token.substring(0, eq), unescape(token.substring(eq + 1)));
|
||||
}
|
||||
entries.add(map);
|
||||
}
|
||||
return entries;
|
||||
}
|
||||
|
||||
static String escape(String value) {
|
||||
StringBuilder sb = new StringBuilder(value.length());
|
||||
for (char c : value.toCharArray()) {
|
||||
switch (c) {
|
||||
case '\\' -> sb.append("\\\\");
|
||||
case '/' -> sb.append("\\/");
|
||||
case ' ' -> sb.append("\\s");
|
||||
case '|' -> sb.append("\\p");
|
||||
case '\n' -> sb.append("\\n");
|
||||
case '\r' -> sb.append("\\r");
|
||||
case '\t' -> sb.append("\\t");
|
||||
default -> sb.append(c);
|
||||
}
|
||||
}
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
static String unescape(String value) {
|
||||
StringBuilder sb = new StringBuilder(value.length());
|
||||
for (int i = 0; i < value.length(); i++) {
|
||||
char c = value.charAt(i);
|
||||
if (c != '\\' || i + 1 == value.length()) {
|
||||
sb.append(c);
|
||||
continue;
|
||||
}
|
||||
char next = value.charAt(++i);
|
||||
switch (next) {
|
||||
case 's' -> sb.append(' ');
|
||||
case 'p' -> sb.append('|');
|
||||
case '/' -> sb.append('/');
|
||||
case 'n' -> sb.append('\n');
|
||||
case 'r' -> sb.append('\r');
|
||||
case 't' -> sb.append('\t');
|
||||
case '\\' -> sb.append('\\');
|
||||
default -> sb.append(next);
|
||||
}
|
||||
}
|
||||
return sb.toString();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
package eu.mhsl.voicebridge.bridge.teamspeak;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
class TsQueryTest {
|
||||
|
||||
@Test
|
||||
void escapeRoundTrip() {
|
||||
String value = "Crosstalk 1 (MC/DC) | a\\b";
|
||||
assertEquals("Crosstalk\\s1\\s(MC\\/DC)\\s\\p\\sa\\\\b", TsQuery.escape(value));
|
||||
assertEquals(value, TsQuery.unescape(TsQuery.escape(value)));
|
||||
}
|
||||
|
||||
@Test
|
||||
void parsesListEntries() {
|
||||
List<Map<String, String>> entries = TsQuery.parse(
|
||||
"clid=5 cid=265 client_nickname=Crosstalk\\s1\\s(MC\\/DC) client_type=0|clid=7 cid=1 client_type=1");
|
||||
assertEquals(2, entries.size());
|
||||
assertEquals("Crosstalk 1 (MC/DC)", entries.get(0).get("client_nickname"));
|
||||
assertEquals("265", entries.get(0).get("cid"));
|
||||
assertEquals("1", entries.get(1).get("client_type"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void parsesFlagsWithoutValue() {
|
||||
assertEquals("", TsQuery.parse("cid=3 channel_flag_permanent").getFirst().get("channel_flag_permanent"));
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user