forked from quic-go/quic-go
-
Notifications
You must be signed in to change notification settings - Fork 0
/
streams_map.go
172 lines (153 loc) · 5.37 KB
/
streams_map.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
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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
package quic
import (
"fmt"
"github.com/boisjacques/qed/internal/flowcontrol"
"github.com/boisjacques/qed/internal/handshake"
"github.com/boisjacques/qed/internal/protocol"
"github.com/boisjacques/qed/internal/wire"
)
type streamsMap struct {
perspective protocol.Perspective
sender streamSender
newFlowController func(protocol.StreamID) flowcontrol.StreamFlowController
outgoingBidiStreams *outgoingBidiStreamsMap
outgoingUniStreams *outgoingUniStreamsMap
incomingBidiStreams *incomingBidiStreamsMap
incomingUniStreams *incomingUniStreamsMap
}
var _ streamManager = &streamsMap{}
func newStreamsMap(
sender streamSender,
newFlowController func(protocol.StreamID) flowcontrol.StreamFlowController,
maxIncomingStreams uint64,
maxIncomingUniStreams uint64,
perspective protocol.Perspective,
version protocol.VersionNumber,
) streamManager {
m := &streamsMap{
perspective: perspective,
newFlowController: newFlowController,
sender: sender,
}
newBidiStream := func(id protocol.StreamID) streamI {
return newStream(id, m.sender, m.newFlowController(id), version)
}
newUniSendStream := func(id protocol.StreamID) sendStreamI {
return newSendStream(id, m.sender, m.newFlowController(id), version)
}
newUniReceiveStream := func(id protocol.StreamID) receiveStreamI {
return newReceiveStream(id, m.sender, m.newFlowController(id), version)
}
m.outgoingBidiStreams = newOutgoingBidiStreamsMap(
protocol.FirstStream(protocol.StreamTypeBidi, perspective),
newBidiStream,
sender.queueControlFrame,
)
m.incomingBidiStreams = newIncomingBidiStreamsMap(
protocol.FirstStream(protocol.StreamTypeBidi, perspective.Opposite()),
protocol.MaxStreamID(protocol.StreamTypeBidi, maxIncomingStreams, perspective.Opposite()),
maxIncomingStreams,
sender.queueControlFrame,
newBidiStream,
)
m.outgoingUniStreams = newOutgoingUniStreamsMap(
protocol.FirstStream(protocol.StreamTypeUni, perspective),
newUniSendStream,
sender.queueControlFrame,
)
m.incomingUniStreams = newIncomingUniStreamsMap(
protocol.FirstStream(protocol.StreamTypeUni, perspective.Opposite()),
protocol.MaxStreamID(protocol.StreamTypeUni, maxIncomingUniStreams, perspective.Opposite()),
maxIncomingUniStreams,
sender.queueControlFrame,
newUniReceiveStream,
)
return m
}
func (m *streamsMap) OpenStream() (Stream, error) {
return m.outgoingBidiStreams.OpenStream()
}
func (m *streamsMap) OpenStreamSync() (Stream, error) {
return m.outgoingBidiStreams.OpenStreamSync()
}
func (m *streamsMap) OpenUniStream() (SendStream, error) {
return m.outgoingUniStreams.OpenStream()
}
func (m *streamsMap) OpenUniStreamSync() (SendStream, error) {
return m.outgoingUniStreams.OpenStreamSync()
}
func (m *streamsMap) AcceptStream() (Stream, error) {
return m.incomingBidiStreams.AcceptStream()
}
func (m *streamsMap) AcceptUniStream() (ReceiveStream, error) {
return m.incomingUniStreams.AcceptStream()
}
func (m *streamsMap) DeleteStream(id protocol.StreamID) error {
switch id.Type() {
case protocol.StreamTypeUni:
if id.InitiatedBy() == m.perspective {
return m.outgoingUniStreams.DeleteStream(id)
}
return m.incomingUniStreams.DeleteStream(id)
case protocol.StreamTypeBidi:
if id.InitiatedBy() == m.perspective {
return m.outgoingBidiStreams.DeleteStream(id)
}
return m.incomingBidiStreams.DeleteStream(id)
}
panic("")
}
func (m *streamsMap) GetOrOpenReceiveStream(id protocol.StreamID) (receiveStreamI, error) {
switch id.Type() {
case protocol.StreamTypeUni:
if id.InitiatedBy() == m.perspective {
// an outgoing unidirectional stream is a send stream, not a receive stream
return nil, fmt.Errorf("peer attempted to open receive stream %d", id)
}
return m.incomingUniStreams.GetOrOpenStream(id)
case protocol.StreamTypeBidi:
if id.InitiatedBy() == m.perspective {
return m.outgoingBidiStreams.GetStream(id)
}
return m.incomingBidiStreams.GetOrOpenStream(id)
}
panic("")
}
func (m *streamsMap) GetOrOpenSendStream(id protocol.StreamID) (sendStreamI, error) {
switch id.Type() {
case protocol.StreamTypeUni:
if id.InitiatedBy() == m.perspective {
return m.outgoingUniStreams.GetStream(id)
}
// an incoming unidirectional stream is a receive stream, not a send stream
return nil, fmt.Errorf("peer attempted to open send stream %d", id)
case protocol.StreamTypeBidi:
if id.InitiatedBy() == m.perspective {
return m.outgoingBidiStreams.GetStream(id)
}
return m.incomingBidiStreams.GetOrOpenStream(id)
}
panic("")
}
func (m *streamsMap) HandleMaxStreamsFrame(f *wire.MaxStreamsFrame) error {
id := protocol.MaxStreamID(f.Type, f.MaxStreams, m.perspective)
switch id.Type() {
case protocol.StreamTypeUni:
m.outgoingUniStreams.SetMaxStream(id)
case protocol.StreamTypeBidi:
fmt.Printf("")
m.outgoingBidiStreams.SetMaxStream(id)
}
return nil
}
func (m *streamsMap) UpdateLimits(p *handshake.TransportParameters) {
// Max{Uni,Bidi}StreamID returns the highest stream ID that the peer is allowed to open.
m.outgoingBidiStreams.SetMaxStream(protocol.MaxStreamID(protocol.StreamTypeBidi, p.MaxBidiStreams, m.perspective))
m.outgoingUniStreams.SetMaxStream(protocol.MaxStreamID(protocol.StreamTypeUni, p.MaxUniStreams, m.perspective))
}
func (m *streamsMap) CloseWithError(err error) {
m.outgoingBidiStreams.CloseWithError(err)
m.outgoingUniStreams.CloseWithError(err)
m.incomingBidiStreams.CloseWithError(err)
m.incomingUniStreams.CloseWithError(err)
}