make the flow control package internal

internal/flowcontrol/flow_control_manager.go

@@ -0,0 +1,240 @@
+package flowcontrol
+
+import (
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/lucas-clemente/quic-go/congestion"
+	"github.com/lucas-clemente/quic-go/handshake"
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/internal/utils"
+	"github.com/lucas-clemente/quic-go/qerr"
+)
+
+type flowControlManager struct {
+	connectionParameters handshake.ConnectionParametersManager
+	rttStats             *congestion.RTTStats
+
+	streamFlowController map[protocol.StreamID]*flowController
+	connFlowController   *flowController
+	mutex                sync.RWMutex
+}
+
+var _ FlowControlManager = &flowControlManager{}
+
+var errMapAccess = errors.New("Error accessing the flowController map.")
+
+// NewFlowControlManager creates a new flow control manager
+func NewFlowControlManager(connectionParameters handshake.ConnectionParametersManager, rttStats *congestion.RTTStats) FlowControlManager {
+	return &flowControlManager{
+		connectionParameters: connectionParameters,
+		rttStats:             rttStats,
+		streamFlowController: make(map[protocol.StreamID]*flowController),
+		connFlowController:   newFlowController(0, false, connectionParameters, rttStats),
+	}
+}
+
+// NewStream creates new flow controllers for a stream
+// it does nothing if the stream already exists
+func (f *flowControlManager) NewStream(streamID protocol.StreamID, contributesToConnection bool) {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	if _, ok := f.streamFlowController[streamID]; ok {
+		return
+	}
+
+	f.streamFlowController[streamID] = newFlowController(streamID, contributesToConnection, f.connectionParameters, f.rttStats)
+}
+
+// RemoveStream removes a closed stream from flow control
+func (f *flowControlManager) RemoveStream(streamID protocol.StreamID) {
+	f.mutex.Lock()
+	delete(f.streamFlowController, streamID)
+	f.mutex.Unlock()
+}
+
+// ResetStream should be called when receiving a RstStreamFrame
+// it updates the byte offset to the value in the RstStreamFrame
+// streamID must not be 0 here
+func (f *flowControlManager) ResetStream(streamID protocol.StreamID, byteOffset protocol.ByteCount) error {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	streamFlowController, err := f.getFlowController(streamID)
+	if err != nil {
+		return err
+	}
+	increment, err := streamFlowController.UpdateHighestReceived(byteOffset)
+	if err != nil {
+		return qerr.StreamDataAfterTermination
+	}
+
+	if streamFlowController.CheckFlowControlViolation() {
+		return qerr.Error(qerr.FlowControlReceivedTooMuchData, fmt.Sprintf("Received %d bytes on stream %d, allowed %d bytes", byteOffset, streamID, streamFlowController.receiveWindow))
+	}
+
+	if streamFlowController.ContributesToConnection() {
+		f.connFlowController.IncrementHighestReceived(increment)
+		if f.connFlowController.CheckFlowControlViolation() {
+			return qerr.Error(qerr.FlowControlReceivedTooMuchData, fmt.Sprintf("Received %d bytes for the connection, allowed %d bytes", f.connFlowController.highestReceived, f.connFlowController.receiveWindow))
+		}
+	}
+
+	return nil
+}
+
+// UpdateHighestReceived updates the highest received byte offset for a stream
+// it adds the number of additional bytes to connection level flow control
+// streamID must not be 0 here
+func (f *flowControlManager) UpdateHighestReceived(streamID protocol.StreamID, byteOffset protocol.ByteCount) error {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	streamFlowController, err := f.getFlowController(streamID)
+	if err != nil {
+		return err
+	}
+	// UpdateHighestReceived returns an ErrReceivedSmallerByteOffset when StreamFrames got reordered
+	// this error can be ignored here
+	increment, _ := streamFlowController.UpdateHighestReceived(byteOffset)
+
+	if streamFlowController.CheckFlowControlViolation() {
+		return qerr.Error(qerr.FlowControlReceivedTooMuchData, fmt.Sprintf("Received %d bytes on stream %d, allowed %d bytes", byteOffset, streamID, streamFlowController.receiveWindow))
+	}
+
+	if streamFlowController.ContributesToConnection() {
+		f.connFlowController.IncrementHighestReceived(increment)
+		if f.connFlowController.CheckFlowControlViolation() {
+			return qerr.Error(qerr.FlowControlReceivedTooMuchData, fmt.Sprintf("Received %d bytes for the connection, allowed %d bytes", f.connFlowController.highestReceived, f.connFlowController.receiveWindow))
+		}
+	}
+
+	return nil
+}
+
+// streamID must not be 0 here
+func (f *flowControlManager) AddBytesRead(streamID protocol.StreamID, n protocol.ByteCount) error {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	fc, err := f.getFlowController(streamID)
+	if err != nil {
+		return err
+	}
+
+	fc.AddBytesRead(n)
+	if fc.ContributesToConnection() {
+		f.connFlowController.AddBytesRead(n)
+	}
+
+	return nil
+}
+
+func (f *flowControlManager) GetWindowUpdates() (res []WindowUpdate) {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	// get WindowUpdates for streams
+	for id, fc := range f.streamFlowController {
+		if necessary, newIncrement, offset := fc.MaybeUpdateWindow(); necessary {
+			res = append(res, WindowUpdate{StreamID: id, Offset: offset})
+			if fc.ContributesToConnection() && newIncrement != 0 {
+				f.connFlowController.EnsureMinimumWindowIncrement(protocol.ByteCount(float64(newIncrement) * protocol.ConnectionFlowControlMultiplier))
+			}
+		}
+	}
+	// get a WindowUpdate for the connection
+	if necessary, _, offset := f.connFlowController.MaybeUpdateWindow(); necessary {
+		res = append(res, WindowUpdate{StreamID: 0, Offset: offset})
+	}
+
+	return
+}
+
+func (f *flowControlManager) GetReceiveWindow(streamID protocol.StreamID) (protocol.ByteCount, error) {
+	f.mutex.RLock()
+	defer f.mutex.RUnlock()
+
+	// StreamID can be 0 when retransmitting
+	if streamID == 0 {
+		return f.connFlowController.receiveWindow, nil
+	}
+
+	flowController, err := f.getFlowController(streamID)
+	if err != nil {
+		return 0, err
+	}
+	return flowController.receiveWindow, nil
+}
+
+// streamID must not be 0 here
+func (f *flowControlManager) AddBytesSent(streamID protocol.StreamID, n protocol.ByteCount) error {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	fc, err := f.getFlowController(streamID)
+	if err != nil {
+		return err
+	}
+
+	fc.AddBytesSent(n)
+	if fc.ContributesToConnection() {
+		f.connFlowController.AddBytesSent(n)
+	}
+
+	return nil
+}
+
+// must not be called with StreamID 0
+func (f *flowControlManager) SendWindowSize(streamID protocol.StreamID) (protocol.ByteCount, error) {
+	f.mutex.RLock()
+	defer f.mutex.RUnlock()
+
+	fc, err := f.getFlowController(streamID)
+	if err != nil {
+		return 0, err
+	}
+	res := fc.SendWindowSize()
+
+	if fc.ContributesToConnection() {
+		res = utils.MinByteCount(res, f.connFlowController.SendWindowSize())
+	}
+
+	return res, nil
+}
+
+func (f *flowControlManager) RemainingConnectionWindowSize() protocol.ByteCount {
+	f.mutex.RLock()
+	defer f.mutex.RUnlock()
+
+	return f.connFlowController.SendWindowSize()
+}
+
+// streamID may be 0 here
+func (f *flowControlManager) UpdateWindow(streamID protocol.StreamID, offset protocol.ByteCount) (bool, error) {
+	f.mutex.Lock()
+	defer f.mutex.Unlock()
+
+	var fc *flowController
+	if streamID == 0 {
+		fc = f.connFlowController
+	} else {
+		var err error
+		fc, err = f.getFlowController(streamID)
+		if err != nil {
+			return false, err
+		}
+	}
+
+	return fc.UpdateSendWindow(offset), nil
+}
+
+func (f *flowControlManager) getFlowController(streamID protocol.StreamID) (*flowController, error) {
+	streamFlowController, ok := f.streamFlowController[streamID]
+	if !ok {
+		return nil, errMapAccess
+	}
+	return streamFlowController, nil
+}

