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
|
package report
import (
"sync"
"time"
"github.com/pion/interceptor"
"github.com/pion/logging"
"github.com/pion/rtcp"
"github.com/pion/rtp"
)
// SenderInterceptorFactory is a interceptor.Factory for a SenderInterceptor
type SenderInterceptorFactory struct {
opts []SenderOption
}
// NewInterceptor constructs a new SenderInterceptor
func (s *SenderInterceptorFactory) NewInterceptor(id string) (interceptor.Interceptor, error) {
i := &SenderInterceptor{
interval: 1 * time.Second,
now: time.Now,
log: logging.NewDefaultLoggerFactory().NewLogger("sender_interceptor"),
close: make(chan struct{}),
}
for _, opt := range s.opts {
if err := opt(i); err != nil {
return nil, err
}
}
return i, nil
}
// NewSenderInterceptor returns a new SenderInterceptorFactory
func NewSenderInterceptor(opts ...SenderOption) (*SenderInterceptorFactory, error) {
return &SenderInterceptorFactory{opts}, nil
}
// SenderInterceptor interceptor generates sender reports.
type SenderInterceptor 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 (s *SenderInterceptor) isClosed() bool {
select {
case <-s.close:
return true
default:
return false
}
}
// Close closes the interceptor.
func (s *SenderInterceptor) Close() error {
defer s.wg.Wait()
s.m.Lock()
defer s.m.Unlock()
if !s.isClosed() {
close(s.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 (s *SenderInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter {
s.m.Lock()
defer s.m.Unlock()
if s.isClosed() {
return writer
}
s.wg.Add(1)
go s.loop(writer)
return writer
}
func (s *SenderInterceptor) loop(rtcpWriter interceptor.RTCPWriter) {
defer s.wg.Done()
ticker := time.NewTicker(s.interval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
now := s.now()
s.streams.Range(func(key, value interface{}) bool {
if stream, ok := value.(*senderStream); !ok {
s.log.Warnf("failed to cast SenderInterceptor stream")
} else if _, err := rtcpWriter.Write([]rtcp.Packet{stream.generateReport(now)}, interceptor.Attributes{}); err != nil {
s.log.Warnf("failed sending: %+v", err)
}
return true
})
case <-s.close:
return
}
}
}
// 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 (s *SenderInterceptor) BindLocalStream(info *interceptor.StreamInfo, writer interceptor.RTPWriter) interceptor.RTPWriter {
stream := newSenderStream(info.SSRC, info.ClockRate)
s.streams.Store(info.SSRC, stream)
return interceptor.RTPWriterFunc(func(header *rtp.Header, payload []byte, a interceptor.Attributes) (int, error) {
stream.processRTP(s.now(), header, payload)
return writer.Write(header, payload, a)
})
}
|
