1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
package nack
import (
"sync"
"github.com/pion/interceptor"
"github.com/pion/logging"
"github.com/pion/rtcp"
"github.com/pion/rtp"
)
// ResponderInterceptorFactory is a interceptor.Factory for a ResponderInterceptor
type ResponderInterceptorFactory struct {
opts []ResponderOption
}
type packetFactory interface {
NewPacket(header *rtp.Header, payload []byte) (*retainablePacket, error)
}
// NewInterceptor constructs a new ResponderInterceptor
func (r *ResponderInterceptorFactory) NewInterceptor(id string) (interceptor.Interceptor, error) {
i := &ResponderInterceptor{
size: 1024,
log: logging.NewDefaultLoggerFactory().NewLogger("nack_responder"),
streams: map[uint32]*localStream{},
}
for _, opt := range r.opts {
if err := opt(i); err != nil {
return nil, err
}
}
if i.packetFactory == nil {
i.packetFactory = newPacketManager()
}
if _, err := newSendBuffer(i.size); err != nil {
return nil, err
}
return i, nil
}
// ResponderInterceptor responds to nack feedback messages
type ResponderInterceptor struct {
interceptor.NoOp
size uint16
log logging.LeveledLogger
packetFactory packetFactory
streams map[uint32]*localStream
streamsMu sync.Mutex
}
type localStream struct {
sendBuffer *sendBuffer
rtpWriter interceptor.RTPWriter
}
// NewResponderInterceptor returns a new ResponderInterceptorFactor
func NewResponderInterceptor(opts ...ResponderOption) (*ResponderInterceptorFactory, error) {
return &ResponderInterceptorFactory{opts}, nil
}
// BindRTCPReader lets you modify any incoming RTCP packets. It is called once per sender/receiver, however this might
// change in the future. The returned method will be called once per packet batch.
func (n *ResponderInterceptor) BindRTCPReader(reader interceptor.RTCPReader) interceptor.RTCPReader {
return interceptor.RTCPReaderFunc(func(b []byte, a interceptor.Attributes) (int, interceptor.Attributes, error) {
i, attr, err := reader.Read(b, a)
if err != nil {
return 0, nil, err
}
if attr == nil {
attr = make(interceptor.Attributes)
}
pkts, err := attr.GetRTCPPackets(b[:i])
if err != nil {
return 0, nil, err
}
for _, rtcpPacket := range pkts {
nack, ok := rtcpPacket.(*rtcp.TransportLayerNack)
if !ok {
continue
}
go n.resendPackets(nack)
}
return i, attr, err
})
}
// BindLocalStream lets you modify any outgoing RTP packets. It is called once for per LocalStream. The returned method
// will be called once per rtp packet.
func (n *ResponderInterceptor) BindLocalStream(info *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
if !streamSupportNack(info) {
return writer
}
// error is already checked in NewGeneratorInterceptor
sendBuffer, _ := newSendBuffer(n.size)
n.streamsMu.Lock()
n.streams[info.SSRC] = &localStream{sendBuffer: sendBuffer, rtpWriter: writer}
n.streamsMu.Unlock()
return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, attributes interceptor.Attributes) (int, error) {
pkt, err := n.packetFactory.NewPacket(header, payload)
if err != nil {
return 0, err
}
sendBuffer.add(pkt)
return writer.Write(header, payload, attributes)
})
}
// UnbindLocalStream is called when the Stream is removed. It can be used to clean up any data related to that track.
func (n *ResponderInterceptor) UnbindLocalStream(info *interceptor.StreamInfo) {
n.streamsMu.Lock()
delete(n.streams, info.SSRC)
n.streamsMu.Unlock()
}
func (n *ResponderInterceptor) resendPackets(nack *rtcp.TransportLayerNack) {
n.streamsMu.Lock()
stream, ok := n.streams[nack.MediaSSRC]
n.streamsMu.Unlock()
if !ok {
return
}
for i := range nack.Nacks {
nack.Nacks[i].Range(func(seq uint16) bool {
if p := stream.sendBuffer.get(seq); p != nil {
if _, err := stream.rtpWriter.Write(p.Header(), p.Payload(), interceptor.Attributes{}); err != nil {
n.log.Warnf("failed resending nacked packet: %+v", err)
}
p.Release()
}
return true
})
}
}
|