internal/flowcontrol/flow_control_manager_test.go

@@ -0,0 +1,335 @@
+package flowcontrol
+
+import (
+	"time"
+
+	"github.com/lucas-clemente/quic-go/congestion"
+	"github.com/lucas-clemente/quic-go/internal/mocks"
+	"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("Flow Control Manager", func() {
+	var fcm *flowControlManager
+
+	BeforeEach(func() {
+		mockCpm := mocks.NewMockConnectionParametersManager(mockCtrl)
+		mockCpm.EXPECT().GetReceiveStreamFlowControlWindow().AnyTimes().Return(protocol.ByteCount(100))
+		mockCpm.EXPECT().GetReceiveConnectionFlowControlWindow().AnyTimes().Return(protocol.ByteCount(200))
+		mockCpm.EXPECT().GetMaxReceiveStreamFlowControlWindow().AnyTimes().Return(protocol.MaxByteCount)
+		mockCpm.EXPECT().GetMaxReceiveConnectionFlowControlWindow().AnyTimes().Return(protocol.MaxByteCount)
+		fcm = NewFlowControlManager(mockCpm, &congestion.RTTStats{}).(*flowControlManager)
+	})
+
+	It("creates a connection level flow controller", func() {
+		Expect(fcm.streamFlowController).ToNot(HaveKey(protocol.StreamID(0)))
+		Expect(fcm.connFlowController.ContributesToConnection()).To(BeFalse())
+	})
+
+	Context("creating new streams", func() {
+		It("creates a new stream", func() {
+			fcm.NewStream(5, false)
+			Expect(fcm.streamFlowController).To(HaveKey(protocol.StreamID(5)))
+			fc := fcm.streamFlowController[5]
+			Expect(fc.streamID).To(Equal(protocol.StreamID(5)))
+			Expect(fc.ContributesToConnection()).To(BeFalse())
+		})
+
+		It("doesn't create a new flow controller if called for an existing stream", func() {
+			fcm.NewStream(5, true)
+			Expect(fcm.streamFlowController).To(HaveKey(protocol.StreamID(5)))
+			fcm.streamFlowController[5].bytesRead = 0x1337
+			fcm.NewStream(5, false)
+			fc := fcm.streamFlowController[5]
+			Expect(fc.bytesRead).To(BeEquivalentTo(0x1337))
+			Expect(fc.ContributesToConnection()).To(BeTrue())
+		})
+	})
+
+	It("removes streams", func() {
+		fcm.NewStream(5, true)
+		Expect(fcm.streamFlowController).To(HaveKey(protocol.StreamID(5)))
+		fcm.RemoveStream(5)
+		Expect(fcm.streamFlowController).ToNot(HaveKey(protocol.StreamID(5)))
+	})
+
+	Context("receiving data", func() {
+		BeforeEach(func() {
+			fcm.NewStream(1, false)
+			fcm.NewStream(4, true)
+			fcm.NewStream(6, true)
+		})
+
+		It("updates the connection level flow controller if the stream contributes", func() {
+			err := fcm.UpdateHighestReceived(4, 100)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(fcm.connFlowController.highestReceived).To(Equal(protocol.ByteCount(100)))
+			Expect(fcm.streamFlowController[4].highestReceived).To(Equal(protocol.ByteCount(100)))
+		})
+
+		It("adds the offsets of multiple streams for the connection flow control window", func() {
+			err := fcm.UpdateHighestReceived(4, 100)
+			Expect(err).ToNot(HaveOccurred())
+			err = fcm.UpdateHighestReceived(6, 50)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(fcm.connFlowController.highestReceived).To(Equal(protocol.ByteCount(100 + 50)))
+		})
+
+		It("does not update the connection level flow controller if the stream does not contribute", func() {
+			err := fcm.UpdateHighestReceived(1, 100)
+			// fcm.streamFlowController[4].receiveWindow = 0x1000
+			Expect(err).ToNot(HaveOccurred())
+			Expect(fcm.connFlowController.highestReceived).To(BeZero())
+			Expect(fcm.streamFlowController[1].highestReceived).To(Equal(protocol.ByteCount(100)))
+		})
+
+		It("returns an error when called with an unknown stream", func() {
+			err := fcm.UpdateHighestReceived(1337, 0x1337)
+			Expect(err).To(MatchError(errMapAccess))
+		})
+
+		It("gets the offset of the receive window", func() {
+			offset, err := fcm.GetReceiveWindow(4)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(offset).To(Equal(protocol.ByteCount(100)))
+		})
+
+		It("errors when asked for the receive window of a stream that doesn't exist", func() {
+			_, err := fcm.GetReceiveWindow(17)
+			Expect(err).To(MatchError(errMapAccess))
+		})
+
+		It("gets the offset of the connection-level receive window", func() {
+			offset, err := fcm.GetReceiveWindow(0)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(offset).To(Equal(protocol.ByteCount(200)))
+		})
+
+		Context("flow control violations", func() {
+			It("errors when encountering a stream level flow control violation", func() {
+				err := fcm.UpdateHighestReceived(4, 101)
+				Expect(err).To(MatchError(qerr.Error(qerr.FlowControlReceivedTooMuchData, "Received 101 bytes on stream 4, allowed 100 bytes")))
+			})
+
+			It("errors when encountering a connection-level flow control violation", func() {
+				fcm.streamFlowController[4].receiveWindow = 300
+				fcm.streamFlowController[6].receiveWindow = 300
+				err := fcm.UpdateHighestReceived(6, 100)
+				Expect(err).ToNot(HaveOccurred())
+				err = fcm.UpdateHighestReceived(4, 103)
+				Expect(err).To(MatchError(qerr.Error(qerr.FlowControlReceivedTooMuchData, "Received 203 bytes for the connection, allowed 200 bytes")))
+			})
+		})
+
+		Context("window updates", func() {
+			// update the congestion such that it returns a given value for the smoothed RTT
+			setRtt := func(t time.Duration) {
+				for _, controller := range fcm.streamFlowController {
+					controller.rttStats.UpdateRTT(t, 0, time.Now())
+					Expect(controller.rttStats.SmoothedRTT()).To(Equal(t)) // make sure it worked
+				}
+			}
+
+			It("gets stream level window updates", func() {
+				err := fcm.UpdateHighestReceived(4, 100)
+				Expect(err).ToNot(HaveOccurred())
+				err = fcm.AddBytesRead(4, 90)
+				Expect(err).ToNot(HaveOccurred())
+				updates := fcm.GetWindowUpdates()
+				Expect(updates).To(HaveLen(1))
+				Expect(updates[0]).To(Equal(WindowUpdate{StreamID: 4, Offset: 190}))
+			})
+
+			It("gets connection level window updates", func() {
+				err := fcm.UpdateHighestReceived(4, 100)
+				Expect(err).ToNot(HaveOccurred())
+				err = fcm.UpdateHighestReceived(6, 100)
+				Expect(err).ToNot(HaveOccurred())
+				err = fcm.AddBytesRead(4, 90)
+				Expect(err).ToNot(HaveOccurred())
+				err = fcm.AddBytesRead(6, 90)
+				Expect(err).ToNot(HaveOccurred())
+				updates := fcm.GetWindowUpdates()
+				Expect(updates).To(HaveLen(3))
+				Expect(updates).ToNot(ContainElement(WindowUpdate{StreamID: 0, Offset: 200}))
+			})
+
+			It("errors when AddBytesRead is called for a stream doesn't exist", func() {
+				err := fcm.AddBytesRead(17, 1000)
+				Expect(err).To(MatchError(errMapAccess))
+			})
+
+			It("increases the connection-level window, when a stream window was increased by autotuning", func() {
+				setRtt(10 * time.Millisecond)
+				fcm.streamFlowController[4].lastWindowUpdateTime = time.Now().Add(-1 * time.Millisecond)
+				err := fcm.UpdateHighestReceived(4, 100)
+				Expect(err).ToNot(HaveOccurred())
+				err = fcm.AddBytesRead(4, 90)
+				Expect(err).ToNot(HaveOccurred())
+				updates := fcm.GetWindowUpdates()
+				Expect(updates).To(HaveLen(2))
+				connLevelIncrement := protocol.ByteCount(protocol.ConnectionFlowControlMultiplier * 200) // 300
+				Expect(updates).To(ContainElement(WindowUpdate{StreamID: 4, Offset: 290}))
+				Expect(updates).To(ContainElement(WindowUpdate{StreamID: 0, Offset: 90 + connLevelIncrement}))
+			})
+
+			It("doesn't increase the connection-level window, when a non-contributing stream window was increased by autotuning", func() {
+				setRtt(10 * time.Millisecond)
+				fcm.streamFlowController[1].lastWindowUpdateTime = time.Now().Add(-1 * time.Millisecond)
+				err := fcm.UpdateHighestReceived(1, 100)
+				Expect(err).ToNot(HaveOccurred())
+				err = fcm.AddBytesRead(1, 90)
+				Expect(err).ToNot(HaveOccurred())
+				updates := fcm.GetWindowUpdates()
+				Expect(updates).To(HaveLen(1))
+				Expect(updates).To(ContainElement(WindowUpdate{StreamID: 1, Offset: 290}))
+				// the only window update is for stream 1, thus there's no connection-level window update
+			})
+		})
+	})
+
+	Context("resetting a stream", func() {
+		BeforeEach(func() {
+			fcm.NewStream(1, false)
+			fcm.NewStream(4, true)
+			fcm.NewStream(6, true)
+			fcm.streamFlowController[1].bytesSent = 41
+			fcm.streamFlowController[4].bytesSent = 42
+		})
+
+		It("updates the connection level flow controller if the stream contributes", func() {
+			err := fcm.ResetStream(4, 100)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(fcm.connFlowController.highestReceived).To(Equal(protocol.ByteCount(100)))
+			Expect(fcm.streamFlowController[4].highestReceived).To(Equal(protocol.ByteCount(100)))
+		})
+
+		It("does not update the connection level flow controller if the stream does not contribute", func() {
+			err := fcm.ResetStream(1, 100)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(fcm.connFlowController.highestReceived).To(BeZero())
+			Expect(fcm.streamFlowController[1].highestReceived).To(Equal(protocol.ByteCount(100)))
+		})
+
+		It("errors if the byteOffset is smaller than a byteOffset that set earlier", func() {
+			err := fcm.UpdateHighestReceived(4, 100)
+			Expect(err).ToNot(HaveOccurred())
+			err = fcm.ResetStream(4, 50)
+			Expect(err).To(MatchError(qerr.StreamDataAfterTermination))
+		})
+
+		It("returns an error when called with an unknown stream", func() {
+			err := fcm.ResetStream(1337, 0x1337)
+			Expect(err).To(MatchError(errMapAccess))
+		})
+
+		Context("flow control violations", func() {
+			It("errors when encountering a stream level flow control violation", func() {
+				err := fcm.ResetStream(4, 101)
+				Expect(err).To(MatchError(qerr.Error(qerr.FlowControlReceivedTooMuchData, "Received 101 bytes on stream 4, allowed 100 bytes")))
+			})
+
+			It("errors when encountering a connection-level flow control violation", func() {
+				fcm.streamFlowController[4].receiveWindow = 300
+				fcm.streamFlowController[6].receiveWindow = 300
+				err := fcm.ResetStream(4, 100)
+				Expect(err).ToNot(HaveOccurred())
+				err = fcm.ResetStream(6, 101)
+				Expect(err).To(MatchError(qerr.Error(qerr.FlowControlReceivedTooMuchData, "Received 201 bytes for the connection, allowed 200 bytes")))
+			})
+		})
+	})
+
+	Context("sending data", func() {
+		It("adds bytes sent for all stream contributing to connection level flow control", func() {
+			fcm.NewStream(1, false)
+			fcm.NewStream(3, true)
+			fcm.NewStream(5, true)
+			err := fcm.AddBytesSent(1, 100)
+			Expect(err).ToNot(HaveOccurred())
+			err = fcm.AddBytesSent(3, 200)
+			Expect(err).ToNot(HaveOccurred())
+			err = fcm.AddBytesSent(5, 500)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(fcm.connFlowController.bytesSent).To(Equal(protocol.ByteCount(200 + 500)))
+		})
+
+		It("errors when called for a stream doesn't exist", func() {
+			err := fcm.AddBytesSent(17, 1000)
+			Expect(err).To(MatchError(errMapAccess))
+		})
+
+		Context("window updates", func() {
+			It("updates the window for a normal stream", func() {
+				fcm.NewStream(5, true)
+				updated, err := fcm.UpdateWindow(5, 1000)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(updated).To(BeTrue())
+			})
+
+			It("updates the connection level window", func() {
+				updated, err := fcm.UpdateWindow(0, 1000)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(updated).To(BeTrue())
+			})
+
+			It("errors when called for a stream that doesn't exist", func() {
+				_, err := fcm.UpdateWindow(17, 1000)
+				Expect(err).To(MatchError(errMapAccess))
+			})
+		})
+
+		Context("window sizes", func() {
+			It("gets the window size of a stream", func() {
+				fcm.NewStream(5, false)
+				updated, err := fcm.UpdateWindow(5, 1000)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(updated).To(BeTrue())
+				fcm.AddBytesSent(5, 500)
+				size, err := fcm.SendWindowSize(5)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(size).To(Equal(protocol.ByteCount(1000 - 500)))
+			})
+
+			It("gets the connection window size", func() {
+				fcm.NewStream(5, true)
+				updated, err := fcm.UpdateWindow(0, 1000)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(updated).To(BeTrue())
+				fcm.AddBytesSent(5, 500)
+				size := fcm.RemainingConnectionWindowSize()
+				Expect(size).To(Equal(protocol.ByteCount(1000 - 500)))
+			})
+
+			It("erros when asked for the send window size of a stream that doesn't exist", func() {
+				_, err := fcm.SendWindowSize(17)
+				Expect(err).To(MatchError(errMapAccess))
+			})
+
+			It("limits the stream window size by the connection window size", func() {
+				fcm.NewStream(5, true)
+				updated, err := fcm.UpdateWindow(0, 500)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(updated).To(BeTrue())
+				updated, err = fcm.UpdateWindow(5, 1000)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(updated).To(BeTrue())
+				size, err := fcm.SendWindowSize(5)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(size).To(Equal(protocol.ByteCount(500)))
+			})
+
+			It("does not reduce the size of the connection level window, if the stream does not contribute", func() {
+				fcm.NewStream(3, false)
+				updated, err := fcm.UpdateWindow(0, 1000)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(updated).To(BeTrue())
+				fcm.AddBytesSent(3, 456) // WindowSize should return the same value no matter how much was sent
+				size := fcm.RemainingConnectionWindowSize()
+				Expect(size).To(Equal(protocol.ByteCount(1000)))
+			})
+		})
+	})
+})

