forked from ionorg/ion-sfu
-
Notifications
You must be signed in to change notification settings - Fork 0
/
factory.go
95 lines (86 loc) · 2.01 KB
/
factory.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
package buffer
import (
"io"
"sync"
"github.com/go-logr/logr"
"github.com/pion/transport/packetio"
)
type Factory struct {
sync.RWMutex
videoPool *sync.Pool
audioPool *sync.Pool
rtpBuffers map[uint32]*Buffer
rtcpReaders map[uint32]*RTCPReader
logger logr.Logger
}
func NewBufferFactory(trackingPackets int, logger logr.Logger) *Factory {
// Enable package wide logging for non-method functions.
// If logger is nil - buffer logs will be disabled.
// Logger is a public variable in buffer package.
if logger != nil {
Logger = logger
} else {
logger = Logger
}
return &Factory{
videoPool: &sync.Pool{
New: func() interface{} {
return make([]byte, trackingPackets*maxPktSize)
},
},
audioPool: &sync.Pool{
New: func() interface{} {
return make([]byte, maxPktSize*25)
},
},
rtpBuffers: make(map[uint32]*Buffer),
rtcpReaders: make(map[uint32]*RTCPReader),
logger: logger,
}
}
func (f *Factory) GetOrNew(packetType packetio.BufferPacketType, ssrc uint32) io.ReadWriteCloser {
f.Lock()
defer f.Unlock()
switch packetType {
case packetio.RTCPBufferPacket:
if reader, ok := f.rtcpReaders[ssrc]; ok {
return reader
}
reader := NewRTCPReader(ssrc)
f.rtcpReaders[ssrc] = reader
reader.OnClose(func() {
f.Lock()
delete(f.rtcpReaders, ssrc)
f.Unlock()
})
return reader
case packetio.RTPBufferPacket:
if reader, ok := f.rtpBuffers[ssrc]; ok {
return reader
}
buffer := NewBuffer(ssrc, f.videoPool, f.audioPool, f.logger)
f.rtpBuffers[ssrc] = buffer
buffer.OnClose(func() {
f.Lock()
delete(f.rtpBuffers, ssrc)
f.Unlock()
})
return buffer
}
return nil
}
func (f *Factory) GetBufferPair(ssrc uint32) (*Buffer, *RTCPReader) {
f.RLock()
defer f.RUnlock()
return f.rtpBuffers[ssrc], f.rtcpReaders[ssrc]
}
func (f *Factory) GetBuffer(ssrc uint32) *Buffer {
f.RLock()
defer f.RUnlock()
return f.rtpBuffers[ssrc]
}
func (f *Factory) GetRTCPReader(ssrc uint32) *RTCPReader {
f.RLock()
defer f.RUnlock()
return f.rtcpReaders[ssrc]
}