diff --git a/calls/audio/xmpp_to_tg.go b/calls/audio/xmpp_to_tg.go index c2f0865..5a3996d 100644 --- a/calls/audio/xmpp_to_tg.go +++ b/calls/audio/xmpp_to_tg.go @@ -50,6 +50,7 @@ type XmppToTg struct { cShipSilence atomic.Int64 // of those, silence frames (underrun/priming) cPlcFrames atomic.Int64 // PLC 10ms frames synthesised over gaps cGapEvents atomic.Int64 // ticks where a seq gap was skipped + cDtxFrames atomic.Int64 // empty (DTX) packets concealed as comfort noise cSendErrs atomic.Int64 // SendMicrophonePCM errors capture *captureWriter @@ -57,7 +58,7 @@ type XmppToTg struct { // counter snapshot for computing per-interval deltas in the stats log type xmppToTgSnap struct { - rtpIn, rtpSkippedPT, shipTotal, shipSilence, plcFrames, gapEvents, sendErrs int64 + rtpIn, rtpSkippedPT, shipTotal, shipSilence, plcFrames, gapEvents, dtxFrames, sendErrs int64 } func (h *XmppToTg) snap() xmppToTgSnap { @@ -68,6 +69,7 @@ func (h *XmppToTg) snap() xmppToTgSnap { shipSilence: h.cShipSilence.Load(), plcFrames: h.cPlcFrames.Load(), gapEvents: h.cGapEvents.Load(), + dtxFrames: h.cDtxFrames.Load(), sendErrs: h.cSendErrs.Load(), } } @@ -273,6 +275,7 @@ func (h *XmppToTg) logStats(elapsed, sinceStart time.Duration, ticks, stalls, ma "feed_drift_ms": feedDriftMs, "plc_frames": cur.plcFrames - prev.plcFrames, "gap_events": cur.gapEvents - prev.gapEvents, + "dtx_frames": cur.dtxFrames - prev.dtxFrames, "send_errs": cur.sendErrs - prev.sendErrs, "depth": ps.Depth, "depth_ms": ps.Depth * OpusFrameMs, @@ -314,15 +317,29 @@ func (h *XmppToTg) processTick(st *tickState) { st.pending10ms = appendChunks10ms(st.pending10ms, plcOut) h.cPlcFrames.Add(int64(len(st.pending10ms) - before)) } - n, err := h.dec.Decode(pkt.Payload, h.pcmBuf) - if err != nil { - h.log.WithError(err).Warn("opus decode failed") - h.cShipSilence.Add(1) - h.sendToNtg(h.silence) - return + if len(pkt.Payload) == 0 { + // empty payload = opus DTX/comfort-noise frame during silence, not + // a decode error. Conceal for the last frame's duration (libopus + // synthesises comfort noise) rather than emitting hard silence. + h.cDtxFrames.Add(1) + plcOut := h.pcmBuf[:st.lastDecodedSamples*Channels] + if err := h.dec.DecodePLC(plcOut); err != nil { + h.cShipSilence.Add(1) + h.sendToNtg(h.silence) + return + } + st.pending10ms = appendChunks10ms(st.pending10ms, plcOut) + } else { + n, err := h.dec.Decode(pkt.Payload, h.pcmBuf) + if err != nil { + h.log.WithError(err).Warn("opus decode failed") + h.cShipSilence.Add(1) + h.sendToNtg(h.silence) + return + } + st.lastDecodedSamples = n + st.pending10ms = appendChunks10ms(st.pending10ms, h.pcmBuf[:n*Channels]) } - st.lastDecodedSamples = n - st.pending10ms = appendChunks10ms(st.pending10ms, h.pcmBuf[:n*Channels]) } chunk := st.pending10ms[0] st.pending10ms = st.pending10ms[1:] diff --git a/calls/audio/xmpp_to_tg_test.go b/calls/audio/xmpp_to_tg_test.go index 93bfeca..41a7105 100644 --- a/calls/audio/xmpp_to_tg_test.go +++ b/calls/audio/xmpp_to_tg_test.go @@ -119,6 +119,35 @@ func TestProcessTickDecodesAndSplits(t *testing.T) { } } +func TestProcessTickConcealsDtxEmptyPacket(t *testing.T) { + h, n := newTestXmppToTg(t) + enc, err := NewEncoder() + if err != nil { + t.Fatalf("NewEncoder: %v", err) + } + // Prime with a real packet, then an empty (DTX) packet the peer sends + // during silence. + h.playout.Push(rtpWithSeq(200, encodeOpusFrame(t, enc))) + h.playout.Push(rtpWithSeq(201, nil)) + + st := newTickState() + // real packet -> 2 chunks (ticks 1-2), DTX packet -> concealed (ticks 3+). + for i := 0; i < 4; i++ { + h.processTick(st) + } + + if got := h.cDtxFrames.Load(); got != 1 { + t.Errorf("dtx frames = %d, want 1 (empty packet must be concealed, not errored)", got) + } + // DTX must not fall through to the silence path, and every tick ships a frame. + if got := h.cShipSilence.Load(); got != 0 { + t.Errorf("ship_silence = %d, want 0 (DTX conceals rather than emitting silence)", got) + } + if got := n.sentCount(); got != 4 { + t.Errorf("shipped %d frames, want 4", got) + } +} + func TestProcessTickRunsPLCBeforeGapRecovery(t *testing.T) { h, _ := newTestXmppToTg(t) enc, err := NewEncoder()