internal/flowcontrol/flow_controller.go

@@ -0,0 +1,198 @@
+package flowcontrol
+
+import (
+	"errors"
+	"time"
+
+	"github.com/lucas-clemente/quic-go/congestion"
+	"github.com/lucas-clemente/quic-go/handshake"
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	"github.com/lucas-clemente/quic-go/internal/utils"
+)
+
+type flowController struct {
+	streamID                protocol.StreamID
+	contributesToConnection bool // does the stream contribute to connection level flow control
+
+	connectionParameters handshake.ConnectionParametersManager
+	rttStats             *congestion.RTTStats
+
+	bytesSent  protocol.ByteCount
+	sendWindow protocol.ByteCount
+
+	lastWindowUpdateTime time.Time
+
+	bytesRead                 protocol.ByteCount
+	highestReceived           protocol.ByteCount
+	receiveWindow             protocol.ByteCount
+	receiveWindowIncrement    protocol.ByteCount
+	maxReceiveWindowIncrement protocol.ByteCount
+}
+
+// ErrReceivedSmallerByteOffset occurs if the ByteOffset received is smaller than a ByteOffset that was set previously
+var ErrReceivedSmallerByteOffset = errors.New("Received a smaller byte offset")
+
+// newFlowController gets a new flow controller
+func newFlowController(streamID protocol.StreamID, contributesToConnection bool, connectionParameters handshake.ConnectionParametersManager, rttStats *congestion.RTTStats) *flowController {
+	fc := flowController{
+		streamID:                streamID,
+		contributesToConnection: contributesToConnection,
+		connectionParameters:    connectionParameters,
+		rttStats:                rttStats,
+	}
+
+	if streamID == 0 {
+		fc.receiveWindow = connectionParameters.GetReceiveConnectionFlowControlWindow()
+		fc.receiveWindowIncrement = fc.receiveWindow
+		fc.maxReceiveWindowIncrement = connectionParameters.GetMaxReceiveConnectionFlowControlWindow()
+	} else {
+		fc.receiveWindow = connectionParameters.GetReceiveStreamFlowControlWindow()
+		fc.receiveWindowIncrement = fc.receiveWindow
+		fc.maxReceiveWindowIncrement = connectionParameters.GetMaxReceiveStreamFlowControlWindow()
+	}
+
+	return &fc
+}
+
+func (c *flowController) ContributesToConnection() bool {
+	return c.contributesToConnection
+}
+
+func (c *flowController) getSendWindow() protocol.ByteCount {
+	if c.sendWindow == 0 {
+		if c.streamID == 0 {
+			return c.connectionParameters.GetSendConnectionFlowControlWindow()
+		}
+		return c.connectionParameters.GetSendStreamFlowControlWindow()
+	}
+	return c.sendWindow
+}
+
+func (c *flowController) AddBytesSent(n protocol.ByteCount) {
+	c.bytesSent += n
+}
+
+// UpdateSendWindow should be called after receiving a WindowUpdateFrame
+// it returns true if the window was actually updated
+func (c *flowController) UpdateSendWindow(newOffset protocol.ByteCount) bool {
+	if newOffset > c.sendWindow {
+		c.sendWindow = newOffset
+		return true
+	}
+	return false
+}
+
+func (c *flowController) SendWindowSize() protocol.ByteCount {
+	sendWindow := c.getSendWindow()
+
+	if c.bytesSent > sendWindow { // should never happen, but make sure we don't do an underflow here
+		return 0
+	}
+	return sendWindow - c.bytesSent
+}
+
+func (c *flowController) SendWindowOffset() protocol.ByteCount {
+	return c.getSendWindow()
+}
+
+// UpdateHighestReceived updates the highestReceived value, if the byteOffset is higher
+// Should **only** be used for the stream-level FlowController
+// it returns an ErrReceivedSmallerByteOffset if the received byteOffset is smaller than any byteOffset received before
+// This error occurs every time StreamFrames get reordered and has to be ignored in that case
+// It should only be treated as an error when resetting a stream
+func (c *flowController) UpdateHighestReceived(byteOffset protocol.ByteCount) (protocol.ByteCount, error) {
+	if byteOffset == c.highestReceived {
+		return 0, nil
+	}
+	if byteOffset > c.highestReceived {
+		increment := byteOffset - c.highestReceived
+		c.highestReceived = byteOffset
+		return increment, nil
+	}
+	return 0, ErrReceivedSmallerByteOffset
+}
+
+// IncrementHighestReceived adds an increment to the highestReceived value
+// Should **only** be used for the connection-level FlowController
+func (c *flowController) IncrementHighestReceived(increment protocol.ByteCount) {
+	c.highestReceived += increment
+}
+
+func (c *flowController) AddBytesRead(n protocol.ByteCount) {
+	// pretend we sent a WindowUpdate when reading the first byte
+	// this way auto-tuning of the window increment already works for the first WindowUpdate
+	if c.bytesRead == 0 {
+		c.lastWindowUpdateTime = time.Now()
+	}
+	c.bytesRead += n
+}
+
+// MaybeUpdateWindow updates the receive window, if necessary
+// if the receive window increment is changed, the new value is returned, otherwise a 0
+// the last return value is the new offset of the receive window
+func (c *flowController) MaybeUpdateWindow() (bool, protocol.ByteCount /* new increment */, protocol.ByteCount /* new offset */) {
+	diff := c.receiveWindow - c.bytesRead
+
+	// Chromium implements the same threshold
+	if diff < (c.receiveWindowIncrement / 2) {
+		var newWindowIncrement protocol.ByteCount
+		oldWindowIncrement := c.receiveWindowIncrement
+
+		c.maybeAdjustWindowIncrement()
+		if c.receiveWindowIncrement != oldWindowIncrement {
+			newWindowIncrement = c.receiveWindowIncrement
+		}
+
+		c.lastWindowUpdateTime = time.Now()
+		c.receiveWindow = c.bytesRead + c.receiveWindowIncrement
+		return true, newWindowIncrement, c.receiveWindow
+	}
+
+	return false, 0, 0
+}
+
+// maybeAdjustWindowIncrement increases the receiveWindowIncrement if we're sending WindowUpdates too often
+func (c *flowController) maybeAdjustWindowIncrement() {
+	if c.lastWindowUpdateTime.IsZero() {
+		return
+	}
+
+	rtt := c.rttStats.SmoothedRTT()
+	if rtt == 0 {
+		return
+	}
+
+	timeSinceLastWindowUpdate := time.Since(c.lastWindowUpdateTime)
+
+	// interval between the window updates is sufficiently large, no need to increase the increment
+	if timeSinceLastWindowUpdate >= 2*rtt {
+		return
+	}
+
+	oldWindowSize := c.receiveWindowIncrement
+	c.receiveWindowIncrement = utils.MinByteCount(2*c.receiveWindowIncrement, c.maxReceiveWindowIncrement)
+
+	// debug log, if the window size was actually increased
+	if oldWindowSize < c.receiveWindowIncrement {
+		newWindowSize := c.receiveWindowIncrement / (1 << 10)
+		if c.streamID == 0 {
+			utils.Debugf("Increasing receive flow control window for the connection to %d kB", newWindowSize)
+		} else {
+			utils.Debugf("Increasing receive flow control window increment for stream %d to %d kB", c.streamID, newWindowSize)
+		}
+	}
+}
+
+// EnsureMinimumWindowIncrement sets a minimum window increment
+// it is intended be used for the connection-level flow controller
+// it should make sure that the connection-level window is increased when a stream-level window grows
+func (c *flowController) EnsureMinimumWindowIncrement(inc protocol.ByteCount) {
+	if inc > c.receiveWindowIncrement {
+		c.receiveWindowIncrement = utils.MinByteCount(inc, c.maxReceiveWindowIncrement)
+		c.lastWindowUpdateTime = time.Time{} // disables autotuning for the next window update
+	}
+}
+
+func (c *flowController) CheckFlowControlViolation() bool {
+	return c.highestReceived > c.receiveWindow
+}

