From 5a18a3bc197c2592f14dd88dd458010627db109f Mon Sep 17 00:00:00 2001 From: David Chen Date: Tue, 22 Sep 2026 13:27:50 -0700 Subject: [PATCH] refactor common code into unlockAndInvokeFECRecovery --- pkg/sfu/buffer/buffer.go | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index b77f24642..5ff0addfb 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -103,6 +103,14 @@ type Buffer struct { warnedPendingOverflow bool } +func (b *Buffer) unlockAndInvokeFECRecovery( + delta fecRecoveryDelta, + callback func(received int, recovered int, discarded int, bytesReceived int), +) { + b.Unlock() + delta.invoke(callback) +} + // StreamInfoProbe identifies a stream from the mid/rid/rsid header extensions of its // packets. It runs on the write path, i. e. as SRTP pushes into this buffer, because // nothing reads remote streams through pion's interceptor chain. @@ -287,8 +295,7 @@ func (b *Buffer) Write(pkt []byte) (n int, err error) { // previously unrecoverable FEC window fecDelta, onFECRecovery = b.feedFECLocked(&rtpPacket, now) } - b.Unlock() - fecDelta.invoke(onFECRecovery) + b.unlockAndInvokeFECRecovery(fecDelta, onFECRecovery) if len(rtcpPackets) != 0 { if cb := b.getOnRtcpFeedback(); cb != nil { @@ -418,8 +425,7 @@ func (b *Buffer) writeRTX(rtxPkt *rtp.Packet, arrivalTime int64) { if b.fecDecoder != nil { fecDelta, onFECRecovery = b.feedFECLocked(&repairedPkt, arrivalTime) } - b.Unlock() - fecDelta.invoke(onFECRecovery) + b.unlockAndInvokeFECRecovery(fecDelta, onFECRecovery) } func (b *Buffer) SetPrimaryBufferForFEC(primaryBuffer *Buffer) { @@ -565,8 +571,7 @@ func (b *Buffer) writeFEC(fecPkt *rtp.Packet, arrivalTime int64) { } fecDelta, onFECRecovery := b.feedFECLocked(fecPkt, arrivalTime) - b.Unlock() - fecDelta.invoke(onFECRecovery) + b.unlockAndInvokeFECRecovery(fecDelta, onFECRecovery) } // feedFECLocked runs a media or FEC packet through the FEC decoder and