-
Notifications
You must be signed in to change notification settings - Fork 55
/
receiver_interceptor.go
184 lines (151 loc) · 4.57 KB
/
receiver_interceptor.go
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
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
// SPDX-License-Identifier: MIT
package report
import (
"sync"
"time"
"github.com/pion/interceptor"
"github.com/pion/logging"
"github.com/pion/rtcp"
)
// ReceiverInterceptorFactory is a interceptor.Factory for a ReceiverInterceptor
type ReceiverInterceptorFactory struct {
opts []ReceiverOption
}
// NewInterceptor constructs a new ReceiverInterceptor
func (r *ReceiverInterceptorFactory) NewInterceptor(_ string) (interceptor.Interceptor, error) {
i := &ReceiverInterceptor{
interval: 1 * time.Second,
now: time.Now,
log: logging.NewDefaultLoggerFactory().NewLogger("receiver_interceptor"),
close: make(chan struct{}),
}
for _, opt := range r.opts {
if err := opt(i); err != nil {
return nil, err
}
}
return i, nil
}
// NewReceiverInterceptor returns a new ReceiverInterceptorFactory
func NewReceiverInterceptor(opts ...ReceiverOption) (*ReceiverInterceptorFactory, error) {
return &ReceiverInterceptorFactory{opts}, nil
}
// ReceiverInterceptor interceptor generates receiver reports.
type ReceiverInterceptor struct {
interceptor.NoOp
interval time.Duration
now func() time.Time
streams sync.Map
log logging.LeveledLogger
m sync.Mutex
wg sync.WaitGroup
close chan struct{}
}
func (r *ReceiverInterceptor) isClosed() bool {
select {
case <-r.close:
return true
default:
return false
}
}
// Close closes the interceptor.
func (r *ReceiverInterceptor) Close() error {
defer r.wg.Wait()
r.m.Lock()
defer r.m.Unlock()
if !r.isClosed() {
close(r.close)
}
return nil
}
// BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method
// will be called once per packet batch.
func (r *ReceiverInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
r.m.Lock()
defer r.m.Unlock()
if r.isClosed() {
return writer
}
r.wg.Add(1)
go r.loop(writer)
return writer
}
func (r *ReceiverInterceptor) loop(rtcpWriter interceptor.RTCPWriter) {
defer r.wg.Done()
ticker := time.NewTicker(r.interval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
now := r.now()
r.streams.Range(func(_, value interface{}) bool {
if stream, ok := value.(*receiverStream); !ok {
r.log.Warnf("failed to cast ReceiverInterceptor stream")
} else if _, err := rtcpWriter.Write([]rtcp.Packet{stream.generateReport(now)}, interceptor.Attributes{}); err != nil {
r.log.Warnf("failed sending: %+v", err)
}
return true
})
case <-r.close:
return
}
}
}
// BindRemoteStream lets you modify any incoming RTP packets. It is called once for per RemoteStream. The returned method
// will be called once per rtp packet.
func (r *ReceiverInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
stream := newReceiverStream(info.SSRC, info.ClockRate)
r.streams.Store(info.SSRC, stream)
return interceptor.RTPReaderFunc(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)
}
header, err := attr.GetRTPHeader(b[:i])
if err != nil {
return 0, nil, err
}
stream.processRTP(r.now(), header)
return i, attr, nil
})
}
// UnbindRemoteStream is called when the Stream is removed. It can be used to clean up any data related to that track.
func (r *ReceiverInterceptor) UnbindRemoteStream(info *interceptor.StreamInfo) {
r.streams.Delete(info.SSRC)
}
// 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 (r *ReceiverInterceptor) 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 _, pkt := range pkts {
if sr, ok := (pkt).(*rtcp.SenderReport); ok {
value, ok := r.streams.Load(sr.SSRC)
if !ok {
continue
}
if stream, ok := value.(*receiverStream); !ok {
r.log.Warnf("failed to cast ReceiverInterceptor stream")
} else {
stream.processSenderReport(r.now(), sr)
}
}
}
return i, attr, nil
})
}