diff --git a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/Bridge.java b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/Bridge.java index a057e86..dd30a56 100644 --- a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/Bridge.java +++ b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/Bridge.java @@ -23,6 +23,7 @@ import java.util.concurrent.TimeUnit; */ public final class Bridge { private static final Logger LOG = LoggerFactory.getLogger(Bridge.class); + private static final long STATS_MILLIS = 30_000; private final BridgeConfig config; private final Map rooms = new LinkedHashMap<>(); @@ -39,6 +40,7 @@ public final class Bridge { private static final class Presence { long roomWanted, teamspeakWanted, discordWanted; String lastCounts = ""; + long lastStats; boolean teamspeakConnected, discordConnected; } @@ -108,6 +110,14 @@ public final class Bridge { if (bot != null && dcConnect != p.discordConnected) bot.setConnected(dcConnect); p.discordConnected = dcConnect; + if (active && now - p.lastStats >= STATS_MILLIS) { + if (p.lastStats > 0) LOG.info("{}: {}", room.name(), room.takeStats()); + else room.takeStats(); + p.lastStats = now; + } else if (!active) { + p.lastStats = 0; + } + room.dropIdleSources(); } roomInfos = buildInfos(); diff --git a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/Room.java b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/Room.java index f29997f..c056f14 100644 --- a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/Room.java +++ b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/Room.java @@ -10,10 +10,12 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; /** * Ein Crosstalk-Raum aus drei Seiten. Jede Seite bekommt das Audio der beiden anderen: @@ -22,6 +24,8 @@ import java.util.concurrent.ConcurrentHashMap; public final class Room { private static final Logger LOG = LoggerFactory.getLogger(Room.class); private static final long SOURCE_IDLE_MILLIS = 5000; + /** Größere Lücken in der Sequenz sind eine neue Sprechphase, keine verlorenen Pakete. */ + private static final int MAX_CONCEALED = 5; public enum Side { MINECRAFT, TEAMSPEAK, DISCORD } @@ -35,6 +39,8 @@ public final class Room { private volatile MinecraftSink minecraft = (room, source, opus) -> {}; private volatile boolean active; + private final AtomicLong concealed = new AtomicLong(); + private final AtomicLong late = new AtomicLong(); public interface MinecraftSink { void send(String room, UUID source, byte[] opus); @@ -49,6 +55,7 @@ public final class Room { final FrameSplitter splitter = new FrameSplitter(); OpusEncoder minecraftEncoder; volatile long lastSeen; + int lastSequence = -1; Source(Side side, String key) { this.side = side; @@ -83,26 +90,41 @@ public final class Room { LOG.info("{}: {}", name(), active ? "aktiv" : "inaktiv"); } + /** Simple Voice Chat gibt die Sequenznummer nicht heraus, daher ohne Lückenerkennung. */ public void fromMinecraft(UUID player, byte[] opus) { - receive(Side.MINECRAFT, "mc:" + player, opus); + receive(Side.MINECRAFT, "mc:" + player, opus, -1); } - public void fromTeamspeak(int clientId, byte[] opus) { - receive(Side.TEAMSPEAK, "ts:" + clientId, opus); + /** @param sequence 16-Bit-Zähler des Senders */ + public void fromTeamspeak(int clientId, byte[] opus, int sequence) { + receive(Side.TEAMSPEAK, "ts:" + clientId, opus, sequence); } - public void fromDiscord(long userId, byte[] opus) { - receive(Side.DISCORD, "dc:" + userId, opus); + /** @param sequence RTP-Sequenz (16 Bit) */ + public void fromDiscord(long userId, byte[] opus, int sequence) { + receive(Side.DISCORD, "dc:" + userId, opus, sequence); } - private void receive(Side side, String key, byte[] opus) { + private void receive(Side side, String key, byte[] opus, int sequence) { if (!active || opus.length == 0) return; Source source = sources.computeIfAbsent(key, k -> new Source(side, k)); source.lastSeen = System.currentTimeMillis(); - List frames; + List frames = new ArrayList<>(2); try { - frames = source.splitter.push(Opus.decode(source.decoder, opus)); + if (sequence >= 0 && source.lastSequence >= 0) { + int gap = (sequence - source.lastSequence) & 0xFFFF; + if (gap == 0 || gap > 0x8000) { + late.incrementAndGet(); + return; // doppelt oder zu spät: der Decoder ist schon weiter + } + if (gap > 1 && gap <= MAX_CONCEALED + 1) { + for (int i = 1; i < gap; i++) frames.addAll(source.splitter.push(Opus.conceal(source.decoder))); + concealed.addAndGet(gap - 1); + } + } + if (sequence >= 0) source.lastSequence = sequence; + frames.addAll(source.splitter.push(Opus.decode(source.decoder, opus))); } catch (OpusException e) { LOG.debug("{}: Opus-Paket von {} nicht dekodierbar: {}", name(), key, e.getMessage()); return; @@ -124,6 +146,15 @@ public final class Room { } } + /** Zählerstand seit dem letzten Aufruf, fürs Log. */ + String takeStats() { + long[] ts = toTeamspeak.takeStats(); + long[] dc = toDiscord.takeStats(); + return String.format("eingehend %d Pakete verloren (überbrückt), %d zu spät | an TS %d Frames, %d Lücken, %d verworfen" + + " | an Discord %d Frames, %d Lücken, %d verworfen", + concealed.getAndSet(0), late.getAndSet(0), ts[0], ts[1], ts[2], dc[0], dc[1], dc[2]); + } + void dropIdleSources() { long now = System.currentTimeMillis(); sources.values().removeIf(source -> now - source.lastSeen > SOURCE_IDLE_MILLIS); diff --git a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/audio/Mixer.java b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/audio/Mixer.java index e00e187..0a96a07 100644 --- a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/audio/Mixer.java +++ b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/audio/Mixer.java @@ -7,32 +7,43 @@ 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. + * Jeder Sprecher hat einen Jitter-Puffer: abgespielt wird erst ab {@link #PRIME} Frames, über + * {@link #MAX} Frames werden die ältesten verworfen, damit die Verzögerung nicht wächst. Fehlt einmal + * ein Frame, wird der letzte leiser wiederholt, statt sofort hart abzubrechen. */ public final class Mixer { - static final int PRIME = 2; - static final int MAX = 8; + static final int PRIME = 3; + static final int MAX = 10; + /** So viele fehlende Frames hintereinander werden überbrückt, danach gilt der Sprecher als fertig. */ + static final int CONCEAL = 2; private static final long IDLE_MILLIS = 2000; private static final class Queue { final ArrayDeque frames = new ArrayDeque<>(); boolean playing; + int missing; + short[] last; long lastPush; } private final Map queues = new HashMap<>(); + private long underruns; + private long dropped; + private long played; 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(); + while (queue.frames.size() > MAX) { + queue.frames.removeFirst(); + dropped++; + } 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; + if (queue.playing || queue.frames.size() >= PRIME) return true; } return false; } @@ -44,16 +55,30 @@ public final class Mixer { Iterator it = queues.values().iterator(); while (it.hasNext()) { Queue queue = it.next(); - if (!queue.playing && queue.frames.size() >= PRIME) queue.playing = true; + if (!queue.playing && queue.frames.size() >= PRIME) { + queue.playing = true; + queue.missing = 0; + } if (!queue.playing) { if (queue.frames.isEmpty() && now - queue.lastPush > IDLE_MILLIS) it.remove(); continue; } + short[] frame = queue.frames.pollFirst(); - if (frame == null) { + if (frame != null) { + queue.missing = 0; + queue.last = frame; + played++; + } else if (++queue.missing <= CONCEAL && queue.last != null) { + underruns++; + frame = fade(queue.last, queue.missing); + queue.last = frame; + } else { queue.playing = false; + queue.last = null; continue; } + if (sum == null) sum = new int[Opus.FRAME_SAMPLES]; for (int i = 0; i < Opus.FRAME_SAMPLES; i++) sum[i] += frame[i]; } @@ -66,6 +91,20 @@ public final class Mixer { return out; } + /** Den letzten Frame mit halber Lautstärke wiederholen, damit eine Lücke nicht als Knacken hörbar wird. */ + private static short[] fade(short[] frame, int step) { + short[] out = new short[frame.length]; + for (int i = 0; i < frame.length; i++) out[i] = (short) (frame[i] >> step); + return out; + } + + /** Zähler seit dem letzten Aufruf: gespielt, überbrückt, verworfen. */ + public synchronized long[] takeStats() { + long[] stats = {played, underruns, dropped}; + played = underruns = dropped = 0; + return stats; + } + public synchronized void clear() { queues.clear(); } diff --git a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/audio/Opus.java b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/audio/Opus.java index 978172d..ad75455 100644 --- a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/audio/Opus.java +++ b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/audio/Opus.java @@ -47,6 +47,13 @@ public final class Opus { return Arrays.copyOf(pcm, samples); } + /** Ersatz für ein verlorenes Paket: Opus rechnet aus dem bisherigen Verlauf 20 ms weiter. */ + public static short[] conceal(OpusDecoder decoder) throws OpusException { + short[] pcm = new short[FRAME_SAMPLES]; + int samples = decoder.decode(null, 0, 0, pcm, 0, FRAME_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); diff --git a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/discord/DiscordRoomBot.java b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/discord/DiscordRoomBot.java index 7b4facf..a1ba5d1 100644 --- a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/discord/DiscordRoomBot.java +++ b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/discord/DiscordRoomBot.java @@ -186,6 +186,6 @@ public final class DiscordRoomBot implements AudioSendHandler, AudioReceiveHandl if (guild == null) return; Member member = guild.getMemberById(packet.getUserId()); if (member != null && member.getUser().isBot()) return; - room.fromDiscord(packet.getUserId(), packet.getOpusAudio()); + room.fromDiscord(packet.getUserId(), packet.getOpusAudio(), packet.getSequence()); } } diff --git a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/teamspeak/TeamspeakRoomClient.java b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/teamspeak/TeamspeakRoomClient.java index f27f0e3..44897f6 100644 --- a/bridge/src/main/java/eu/mhsl/voicebridge/bridge/teamspeak/TeamspeakRoomClient.java +++ b/bridge/src/main/java/eu/mhsl/voicebridge/bridge/teamspeak/TeamspeakRoomClient.java @@ -79,7 +79,7 @@ final class TeamspeakRoomClient implements Microphone, TS3Listener { 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); + room.fromTeamspeak(packet.getClientId(), data, packet.getPacketId() & 0xFFFF); } @Override diff --git a/bridge/src/test/java/eu/mhsl/voicebridge/bridge/audio/MixerTest.java b/bridge/src/test/java/eu/mhsl/voicebridge/bridge/audio/MixerTest.java index bf287be..787bcc1 100644 --- a/bridge/src/test/java/eu/mhsl/voicebridge/bridge/audio/MixerTest.java +++ b/bridge/src/test/java/eu/mhsl/voicebridge/bridge/audio/MixerTest.java @@ -15,29 +15,38 @@ class MixerTest { return frame; } + private static void pushFrames(Mixer mixer, String source, int value, int count) { + for (int i = 0; i < count; i++) mixer.push(source, frame(value)); + } + @Test void waitsForJitterBufferThenMixes() { Mixer mixer = new Mixer(); - mixer.push("a", frame(100)); + pushFrames(mixer, "a", 100, Mixer.PRIME - 1); assertFalse(mixer.ready()); assertNull(mixer.pull()); mixer.push("a", frame(100)); - mixer.push("b", frame(50)); - mixer.push("b", frame(50)); + pushFrames(mixer, "b", 50, Mixer.PRIME); assertTrue(mixer.ready()); - assertEquals(150, mixer.pull()[0]); - assertEquals(150, mixer.pull()[0]); + for (int i = 0; i < Mixer.PRIME; i++) assertEquals(150, mixer.pull()[0]); + } + + @Test + void concealsShortGapsThenStops() { + Mixer mixer = new Mixer(); + pushFrames(mixer, "a", 1000, Mixer.PRIME); + for (int i = 0; i < Mixer.PRIME; i++) assertEquals(1000, mixer.pull()[0]); + assertEquals(500, mixer.pull()[0]); + assertEquals(125, mixer.pull()[0]); assertNull(mixer.pull()); + assertArrayEquals(new long[]{Mixer.PRIME, Mixer.CONCEAL, 0}, mixer.takeStats()); } @Test void clipsInsteadOfOverflowing() { Mixer mixer = new Mixer(); - for (String source : List.of("a", "b")) { - mixer.push(source, frame(30000)); - mixer.push(source, frame(30000)); - } + for (String source : List.of("a", "b")) pushFrames(mixer, source, 30000, Mixer.PRIME); assertEquals(Short.MAX_VALUE, mixer.pull()[0]); }