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
|
package srtp
import (
"errors"
"io"
"sync"
"time"
"github.com/pion/rtp"
"github.com/pion/transport/packetio"
)
// Limit the buffer size to 1MB
const srtpBufferSize = 1000 * 1000
// ReadStreamSRTP handles decryption for a single RTP SSRC
type ReadStreamSRTP struct {
mu sync.Mutex
isInited bool
isClosed chan bool
session *SessionSRTP
ssrc uint32
buffer io.ReadWriteCloser
}
// Used by getOrCreateReadStream
func newReadStreamSRTP() readStream {
return &ReadStreamSRTP{}
}
func (r *ReadStreamSRTP) init(child streamSession, ssrc uint32) error {
sessionSRTP, ok := child.(*SessionSRTP)
r.mu.Lock()
defer r.mu.Unlock()
if !ok {
return errFailedTypeAssertion
} else if r.isInited {
return errStreamAlreadyInited
}
r.session = sessionSRTP
r.ssrc = ssrc
r.isInited = true
r.isClosed = make(chan bool)
// Create a buffer with a 1MB limit
if r.session.bufferFactory != nil {
r.buffer = r.session.bufferFactory(packetio.RTPBufferPacket, ssrc)
} else {
buff := packetio.NewBuffer()
buff.SetLimitSize(srtpBufferSize)
r.buffer = buff
}
return nil
}
func (r *ReadStreamSRTP) write(buf []byte) (n int, err error) {
n, err = r.buffer.Write(buf)
if errors.Is(err, packetio.ErrFull) {
// Silently drop data when the buffer is full.
return len(buf), nil
}
return n, err
}
// Read reads and decrypts full RTP packet from the nextConn
func (r *ReadStreamSRTP) Read(buf []byte) (int, error) {
return r.buffer.Read(buf)
}
// ReadRTP reads and decrypts full RTP packet and its header from the nextConn
func (r *ReadStreamSRTP) ReadRTP(buf []byte) (int, *rtp.Header, error) {
n, err := r.Read(buf)
if err != nil {
return 0, nil, err
}
header := &rtp.Header{}
err = header.Unmarshal(buf[:n])
if err != nil {
return 0, nil, err
}
return n, header, nil
}
// SetReadDeadline sets the deadline for the Read operation.
// Setting to zero means no deadline.
func (r *ReadStreamSRTP) SetReadDeadline(t time.Time) error {
if b, ok := r.buffer.(interface {
SetReadDeadline(time.Time) error
}); ok {
return b.SetReadDeadline(t)
}
return nil
}
// Close removes the ReadStream from the session and cleans up any associated state
func (r *ReadStreamSRTP) Close() error {
r.mu.Lock()
defer r.mu.Unlock()
if !r.isInited {
return errStreamNotInited
}
select {
case <-r.isClosed:
return errStreamAlreadyClosed
default:
err := r.buffer.Close()
if err != nil {
return err
}
r.session.removeReadStream(r.ssrc)
return nil
}
}
// GetSSRC returns the SSRC we are demuxing for
func (r *ReadStreamSRTP) GetSSRC() uint32 {
return r.ssrc
}
// WriteStreamSRTP is stream for a single Session that is used to encrypt RTP
type WriteStreamSRTP struct {
session *SessionSRTP
}
// WriteRTP encrypts a RTP packet and writes to the connection
func (w *WriteStreamSRTP) WriteRTP(header *rtp.Header, payload []byte) (int, error) {
return w.session.writeRTP(header, payload)
}
// Write encrypts and writes a full RTP packets to the nextConn
func (w *WriteStreamSRTP) Write(b []byte) (int, error) {
return w.session.write(b)
}
// SetWriteDeadline sets the deadline for the Write operation.
// Setting to zero means no deadline.
func (w *WriteStreamSRTP) SetWriteDeadline(t time.Time) error {
return w.session.setWriteDeadline(t)
}
|