internal/flowcontrol/flow_controller_test.go

@@ -0,0 +1,340 @@
+package flowcontrol
+
+import (
+	"time"
+
+	"github.com/lucas-clemente/quic-go/congestion"
+	"github.com/lucas-clemente/quic-go/internal/mocks"
+	"github.com/lucas-clemente/quic-go/internal/protocol"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Flow controller", func() {
+	var controller *flowController
+
+	BeforeEach(func() {
+		controller = &flowController{}
+		controller.rttStats = &congestion.RTTStats{}
+	})
+
+	Context("Constructor", func() {
+		var rttStats *congestion.RTTStats
+		var mockCpm *mocks.MockConnectionParametersManager
+
+		BeforeEach(func() {
+			mockCpm = mocks.NewMockConnectionParametersManager(mockCtrl)
+			mockCpm.EXPECT().GetSendStreamFlowControlWindow().AnyTimes().Return(protocol.ByteCount(1000))
+			mockCpm.EXPECT().GetReceiveStreamFlowControlWindow().AnyTimes().Return(protocol.ByteCount(2000))
+			mockCpm.EXPECT().GetSendConnectionFlowControlWindow().AnyTimes().Return(protocol.ByteCount(3000))
+			mockCpm.EXPECT().GetReceiveConnectionFlowControlWindow().AnyTimes().Return(protocol.ByteCount(4000))
+			mockCpm.EXPECT().GetMaxReceiveStreamFlowControlWindow().AnyTimes().Return(protocol.ByteCount(8000))
+			mockCpm.EXPECT().GetMaxReceiveConnectionFlowControlWindow().AnyTimes().Return(protocol.ByteCount(9000))
+			rttStats = &congestion.RTTStats{}
+		})
+
+		It("reads the stream send and receive windows when acting as stream-level flow controller", func() {
+			fc := newFlowController(5, true, mockCpm, rttStats)
+			Expect(fc.streamID).To(Equal(protocol.StreamID(5)))
+			Expect(fc.receiveWindow).To(Equal(protocol.ByteCount(2000)))
+			Expect(fc.maxReceiveWindowIncrement).To(Equal(mockCpm.GetMaxReceiveStreamFlowControlWindow()))
+		})
+
+		It("reads the stream send and receive windows when acting as connection-level flow controller", func() {
+			fc := newFlowController(0, false, mockCpm, rttStats)
+			Expect(fc.streamID).To(Equal(protocol.StreamID(0)))
+			Expect(fc.receiveWindow).To(Equal(protocol.ByteCount(4000)))
+			Expect(fc.maxReceiveWindowIncrement).To(Equal(mockCpm.GetMaxReceiveConnectionFlowControlWindow()))
+		})
+
+		It("does not set the stream flow control windows for sending", func() {
+			fc := newFlowController(5, true, mockCpm, rttStats)
+			Expect(fc.sendWindow).To(BeZero())
+		})
+
+		It("does not set the connection flow control windows for sending", func() {
+			fc := newFlowController(0, false, mockCpm, rttStats)
+			Expect(fc.sendWindow).To(BeZero())
+		})
+
+		It("says if it contributes to connection-level flow control", func() {
+			fc := newFlowController(1, false, mockCpm, rttStats)
+			Expect(fc.ContributesToConnection()).To(BeFalse())
+			fc = newFlowController(5, true, mockCpm, rttStats)
+			Expect(fc.ContributesToConnection()).To(BeTrue())
+		})
+	})
+
+	Context("send flow control", func() {
+		var mockCpm *mocks.MockConnectionParametersManager
+
+		BeforeEach(func() {
+			mockCpm = mocks.NewMockConnectionParametersManager(mockCtrl)
+			controller.connectionParameters = mockCpm
+		})
+
+		It("adds bytes sent", func() {
+			controller.bytesSent = 5
+			controller.AddBytesSent(6)
+			Expect(controller.bytesSent).To(Equal(protocol.ByteCount(5 + 6)))
+		})
+
+		It("gets the size of the remaining flow control window", func() {
+			controller.bytesSent = 5
+			controller.sendWindow = 12
+			Expect(controller.SendWindowSize()).To(Equal(protocol.ByteCount(12 - 5)))
+		})
+
+		It("gets the offset of the flow control window", func() {
+			controller.bytesSent = 5
+			controller.sendWindow = 12
+			Expect(controller.SendWindowOffset()).To(Equal(protocol.ByteCount(12)))
+		})
+
+		It("updates the size of the flow control window", func() {
+			controller.bytesSent = 5
+			updateSuccessful := controller.UpdateSendWindow(15)
+			Expect(updateSuccessful).To(BeTrue())
+			Expect(controller.SendWindowOffset()).To(Equal(protocol.ByteCount(15)))
+			Expect(controller.SendWindowSize()).To(Equal(protocol.ByteCount(15 - 5)))
+		})
+
+		It("does not decrease the flow control window", func() {
+			updateSuccessful := controller.UpdateSendWindow(20)
+			Expect(updateSuccessful).To(BeTrue())
+			Expect(controller.SendWindowSize()).To(Equal(protocol.ByteCount(20)))
+			updateSuccessful = controller.UpdateSendWindow(10)
+			Expect(updateSuccessful).To(BeFalse())
+			Expect(controller.SendWindowSize()).To(Equal(protocol.ByteCount(20)))
+		})
+
+		It("asks the ConnectionParametersManager for the stream flow control window size", func() {
+			controller.streamID = 5
+			mockCpm.EXPECT().GetSendStreamFlowControlWindow().Return(protocol.ByteCount(1000))
+			Expect(controller.getSendWindow()).To(Equal(protocol.ByteCount(1000)))
+			// make sure the value is not cached
+			mockCpm.EXPECT().GetSendStreamFlowControlWindow().Return(protocol.ByteCount(2000))
+			Expect(controller.getSendWindow()).To(Equal(protocol.ByteCount(2000)))
+		})
+
+		It("stops asking the ConnectionParametersManager for the flow control stream window size once a window update has arrived", func() {
+			controller.streamID = 5
+			Expect(controller.UpdateSendWindow(8000))
+			Expect(controller.getSendWindow()).To(Equal(protocol.ByteCount(8000)))
+		})
+
+		It("asks the ConnectionParametersManager for the connection flow control window size", func() {
+			controller.streamID = 0
+			mockCpm.EXPECT().GetSendConnectionFlowControlWindow().Return(protocol.ByteCount(3000))
+			Expect(controller.getSendWindow()).To(Equal(protocol.ByteCount(3000)))
+			// make sure the value is not cached
+			mockCpm.EXPECT().GetSendConnectionFlowControlWindow().Return(protocol.ByteCount(5000))
+			Expect(controller.getSendWindow()).To(Equal(protocol.ByteCount(5000)))
+		})
+
+		It("stops asking the ConnectionParametersManager for the connection flow control window size once a window update has arrived", func() {
+			controller.streamID = 0
+			Expect(controller.UpdateSendWindow(7000))
+			Expect(controller.getSendWindow()).To(Equal(protocol.ByteCount(7000)))
+		})
+	})
+
+	Context("receive flow control", func() {
+		var receiveWindow protocol.ByteCount = 10000
+		var receiveWindowIncrement protocol.ByteCount = 600
+
+		BeforeEach(func() {
+			controller.receiveWindow = receiveWindow
+			controller.receiveWindowIncrement = receiveWindowIncrement
+		})
+
+		It("adds bytes read", func() {
+			controller.bytesRead = 5
+			controller.AddBytesRead(6)
+			Expect(controller.bytesRead).To(Equal(protocol.ByteCount(5 + 6)))
+		})
+
+		It("triggers a window update when necessary", func() {
+			controller.lastWindowUpdateTime = time.Now().Add(-time.Hour)
+			readPosition := receiveWindow - receiveWindowIncrement/2 + 1
+			controller.bytesRead = readPosition
+			updateNecessary, _, offset := controller.MaybeUpdateWindow()
+			Expect(updateNecessary).To(BeTrue())
+			Expect(offset).To(Equal(readPosition + receiveWindowIncrement))
+			Expect(controller.receiveWindow).To(Equal(readPosition + receiveWindowIncrement))
+			Expect(controller.lastWindowUpdateTime).To(BeTemporally("~", time.Now(), 20*time.Millisecond))
+		})
+
+		It("doesn't trigger a window update when not necessary", func() {
+			lastWindowUpdateTime := time.Now().Add(-time.Hour)
+			controller.lastWindowUpdateTime = lastWindowUpdateTime
+			readPosition := receiveWindow - receiveWindow/2 - 1
+			controller.bytesRead = readPosition
+			updateNecessary, _, _ := controller.MaybeUpdateWindow()
+			Expect(updateNecessary).To(BeFalse())
+			Expect(controller.lastWindowUpdateTime).To(Equal(lastWindowUpdateTime))
+		})
+
+		It("updates the highestReceived", func() {
+			controller.highestReceived = 1337
+			increment, err := controller.UpdateHighestReceived(1338)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(increment).To(Equal(protocol.ByteCount(1338 - 1337)))
+			Expect(controller.highestReceived).To(Equal(protocol.ByteCount(1338)))
+		})
+
+		It("does not decrease the highestReceived", func() {
+			controller.highestReceived = 1337
+			increment, err := controller.UpdateHighestReceived(1000)
+			Expect(err).To(MatchError(ErrReceivedSmallerByteOffset))
+			Expect(increment).To(BeZero())
+			Expect(controller.highestReceived).To(Equal(protocol.ByteCount(1337)))
+		})
+
+		It("does not error when setting the same byte offset", func() {
+			controller.highestReceived = 1337
+			increment, err := controller.UpdateHighestReceived(1337)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(increment).To(BeZero())
+		})
+
+		It("increases the highestReceived by a given increment", func() {
+			controller.highestReceived = 1337
+			controller.IncrementHighestReceived(123)
+			Expect(controller.highestReceived).To(Equal(protocol.ByteCount(1337 + 123)))
+		})
+
+		It("detects a flow control violation", func() {
+			controller.UpdateHighestReceived(receiveWindow + 1)
+			Expect(controller.CheckFlowControlViolation()).To(BeTrue())
+		})
+
+		It("does not give a flow control violation when using the window completely", func() {
+			controller.UpdateHighestReceived(receiveWindow)
+			Expect(controller.CheckFlowControlViolation()).To(BeFalse())
+		})
+
+		Context("receive window increment auto-tuning", func() {
+			var oldIncrement protocol.ByteCount
+
+			BeforeEach(func() {
+				oldIncrement = controller.receiveWindowIncrement
+				controller.maxReceiveWindowIncrement = 3000
+			})
+
+			// update the congestion such that it returns a given value for the smoothed RTT
+			setRtt := func(t time.Duration) {
+				controller.rttStats.UpdateRTT(t, 0, time.Now())
+				Expect(controller.rttStats.SmoothedRTT()).To(Equal(t)) // make sure it worked
+			}
+
+			It("doesn't increase the increment for a new stream", func() {
+				controller.maybeAdjustWindowIncrement()
+				Expect(controller.receiveWindowIncrement).To(Equal(oldIncrement))
+			})
+
+			It("doesn't increase the increment when no RTT estimate is available", func() {
+				setRtt(0)
+				controller.lastWindowUpdateTime = time.Now()
+				controller.maybeAdjustWindowIncrement()
+				Expect(controller.receiveWindowIncrement).To(Equal(oldIncrement))
+			})
+
+			It("increases the increment when the last WindowUpdate was sent less than two RTTs ago", func() {
+				setRtt(20 * time.Millisecond)
+				controller.lastWindowUpdateTime = time.Now().Add(-35 * time.Millisecond)
+				controller.maybeAdjustWindowIncrement()
+				Expect(controller.receiveWindowIncrement).To(Equal(2 * oldIncrement))
+			})
+
+			It("doesn't increase the increase increment when the last WindowUpdate was sent more than two RTTs ago", func() {
+				setRtt(20 * time.Millisecond)
+				controller.lastWindowUpdateTime = time.Now().Add(-45 * time.Millisecond)
+				controller.maybeAdjustWindowIncrement()
+				Expect(controller.receiveWindowIncrement).To(Equal(oldIncrement))
+			})
+
+			It("doesn't increase the increment to a value higher than the maxReceiveWindowIncrement", func() {
+				setRtt(20 * time.Millisecond)
+				controller.lastWindowUpdateTime = time.Now().Add(-35 * time.Millisecond)
+				controller.maybeAdjustWindowIncrement()
+				Expect(controller.receiveWindowIncrement).To(Equal(2 * oldIncrement)) // 1200
+				// because the lastWindowUpdateTime is updated by MaybeTriggerWindowUpdate(), we can just call maybeAdjustWindowIncrement() multiple times and get an increase of the increment every time
+				controller.maybeAdjustWindowIncrement()
+				Expect(controller.receiveWindowIncrement).To(Equal(2 * 2 * oldIncrement)) // 2400
+				controller.maybeAdjustWindowIncrement()
+				Expect(controller.receiveWindowIncrement).To(Equal(controller.maxReceiveWindowIncrement)) // 3000
+				controller.maybeAdjustWindowIncrement()
+				Expect(controller.receiveWindowIncrement).To(Equal(controller.maxReceiveWindowIncrement)) // 3000
+			})
+
+			It("returns the new increment when updating the window", func() {
+				setRtt(20 * time.Millisecond)
+				controller.AddBytesRead(9900) // receive window is 10000
+				controller.lastWindowUpdateTime = time.Now().Add(-35 * time.Millisecond)
+				necessary, newIncrement, offset := controller.MaybeUpdateWindow()
+				Expect(necessary).To(BeTrue())
+				Expect(newIncrement).To(Equal(2 * oldIncrement))
+				Expect(controller.receiveWindowIncrement).To(Equal(newIncrement))
+				Expect(offset).To(Equal(protocol.ByteCount(9900 + newIncrement)))
+			})
+
+			It("increases the increment sent in the first WindowUpdate, if data is read fast enough", func() {
+				setRtt(20 * time.Millisecond)
+				controller.AddBytesRead(9900)
+				necessary, newIncrement, _ := controller.MaybeUpdateWindow()
+				Expect(necessary).To(BeTrue())
+				Expect(newIncrement).To(Equal(2 * oldIncrement))
+			})
+
+			It("doesn't increamse the increment sent in the first WindowUpdate, if data is read slowly", func() {
+				setRtt(5 * time.Millisecond)
+				controller.AddBytesRead(9900)
+				time.Sleep(15 * time.Millisecond) // more than 2x RTT
+				necessary, newIncrement, _ := controller.MaybeUpdateWindow()
+				Expect(necessary).To(BeTrue())
+				Expect(newIncrement).To(BeZero())
+			})
+
+			It("only returns the increment if it was increased", func() {
+				setRtt(20 * time.Millisecond)
+				controller.AddBytesRead(9900) // receive window is 10000
+				controller.lastWindowUpdateTime = time.Now().Add(-45 * time.Millisecond)
+				necessary, newIncrement, offset := controller.MaybeUpdateWindow()
+				Expect(necessary).To(BeTrue())
+				Expect(newIncrement).To(BeZero())
+				Expect(controller.receiveWindowIncrement).To(Equal(oldIncrement))
+				Expect(offset).To(Equal(protocol.ByteCount(9900 + oldIncrement)))
+			})
+
+			Context("setting the minimum increment", func() {
+				It("sets the minimum window increment", func() {
+					controller.EnsureMinimumWindowIncrement(1000)
+					Expect(controller.receiveWindowIncrement).To(Equal(protocol.ByteCount(1000)))
+				})
+
+				It("doesn't reduce the window increment", func() {
+					controller.EnsureMinimumWindowIncrement(1)
+					Expect(controller.receiveWindowIncrement).To(Equal(oldIncrement))
+				})
+
+				It("doens't increase the increment beyong the maxReceiveWindowIncrement", func() {
+					max := controller.maxReceiveWindowIncrement
+					controller.EnsureMinimumWindowIncrement(2 * max)
+					Expect(controller.receiveWindowIncrement).To(Equal(max))
+				})
+
+				It("doesn't auto-tune the window after the increment was increased", func() {
+					setRtt(20 * time.Millisecond)
+					controller.bytesRead = 9900 // receive window is 10000
+					controller.lastWindowUpdateTime = time.Now().Add(-20 * time.Millisecond)
+					controller.EnsureMinimumWindowIncrement(912)
+					necessary, newIncrement, offset := controller.MaybeUpdateWindow()
+					Expect(necessary).To(BeTrue())
+					Expect(newIncrement).To(BeZero()) // no auto-tuning
+					Expect(offset).To(Equal(protocol.ByteCount(9900 + 912)))
+				})
+			})
+		})
+	})
+})

