Voice-Bridge: Plugin, Bridge-Dienst und Protokoll
This commit is contained in:
MineTec committed
commit
c0c290d3dc
35 files changed
+2497
No files matched your search
@@ -0,0 +1,128 @@
|
||||
package eu.mhsl.voicebridge.bridge;
|
||||
|
||||
import eu.mhsl.voicebridge.bridge.discord.DiscordRoomBot;
|
||||
import eu.mhsl.voicebridge.bridge.teamspeak.TeamspeakSide;
|
||||
import eu.mhsl.voicebridge.protocol.Message;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* Steuert die Räume: jede Sekunde wird gezählt, wer auf welcher Seite ist. Ein Raum ist aktiv, sobald
|
||||
* auf mindestens zwei Seiten jemand ist. Die Bots einer Seite verbinden sich nur, wenn der Raum aktiv
|
||||
* ist und auf ihrer Seite jemand sitzt; danach bleiben sie noch {@code lingerSeconds} verbunden.
|
||||
*/
|
||||
public final class Bridge {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Bridge.class);
|
||||
|
||||
private final BridgeConfig config;
|
||||
private final Map<String, Room> rooms = new LinkedHashMap<>();
|
||||
private final Map<String, DiscordRoomBot> discordBots = new HashMap<>();
|
||||
private final Map<String, Presence> presence = new HashMap<>();
|
||||
private final PluginLink plugin;
|
||||
private final TeamspeakSide teamspeak;
|
||||
private final ScheduledExecutorService ticker = Executors.newSingleThreadScheduledExecutor(
|
||||
Thread.ofPlatform().name("bridge-tick").factory());
|
||||
|
||||
private volatile List<Message.RoomInfo> roomInfos = List.of();
|
||||
|
||||
/** Zeitpunkte, bis zu denen Raum bzw. Bot noch gebraucht werden (für das Nachlaufen). */
|
||||
private static final class Presence {
|
||||
long roomWanted, teamspeakWanted, discordWanted;
|
||||
boolean teamspeakConnected, discordConnected;
|
||||
}
|
||||
|
||||
public Bridge(BridgeConfig config) throws IOException {
|
||||
this.config = config;
|
||||
for (BridgeConfig.RoomConfig roomConfig : config.rooms()) {
|
||||
Room room = new Room(roomConfig);
|
||||
rooms.put(room.key(), room);
|
||||
presence.put(room.key(), new Presence());
|
||||
if (roomConfig.discordEnabled()) discordBots.put(room.key(), new DiscordRoomBot(config.discord(), room));
|
||||
}
|
||||
this.plugin = new PluginLink(config.pluginBind(), config.pluginPort(), rooms, () -> roomInfos);
|
||||
for (Room room : rooms.values()) room.setMinecraftSink(plugin::sendAudio);
|
||||
this.teamspeak = config.teamspeak().enabled()
|
||||
? new TeamspeakSide(config.teamspeak(), config.dataDir(), List.copyOf(rooms.values()))
|
||||
: null;
|
||||
roomInfos = buildInfos();
|
||||
}
|
||||
|
||||
public void start() {
|
||||
LOG.info("Starte Bridge mit {} Räumen (TeamSpeak {}, Discord-Bots {})", rooms.size(),
|
||||
teamspeak != null ? "an" : "aus", discordBots.size());
|
||||
plugin.start();
|
||||
if (teamspeak != null) teamspeak.start();
|
||||
discordBots.values().forEach(DiscordRoomBot::start);
|
||||
ticker.scheduleAtFixedRate(this::safeTick, 1, 1, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
private void safeTick() {
|
||||
try {
|
||||
tick();
|
||||
} catch (RuntimeException e) {
|
||||
LOG.error("Fehler im Bridge-Takt", e);
|
||||
}
|
||||
}
|
||||
|
||||
private void tick() {
|
||||
long now = System.currentTimeMillis();
|
||||
long linger = config.lingerMillis();
|
||||
|
||||
for (Room room : rooms.values()) {
|
||||
Presence p = presence.get(room.key());
|
||||
DiscordRoomBot bot = discordBots.get(room.key());
|
||||
int mc = plugin.minecraftMembers(room.key());
|
||||
int ts = teamspeak != null ? teamspeak.humansIn(room) : 0;
|
||||
int dc = bot != null ? bot.humans() : 0;
|
||||
int sides = (mc > 0 ? 1 : 0) + (ts > 0 ? 1 : 0) + (dc > 0 ? 1 : 0);
|
||||
|
||||
if (sides >= 2) p.roomWanted = now;
|
||||
boolean active = now - p.roomWanted < linger;
|
||||
room.setActive(active);
|
||||
|
||||
if (active && ts > 0) p.teamspeakWanted = now;
|
||||
boolean tsConnect = active && now - p.teamspeakWanted < linger;
|
||||
if (teamspeak != null && (tsConnect || p.teamspeakConnected)) {
|
||||
teamspeak.setRoomConnected(room, tsConnect);
|
||||
}
|
||||
p.teamspeakConnected = tsConnect;
|
||||
|
||||
if (active && dc > 0) p.discordWanted = now;
|
||||
boolean dcConnect = active && now - p.discordWanted < linger;
|
||||
if (bot != null && dcConnect != p.discordConnected) bot.setConnected(dcConnect);
|
||||
p.discordConnected = dcConnect;
|
||||
|
||||
room.dropIdleSources();
|
||||
}
|
||||
roomInfos = buildInfos();
|
||||
}
|
||||
|
||||
private List<Message.RoomInfo> buildInfos() {
|
||||
List<Message.RoomInfo> infos = new ArrayList<>(rooms.size());
|
||||
for (Room room : rooms.values()) {
|
||||
DiscordRoomBot bot = discordBots.get(room.key());
|
||||
infos.add(new Message.RoomInfo(room.key(), room.name(), room.groupId(),
|
||||
teamspeak != null ? teamspeak.humansIn(room) : 0,
|
||||
bot != null ? bot.humans() : 0,
|
||||
room.active()));
|
||||
}
|
||||
return infos;
|
||||
}
|
||||
|
||||
public void stop() {
|
||||
ticker.shutdownNow();
|
||||
plugin.close();
|
||||
if (teamspeak != null) teamspeak.stop();
|
||||
discordBots.values().forEach(DiscordRoomBot::stop);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
package eu.mhsl.voicebridge.bridge;
|
||||
|
||||
import org.yaml.snakeyaml.Yaml;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.io.Reader;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
public record BridgeConfig(
|
||||
String pluginBind,
|
||||
int pluginPort,
|
||||
Path dataDir,
|
||||
long lingerMillis,
|
||||
Teamspeak teamspeak,
|
||||
Discord discord,
|
||||
List<RoomConfig> rooms
|
||||
) {
|
||||
public record Teamspeak(String host, int port, String password, int parentChannel, String nickname,
|
||||
String roomNickname, int securityLevel, ClientVersion clientVersion) {
|
||||
public boolean enabled() { return !host.isBlank(); }
|
||||
}
|
||||
|
||||
public record ClientVersion(String platform, String version, String sign) {}
|
||||
|
||||
public record Discord(String guild, String category, String nickname) {}
|
||||
|
||||
public record RoomConfig(String key, String name, String discordToken, String discordChannel,
|
||||
int teamspeakChannel) {
|
||||
public boolean discordEnabled() { return !discordToken.isBlank(); }
|
||||
}
|
||||
|
||||
public static BridgeConfig load(Path file) throws IOException {
|
||||
Map<String, Object> root;
|
||||
try (Reader reader = Files.newBufferedReader(file)) {
|
||||
root = new Yaml().load(reader);
|
||||
}
|
||||
if (root == null) throw new IOException(file + " ist leer");
|
||||
|
||||
Map<String, Object> plugin = section(root, "plugin");
|
||||
Map<String, Object> ts = section(root, "teamspeak");
|
||||
Map<String, Object> dc = section(root, "discord");
|
||||
|
||||
ClientVersion version = null;
|
||||
if (ts.get("clientVersion") instanceof Map<?, ?> v) {
|
||||
@SuppressWarnings("unchecked") Map<String, Object> cv = (Map<String, Object>) v;
|
||||
version = new ClientVersion(string(cv, "platform", ""), string(cv, "version", ""), string(cv, "sign", ""));
|
||||
}
|
||||
|
||||
List<RoomConfig> rooms = new ArrayList<>();
|
||||
Set<String> keys = new HashSet<>();
|
||||
if (root.get("rooms") instanceof List<?> list) {
|
||||
for (Object entry : list) {
|
||||
if (!(entry instanceof Map<?, ?> raw)) continue;
|
||||
@SuppressWarnings("unchecked") Map<String, Object> room = (Map<String, Object>) raw;
|
||||
String key = string(room, "key", "");
|
||||
if (key.isBlank()) throw new IOException("Raum ohne key in " + file);
|
||||
if (!keys.add(key)) throw new IOException("Raum-key doppelt: " + key);
|
||||
rooms.add(new RoomConfig(key, string(room, "name", key), string(room, "discordToken", ""),
|
||||
string(room, "discordChannel", ""), integer(room, "teamspeakChannel", 0)));
|
||||
}
|
||||
}
|
||||
|
||||
return new BridgeConfig(
|
||||
string(plugin, "bind", "127.0.0.1"),
|
||||
integer(plugin, "port", 24470),
|
||||
Path.of(string(root, "dataDir", "data")),
|
||||
integer(root, "lingerSeconds", 15) * 1000L,
|
||||
new Teamspeak(string(ts, "host", ""), integer(ts, "port", 9987), string(ts, "password", ""),
|
||||
integer(ts, "parentChannel", 0), string(ts, "nickname", "Crosstalk"),
|
||||
string(ts, "roomNickname", "%room% (Minecraft/Discord)"), integer(ts, "securityLevel", 8), version),
|
||||
new Discord(string(dc, "guild", ""), string(dc, "category", ""),
|
||||
string(dc, "nickname", "%room% (Minecraft/TS)")),
|
||||
List.copyOf(rooms));
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static Map<String, Object> section(Map<String, Object> map, String key) {
|
||||
return map.get(key) instanceof Map<?, ?> m ? (Map<String, Object>) m : Map.of();
|
||||
}
|
||||
|
||||
private static String string(Map<String, Object> map, String key, String def) {
|
||||
Object value = map.get(key);
|
||||
return value == null ? def : value.toString().trim();
|
||||
}
|
||||
|
||||
private static int integer(Map<String, Object> map, String key, int def) {
|
||||
Object value = map.get(key);
|
||||
if (value == null) return def;
|
||||
if (value instanceof Number n) return n.intValue();
|
||||
return Integer.parseInt(value.toString().trim());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package eu.mhsl.voicebridge.bridge;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
|
||||
public final class Main {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Main.class);
|
||||
|
||||
private Main() {}
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
Path configFile = Path.of(args.length > 0 ? args[0] : "config.yml");
|
||||
if (!Files.exists(configFile)) {
|
||||
LOG.error("{} fehlt. Vorlage: bridge/config.example.yml", configFile.toAbsolutePath());
|
||||
System.exit(1);
|
||||
}
|
||||
|
||||
Bridge bridge = new Bridge(BridgeConfig.load(configFile));
|
||||
Runtime.getRuntime().addShutdownHook(new Thread(bridge::stop, "shutdown"));
|
||||
bridge.start();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,114 @@
|
||||
package eu.mhsl.voicebridge.bridge;
|
||||
|
||||
import eu.mhsl.voicebridge.protocol.Message;
|
||||
import eu.mhsl.voicebridge.protocol.Protocol;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.DatagramPacket;
|
||||
import java.net.DatagramSocket;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.net.SocketAddress;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
/** UDP-Gegenstelle zum Paper-Plugin. Das Plugin meldet sich jede Sekunde mit einem Hello. */
|
||||
final class PluginLink {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(PluginLink.class);
|
||||
private static final long ALIVE_MILLIS = 5000;
|
||||
|
||||
private final DatagramSocket socket;
|
||||
private final Map<String, Room> rooms;
|
||||
private final Supplier<List<Message.RoomInfo>> roomInfos;
|
||||
private final Map<String, Integer> minecraftMembers = new ConcurrentHashMap<>();
|
||||
|
||||
private volatile SocketAddress plugin;
|
||||
private volatile long lastHello;
|
||||
private boolean warnedVersion;
|
||||
|
||||
PluginLink(String bind, int port, Map<String, Room> rooms, Supplier<List<Message.RoomInfo>> roomInfos)
|
||||
throws IOException {
|
||||
this.socket = new DatagramSocket(new InetSocketAddress(bind, port));
|
||||
this.rooms = rooms;
|
||||
this.roomInfos = roomInfos;
|
||||
}
|
||||
|
||||
void start() {
|
||||
Thread.ofPlatform().daemon().name("plugin-link").start(this::receiveLoop);
|
||||
LOG.info("Warte auf das Paper-Plugin auf {}", socket.getLocalSocketAddress());
|
||||
}
|
||||
|
||||
boolean alive() {
|
||||
return System.currentTimeMillis() - lastHello < ALIVE_MILLIS;
|
||||
}
|
||||
|
||||
int minecraftMembers(String room) {
|
||||
return alive() ? minecraftMembers.getOrDefault(room, 0) : 0;
|
||||
}
|
||||
|
||||
void sendAudio(String room, UUID source, byte[] opus) {
|
||||
SocketAddress target = plugin;
|
||||
if (target == null || !alive()) return;
|
||||
send(new Message.Audio(room, source, opus), target);
|
||||
}
|
||||
|
||||
private void send(Message message, SocketAddress target) {
|
||||
byte[] data = Protocol.encode(message);
|
||||
try {
|
||||
socket.send(new DatagramPacket(data, data.length, target));
|
||||
} catch (IOException e) {
|
||||
if (!socket.isClosed()) LOG.debug("Senden ans Plugin fehlgeschlagen: {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private void receiveLoop() {
|
||||
byte[] buffer = new byte[Protocol.MAX_DATAGRAM];
|
||||
DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
|
||||
while (!socket.isClosed()) {
|
||||
try {
|
||||
packet.setLength(buffer.length);
|
||||
socket.receive(packet);
|
||||
handle(Protocol.decode(packet.getData(), packet.getOffset(), packet.getLength()),
|
||||
packet.getSocketAddress());
|
||||
} catch (IOException e) {
|
||||
if (!socket.isClosed()) LOG.warn("Ungültiges Paket vom Plugin: {}", e.getMessage());
|
||||
} catch (RuntimeException e) {
|
||||
LOG.warn("Fehler beim Verarbeiten eines Plugin-Pakets", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void handle(Message message, SocketAddress from) {
|
||||
switch (message) {
|
||||
case Message.Hello hello -> {
|
||||
if (hello.version() != Protocol.VERSION) {
|
||||
if (!warnedVersion) LOG.error("Plugin spricht Protokoll {}, Bridge {}: bitte beide aktualisieren",
|
||||
hello.version(), Protocol.VERSION);
|
||||
warnedVersion = true;
|
||||
return;
|
||||
}
|
||||
if (!alive() || !from.equals(plugin)) LOG.info("Paper-Plugin verbunden von {}", from);
|
||||
plugin = from;
|
||||
lastHello = System.currentTimeMillis();
|
||||
minecraftMembers.clear();
|
||||
minecraftMembers.putAll(hello.members());
|
||||
send(new Message.Rooms(new ArrayList<>(roomInfos.get())), from);
|
||||
}
|
||||
case Message.Audio audio -> {
|
||||
if (!from.equals(plugin)) return;
|
||||
Room room = rooms.get(audio.room());
|
||||
if (room != null) room.fromMinecraft(audio.source(), audio.opus());
|
||||
}
|
||||
case Message.Rooms ignored -> {}
|
||||
}
|
||||
}
|
||||
|
||||
void close() {
|
||||
socket.close();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
package eu.mhsl.voicebridge.bridge;
|
||||
|
||||
import eu.mhsl.voicebridge.bridge.audio.FrameSplitter;
|
||||
import eu.mhsl.voicebridge.bridge.audio.Mixer;
|
||||
import eu.mhsl.voicebridge.bridge.audio.Opus;
|
||||
import io.github.jaredmdobson.concentus.OpusDecoder;
|
||||
import io.github.jaredmdobson.concentus.OpusEncoder;
|
||||
import io.github.jaredmdobson.concentus.OpusException;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
/**
|
||||
* Ein Crosstalk-Raum aus drei Seiten. Jede Seite bekommt das Audio der beiden anderen:
|
||||
* Discord und TeamSpeak je einen gemischten Stream, Minecraft jeden Sprecher einzeln.
|
||||
*/
|
||||
public final class Room {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(Room.class);
|
||||
private static final long SOURCE_IDLE_MILLIS = 5000;
|
||||
|
||||
public enum Side { MINECRAFT, TEAMSPEAK, DISCORD }
|
||||
|
||||
private final BridgeConfig.RoomConfig config;
|
||||
private final UUID groupId;
|
||||
private final Mixer toTeamspeak = new Mixer();
|
||||
private final Mixer toDiscord = new Mixer();
|
||||
private final Map<String, Source> sources = new ConcurrentHashMap<>();
|
||||
|
||||
/** Wohin Frames für Minecraft gehen (Plugin-Verbindung). */
|
||||
private volatile MinecraftSink minecraft = (room, source, opus) -> {};
|
||||
|
||||
private volatile boolean active;
|
||||
|
||||
public interface MinecraftSink {
|
||||
void send(String room, UUID source, byte[] opus);
|
||||
}
|
||||
|
||||
/** Ein Sprecher: eigener Decoder (Opus ist zustandsbehaftet) und für Minecraft ein eigener Encoder. */
|
||||
private final class Source {
|
||||
final Side side;
|
||||
final String key;
|
||||
final UUID uuid;
|
||||
final OpusDecoder decoder = Opus.decoder();
|
||||
final FrameSplitter splitter = new FrameSplitter();
|
||||
OpusEncoder minecraftEncoder;
|
||||
volatile long lastSeen;
|
||||
|
||||
Source(Side side, String key) {
|
||||
this.side = side;
|
||||
this.key = key;
|
||||
this.uuid = UUID.nameUUIDFromBytes(("voicebridge:" + config.key() + ":" + key).getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
}
|
||||
|
||||
public Room(BridgeConfig.RoomConfig config) {
|
||||
this.config = config;
|
||||
this.groupId = UUID.nameUUIDFromBytes(("voicebridge-group:" + config.key()).getBytes(StandardCharsets.UTF_8));
|
||||
}
|
||||
|
||||
public BridgeConfig.RoomConfig config() { return config; }
|
||||
public String key() { return config.key(); }
|
||||
public String name() { return config.name(); }
|
||||
public UUID groupId() { return groupId; }
|
||||
public boolean active() { return active; }
|
||||
public Mixer toTeamspeak() { return toTeamspeak; }
|
||||
public Mixer toDiscord() { return toDiscord; }
|
||||
|
||||
public void setMinecraftSink(MinecraftSink sink) { this.minecraft = sink; }
|
||||
|
||||
void setActive(boolean active) {
|
||||
if (this.active == active) return;
|
||||
this.active = active;
|
||||
if (!active) {
|
||||
toTeamspeak.clear();
|
||||
toDiscord.clear();
|
||||
sources.clear();
|
||||
}
|
||||
LOG.info("{}: {}", name(), active ? "aktiv" : "inaktiv");
|
||||
}
|
||||
|
||||
public void fromMinecraft(UUID player, byte[] opus) {
|
||||
receive(Side.MINECRAFT, "mc:" + player, opus);
|
||||
}
|
||||
|
||||
public void fromTeamspeak(int clientId, byte[] opus) {
|
||||
receive(Side.TEAMSPEAK, "ts:" + clientId, opus);
|
||||
}
|
||||
|
||||
public void fromDiscord(long userId, byte[] opus) {
|
||||
receive(Side.DISCORD, "dc:" + userId, opus);
|
||||
}
|
||||
|
||||
private void receive(Side side, String key, byte[] opus) {
|
||||
if (!active || opus.length == 0) return;
|
||||
Source source = sources.computeIfAbsent(key, k -> new Source(side, k));
|
||||
source.lastSeen = System.currentTimeMillis();
|
||||
|
||||
List<short[]> frames;
|
||||
try {
|
||||
frames = source.splitter.push(Opus.decode(source.decoder, opus));
|
||||
} catch (OpusException e) {
|
||||
LOG.debug("{}: Opus-Paket von {} nicht dekodierbar: {}", name(), key, e.getMessage());
|
||||
return;
|
||||
}
|
||||
|
||||
for (short[] frame : frames) {
|
||||
if (side != Side.TEAMSPEAK) toTeamspeak.push(key, frame);
|
||||
if (side != Side.DISCORD) toDiscord.push(key, frame);
|
||||
if (side != Side.MINECRAFT) sendToMinecraft(source, frame);
|
||||
}
|
||||
}
|
||||
|
||||
private void sendToMinecraft(Source source, short[] frame) {
|
||||
try {
|
||||
if (source.minecraftEncoder == null) source.minecraftEncoder = Opus.encoder(1, 48000);
|
||||
minecraft.send(key(), source.uuid, Opus.encodeMono(source.minecraftEncoder, frame));
|
||||
} catch (OpusException e) {
|
||||
LOG.debug("{}: Kodieren für Minecraft fehlgeschlagen: {}", name(), e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
void dropIdleSources() {
|
||||
long now = System.currentTimeMillis();
|
||||
sources.values().removeIf(source -> now - source.lastSeen > SOURCE_IDLE_MILLIS);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package eu.mhsl.voicebridge.bridge.audio;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/** Zerlegt dekodiertes PCM beliebiger Länge in Frames zu genau 20 ms. */
|
||||
public final class FrameSplitter {
|
||||
private final short[] pending = new short[Opus.FRAME_SAMPLES];
|
||||
private int filled;
|
||||
|
||||
public List<short[]> push(short[] pcm) {
|
||||
List<short[]> frames = new ArrayList<>(pcm.length / Opus.FRAME_SAMPLES + 1);
|
||||
int offset = 0;
|
||||
while (offset < pcm.length) {
|
||||
int take = Math.min(Opus.FRAME_SAMPLES - filled, pcm.length - offset);
|
||||
System.arraycopy(pcm, offset, pending, filled, take);
|
||||
filled += take;
|
||||
offset += take;
|
||||
if (filled == Opus.FRAME_SAMPLES) {
|
||||
frames.add(pending.clone());
|
||||
filled = 0;
|
||||
}
|
||||
}
|
||||
return frames;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
package eu.mhsl.voicebridge.bridge.audio;
|
||||
|
||||
import java.util.ArrayDeque;
|
||||
import java.util.HashMap;
|
||||
import java.util.Iterator;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Mischt die Sprecher für eine Ausgabe (Discord- oder TS-Bot), die alle 20 ms einen Frame abholt.
|
||||
* Jeder Sprecher hat einen kleinen Jitter-Puffer: abgespielt wird erst ab {@link #PRIME} Frames,
|
||||
* über {@link #MAX} Frames werden die ältesten verworfen, damit die Verzögerung nicht wächst.
|
||||
*/
|
||||
public final class Mixer {
|
||||
static final int PRIME = 2;
|
||||
static final int MAX = 8;
|
||||
private static final long IDLE_MILLIS = 2000;
|
||||
|
||||
private static final class Queue {
|
||||
final ArrayDeque<short[]> frames = new ArrayDeque<>();
|
||||
boolean playing;
|
||||
long lastPush;
|
||||
}
|
||||
|
||||
private final Map<String, Queue> queues = new HashMap<>();
|
||||
|
||||
public synchronized void push(String source, short[] frame) {
|
||||
Queue queue = queues.computeIfAbsent(source, k -> new Queue());
|
||||
queue.frames.addLast(frame);
|
||||
while (queue.frames.size() > MAX) queue.frames.removeFirst();
|
||||
queue.lastPush = System.currentTimeMillis();
|
||||
}
|
||||
|
||||
public synchronized boolean ready() {
|
||||
for (Queue queue : queues.values()) {
|
||||
if (queue.playing ? !queue.frames.isEmpty() : queue.frames.size() >= PRIME) return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/** Nächster gemischter Frame, oder {@code null}, wenn gerade niemand spricht. */
|
||||
public synchronized short[] pull() {
|
||||
int[] sum = null;
|
||||
long now = System.currentTimeMillis();
|
||||
Iterator<Queue> it = queues.values().iterator();
|
||||
while (it.hasNext()) {
|
||||
Queue queue = it.next();
|
||||
if (!queue.playing && queue.frames.size() >= PRIME) queue.playing = true;
|
||||
if (!queue.playing) {
|
||||
if (queue.frames.isEmpty() && now - queue.lastPush > IDLE_MILLIS) it.remove();
|
||||
continue;
|
||||
}
|
||||
short[] frame = queue.frames.pollFirst();
|
||||
if (frame == null) {
|
||||
queue.playing = false;
|
||||
continue;
|
||||
}
|
||||
if (sum == null) sum = new int[Opus.FRAME_SAMPLES];
|
||||
for (int i = 0; i < Opus.FRAME_SAMPLES; i++) sum[i] += frame[i];
|
||||
}
|
||||
if (sum == null) return null;
|
||||
|
||||
short[] out = new short[Opus.FRAME_SAMPLES];
|
||||
for (int i = 0; i < Opus.FRAME_SAMPLES; i++) {
|
||||
out[i] = (short) Math.max(Short.MIN_VALUE, Math.min(Short.MAX_VALUE, sum[i]));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
public synchronized void clear() {
|
||||
queues.clear();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,67 @@
|
||||
package eu.mhsl.voicebridge.bridge.audio;
|
||||
|
||||
import io.github.jaredmdobson.concentus.OpusApplication;
|
||||
import io.github.jaredmdobson.concentus.OpusDecoder;
|
||||
import io.github.jaredmdobson.concentus.OpusEncoder;
|
||||
import io.github.jaredmdobson.concentus.OpusException;
|
||||
import io.github.jaredmdobson.concentus.OpusSignal;
|
||||
|
||||
import java.util.Arrays;
|
||||
|
||||
/** Alles läuft intern mit 48 kHz mono in Frames zu 20 ms. */
|
||||
public final class Opus {
|
||||
public static final int SAMPLE_RATE = 48000;
|
||||
public static final int FRAME_SAMPLES = 960;
|
||||
/** Längstes Opus-Paket: 120 ms. */
|
||||
private static final int MAX_PACKET_SAMPLES = 5760;
|
||||
private static final int MAX_PACKET_BYTES = 1275;
|
||||
|
||||
private Opus() {}
|
||||
|
||||
public static OpusDecoder decoder() {
|
||||
try {
|
||||
return new OpusDecoder(SAMPLE_RATE, 1);
|
||||
} catch (OpusException e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
|
||||
/** Encoder für Sprache, mono oder stereo. */
|
||||
public static OpusEncoder encoder(int channels, int bitrate) {
|
||||
try {
|
||||
OpusEncoder encoder = new OpusEncoder(SAMPLE_RATE, channels, OpusApplication.OPUS_APPLICATION_VOIP);
|
||||
encoder.setBitrate(bitrate);
|
||||
encoder.setSignalType(OpusSignal.OPUS_SIGNAL_VOICE);
|
||||
encoder.setComplexity(5);
|
||||
return encoder;
|
||||
} catch (OpusException e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
|
||||
/** Dekodiert ein Paket zu PCM (mono); leere Pakete ergeben ein leeres Array. */
|
||||
public static short[] decode(OpusDecoder decoder, byte[] packet) throws OpusException {
|
||||
if (packet.length == 0) return new short[0];
|
||||
short[] pcm = new short[MAX_PACKET_SAMPLES];
|
||||
int samples = decoder.decode(packet, 0, packet.length, pcm, 0, MAX_PACKET_SAMPLES, false);
|
||||
return Arrays.copyOf(pcm, samples);
|
||||
}
|
||||
|
||||
public static byte[] encodeMono(OpusEncoder encoder, short[] frame) throws OpusException {
|
||||
byte[] out = new byte[MAX_PACKET_BYTES];
|
||||
int length = encoder.encode(frame, 0, FRAME_SAMPLES, out, 0, out.length);
|
||||
return Arrays.copyOf(out, length);
|
||||
}
|
||||
|
||||
/** Für Discord: mono auf beide Kanäle verteilen. */
|
||||
public static byte[] encodeStereo(OpusEncoder encoder, short[] frame) throws OpusException {
|
||||
short[] stereo = new short[FRAME_SAMPLES * 2];
|
||||
for (int i = 0; i < FRAME_SAMPLES; i++) {
|
||||
stereo[2 * i] = frame[i];
|
||||
stereo[2 * i + 1] = frame[i];
|
||||
}
|
||||
byte[] out = new byte[MAX_PACKET_BYTES];
|
||||
int length = encoder.encode(stereo, 0, FRAME_SAMPLES, out, 0, out.length);
|
||||
return Arrays.copyOf(out, length);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,190 @@
|
||||
package eu.mhsl.voicebridge.bridge.discord;
|
||||
|
||||
import club.minnced.discord.jdave.interop.JDaveSessionFactory;
|
||||
import eu.mhsl.voicebridge.bridge.BridgeConfig;
|
||||
import eu.mhsl.voicebridge.bridge.Room;
|
||||
import eu.mhsl.voicebridge.bridge.audio.Opus;
|
||||
import io.github.jaredmdobson.concentus.OpusEncoder;
|
||||
import io.github.jaredmdobson.concentus.OpusException;
|
||||
import net.dv8tion.jda.api.JDA;
|
||||
import net.dv8tion.jda.api.JDABuilder;
|
||||
import net.dv8tion.jda.api.audio.AudioModuleConfig;
|
||||
import net.dv8tion.jda.api.audio.AudioReceiveHandler;
|
||||
import net.dv8tion.jda.api.audio.AudioSendHandler;
|
||||
import net.dv8tion.jda.api.audio.OpusPacket;
|
||||
import net.dv8tion.jda.api.entities.Guild;
|
||||
import net.dv8tion.jda.api.entities.Member;
|
||||
import net.dv8tion.jda.api.entities.channel.concrete.Category;
|
||||
import net.dv8tion.jda.api.entities.channel.concrete.VoiceChannel;
|
||||
import net.dv8tion.jda.api.managers.AudioManager;
|
||||
import net.dv8tion.jda.api.requests.GatewayIntent;
|
||||
import net.dv8tion.jda.api.utils.MemberCachePolicy;
|
||||
import net.dv8tion.jda.api.utils.cache.CacheFlag;
|
||||
import org.jetbrains.annotations.NotNull;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.nio.ByteBuffer;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
|
||||
/**
|
||||
* Ein Discord-Bot pro Raum (ein Bot kann pro Server nur in einem Voice-Channel sein).
|
||||
* Der Bot ist dauerhaft am Gateway, sieht so wer im Channel ist, und tritt nur bei Bedarf bei.
|
||||
*/
|
||||
public final class DiscordRoomBot implements AudioSendHandler, AudioReceiveHandler {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(DiscordRoomBot.class);
|
||||
|
||||
private final BridgeConfig.Discord config;
|
||||
private final Room room;
|
||||
private final OpusEncoder encoder = Opus.encoder(2, 64000);
|
||||
private final ExecutorService executor;
|
||||
|
||||
private volatile JDA jda;
|
||||
private volatile Guild guild;
|
||||
private volatile long channelId;
|
||||
private volatile boolean joined;
|
||||
|
||||
public DiscordRoomBot(BridgeConfig.Discord config, Room room) {
|
||||
this.config = config;
|
||||
this.room = room;
|
||||
this.executor = Executors.newSingleThreadExecutor(
|
||||
Thread.ofPlatform().daemon().name("discord-" + room.key()).factory());
|
||||
}
|
||||
|
||||
public void start() {
|
||||
CompletableFuture.runAsync(this::login, executor).exceptionally(e -> {
|
||||
LOG.error("{}: Discord-Bot konnte nicht starten: {}", room.name(), e.toString());
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
private void login() {
|
||||
try {
|
||||
jda = JDABuilder.createLight(room.config().discordToken(), GatewayIntent.GUILD_VOICE_STATES)
|
||||
.enableCache(CacheFlag.VOICE_STATE)
|
||||
.setMemberCachePolicy(MemberCachePolicy.VOICE)
|
||||
.setAudioModuleConfig(new AudioModuleConfig().withDaveSessionFactory(new JDaveSessionFactory()))
|
||||
.build()
|
||||
.awaitReady();
|
||||
} catch (InterruptedException e) {
|
||||
Thread.currentThread().interrupt();
|
||||
return;
|
||||
}
|
||||
|
||||
guild = jda.getGuildById(config.guild());
|
||||
if (guild == null) {
|
||||
LOG.error("{}: Bot {} ist nicht auf dem Discord-Server {}", room.name(), jda.getSelfUser().getName(), config.guild());
|
||||
return;
|
||||
}
|
||||
VoiceChannel channel = resolveChannel(guild);
|
||||
channelId = channel.getIdLong();
|
||||
LOG.info("{}: Discord-Bot {} bereit, Channel #{}", room.name(), jda.getSelfUser().getName(), channel.getName());
|
||||
|
||||
String nickname = config.nickname().replace("%room%", room.name());
|
||||
if (!nickname.equals(guild.getSelfMember().getNickname())) {
|
||||
guild.getSelfMember().modifyNickname(nickname).queue(null,
|
||||
e -> LOG.debug("{}: Nickname nicht gesetzt: {}", room.name(), e.getMessage()));
|
||||
}
|
||||
}
|
||||
|
||||
private VoiceChannel resolveChannel(Guild guild) {
|
||||
String configured = room.config().discordChannel();
|
||||
if (!configured.isBlank()) {
|
||||
VoiceChannel channel = guild.getVoiceChannelById(configured);
|
||||
if (channel == null) throw new IllegalStateException("Voice-Channel " + configured + " gibt es nicht");
|
||||
return channel;
|
||||
}
|
||||
|
||||
Category category = config.category().isBlank() ? null : guild.getCategoryById(config.category());
|
||||
if (!config.category().isBlank() && category == null) {
|
||||
throw new IllegalStateException("Kategorie " + config.category() + " gibt es nicht");
|
||||
}
|
||||
for (VoiceChannel channel : category != null ? category.getVoiceChannels() : guild.getVoiceChannels()) {
|
||||
if (channel.getName().equals(room.name()) && (category != null || channel.getParentCategory() == null)) {
|
||||
return channel;
|
||||
}
|
||||
}
|
||||
LOG.info("{}: lege Discord-Voice-Channel an", room.name());
|
||||
return category != null
|
||||
? category.createVoiceChannel(room.name()).complete()
|
||||
: guild.createVoiceChannel(room.name()).complete();
|
||||
}
|
||||
|
||||
/** Menschen im Raum-Channel (Bots zählen nicht). */
|
||||
public int humans() {
|
||||
Guild guild = this.guild;
|
||||
if (guild == null || channelId == 0) return 0;
|
||||
VoiceChannel channel = guild.getVoiceChannelById(channelId);
|
||||
if (channel == null) return 0;
|
||||
int count = 0;
|
||||
for (Member member : channel.getMembers()) if (!member.getUser().isBot()) count++;
|
||||
return count;
|
||||
}
|
||||
|
||||
public void setConnected(boolean connected) {
|
||||
Guild guild = this.guild;
|
||||
if (guild == null || connected == joined) return;
|
||||
joined = connected;
|
||||
AudioManager audio = guild.getAudioManager();
|
||||
if (connected) {
|
||||
VoiceChannel channel = guild.getVoiceChannelById(channelId);
|
||||
if (channel == null) return;
|
||||
audio.setSendingHandler(this);
|
||||
audio.setReceivingHandler(this);
|
||||
audio.setSelfDeafened(false);
|
||||
audio.setSelfMuted(false);
|
||||
audio.openAudioConnection(channel);
|
||||
LOG.info("{}: Discord-Bot betritt #{}", room.name(), channel.getName());
|
||||
} else {
|
||||
audio.closeAudioConnection();
|
||||
LOG.info("{}: Discord-Bot verlässt den Channel", room.name());
|
||||
}
|
||||
}
|
||||
|
||||
public void stop() {
|
||||
JDA jda = this.jda;
|
||||
if (jda != null) jda.shutdown();
|
||||
executor.shutdownNow();
|
||||
}
|
||||
|
||||
// --- Senden: JDA fragt alle 20 ms ---
|
||||
|
||||
@Override
|
||||
public boolean canProvide() {
|
||||
return room.toDiscord().ready();
|
||||
}
|
||||
|
||||
@Override
|
||||
public ByteBuffer provide20MsAudio() {
|
||||
short[] frame = room.toDiscord().pull();
|
||||
if (frame == null) return null;
|
||||
try {
|
||||
return ByteBuffer.wrap(Opus.encodeStereo(encoder, frame));
|
||||
} catch (OpusException e) {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isOpus() {
|
||||
return true;
|
||||
}
|
||||
|
||||
// --- Empfangen: Opus pro User, bereits entschlüsselt ---
|
||||
|
||||
@Override
|
||||
public boolean canReceiveEncoded() {
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handleEncodedAudio(@NotNull OpusPacket packet) {
|
||||
Guild guild = this.guild;
|
||||
if (guild == null) return;
|
||||
Member member = guild.getMemberById(packet.getUserId());
|
||||
if (member != null && member.getUser().isBot()) return;
|
||||
room.fromDiscord(packet.getUserId(), packet.getOpusAudio());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,112 @@
|
||||
package eu.mhsl.voicebridge.bridge.teamspeak;
|
||||
|
||||
import com.github.manevolent.ts3j.audio.Microphone;
|
||||
import com.github.manevolent.ts3j.enums.CodecType;
|
||||
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.packet.PacketBody0Voice;
|
||||
import com.github.manevolent.ts3j.protocol.socket.client.LocalTeamspeakClientSocket;
|
||||
import eu.mhsl.voicebridge.bridge.BridgeConfig;
|
||||
import eu.mhsl.voicebridge.bridge.Room;
|
||||
import eu.mhsl.voicebridge.bridge.audio.Opus;
|
||||
import io.github.jaredmdobson.concentus.OpusEncoder;
|
||||
import io.github.jaredmdobson.concentus.OpusException;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.function.Consumer;
|
||||
|
||||
/** Ein TS-Client im Raum-Channel: spielt die Mischung aus Minecraft und Discord ab und hört zu. */
|
||||
final class TeamspeakRoomClient implements Microphone, TS3Listener {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(TeamspeakRoomClient.class);
|
||||
|
||||
private final BridgeConfig.Teamspeak config;
|
||||
private final Room room;
|
||||
private final int channel;
|
||||
private final LocalTeamspeakClientSocket socket;
|
||||
private final Consumer<TeamspeakRoomClient> onLost;
|
||||
private final OpusEncoder encoder = Opus.encoder(1, 48000);
|
||||
private volatile boolean closing;
|
||||
|
||||
TeamspeakRoomClient(BridgeConfig.Teamspeak config, Room room, int channel, LocalIdentity identity,
|
||||
Consumer<TeamspeakRoomClient> onLost) {
|
||||
this.config = config;
|
||||
this.room = room;
|
||||
this.channel = channel;
|
||||
this.onLost = onLost;
|
||||
this.socket = newSocket(config, identity, config.roomNickname().replace("%room%", room.name()));
|
||||
socket.addListener(this);
|
||||
socket.setVoiceHandler(this::onVoice);
|
||||
socket.setWhisperHandler(packet -> {});
|
||||
socket.setMicrophone(this);
|
||||
}
|
||||
|
||||
static LocalTeamspeakClientSocket newSocket(BridgeConfig.Teamspeak config, LocalIdentity identity, String nickname) {
|
||||
LocalTeamspeakClientSocket socket = new LocalTeamspeakClientSocket();
|
||||
socket.setIdentity(identity);
|
||||
socket.setNickname(nickname);
|
||||
BridgeConfig.ClientVersion version = config.clientVersion();
|
||||
if (version != null) socket.setClientVersion(version.platform(), version.version(), version.sign());
|
||||
socket.setExceptionHandler(e -> LOG.debug("ts3j: {}", e.toString()));
|
||||
return socket;
|
||||
}
|
||||
|
||||
void connect() throws Exception {
|
||||
socket.connect(new InetSocketAddress(config.host(), config.port()),
|
||||
TeamspeakSide.emptyToNull(config.password()), 15000L);
|
||||
socket.joinChannel(channel, null);
|
||||
}
|
||||
|
||||
int clientId() {
|
||||
return socket.getClientId();
|
||||
}
|
||||
|
||||
void disconnect() {
|
||||
closing = true;
|
||||
try {
|
||||
socket.disconnect();
|
||||
} catch (Exception ignored) {
|
||||
// Verbindung ist ohnehin weg
|
||||
}
|
||||
}
|
||||
|
||||
private void onVoice(PacketBody0Voice packet) {
|
||||
CodecType codec = packet.getCodecType();
|
||||
if (codec != CodecType.OPUS_VOICE && codec != CodecType.OPUS_MUSIC) return;
|
||||
byte[] data = packet.getCodecData();
|
||||
if (data == null || data.length == 0) return;
|
||||
room.fromTeamspeak(packet.getClientId(), data);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onDisconnected(DisconnectedEvent e) {
|
||||
if (closing) return;
|
||||
LOG.warn("{}: TeamSpeak-Client vom Server getrennt", room.name());
|
||||
onLost.accept(this);
|
||||
}
|
||||
|
||||
// --- Microphone: ts3j fragt alle 20 ms ---
|
||||
|
||||
@Override
|
||||
public boolean isReady() {
|
||||
return room.toTeamspeak().ready();
|
||||
}
|
||||
|
||||
@Override
|
||||
public CodecType getCodec() {
|
||||
return CodecType.OPUS_VOICE;
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] provide() {
|
||||
short[] frame = room.toTeamspeak().pull();
|
||||
if (frame == null) return new byte[0];
|
||||
try {
|
||||
return Opus.encodeMono(encoder, frame);
|
||||
} catch (OpusException e) {
|
||||
return new byte[0];
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,289 @@
|
||||
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;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.Executors;
|
||||
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.
|
||||
*/
|
||||
public final class TeamspeakSide implements TS3Listener {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(TeamspeakSide.class);
|
||||
private static final long RECONNECT_SECONDS = 30;
|
||||
/** client_type 0 = normaler Client, 1 = ServerQuery. */
|
||||
private static final int CLIENT_TYPE_VOICE = 0;
|
||||
|
||||
private record TsClient(int channel, int type) {}
|
||||
private record TsChannel(int parent, String name) {}
|
||||
|
||||
private final BridgeConfig.Teamspeak config;
|
||||
private final Path dataDir;
|
||||
private final List<Room> rooms;
|
||||
private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(
|
||||
Thread.ofPlatform().daemon().name("teamspeak").factory());
|
||||
|
||||
private final Map<Integer, TsClient> clients = new ConcurrentHashMap<>();
|
||||
private final Map<Integer, TsChannel> channels = new ConcurrentHashMap<>();
|
||||
private final Map<String, Integer> roomChannels = new ConcurrentHashMap<>();
|
||||
private final Map<String, TeamspeakRoomClient> roomClients = new ConcurrentHashMap<>();
|
||||
|
||||
private volatile LocalTeamspeakClientSocket observer;
|
||||
private volatile boolean ready;
|
||||
|
||||
public TeamspeakSide(BridgeConfig.Teamspeak config, Path dataDir, List<Room> rooms) {
|
||||
this.config = config;
|
||||
this.dataDir = dataDir;
|
||||
this.rooms = rooms;
|
||||
}
|
||||
|
||||
public void start() {
|
||||
executor.execute(this::connectObserver);
|
||||
}
|
||||
|
||||
public boolean ready() {
|
||||
return ready;
|
||||
}
|
||||
|
||||
/** Menschen im Raum-Channel, ohne unsere eigenen Clients und ohne ServerQuery. */
|
||||
public int humansIn(Room room) {
|
||||
Integer channel = roomChannels.get(room.key());
|
||||
if (!ready || channel == null) return 0;
|
||||
Set<Integer> own = ownClientIds();
|
||||
int count = 0;
|
||||
for (Map.Entry<Integer, TsClient> e : clients.entrySet()) {
|
||||
TsClient client = e.getValue();
|
||||
if (client.channel() == channel && client.type() == CLIENT_TYPE_VOICE && !own.contains(e.getKey())) count++;
|
||||
}
|
||||
return count;
|
||||
}
|
||||
|
||||
/** Raum-Client verbinden oder trennen. Läuft asynchron, mehrfaches Aufrufen schadet nicht. */
|
||||
public void setRoomConnected(Room room, boolean connected) {
|
||||
executor.execute(() -> {
|
||||
TeamspeakRoomClient client = roomClients.get(room.key());
|
||||
if (connected && client == null && ready) {
|
||||
Integer channel = roomChannels.get(room.key());
|
||||
if (channel == null) return;
|
||||
try {
|
||||
client = new TeamspeakRoomClient(config, room, channel,
|
||||
loadIdentity("ts-" + room.key() + ".ini"), this::onRoomClientLost);
|
||||
roomClients.put(room.key(), client);
|
||||
client.connect();
|
||||
LOG.info("{}: TeamSpeak-Client im Channel {}", room.name(), channel);
|
||||
} catch (Exception e) {
|
||||
LOG.warn("{}: TeamSpeak-Client konnte nicht verbinden: {}", room.name(), e.toString());
|
||||
roomClients.remove(room.key());
|
||||
if (client != null) client.disconnect();
|
||||
}
|
||||
} else if (!connected && client != null) {
|
||||
roomClients.remove(room.key());
|
||||
client.disconnect();
|
||||
LOG.info("{}: TeamSpeak-Client getrennt", room.name());
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void onRoomClientLost(TeamspeakRoomClient client) {
|
||||
executor.execute(() -> roomClients.values().remove(client));
|
||||
}
|
||||
|
||||
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() {
|
||||
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);
|
||||
LOG.info("TeamSpeak verbunden mit {}:{} ({} Channels, {} Clients sichtbar)",
|
||||
config.host(), config.port(), channels.size(), clients.size());
|
||||
resolveChannels(socket);
|
||||
if (config.parentChannel() > 0) socket.joinChannel(config.parentChannel(), null);
|
||||
ready = true;
|
||||
} 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);
|
||||
}
|
||||
}
|
||||
|
||||
private void resolveChannels(LocalTeamspeakClientSocket socket) throws Exception {
|
||||
for (Room room : rooms) {
|
||||
int configured = room.config().teamspeakChannel();
|
||||
if (configured > 0) {
|
||||
roomChannels.put(room.key(), configured);
|
||||
continue;
|
||||
}
|
||||
Integer found = findChannel(room.name());
|
||||
if (found == null) {
|
||||
socket.executeCommand(new SingleCommand("channelcreate", ProtocolRole.CLIENT,
|
||||
new CommandSingleParameter("channel_name", room.name()),
|
||||
new CommandSingleParameter("cpid", Integer.toString(config.parentChannel())),
|
||||
new CommandSingleParameter("channel_flag_permanent", "1"),
|
||||
new CommandSingleParameter("channel_codec", "4"))).get();
|
||||
// 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) throw new IOException("Channel " + room.name() + " wurde nicht angelegt");
|
||||
LOG.info("TeamSpeak-Channel angelegt: {} ({})", room.name(), found);
|
||||
}
|
||||
roomChannels.put(room.key(), found);
|
||||
}
|
||||
}
|
||||
|
||||
private Integer findChannel(String name) {
|
||||
for (Map.Entry<Integer, TsChannel> e : channels.entrySet()) {
|
||||
if (e.getValue().parent() == config.parentChannel() && e.getValue().name().equals(name)) return e.getKey();
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
private void disconnectObserver() {
|
||||
ready = false;
|
||||
LocalTeamspeakClientSocket socket = observer;
|
||||
observer = null;
|
||||
if (socket != null) {
|
||||
try {
|
||||
socket.disconnect();
|
||||
} catch (Exception ignored) {
|
||||
// Verbindung ist ohnehin weg
|
||||
}
|
||||
}
|
||||
for (TeamspeakRoomClient client : roomClients.values()) client.disconnect();
|
||||
roomClients.clear();
|
||||
clients.clear();
|
||||
channels.clear();
|
||||
}
|
||||
|
||||
LocalIdentity loadIdentity(String file) throws IOException, GeneralSecurityException {
|
||||
Path path = dataDir.resolve(file);
|
||||
if (Files.exists(path)) return LocalIdentity.read(path.toFile());
|
||||
Files.createDirectories(dataDir);
|
||||
LOG.info("Erzeuge TeamSpeak-Identität {} (Security-Level {})", file, config.securityLevel());
|
||||
LocalIdentity identity = LocalIdentity.generateNew(config.securityLevel());
|
||||
identity.save(path.toFile());
|
||||
return identity;
|
||||
}
|
||||
|
||||
static String emptyToNull(String value) {
|
||||
return value == null || value.isEmpty() ? null : value;
|
||||
}
|
||||
|
||||
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) {
|
||||
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", "")));
|
||||
}
|
||||
|
||||
private static int parse(String value, int def) {
|
||||
if (value == null) return def;
|
||||
try {
|
||||
return Integer.parseInt(value.trim());
|
||||
} catch (NumberFormatException e) {
|
||||
return def;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
package eu.mhsl.voicebridge.bridge.audio;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
class MixerTest {
|
||||
|
||||
private static short[] frame(int value) {
|
||||
short[] frame = new short[Opus.FRAME_SAMPLES];
|
||||
Arrays.fill(frame, (short) value);
|
||||
return frame;
|
||||
}
|
||||
|
||||
@Test
|
||||
void waitsForJitterBufferThenMixes() {
|
||||
Mixer mixer = new Mixer();
|
||||
mixer.push("a", frame(100));
|
||||
assertFalse(mixer.ready());
|
||||
assertNull(mixer.pull());
|
||||
|
||||
mixer.push("a", frame(100));
|
||||
mixer.push("b", frame(50));
|
||||
mixer.push("b", frame(50));
|
||||
assertTrue(mixer.ready());
|
||||
assertEquals(150, mixer.pull()[0]);
|
||||
assertEquals(150, mixer.pull()[0]);
|
||||
assertNull(mixer.pull());
|
||||
}
|
||||
|
||||
@Test
|
||||
void clipsInsteadOfOverflowing() {
|
||||
Mixer mixer = new Mixer();
|
||||
for (String source : List.of("a", "b")) {
|
||||
mixer.push(source, frame(30000));
|
||||
mixer.push(source, frame(30000));
|
||||
}
|
||||
assertEquals(Short.MAX_VALUE, mixer.pull()[0]);
|
||||
}
|
||||
|
||||
@Test
|
||||
void dropsOldestFramesWhenBehind() {
|
||||
Mixer mixer = new Mixer();
|
||||
for (int i = 0; i < Mixer.MAX + 5; i++) mixer.push("a", frame(i));
|
||||
assertEquals(5, mixer.pull()[0]);
|
||||
}
|
||||
|
||||
@Test
|
||||
void splitterProducesExactFrames() {
|
||||
FrameSplitter splitter = new FrameSplitter();
|
||||
assertEquals(0, splitter.push(new short[500]).size());
|
||||
assertEquals(1, splitter.push(new short[500]).size());
|
||||
assertEquals(2, splitter.push(new short[1920]).size());
|
||||
}
|
||||
|
||||
@Test
|
||||
void opusRoundTrip() throws Exception {
|
||||
short[] tone = new short[Opus.FRAME_SAMPLES];
|
||||
for (int i = 0; i < tone.length; i++) tone[i] = (short) (8000 * Math.sin(i * 2 * Math.PI * 440 / Opus.SAMPLE_RATE));
|
||||
byte[] packet = Opus.encodeMono(Opus.encoder(1, 48000), tone);
|
||||
assertTrue(packet.length > 0);
|
||||
assertEquals(Opus.FRAME_SAMPLES, Opus.decode(Opus.decoder(), packet).length);
|
||||
|
||||
byte[] stereo = Opus.encodeStereo(Opus.encoder(2, 64000), tone);
|
||||
assertEquals(Opus.FRAME_SAMPLES, Opus.decode(Opus.decoder(), stereo).length);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user