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
|
// Package mux multiplexes packets on a single socket (RFC7983)
package mux
import (
"net"
"sync"
"github.com/pion/logging"
"github.com/pion/transport/packetio"
)
// The maximum amount of data that can be buffered before returning errors.
const maxBufferSize = 1000 * 1000 // 1MB
// Config collects the arguments to mux.Mux construction into
// a single structure
type Config struct {
Conn net.Conn
BufferSize int
LoggerFactory logging.LoggerFactory
}
// Mux allows multiplexing
type Mux struct {
lock sync.RWMutex
nextConn net.Conn
endpoints map[*Endpoint]MatchFunc
bufferSize int
closedCh chan struct{}
log logging.LeveledLogger
}
// NewMux creates a new Mux
func NewMux(config Config) *Mux {
m := &Mux{
nextConn: config.Conn,
endpoints: make(map[*Endpoint]MatchFunc),
bufferSize: config.BufferSize,
closedCh: make(chan struct{}),
log: config.LoggerFactory.NewLogger("mux"),
}
go m.readLoop()
return m
}
// NewEndpoint creates a new Endpoint
func (m *Mux) NewEndpoint(f MatchFunc) *Endpoint {
e := &Endpoint{
mux: m,
buffer: packetio.NewBuffer(),
}
// Set a maximum size of the buffer in bytes.
// NOTE: We actually won't get anywhere close to this limit.
// SRTP will constantly read from the endpoint and drop packets if it's full.
e.buffer.SetLimitSize(maxBufferSize)
m.lock.Lock()
m.endpoints[e] = f
m.lock.Unlock()
return e
}
// RemoveEndpoint removes an endpoint from the Mux
func (m *Mux) RemoveEndpoint(e *Endpoint) {
m.lock.Lock()
defer m.lock.Unlock()
delete(m.endpoints, e)
}
// Close closes the Mux and all associated Endpoints.
func (m *Mux) Close() error {
m.lock.Lock()
for e := range m.endpoints {
err := e.close()
if err != nil {
return err
}
delete(m.endpoints, e)
}
m.lock.Unlock()
err := m.nextConn.Close()
if err != nil {
return err
}
// Wait for readLoop to end
<-m.closedCh
return nil
}
func (m *Mux) readLoop() {
defer func() {
close(m.closedCh)
}()
buf := make([]byte, m.bufferSize)
for {
n, err := m.nextConn.Read(buf)
if err != nil {
return
}
err = m.dispatch(buf[:n])
if err != nil {
return
}
}
}
func (m *Mux) dispatch(buf []byte) error {
var endpoint *Endpoint
m.lock.Lock()
for e, f := range m.endpoints {
if f(buf) {
endpoint = e
break
}
}
m.lock.Unlock()
if endpoint == nil {
if len(buf) > 0 {
m.log.Warnf("Warning: mux: no endpoint for packet starting with %d\n", buf[0])
} else {
m.log.Warnf("Warning: mux: no endpoint for zero length packet")
}
return nil
}
_, err := endpoint.buffer.Write(buf)
if err != nil {
return err
}
return nil
}
|