internal/flowcontrol/flowcontrol_suite_test.go

@@ -0,0 +1,24 @@
+package flowcontrol
+
+import (
+	"github.com/golang/mock/gomock"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	"testing"
+)
+
+func TestCrypto(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "FlowControl Suite")
+}
+
+var mockCtrl *gomock.Controller
+
+var _ = BeforeEach(func() {
+	mockCtrl = gomock.NewController(GinkgoT())
+})
+
+var _ = AfterEach(func() {
+	mockCtrl.Finish()
+})

internal/flowcontrol/interface.go

@@ -0,0 +1,26 @@
+package flowcontrol
+
+import "github.com/lucas-clemente/quic-go/internal/protocol"
+
+// WindowUpdate provides the data for WindowUpdateFrames.
+type WindowUpdate struct {
+	StreamID protocol.StreamID
+	Offset   protocol.ByteCount
+}
+
+// A FlowControlManager manages the flow control
+type FlowControlManager interface {
+	NewStream(streamID protocol.StreamID, contributesToConnectionFlow bool)
+	RemoveStream(streamID protocol.StreamID)
+	// methods needed for receiving data
+	ResetStream(streamID protocol.StreamID, byteOffset protocol.ByteCount) error
+	UpdateHighestReceived(streamID protocol.StreamID, byteOffset protocol.ByteCount) error
+	AddBytesRead(streamID protocol.StreamID, n protocol.ByteCount) error
+	GetWindowUpdates() []WindowUpdate
+	GetReceiveWindow(streamID protocol.StreamID) (protocol.ByteCount, error)
+	// methods needed for sending data
+	AddBytesSent(streamID protocol.StreamID, n protocol.ByteCount) error
+	SendWindowSize(streamID protocol.StreamID) (protocol.ByteCount, error)
+	RemainingConnectionWindowSize() protocol.ByteCount
+	UpdateWindow(streamID protocol.StreamID, offset protocol.ByteCount) (bool, error)
+}