1
0
mirror of https://github.com/v2fly/v2ray-core.git synced 2024-12-22 18:17:52 -05:00
v2fly/vendor/github.com/lucas-clemente/quic-go/streams_map_outgoing_generic.go
Darien Raymond 19926b8e4f
update quic
2018-11-23 17:04:53 +01:00

138 lines
3.5 KiB
Go

package quic
import (
"fmt"
"sync"
"github.com/lucas-clemente/quic-go/internal/protocol"
"github.com/lucas-clemente/quic-go/internal/qerr"
"github.com/lucas-clemente/quic-go/internal/wire"
)
//go:generate genny -in $GOFILE -out streams_map_outgoing_bidi.go gen "item=streamI Item=BidiStream streamTypeGeneric=protocol.StreamTypeBidi"
//go:generate genny -in $GOFILE -out streams_map_outgoing_uni.go gen "item=sendStreamI Item=UniStream streamTypeGeneric=protocol.StreamTypeUni"
type outgoingItemsMap struct {
mutex sync.RWMutex
cond sync.Cond
streams map[protocol.StreamID]item
nextStream protocol.StreamID // stream ID of the stream returned by OpenStream(Sync)
maxStream protocol.StreamID // the maximum stream ID we're allowed to open
maxStreamSet bool // was maxStream set. If not, it's not possible to any stream (also works for stream 0)
blockedSent bool // was a STREAMS_BLOCKED sent for the current maxStream
newStream func(protocol.StreamID) item
queueStreamIDBlocked func(*wire.StreamsBlockedFrame)
closeErr error
}
func newOutgoingItemsMap(
nextStream protocol.StreamID,
newStream func(protocol.StreamID) item,
queueControlFrame func(wire.Frame),
) *outgoingItemsMap {
m := &outgoingItemsMap{
streams: make(map[protocol.StreamID]item),
nextStream: nextStream,
newStream: newStream,
queueStreamIDBlocked: func(f *wire.StreamsBlockedFrame) { queueControlFrame(f) },
}
m.cond.L = &m.mutex
return m
}
func (m *outgoingItemsMap) OpenStream() (item, error) {
m.mutex.Lock()
defer m.mutex.Unlock()
return m.openStreamImpl()
}
func (m *outgoingItemsMap) OpenStreamSync() (item, error) {
m.mutex.Lock()
defer m.mutex.Unlock()
for {
str, err := m.openStreamImpl()
if err == nil {
return str, err
}
if err != nil && err != qerr.TooManyOpenStreams {
return nil, err
}
m.cond.Wait()
}
}
func (m *outgoingItemsMap) openStreamImpl() (item, error) {
if m.closeErr != nil {
return nil, m.closeErr
}
if !m.maxStreamSet || m.nextStream > m.maxStream {
if !m.blockedSent {
if m.maxStreamSet {
m.queueStreamIDBlocked(&wire.StreamsBlockedFrame{
Type: streamTypeGeneric,
StreamLimit: m.maxStream.StreamNum(),
})
} else {
m.queueStreamIDBlocked(&wire.StreamsBlockedFrame{
Type: streamTypeGeneric,
StreamLimit: 0,
})
}
m.blockedSent = true
}
return nil, qerr.TooManyOpenStreams
}
s := m.newStream(m.nextStream)
m.streams[m.nextStream] = s
m.nextStream += 4
return s, nil
}
func (m *outgoingItemsMap) GetStream(id protocol.StreamID) (item, error) {
m.mutex.RLock()
if id >= m.nextStream {
m.mutex.RUnlock()
return nil, qerr.Error(qerr.InvalidStreamID, fmt.Sprintf("peer attempted to open stream %d", id))
}
s := m.streams[id]
m.mutex.RUnlock()
return s, nil
}
func (m *outgoingItemsMap) DeleteStream(id protocol.StreamID) error {
m.mutex.Lock()
defer m.mutex.Unlock()
if _, ok := m.streams[id]; !ok {
return fmt.Errorf("Tried to delete unknown stream %d", id)
}
delete(m.streams, id)
return nil
}
func (m *outgoingItemsMap) SetMaxStream(id protocol.StreamID) {
m.mutex.Lock()
if !m.maxStreamSet || id > m.maxStream {
m.maxStream = id
m.maxStreamSet = true
m.blockedSent = false
m.cond.Broadcast()
}
m.mutex.Unlock()
}
func (m *outgoingItemsMap) CloseWithError(err error) {
m.mutex.Lock()
m.closeErr = err
for _, str := range m.streams {
str.closeForShutdown(err)
}
m.cond.Broadcast()
m.mutex.Unlock()
}