use a generic streams map for incoming streams

streams_map.go

@@ -57,8 +57,8 @@ type streamsMap struct {
 	mutex               sync.Mutex
 	outgoingBidiStreams *outgoingStreamsMap[streamI]
 	outgoingUniStreams  *outgoingStreamsMap[sendStreamI]
-	incomingBidiStreams *incomingBidiStreamsMap
-	incomingUniStreams  *incomingUniStreamsMap
+	incomingBidiStreams *incomingStreamsMap[streamI]
+	incomingUniStreams  *incomingStreamsMap[receiveStreamI]
 	reset               bool
 }
 
@@ -93,7 +93,8 @@ func (m *streamsMap) initMaps() {
 		},
 		m.sender.queueControlFrame,
 	)
-	m.incomingBidiStreams = newIncomingBidiStreamsMap(
+	m.incomingBidiStreams = newIncomingStreamsMap(
+		protocol.StreamTypeBidi,
 		func(num protocol.StreamNum) streamI {
 			id := num.StreamID(protocol.StreamTypeBidi, m.perspective.Opposite())
 			return newStream(id, m.sender, m.newFlowController(id), m.version)
@@ -109,7 +110,8 @@ func (m *streamsMap) initMaps() {
 		},
 		m.sender.queueControlFrame,
 	)
-	m.incomingUniStreams = newIncomingUniStreamsMap(
+	m.incomingUniStreams = newIncomingStreamsMap(
+		protocol.StreamTypeUni,
 		func(num protocol.StreamNum) receiveStreamI {
 			id := num.StreamID(protocol.StreamTypeUni, m.perspective.Opposite())
 			return newReceiveStream(id, m.sender, m.newFlowController(id), m.version)

streams_map_incoming_uni.go → streams_map_incoming.go

@@ -1,7 +1,3 @@
-// This file was automatically generated by genny.
-// Any changes will be lost if this file is regenerated.
-// see https://github.com/cheekybits/genny
-
 package quic
 
 import (
@@ -12,38 +8,45 @@ import (
 	"github.com/lucas-clemente/quic-go/internal/wire"
 )
 
+type incomingStream interface {
+	closeForShutdown(error)
+}
+
 // When a stream is deleted before it was accepted, we can't delete it from the map immediately.
 // We need to wait until the application accepts it, and delete it then.
-type receiveStreamIEntry struct {
-	stream       receiveStreamI
+type incomingStreamEntry[T incomingStream] struct {
+	stream       T
 	shouldDelete bool
 }
 
-type incomingUniStreamsMap struct {
+type incomingStreamsMap[T incomingStream] struct {
 	mutex         sync.RWMutex
 	newStreamChan chan struct{}
 
-	streams map[protocol.StreamNum]receiveStreamIEntry
+	streamType protocol.StreamType
+	streams    map[protocol.StreamNum]incomingStreamEntry[T]
 
 	nextStreamToAccept protocol.StreamNum // the next stream that will be returned by AcceptStream()
 	nextStreamToOpen   protocol.StreamNum // the highest stream that the peer opened
 	maxStream          protocol.StreamNum // the highest stream that the peer is allowed to open
 	maxNumStreams      uint64             // maximum number of streams
 
-	newStream        func(protocol.StreamNum) receiveStreamI
+	newStream        func(protocol.StreamNum) T
 	queueMaxStreamID func(*wire.MaxStreamsFrame)
 
 	closeErr error
 }
 
-func newIncomingUniStreamsMap(
-	newStream func(protocol.StreamNum) receiveStreamI,
+func newIncomingStreamsMap[T incomingStream](
+	streamType protocol.StreamType,
+	newStream func(protocol.StreamNum) T,
 	maxStreams uint64,
 	queueControlFrame func(wire.Frame),
-) *incomingUniStreamsMap {
-	return &incomingUniStreamsMap{
+) *incomingStreamsMap[T] {
+	return &incomingStreamsMap[T]{
 		newStreamChan:      make(chan struct{}, 1),
-		streams:            make(map[protocol.StreamNum]receiveStreamIEntry),
+		streamType:         streamType,
+		streams:            make(map[protocol.StreamNum]incomingStreamEntry[T]),
 		maxStream:          protocol.StreamNum(maxStreams),
 		maxNumStreams:      maxStreams,
 		newStream:          newStream,
@@ -53,7 +56,7 @@ func newIncomingUniStreamsMap(
 	}
 }
 
-func (m *incomingUniStreamsMap) AcceptStream(ctx context.Context) (receiveStreamI, error) {
+func (m *incomingStreamsMap[T]) AcceptStream(ctx context.Context) (T, error) {
 	// drain the newStreamChan, so we don't check the map twice if the stream doesn't exist
 	select {
 	case <-m.newStreamChan:
@@ -63,12 +66,12 @@ func (m *incomingUniStreamsMap) AcceptStream(ctx context.Context) (receiveStream
 	m.mutex.Lock()
 
 	var num protocol.StreamNum
-	var entry receiveStreamIEntry
+	var entry incomingStreamEntry[T]
 	for {
 		num = m.nextStreamToAccept
 		if m.closeErr != nil {
 			m.mutex.Unlock()
-			return nil, m.closeErr
+			return *new(T), m.closeErr
 		}
 		var ok bool
 		entry, ok = m.streams[num]
@@ -78,7 +81,7 @@ func (m *incomingUniStreamsMap) AcceptStream(ctx context.Context) (receiveStream
 		m.mutex.Unlock()
 		select {
 		case <-ctx.Done():
-			return nil, ctx.Err()
+			return *new(T), ctx.Err()
 		case <-m.newStreamChan:
 		}
 		m.mutex.Lock()
@@ -88,18 +91,18 @@ func (m *incomingUniStreamsMap) AcceptStream(ctx context.Context) (receiveStream
 	if entry.shouldDelete {
 		if err := m.deleteStream(num); err != nil {
 			m.mutex.Unlock()
-			return nil, err
+			return *new(T), err
 		}
 	}
 	m.mutex.Unlock()
 	return entry.stream, nil
 }
 
-func (m *incomingUniStreamsMap) GetOrOpenStream(num protocol.StreamNum) (receiveStreamI, error) {
+func (m *incomingStreamsMap[T]) GetOrOpenStream(num protocol.StreamNum) (T, error) {
 	m.mutex.RLock()
 	if num > m.maxStream {
 		m.mutex.RUnlock()
-		return nil, streamError{
+		return *new(T), streamError{
 			message: "peer tried to open stream %d (current limit: %d)",
 			nums:    []protocol.StreamNum{num, m.maxStream},
 		}
@@ -108,7 +111,7 @@ func (m *incomingUniStreamsMap) GetOrOpenStream(num protocol.StreamNum) (receive
 	// * this stream exists in the map, and we can return it, or
 	// * this stream was already closed, then we can return the nil
 	if num < m.nextStreamToOpen {
-		var s receiveStreamI
+		var s T
 		// If the stream was already queued for deletion, and is just waiting to be accepted, don't return it.
 		if entry, ok := m.streams[num]; ok && !entry.shouldDelete {
 			s = entry.stream
@@ -123,7 +126,7 @@ func (m *incomingUniStreamsMap) GetOrOpenStream(num protocol.StreamNum) (receive
 	// * maxStream can only increase, so if the id was valid before, it definitely is valid now
 	// * highestStream is only modified by this function
 	for newNum := m.nextStreamToOpen; newNum <= num; newNum++ {
-		m.streams[newNum] = receiveStreamIEntry{stream: m.newStream(newNum)}
+		m.streams[newNum] = incomingStreamEntry[T]{stream: m.newStream(newNum)}
 		select {
 		case m.newStreamChan <- struct{}{}:
 		default:
@@ -135,14 +138,14 @@ func (m *incomingUniStreamsMap) GetOrOpenStream(num protocol.StreamNum) (receive
 	return entry.stream, nil
 }
 
-func (m *incomingUniStreamsMap) DeleteStream(num protocol.StreamNum) error {
+func (m *incomingStreamsMap[T]) DeleteStream(num protocol.StreamNum) error {
 	m.mutex.Lock()
 	defer m.mutex.Unlock()
 
 	return m.deleteStream(num)
 }
 
-func (m *incomingUniStreamsMap) deleteStream(num protocol.StreamNum) error {
+func (m *incomingStreamsMap[T]) deleteStream(num protocol.StreamNum) error {
 	if _, ok := m.streams[num]; !ok {
 		return streamError{
 			message: "tried to delete unknown incoming stream %d",
@@ -173,15 +176,15 @@ func (m *incomingUniStreamsMap) deleteStream(num protocol.StreamNum) error {
 		if maxStream <= protocol.MaxStreamCount {
 			m.maxStream = maxStream
 			m.queueMaxStreamID(&wire.MaxStreamsFrame{
-				Type:         protocol.StreamTypeUni,
+				Type:         m.streamType,
 				MaxStreamNum: m.maxStream,
 			})
 		}
 	}
 	return nil
 }
 
-func (m *incomingUniStreamsMap) CloseWithError(err error) {
+func (m *incomingStreamsMap[T]) CloseWithError(err error) {
 	m.mutex.Lock()
 	m.closeErr = err
 	for _, entry := range m.streams {