implement a map for outgoing streams

codecov.yml

@@ -1,6 +1,8 @@
 coverage:
   round: nearest
   ignore:
+    - streams_map_outgoing_bidi.go
+    - streams_map_outgoing_uni.go
     - h2quic/gzipreader.go
     - h2quic/response.go
     - internal/ackhandler/packet_linkedlist.go

streams_map_outgoing_bidi.go

@@ -0,0 +1,72 @@
+// 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 (
+	"fmt"
+	"sync"
+
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/qerr"
+)
+
+type outgoingBidiStreamsMap struct {
+	mutex sync.RWMutex
+
+	streams map[protocol.StreamID]streamI
+
+	nextStream protocol.StreamID
+	newStream  func(protocol.StreamID) streamI
+
+	closeErr error
+}
+
+func newOutgoingBidiStreamsMap(nextStream protocol.StreamID, newStream func(protocol.StreamID) streamI) *outgoingBidiStreamsMap {
+	return &outgoingBidiStreamsMap{
+		streams:    make(map[protocol.StreamID]streamI),
+		nextStream: nextStream,
+		newStream:  newStream,
+	}
+}
+
+func (m *outgoingBidiStreamsMap) OpenStream() (streamI, error) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	if m.closeErr != nil {
+		return nil, m.closeErr
+	}
+	s := m.newStream(m.nextStream)
+	m.streams[m.nextStream] = s
+	m.nextStream += 4
+	return s, nil
+}
+
+func (m *outgoingBidiStreamsMap) GetStream(id protocol.StreamID) (streamI, error) {
+	if id >= m.nextStream {
+		return nil, qerr.Error(qerr.InvalidStreamID, fmt.Sprintf("peer attempted to open stream %d", id))
+	}
+	m.mutex.RLock()
+	s := m.streams[id]
+	m.mutex.RUnlock()
+	return s, nil
+}
+
+func (m *outgoingBidiStreamsMap) 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 *outgoingBidiStreamsMap) CloseWithError(err error) {
+	m.mutex.Lock()
+	m.closeErr = err
+	m.mutex.Unlock()
+}

streams_map_outgoing_generic.go

@@ -0,0 +1,73 @@
+package quic
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/cheekybits/genny/generic"
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/qerr"
+)
+
+type item generic.Type
+
+//go:generate genny -in $GOFILE -out streams_map_outgoing_bidi.go gen "item=streamI Item=BidiStream"
+//go:generate genny -in $GOFILE -out streams_map_outgoing_uni.go gen "item=sendStreamI Item=UniStream"
+type outgoingItemsMap struct {
+	mutex sync.RWMutex
+
+	streams map[protocol.StreamID]item
+
+	nextStream protocol.StreamID
+	newStream  func(protocol.StreamID) item
+
+	closeErr error
+}
+
+func newOutgoingItemsMap(nextStream protocol.StreamID, newStream func(protocol.StreamID) item) *outgoingItemsMap {
+	return &outgoingItemsMap{
+		streams:    make(map[protocol.StreamID]item),
+		nextStream: nextStream,
+		newStream:  newStream,
+	}
+}
+
+func (m *outgoingItemsMap) OpenStream() (item, error) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	if m.closeErr != nil {
+		return nil, m.closeErr
+	}
+	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) {
+	if id >= m.nextStream {
+		return nil, qerr.Error(qerr.InvalidStreamID, fmt.Sprintf("peer attempted to open stream %d", id))
+	}
+	m.mutex.RLock()
+	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) CloseWithError(err error) {
+	m.mutex.Lock()
+	m.closeErr = err
+	m.mutex.Unlock()
+}

streams_map_outgoing_generic_test.go

