Handle NACKs locally.

client.go

@@ -16,6 +16,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"sfu/packetlist"
+
 	"github.com/gorilla/websocket"
 	"github.com/pion/rtcp"
 	"github.com/pion/rtp"
@@ -269,8 +271,10 @@ func addUpConn(c *client, id string) (*upConnection, error) {
 			c.mu.Unlock()
 			return
 		}
+		list := packetlist.New(32)
 		track := &upTrack{
 			track:      remote,
+			list:       list,
 			maxBitrate: ^uint64(0),
 		}
 		u.tracks = append(u.tracks, track)
@@ -286,13 +290,13 @@ func addUpConn(c *client, id string) (*upConnection, error) {
 		}
 
 		go func() {
-			buf := make([]byte, 1500)
+			buf := make([]byte, packetlist.BufSize)
 			var packet rtp.Packet
 			var local []*downTrack
 			var localTime time.Time
 			for {
 				now := time.Now()
-				if now.Sub(localTime) > time.Second / 2 {
+				if now.Sub(localTime) > time.Second/2 {
 					local = track.getLocal()
 					localTime = now
 				}
@@ -311,6 +315,8 @@ func addUpConn(c *client, id string) (*upConnection, error) {
 					continue
 				}
 
+				list.Store(packet.SequenceNumber, buf[:i])
+
 				for _, l := range local {
 					if l.muted() {
 						continue
@@ -523,6 +529,8 @@ func rtcpListener(g *group, conn *downConnection, track *downTrack, s *webrtc.RT
 					uint64(ms),
 				)
 			case *rtcp.ReceiverReport:
+			case *rtcp.TransportLayerNack:
+				sendRecovery(p, track)
 			default:
 				log.Printf("RTCP: %T", p)
 			}
@@ -565,7 +573,7 @@ func updateUpBitrate(up *upConnection) {
 			}
 			if bitrate < 9600 ||
 				(l.track.Kind() == webrtc.RTPCodecTypeVideo &&
-				bitrate < 128000) {
+					bitrate < 128000) {
 				l.setMuted(true)
 				continue
 			}
@@ -592,6 +600,25 @@ func sendREMB(pc *webrtc.PeerConnection, ssrc uint32, bitrate uint64) error {
 	})
 }
 
+func sendRecovery(p *rtcp.TransportLayerNack, track *downTrack) {
+	var packet rtp.Packet
+	for _, nack := range p.Nacks {
+		for _, seqno := range nack.PacketList() {
+			raw := track.remote.list.Get(seqno)
+			if raw != nil {
+				err := packet.Unmarshal(raw)
+				if err != nil {
+					continue
+				}
+				err = track.track.WriteRTP(&packet)
+				if err != nil {
+					log.Printf("%v", err)
+				}
+			}
+		}
+	}
+}
+
 func countMediaStreams(data string) (int, error) {
 	desc := sdp.NewJSEPSessionDescription(false)
 	err := desc.Unmarshal(data)

group.go

@@ -15,11 +15,14 @@ import (
 	"sync/atomic"
 	"time"
 
+	"sfu/packetlist"
+
 	"github.com/pion/webrtc/v2"
 )
 
 type upTrack struct {
 	track      *webrtc.Track
+	list       *packetlist.List
 	maxBitrate uint64
 
 	mu    sync.Mutex
@@ -172,6 +175,7 @@ func addGroup(name string, desc *groupDescription) (*group, error) {
 			webrtc.DefaultPayloadTypeVP8, 90000,
 			[]webrtc.RTCPFeedback{
 				{"goog-remb", ""},
+				{"nack", ""},
 				{"nack", "pli"},
 			},
 			"",

packetlist/packetlist.go

+package packetlist
+
+import (
+	"sync"
+)
+
+const BufSize = 1500
+
+type entry struct {
+	seqno  uint16
+	length int
+	buf    [BufSize]byte
+}
+
+type List struct {
+	mu      sync.Mutex
+	tail    int
+	entries []entry
+}
+
+func New(capacity int) *List {
+	return &List{
+		entries: make([]entry, capacity),
+	}
+}
+
+func (list *List) Store(seqno uint16, buf []byte) {
+	list.mu.Lock()
+	defer list.mu.Unlock()
+	list.entries[list.tail].seqno = seqno
+	copy(list.entries[list.tail].buf[:], buf)
+	list.entries[list.tail].length = len(buf)
+	list.tail = (list.tail + 1) % len(list.entries)
+
+}
+
+func (list *List) Get(seqno uint16) []byte {
+	list.mu.Lock()
+	defer list.mu.Unlock()
+
+	for _, entry := range list.entries {
+		if entry.length == 0 || entry.seqno != seqno {
+			continue
+		}
+		buf := make([]byte, entry.length)
+		copy(buf, entry.buf[:entry.length])
+		return buf
+	}
+	return nil
+}