moqtransport/remote_track_map.go at main · mengelbart/moqtransport · GitHub

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
package moqtransport

import (
	"errors"
	"fmt"
	"sync"
)

type errRequestsBlocked struct {
	maxRequestID uint64
}

func (e errRequestsBlocked) Error() string {
	return fmt.Sprintf("too many subscribes, max_request_id=%v", e.maxRequestID)
}

var (
	errUnknownRequestID       = errors.New("unknown request ID")
	errDuplicateRequestIDBug  = errors.New("internal error: duplicate request ID")
	errDuplicateTrackAliasBug = errors.New("internal error: duplicate track alias")
)

type remoteTrackMap struct {
	lock                  sync.Mutex
	nextTrackAlias        uint64
	pending               map[uint64]*RemoteTrack
	open                  map[uint64]*RemoteTrack
	trackAliasToRequestID map[uint64]uint64
}

func newRemoteTrackMap() *remoteTrackMap {
	return &remoteTrackMap{
		lock:                  sync.Mutex{},
		nextTrackAlias:        0,
		pending:               map[uint64]*RemoteTrack{},
		open:                  map[uint64]*RemoteTrack{},
		trackAliasToRequestID: map[uint64]uint64{},
	}
}

func (m *remoteTrackMap) findByRequestID(id uint64) (*RemoteTrack, bool) {
	m.lock.Lock()
	defer m.lock.Unlock()
	sub, ok := m.open[id]
	if !ok {
		sub, ok = m.pending[id]
	}
	if !ok {
		return nil, false
	}
	return sub, true
}

func (m *remoteTrackMap) addPending(requestID uint64, rt *RemoteTrack) error {
	m.lock.Lock()
	defer m.lock.Unlock()
	if _, ok := m.pending[requestID]; ok {
		// Should never happen
		return errDuplicateRequestIDBug
	}
	if _, ok := m.open[requestID]; ok {
		// Should never happen
		return errDuplicateRequestIDBug
	}
	m.pending[requestID] = rt
	return nil
}

func (m *remoteTrackMap) addPendingWithAlias(requestID uint64, rt *RemoteTrack) error {
	m.lock.Lock()
	defer m.lock.Unlock()
	if _, ok := m.pending[requestID]; ok {
		return errDuplicateRequestIDBug
	}
	if _, ok := m.open[requestID]; ok {
		return errDuplicateRequestIDBug
	}
	m.pending[requestID] = rt
	return nil
}

func (m *remoteTrackMap) setAlias(id, alias uint64) error {
	m.lock.Lock()
	defer m.lock.Unlock()
	if _, ok := m.trackAliasToRequestID[alias]; ok {
		return errDuplicateTrackAliasBug
	}
	m.trackAliasToRequestID[alias] = id
	return nil
}

func (m *remoteTrackMap) confirm(id uint64) (*RemoteTrack, error) {
	m.lock.Lock()
	defer m.lock.Unlock()
	s, ok := m.pending[id]
	if !ok {
		return nil, errUnknownRequestID
	}
	delete(m.pending, id)
	m.open[id] = s
	return s, nil
}

func (m *remoteTrackMap) reject(id uint64) (*RemoteTrack, bool) {
	m.lock.Lock()
	defer m.lock.Unlock()
	s, ok := m.pending[id]
	if !ok {
		return nil, false
	}
	delete(m.pending, id)
	return s, true
}

func (m *remoteTrackMap) findByTrackAlias(alias uint64) (*RemoteTrack, bool) {
	m.lock.Lock()
	id, ok := m.trackAliasToRequestID[alias]
	m.lock.Unlock()
	if !ok {
		return nil, false
	}
	return m.findByRequestID(id)
}