Audio: größerer Jitter-Puffer, Lücken überbrücken (Opus-PLC), Sequenzen prüfen, Statistik im Log

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<String, Room> 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();

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) {
+    /** @param sequence 16-Bit-Zähler des Senders */
-        receive(Side.TEAMSPEAK, "ts:" + clientId, opus);
+    public void fromTeamspeak(int clientId, byte[] opus, int sequence) {
+        receive(Side.TEAMSPEAK, "ts:" + clientId, opus, sequence);
     }
 
-    public void fromDiscord(long userId, byte[] opus) {
+    /** @param sequence RTP-Sequenz (16 Bit) */
-        receive(Side.DISCORD, "dc:" + userId, opus);
+    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<short[]> frames;
+        List<short[]> 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);

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,
+ * Jeder Sprecher hat einen Jitter-Puffer: abgespielt wird erst ab {@link #PRIME} Frames, über
- * über {@link #MAX} Frames werden die ältesten verworfen, damit die Verzögerung nicht wächst.
+ * {@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 PRIME = 3;
-    static final int MAX = 8;
+    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<short[]> frames = new ArrayDeque<>();
         boolean playing;
+        int missing;
+        short[] last;
         long lastPush;
     }
 
     private final Map<String, Queue> 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<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 && 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();
     }

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);

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());
     }
 }

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

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));
+        pushFrames(mixer, "b", 50, Mixer.PRIME);
-        mixer.push("b", frame(50));
         assertTrue(mixer.ready());
-        assertEquals(150, mixer.pull()[0]);
+        for (int i = 0; i < Mixer.PRIME; i++) assertEquals(150, mixer.pull()[0]);
-        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")) {
+        for (String source : List.of("a", "b")) pushFrames(mixer, source, 30000, Mixer.PRIME);
-            mixer.push(source, frame(30000));
-            mixer.push(source, frame(30000));
-        }
         assertEquals(Short.MAX_VALUE, mixer.pull()[0]);
     }