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
|
// Package rfc8888 provides an interceptor that generates congestion control
// feedback reports as defined by RFC 8888.
package rfc8888
import (
"sync"
"time"
"github.com/pion/interceptor"
"github.com/pion/logging"
"github.com/pion/rtcp"
)
// TickerFactory is a factory to create new tickers
type TickerFactory func(d time.Duration) ticker
// SenderInterceptorFactory is a interceptor.Factory for a SenderInterceptor
type SenderInterceptorFactory struct {
opts []Option
}
// NewInterceptor constructs a new SenderInterceptor
func (s *SenderInterceptorFactory) NewInterceptor(id string) (interceptor.Interceptor, error) {
i := &SenderInterceptor{
NoOp: interceptor.NoOp{},
log: logging.NewDefaultLoggerFactory().NewLogger("rfc8888_interceptor"),
lock: sync.Mutex{},
wg: sync.WaitGroup{},
recorder: NewRecorder(),
interval: 100 * time.Millisecond,
maxReportSize: 1200,
packetChan: make(chan packet),
newTicker: func(d time.Duration) ticker {
return &timeTicker{time.NewTicker(d)}
},
now: time.Now,
close: make(chan struct{}),
}
for _, opt := range s.opts {
err := opt(i)
if err != nil {
return nil, err
}
}
return i, nil
}
// NewSenderInterceptor returns a new SenderInterceptorFactory configured with the given options.
func NewSenderInterceptor(opts ...Option) (*SenderInterceptorFactory, error) {
return &SenderInterceptorFactory{opts: opts}, nil
}
// SenderInterceptor sends congestion control feedback as specified in RFC 8888.
type SenderInterceptor struct {
interceptor.NoOp
log logging.LeveledLogger
lock sync.Mutex
wg sync.WaitGroup
recorder *Recorder
interval time.Duration
maxReportSize int64
packetChan chan packet
newTicker TickerFactory
now func() time.Time
close chan struct{}
}
type packet struct {
arrival time.Time
ssrc uint32
sequenceNumber uint16
ecn uint8
}
// 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 (s *SenderInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
s.lock.Lock()
defer s.lock.Unlock()
if s.isClosed() {
return writer
}
s.wg.Add(1)
go s.loop(writer)
return writer
}
// 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 (s *SenderInterceptor) BindRemoteStream(info *interceptor.StreamInfo, reader interceptor.RTPReader) interceptor.RTPReader {
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
}
p := packet{
arrival: s.now(),
ssrc: header.SSRC,
sequenceNumber: header.SequenceNumber,
ecn: 0, // ECN is not supported (yet).
}
s.packetChan <- p
return i, attr, nil
})
}
// Close closes the interceptor.
func (s *SenderInterceptor) Close() error {
s.log.Trace("close")
defer s.wg.Wait()
if !s.isClosed() {
close(s.close)
}
return nil
}
func (s *SenderInterceptor) isClosed() bool {
select {
case <-s.close:
return true
default:
return false
}
}
func (s *SenderInterceptor) loop(writer interceptor.RTCPWriter) {
defer s.wg.Done()
select {
case <-s.close:
return
case pkt := <-s.packetChan:
s.log.Tracef("got first packet: %v", pkt)
s.recorder.AddPacket(pkt.arrival, pkt.ssrc, pkt.sequenceNumber, pkt.ecn)
}
s.log.Trace("start loop")
t := s.newTicker(s.interval)
for {
select {
case <-s.close:
t.Stop()
return
case pkt := <-s.packetChan:
s.log.Tracef("got packet: %v", pkt)
s.recorder.AddPacket(pkt.arrival, pkt.ssrc, pkt.sequenceNumber, pkt.ecn)
case now := <-t.Ch():
s.log.Tracef("report triggered at %v", now)
if writer == nil {
s.log.Trace("no writer added, continue")
continue
}
pkts := s.recorder.BuildReport(now, int(s.maxReportSize))
if pkts == nil {
continue
}
s.log.Tracef("got report: %v", pkts)
if _, err := writer.Write([]rtcp.Packet{pkts}, nil); err != nil {
s.log.Error(err.Error())
}
}
}
}
|