@@ -0,0 +1,78 @@
+package quic
+
+import (
+	"errors"
+
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/qerr"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Streams Map (outgoing)", func() {
+	const firstNewStream protocol.StreamID = 10
+	var (
+		m       *outgoingItemsMap
+		newItem func(id protocol.StreamID) item
+	)
+
+	BeforeEach(func() {
+		newItem = func(id protocol.StreamID) item {
+			return id
+		}
+		m = newOutgoingItemsMap(firstNewStream, newItem)
+	})
+
+	It("opens streams", func() {
+		str, err := m.OpenStream()
+		Expect(err).ToNot(HaveOccurred())
+		Expect(str).To(Equal(firstNewStream))
+		str, err = m.OpenStream()
+		Expect(err).ToNot(HaveOccurred())
+		Expect(str).To(Equal(firstNewStream + 4))
+	})
+
+	It("doesn't open streams after it has been closed", func() {
+		testErr := errors.New("close")
+		m.CloseWithError(testErr)
+		_, err := m.OpenStream()
+		Expect(err).To(MatchError(testErr))
+	})
+
+	It("gets streams", func() {
+		_, err := m.OpenStream()
+		Expect(err).ToNot(HaveOccurred())
+		str, err := m.GetStream(firstNewStream)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(str).To(Equal(firstNewStream))
+	})
+
+	It("errors when trying to get a stream that has not yet been opened", func() {
+		_, err := m.GetStream(10)
+		Expect(err).To(MatchError(qerr.Error(qerr.InvalidStreamID, "peer attempted to open stream 10")))
+	})
+
+	It("deletes streams", func() {
+		_, err := m.OpenStream() // opens stream 10
+		Expect(err).ToNot(HaveOccurred())
+		err = m.DeleteStream(10)
+		Expect(err).ToNot(HaveOccurred())
+		str, err := m.GetStream(10)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(str).To(BeNil())
+	})
+
+	It("errors when deleting a non-existing stream", func() {
+		err := m.DeleteStream(1337)
+		Expect(err).To(MatchError("Tried to delete unknown stream 1337"))
+	})
+
+	It("errors when deleting a stream twice", func() {
+		_, err := m.OpenStream() // opens stream 10
+		Expect(err).ToNot(HaveOccurred())
+		err = m.DeleteStream(10)
+		Expect(err).ToNot(HaveOccurred())
+		err = m.DeleteStream(10)
+		Expect(err).To(MatchError("Tried to delete unknown stream 10"))
+	})
+})

streams_map_outgoing_uni.go

@@ -0,0 +1,72 @@
+// 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 (
+	"fmt"
+	"sync"
+
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/qerr"
+)
+
+type outgoingUniStreamsMap struct {
+	mutex sync.RWMutex
+
+	streams map[protocol.StreamID]sendStreamI
+
+	nextStream protocol.StreamID
+	newStream  func(protocol.StreamID) sendStreamI
+
+	closeErr error
+}
+
+func newOutgoingUniStreamsMap(nextStream protocol.StreamID, newStream func(protocol.StreamID) sendStreamI) *outgoingUniStreamsMap {
+	return &outgoingUniStreamsMap{
+		streams:    make(map[protocol.StreamID]sendStreamI),
+		nextStream: nextStream,
+		newStream:  newStream,
+	}
+}
+
+func (m *outgoingUniStreamsMap) OpenStream() (sendStreamI, error) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+
+	if m.closeErr != nil {
+		return nil, m.closeErr
+	}
+	s := m.newStream(m.nextStream)
+	m.streams[m.nextStream] = s
+	m.nextStream += 4
+	return s, nil
+}
+
+func (m *outgoingUniStreamsMap) GetStream(id protocol.StreamID) (sendStreamI, error) {
+	if id >= m.nextStream {
+		return nil, qerr.Error(qerr.InvalidStreamID, fmt.Sprintf("peer attempted to open stream %d", id))
+	}
+	m.mutex.RLock()
+	s := m.streams[id]
+	m.mutex.RUnlock()
+	return s, nil
+}
+
+func (m *outgoingUniStreamsMap) 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 *outgoingUniStreamsMap) CloseWithError(err error) {
+	m.mutex.Lock()
+	m.closeErr = err
+	m.mutex.Unlock()
+}

vendor/github.com/cheekybits/genny/LICENSE

@@ -0,0 +1,22 @@
+The MIT License (MIT)
+
+Copyright (c) 2014 cheekybits
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+

vendor/github.com/cheekybits/genny/generic/doc.go

@@ -0,0 +1,2 @@
+// Package generic contains the generic marker types.
+package generic

vendor/github.com/cheekybits/genny/generic/generic.go

@@ -0,0 +1,13 @@
+package generic
+
+// Type is the placeholder type that indicates a generic value.
+// When genny is executed, variables of this type will be replaced with
+// references to the specific types.
+//      var GenericType generic.Type
+type Type interface{}
+
+// Number is the placehoder type that indiccates a generic numerical value.
+// When genny is executed, variables of this type will be replaced with
+// references to the specific types.
+//      var GenericType generic.Number
+type Number float64

vendor/vendor.json

@@ -14,6 +14,12 @@
 			"revision": "a544bfbca6a083ce9ddeb2c5f570cb240837355a",
 			"revisionTime": "2017-12-09T20:11:46Z"
 		},
+		{
+			"checksumSHA1": "PYXuf7wvcj492uWhjvmXlyyQUYc=",
+			"path": "github.com/cheekybits/genny/generic",
+			"revision": "9127e812e1e9e501ce899a18121d316ecb52e4ba",
+			"revisionTime": "2017-03-28T20:00:08Z"
+		},
 		{
 			"checksumSHA1": "IQkUIOnvlf0tYloFx9mLaXSvXWQ=",
 			"path": "golang.org/x/crypto/curve25519",