send_stream_test.go

package quic

import (
	"bytes"
	"errors"
	"io"
	"runtime"
	"time"

	"github.com/golang/mock/gomock"
	"github.com/lucas-clemente/quic-go/internal/ackhandler"
	"github.com/lucas-clemente/quic-go/internal/mocks"
	"github.com/lucas-clemente/quic-go/internal/protocol"
	"github.com/lucas-clemente/quic-go/internal/wire"

	. "github.com/onsi/ginkgo"
	. "github.com/onsi/gomega"
	"github.com/onsi/gomega/gbytes"
)

var _ = Describe("Send Stream", func() {
	const streamID protocol.StreamID = 1337

	var (
		str            *sendStream
		strWithTimeout io.Writer // str wrapped with gbytes.TimeoutWriter
		mockFC         *mocks.MockStreamFlowController
		mockSender     *MockStreamSender
	)

	BeforeEach(func() {
		mockSender = NewMockStreamSender(mockCtrl)
		mockFC = mocks.NewMockStreamFlowController(mockCtrl)
		str = newSendStream(streamID, mockSender, mockFC, protocol.VersionWhatever)

		timeout := scaleDuration(250 * time.Millisecond)
		strWithTimeout = gbytes.TimeoutWriter(str, timeout)
	})

	waitForWrite := func() {
		EventuallyWithOffset(0, func() []byte {
			str.mutex.Lock()
			data := str.dataForWriting
			str.mutex.Unlock()
			return data
		}).ShouldNot(BeEmpty())
	}

	It("gets stream id", func() {
		Expect(str.StreamID()).To(Equal(protocol.StreamID(1337)))
	})

	Context("writing", func() {
		It("writes and gets all data at once", func() {
			mockSender.EXPECT().onHasStreamData(streamID)
			mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999))
			mockFC.EXPECT().AddBytesSent(protocol.ByteCount(6))
			done := make(chan struct{})
			go func() {
				defer GinkgoRecover()
				n, err := strWithTimeout.Write([]byte("foobar"))
				Expect(err).ToNot(HaveOccurred())
				Expect(n).To(Equal(6))
				close(done)
			}()
			waitForWrite()
			frame, _ := str.popStreamFrame(1000)
			f := frame.Frame.(*wire.StreamFrame)
			Expect(f.Data).To(Equal([]byte("foobar")))
			Expect(f.FinBit).To(BeFalse())
			Expect(f.Offset).To(BeZero())
			Expect(f.DataLenPresent).To(BeTrue())
			Expect(str.writeOffset).To(Equal(protocol.ByteCount(6)))
			Expect(str.dataForWriting).To(BeNil())
			Eventually(done).Should(BeClosed())
		})

		It("writes and gets data in two turns", func() {
			mockSender.EXPECT().onHasStreamData(streamID)
			frameHeaderLen := protocol.ByteCount(4)
			mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999)).Times(2)
			mockFC.EXPECT().AddBytesSent(gomock.Any() /* protocol.ByteCount(3)*/).Times(2)
			done := make(chan struct{})
			go func() {
				defer GinkgoRecover()
				n, err := strWithTimeout.Write([]byte("foobar"))
				Expect(err).ToNot(HaveOccurred())
				Expect(n).To(Equal(6))
				close(done)
			}()
			waitForWrite()
			frame, _ := str.popStreamFrame(3 + frameHeaderLen)
			f := frame.Frame.(*wire.StreamFrame)
			Expect(f.Data).To(Equal([]byte("foo")))
			Expect(f.FinBit).To(BeFalse())
			Expect(f.Offset).To(BeZero())
			Expect(f.DataLenPresent).To(BeTrue())
			frame, _ = str.popStreamFrame(100)
			f = frame.Frame.(*wire.StreamFrame)
			Expect(f.Data).To(Equal([]byte("bar")))
			Expect(f.FinBit).To(BeFalse())
			Expect(f.Offset).To(Equal(protocol.ByteCount(3)))
			Expect(f.DataLenPresent).To(BeTrue())
			Expect(str.popStreamFrame(1000)).To(BeNil())
			Eventually(done).Should(BeClosed())
		})

		It("popStreamFrame returns nil if no data is available", func() {
			frame, hasMoreData := str.popStreamFrame(1000)
			Expect(frame).To(BeNil())
			Expect(hasMoreData).To(BeFalse())
		})

		It("says if it has more data for writing", func() {
			mockSender.EXPECT().onHasStreamData(streamID)
			mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999)).Times(2)
			mockFC.EXPECT().AddBytesSent(gomock.Any()).Times(2)
			done := make(chan struct{})
			go func() {
				defer GinkgoRecover()
				n, err := strWithTimeout.Write(bytes.Repeat([]byte{0}, 100))
				Expect(err).ToNot(HaveOccurred())
				Expect(n).To(Equal(100))
				close(done)
			}()
			waitForWrite()
			frame, hasMoreData := str.popStreamFrame(50)
			Expect(frame).ToNot(BeNil())
			Expect(hasMoreData).To(BeTrue())
			frame, hasMoreData = str.popStreamFrame(1000)
			Expect(frame).ToNot(BeNil())
			Expect(hasMoreData).To(BeFalse())
			frame, _ = str.popStreamFrame(1000)
			Expect(frame).To(BeNil())
			Eventually(done).Should(BeClosed())
		})

		It("copies the slice while writing", func() {
			mockSender.EXPECT().onHasStreamData(streamID)
			frameHeaderSize := protocol.ByteCount(4)
			mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999)).Times(2)
			mockFC.EXPECT().AddBytesSent(protocol.ByteCount(1))
			mockFC.EXPECT().AddBytesSent(protocol.ByteCount(2))
			s := []byte("foo")
			done := make(chan struct{})
			go func() {
				defer GinkgoRecover()
				n, err := strWithTimeout.Write(s)
				Expect(err).ToNot(HaveOccurred())
				Expect(n).To(Equal(3))
				close(done)
			}()
			waitForWrite()
			frame, _ := str.popStreamFrame(frameHeaderSize + 1)
			f := frame.Frame.(*wire.StreamFrame)
			Expect(f.Data).To(Equal([]byte("f")))
			frame, _ = str.popStreamFrame(100)
			Expect(frame).ToNot(BeNil())
			f = frame.Frame.(*wire.StreamFrame)
			Expect(f.Data).To(Equal([]byte("oo")))
			s[1] = 'e'
			Expect(f.Data).To(Equal([]byte("oo")))
			Eventually(done).Should(BeClosed())
		})

		It("returns when given a nil input", func() {
			n, err := strWithTimeout.Write(nil)
			Expect(n).To(BeZero())
			Expect(err).ToNot(HaveOccurred())
		})

		It("returns when given an empty slice", func() {
			n, err := strWithTimeout.Write([]byte(""))
			Expect(n).To(BeZero())
			Expect(err).ToNot(HaveOccurred())
		})

		It("cancels the context when Close is called", func() {
			mockSender.EXPECT().onHasStreamData(streamID)
			Expect(str.Context().Done()).ToNot(BeClosed())
			str.Close()
			Expect(str.Context().Done()).To(BeClosed())
		})

		Context("flow control blocking", func() {
			It("queues a BLOCKED frame if the stream is flow control blocked", func() {
				mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(0))
				mockFC.EXPECT().IsNewlyBlocked().Return(true, protocol.ByteCount(12))
				mockSender.EXPECT().queueControlFrame(&wire.StreamDataBlockedFrame{
					StreamID:  streamID,
					DataLimit: 12,
				})
				mockSender.EXPECT().onHasStreamData(streamID)
				done := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					_, err := str.Write([]byte("foobar"))
					Expect(err).ToNot(HaveOccurred())
					close(done)
				}()
				waitForWrite()
				f, hasMoreData := str.popStreamFrame(1000)
				Expect(f).To(BeNil())
				Expect(hasMoreData).To(BeFalse())
				// make the Write go routine return
				str.closeForShutdown(nil)
				Eventually(done).Should(BeClosed())
			})

			It("says that it doesn't have any more data, when it is flow control blocked", func() {
				frameHeaderSize := protocol.ByteCount(4)
				mockSender.EXPECT().onHasStreamData(streamID)

				done := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					_, err := str.Write([]byte("foobar"))
					Expect(err).ToNot(HaveOccurred())
					close(done)
				}()
				waitForWrite()

				// first pop a STREAM frame of the maximum size allowed by flow control
				mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(3))
				mockFC.EXPECT().AddBytesSent(protocol.ByteCount(3))
				f, hasMoreData := str.popStreamFrame(frameHeaderSize + 3)
				Expect(f).ToNot(BeNil())
				Expect(hasMoreData).To(BeTrue())

				// try to pop again, this time noticing that we're blocked
				mockFC.EXPECT().SendWindowSize()
				// don't use offset 3 here, to make sure the BLOCKED frame contains the number returned by the flow controller
				mockFC.EXPECT().IsNewlyBlocked().Return(true, protocol.ByteCount(10))
				mockSender.EXPECT().queueControlFrame(&wire.StreamDataBlockedFrame{
					StreamID:  streamID,
					DataLimit: 10,
				})
				f, hasMoreData = str.popStreamFrame(1000)
				Expect(f).To(BeNil())
				Expect(hasMoreData).To(BeFalse())
				// make the Write go routine return
				str.closeForShutdown(nil)
				Eventually(done).Should(BeClosed())
			})
		})

		Context("deadlines", func() {
			It("returns an error when Write is called after the deadline", func() {
				str.SetWriteDeadline(time.Now().Add(-time.Second))
				n, err := strWithTimeout.Write([]byte("foobar"))
				Expect(err).To(MatchError(errDeadline))
				Expect(n).To(BeZero())
			})

			It("unblocks after the deadline", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				deadline := time.Now().Add(scaleDuration(50 * time.Millisecond))
				str.SetWriteDeadline(deadline)
				n, err := strWithTimeout.Write([]byte("foobar"))
				Expect(err).To(MatchError(errDeadline))
				Expect(n).To(BeZero())
				Expect(time.Now()).To(BeTemporally("~", deadline, scaleDuration(20*time.Millisecond)))
			})

			It("unblocks when the deadline is changed to the past", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				str.SetWriteDeadline(time.Now().Add(time.Hour))
				done := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					_, err := str.Write([]byte("foobar"))
					Expect(err).To(MatchError(errDeadline))
					close(done)
				}()
				Consistently(done).ShouldNot(BeClosed())
				str.SetWriteDeadline(time.Now().Add(-time.Hour))
				Eventually(done).Should(BeClosed())
			})

			It("returns the number of bytes written, when the deadline expires", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(10000)).AnyTimes()
				mockFC.EXPECT().AddBytesSent(gomock.Any())
				deadline := time.Now().Add(scaleDuration(50 * time.Millisecond))
				str.SetWriteDeadline(deadline)
				var n int
				writeReturned := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					var err error
					n, err = strWithTimeout.Write(bytes.Repeat([]byte{0}, 100))
					Expect(err).To(MatchError(errDeadline))
					Expect(time.Now()).To(BeTemporally("~", deadline, scaleDuration(20*time.Millisecond)))
					close(writeReturned)
				}()
				waitForWrite()
				frame, hasMoreData := str.popStreamFrame(50)
				Expect(frame).ToNot(BeNil())
				Expect(hasMoreData).To(BeTrue())
				Eventually(writeReturned, scaleDuration(80*time.Millisecond)).Should(BeClosed())
				Expect(n).To(BeEquivalentTo(frame.Frame.(*wire.StreamFrame).DataLen()))
			})

			It("doesn't pop any data after the deadline expired", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(10000)).AnyTimes()
				mockFC.EXPECT().AddBytesSent(gomock.Any())
				deadline := time.Now().Add(scaleDuration(50 * time.Millisecond))
				str.SetWriteDeadline(deadline)
				writeReturned := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					_, err := strWithTimeout.Write(bytes.Repeat([]byte{0}, 100))
					Expect(err).To(MatchError(errDeadline))
					close(writeReturned)
				}()
				waitForWrite()
				frame, hasMoreData := str.popStreamFrame(50)
				Expect(frame).ToNot(BeNil())
				Expect(hasMoreData).To(BeTrue())
				Eventually(writeReturned, scaleDuration(80*time.Millisecond)).Should(BeClosed())
				frame, hasMoreData = str.popStreamFrame(50)
				Expect(frame).To(BeNil())
				Expect(hasMoreData).To(BeFalse())
			})

			It("doesn't unblock if the deadline is changed before the first one expires", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				deadline1 := time.Now().Add(scaleDuration(50 * time.Millisecond))
				deadline2 := time.Now().Add(scaleDuration(100 * time.Millisecond))
				str.SetWriteDeadline(deadline1)
				done := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					time.Sleep(scaleDuration(20 * time.Millisecond))
					str.SetWriteDeadline(deadline2)
					// make sure that this was actually execute before the deadline expires
					Expect(time.Now()).To(BeTemporally("<", deadline1))
					close(done)
				}()
				runtime.Gosched()
				n, err := strWithTimeout.Write([]byte("foobar"))
				Expect(err).To(MatchError(errDeadline))
				Expect(n).To(BeZero())
				Expect(time.Now()).To(BeTemporally("~", deadline2, scaleDuration(20*time.Millisecond)))
				Eventually(done).Should(BeClosed())
			})

			It("unblocks earlier, when a new deadline is set", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				deadline1 := time.Now().Add(scaleDuration(200 * time.Millisecond))
				deadline2 := time.Now().Add(scaleDuration(50 * time.Millisecond))
				done := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					time.Sleep(scaleDuration(10 * time.Millisecond))
					str.SetWriteDeadline(deadline2)
					// make sure that this was actually execute before the deadline expires
					Expect(time.Now()).To(BeTemporally("<", deadline2))
					close(done)
				}()
				str.SetWriteDeadline(deadline1)
				runtime.Gosched()
				_, err := strWithTimeout.Write([]byte("foobar"))
				Expect(err).To(MatchError(errDeadline))
				Expect(time.Now()).To(BeTemporally("~", deadline2, scaleDuration(20*time.Millisecond)))
				Eventually(done).Should(BeClosed())
			})

			It("doesn't unblock if the deadline is removed", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				deadline := time.Now().Add(scaleDuration(50 * time.Millisecond))
				str.SetWriteDeadline(deadline)
				deadlineUnset := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					time.Sleep(scaleDuration(20 * time.Millisecond))
					str.SetWriteDeadline(time.Time{})
					// make sure that this was actually execute before the deadline expires
					Expect(time.Now()).To(BeTemporally("<", deadline))
					close(deadlineUnset)
				}()
				done := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					_, err := strWithTimeout.Write([]byte("foobar"))
					Expect(err).To(MatchError("test done"))
					close(done)
				}()
				runtime.Gosched()
				Eventually(deadlineUnset).Should(BeClosed())
				Consistently(done, scaleDuration(100*time.Millisecond)).ShouldNot(BeClosed())
				// make the go routine return
				str.closeForShutdown(errors.New("test done"))
				Eventually(done).Should(BeClosed())
			})
		})

		Context("closing", func() {
			It("doesn't allow writes after it has been closed", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				str.Close()
				_, err := strWithTimeout.Write([]byte("foobar"))
				Expect(err).To(MatchError("write on closed stream 1337"))
			})

			It("allows FIN", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				str.Close()
				frame, hasMoreData := str.popStreamFrame(1000)
				Expect(frame).ToNot(BeNil())
				f := frame.Frame.(*wire.StreamFrame)
				Expect(f.Data).To(BeEmpty())
				Expect(f.FinBit).To(BeTrue())
				Expect(hasMoreData).To(BeFalse())
			})

			It("doesn't send a FIN when there's still data", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				frameHeaderLen := protocol.ByteCount(4)
				mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999)).Times(2)
				mockFC.EXPECT().AddBytesSent(gomock.Any()).Times(2)
				str.dataForWriting = []byte("foobar")
				Expect(str.Close()).To(Succeed())
				frame, _ := str.popStreamFrame(3 + frameHeaderLen)
				Expect(frame).ToNot(BeNil())
				f := frame.Frame.(*wire.StreamFrame)
				Expect(f.Data).To(Equal([]byte("foo")))
				Expect(f.FinBit).To(BeFalse())
				frame, _ = str.popStreamFrame(100)
				f = frame.Frame.(*wire.StreamFrame)
				Expect(f.Data).To(Equal([]byte("bar")))
				Expect(f.FinBit).To(BeTrue())
			})

			It("doesn't allow FIN after it is closed for shutdown", func() {
				str.closeForShutdown(errors.New("test"))
				f, hasMoreData := str.popStreamFrame(1000)
				Expect(f).To(BeNil())
				Expect(hasMoreData).To(BeFalse())
			})

			It("doesn't allow FIN twice", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				str.Close()
				frame, _ := str.popStreamFrame(1000)
				Expect(frame).ToNot(BeNil())
				f := frame.Frame.(*wire.StreamFrame)
				Expect(f.Data).To(BeEmpty())
				Expect(f.FinBit).To(BeTrue())
				frame, hasMoreData := str.popStreamFrame(1000)
				Expect(frame).To(BeNil())
				Expect(hasMoreData).To(BeFalse())
			})
		})

		Context("closing for shutdown", func() {
			testErr := errors.New("test")

			It("returns errors when the stream is cancelled", func() {
				str.closeForShutdown(testErr)
				n, err := strWithTimeout.Write([]byte("foo"))
				Expect(n).To(BeZero())
				Expect(err).To(MatchError(testErr))
			})

			It("doesn't get data for writing if an error occurred", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999))
				mockFC.EXPECT().AddBytesSent(gomock.Any())
				done := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					_, err := strWithTimeout.Write(bytes.Repeat([]byte{0}, 500))
					Expect(err).To(MatchError(testErr))
					close(done)
				}()
				waitForWrite()
				frame, hasMoreData := str.popStreamFrame(50) // get a STREAM frame containing some data, but not all
				Expect(frame).ToNot(BeNil())
				Expect(hasMoreData).To(BeTrue())
				str.closeForShutdown(testErr)
				frame, hasMoreData = str.popStreamFrame(1000)
				Expect(frame).To(BeNil())
				Expect(hasMoreData).To(BeFalse())
				Eventually(done).Should(BeClosed())
			})

			It("cancels the context", func() {
				Expect(str.Context().Done()).ToNot(BeClosed())
				str.closeForShutdown(testErr)
				Expect(str.Context().Done()).To(BeClosed())
			})
		})
	})

	Context("handling MAX_STREAM_DATA frames", func() {
		It("informs the flow controller", func() {
			mockFC.EXPECT().UpdateSendWindow(protocol.ByteCount(0x1337))
			str.handleMaxStreamDataFrame(&wire.MaxStreamDataFrame{
				StreamID:   streamID,
				ByteOffset: 0x1337,
			})
		})

		It("says when it has data for sending", func() {
			mockFC.EXPECT().UpdateSendWindow(gomock.Any())
			mockSender.EXPECT().onHasStreamData(streamID).Times(2) // once for Write, once for the MAX_STREAM_DATA frame
			done := make(chan struct{})
			go func() {
				defer GinkgoRecover()
				_, err := str.Write([]byte("foobar"))
				Expect(err).ToNot(HaveOccurred())
				close(done)
			}()
			waitForWrite()
			str.handleMaxStreamDataFrame(&wire.MaxStreamDataFrame{
				StreamID:   streamID,
				ByteOffset: 42,
			})
			// make sure the Write go routine returns
			str.closeForShutdown(nil)
			Eventually(done).Should(BeClosed())
		})
	})

	Context("stream cancellations", func() {
		Context("canceling writing", func() {
			It("queues a RESET_STREAM frame", func() {
				gomock.InOrder(
					mockSender.EXPECT().queueControlFrame(&wire.ResetStreamFrame{
						StreamID:   streamID,
						ByteOffset: 1234,
						ErrorCode:  9876,
					}),
					mockSender.EXPECT().onStreamCompleted(streamID),
				)
				str.writeOffset = 1234
				str.CancelWrite(9876)
			})

			It("unblocks Write", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				mockSender.EXPECT().queueControlFrame(gomock.Any())
				mockFC.EXPECT().SendWindowSize().Return(protocol.MaxByteCount)
				mockFC.EXPECT().AddBytesSent(gomock.Any())
				writeReturned := make(chan struct{})
				var n int
				go func() {
					defer GinkgoRecover()
					var err error
					n, err = strWithTimeout.Write(bytes.Repeat([]byte{0}, 100))
					Expect(err).To(MatchError("Write on stream 1337 canceled with error code 1234"))
					close(writeReturned)
				}()
				waitForWrite()
				frame, _ := str.popStreamFrame(50)
				Expect(frame).ToNot(BeNil())
				str.CancelWrite(1234)
				Eventually(writeReturned).Should(BeClosed())
				Expect(n).To(BeEquivalentTo(frame.Frame.(*wire.StreamFrame).DataLen()))
			})

			It("doesn't pop STREAM frames after being canceled", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				mockSender.EXPECT().queueControlFrame(gomock.Any())
				mockFC.EXPECT().SendWindowSize().Return(protocol.MaxByteCount)
				mockFC.EXPECT().AddBytesSent(gomock.Any())
				writeReturned := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					_, err := strWithTimeout.Write(bytes.Repeat([]byte{0}, 100))
					Expect(err).To(MatchError("Write on stream 1337 canceled with error code 1234"))
					close(writeReturned)
				}()
				waitForWrite()
				frame, hasMoreData := str.popStreamFrame(50)
				Expect(hasMoreData).To(BeTrue())
				Expect(frame).ToNot(BeNil())
				str.CancelWrite(1234)
				frame, hasMoreData = str.popStreamFrame(10)
				Expect(hasMoreData).To(BeFalse())
				Expect(frame).To(BeNil())
				Eventually(writeReturned).Should(BeClosed())
			})

			It("cancels the context", func() {
				mockSender.EXPECT().queueControlFrame(gomock.Any())
				mockSender.EXPECT().onStreamCompleted(gomock.Any())
				Expect(str.Context().Done()).ToNot(BeClosed())
				str.CancelWrite(1234)
				Expect(str.Context().Done()).To(BeClosed())
			})

			It("doesn't allow further calls to Write", func() {
				mockSender.EXPECT().queueControlFrame(gomock.Any())
				mockSender.EXPECT().onStreamCompleted(gomock.Any())
				str.CancelWrite(1234)
				_, err := strWithTimeout.Write([]byte("foobar"))
				Expect(err).To(MatchError("Write on stream 1337 canceled with error code 1234"))
			})

			It("only cancels once", func() {
				mockSender.EXPECT().queueControlFrame(&wire.ResetStreamFrame{StreamID: streamID, ErrorCode: 1234})
				mockSender.EXPECT().onStreamCompleted(gomock.Any())
				str.CancelWrite(1234)
				str.CancelWrite(4321)
			})

			It("queues a RESET_STREAM frame, even if the stream was already closed", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				mockSender.EXPECT().queueControlFrame(gomock.Any()).Do(func(f wire.Frame) {
					Expect(f).To(BeAssignableToTypeOf(&wire.ResetStreamFrame{}))
				})
				mockSender.EXPECT().onStreamCompleted(gomock.Any())
				Expect(str.Close()).To(Succeed())
				// don't EXPECT any calls to queueControlFrame
				str.CancelWrite(123)
			})
		})

		Context("receiving STOP_SENDING frames", func() {
			It("queues a RESET_STREAM frames, and copies the error code from the STOP_SENDING frame", func() {
				mockSender.EXPECT().queueControlFrame(&wire.ResetStreamFrame{
					StreamID:  streamID,
					ErrorCode: 101,
				})
				mockSender.EXPECT().onStreamCompleted(gomock.Any())

				str.handleStopSendingFrame(&wire.StopSendingFrame{
					StreamID:  streamID,
					ErrorCode: 101,
				})
			})

			It("unblocks Write", func() {
				mockSender.EXPECT().onHasStreamData(streamID)
				mockSender.EXPECT().queueControlFrame(gomock.Any())
				mockSender.EXPECT().onStreamCompleted(gomock.Any())
				done := make(chan struct{})
				go func() {
					defer GinkgoRecover()
					_, err := str.Write([]byte("foobar"))
					Expect(err).To(MatchError("stream 1337 was reset with error code 123"))
					Expect(err).To(BeAssignableToTypeOf(streamCanceledError{}))
					Expect(err.(streamCanceledError).Canceled()).To(BeTrue())
					Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(123)))
					close(done)
				}()
				waitForWrite()
				str.handleStopSendingFrame(&wire.StopSendingFrame{
					StreamID:  streamID,
					ErrorCode: 123,
				})
				Eventually(done).Should(BeClosed())
			})

			It("doesn't allow further calls to Write", func() {
				mockSender.EXPECT().queueControlFrame(gomock.Any())
				mockSender.EXPECT().onStreamCompleted(gomock.Any())
				str.handleStopSendingFrame(&wire.StopSendingFrame{
					StreamID:  streamID,
					ErrorCode: 123,
				})
				_, err := str.Write([]byte("foobar"))
				Expect(err).To(MatchError("stream 1337 was reset with error code 123"))
				Expect(err).To(BeAssignableToTypeOf(streamCanceledError{}))
				Expect(err.(streamCanceledError).Canceled()).To(BeTrue())
				Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(123)))
			})
		})
	})

	Context("retransmissions", func() {
		It("queues and retrieves frames", func() {
			str.numOutstandingFrames = 1
			f := &wire.StreamFrame{
				Data:           []byte("foobar"),
				Offset:         0x42,
				DataLenPresent: false,
			}
			mockSender.EXPECT().onHasStreamData(streamID)
			str.queueRetransmission(f)
			frame, _ := str.popStreamFrame(protocol.MaxByteCount)
			Expect(frame).ToNot(BeNil())
			f = frame.Frame.(*wire.StreamFrame)
			Expect(f.Offset).To(Equal(protocol.ByteCount(0x42)))
			Expect(f.Data).To(Equal([]byte("foobar")))
			Expect(f.DataLenPresent).To(BeTrue())
		})

		It("splits a retransmission", func() {
			str.numOutstandingFrames = 1
			sf := &wire.StreamFrame{
				Data:           []byte("foobar"),
				Offset:         0x42,
				DataLenPresent: false,
			}
			mockSender.EXPECT().onHasStreamData(streamID)
			str.queueRetransmission(sf)
			frame, hasMoreData := str.popStreamFrame(sf.Length(str.version) - 3)
			Expect(frame).ToNot(BeNil())
			f := frame.Frame.(*wire.StreamFrame)
			Expect(hasMoreData).To(BeTrue())
			Expect(f.Offset).To(Equal(protocol.ByteCount(0x42)))
			Expect(f.Data).To(Equal([]byte("foo")))
			Expect(f.DataLenPresent).To(BeTrue())
			frame, _ = str.popStreamFrame(protocol.MaxByteCount)
			Expect(frame).ToNot(BeNil())
			f = frame.Frame.(*wire.StreamFrame)
			Expect(f.Offset).To(Equal(protocol.ByteCount(0x45)))
			Expect(f.Data).To(Equal([]byte("bar")))
			Expect(f.DataLenPresent).To(BeTrue())
		})

		It("returns nil if the size is too small", func() {
			str.numOutstandingFrames = 1
			f := &wire.StreamFrame{
				Data:           []byte("foobar"),
				Offset:         0x42,
				DataLenPresent: false,
			}
			mockSender.EXPECT().onHasStreamData(streamID)
			str.queueRetransmission(f)
			frame, hasMoreData := str.popStreamFrame(2)
			Expect(hasMoreData).To(BeTrue())
			Expect(frame).To(BeNil())
		})

		It("queues lost STREAM frames", func() {
			mockSender.EXPECT().onHasStreamData(streamID).Times(2)
			mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999))
			mockFC.EXPECT().AddBytesSent(protocol.ByteCount(6))
			done := make(chan struct{})
			go func() {
				defer GinkgoRecover()
				_, err := strWithTimeout.Write([]byte("foobar"))
				Expect(err).ToNot(HaveOccurred())
				close(done)
			}()
			waitForWrite()
			frame, _ := str.popStreamFrame(protocol.MaxByteCount)
			Eventually(done).Should(BeClosed())
			Expect(frame).ToNot(BeNil())
			Expect(frame.Frame.(*wire.StreamFrame).Data).To(Equal([]byte("foobar")))

			// now lose the frame
			frame.OnLost(frame.Frame)
			newFrame, _ := str.popStreamFrame(protocol.MaxByteCount)
			Expect(newFrame).ToNot(BeNil())
			Expect(newFrame.Frame.(*wire.StreamFrame).Data).To(Equal([]byte("foobar")))
		})
	})

	Context("determining when a stream is completed", func() {
		BeforeEach(func() {
			mockSender.EXPECT().onHasStreamData(streamID).AnyTimes()
			mockFC.EXPECT().SendWindowSize().Return(protocol.MaxByteCount).AnyTimes()
			mockFC.EXPECT().AddBytesSent(gomock.Any()).AnyTimes()
		})

		It("says when a stream is completed", func() {
			done := make(chan struct{})
			go func() {
				defer GinkgoRecover()
				_, err := strWithTimeout.Write(make([]byte, 1000))
				Expect(err).ToNot(HaveOccurred())
				close(done)
			}()
			waitForWrite()

			// get a bunch of small frames (max. 20 bytes)
			var frames []ackhandler.Frame
			for {
				frame, hasMoreData := str.popStreamFrame(200)
				if frame == nil {
					continue
				}
				frames = append(frames, *frame)
				if !hasMoreData {
					break
				}
			}
			Eventually(done).Should(BeClosed())

			// Acknowledge all frames.
			// We don't expect the stream to be completed, since we still need to send the FIN.
			for _, f := range frames {
				f.OnAcked(f.Frame)
			}

			// Now close the stream and acknowledge the FIN.
			Expect(str.Close()).To(Succeed())
			frame, _ := str.popStreamFrame(protocol.MaxByteCount)
			Expect(frame).ToNot(BeNil())
			mockSender.EXPECT().onStreamCompleted(streamID)
			frame.OnAcked(frame.Frame)
		})

		It("doesn't say it's completed when there are frames waiting to be retransmitted", func() {
			done := make(chan struct{})
			go func() {
				defer GinkgoRecover()
				_, err := strWithTimeout.Write(make([]byte, 10))
				Expect(err).ToNot(HaveOccurred())
				Expect(str.Close()).To(Succeed())
				close(done)
			}()
			waitForWrite()

			// get a bunch of small frames (max. 20 bytes)
			var frames []ackhandler.Frame
			for {
				frame, _ := str.popStreamFrame(protocol.MaxByteCount)
				if frame == nil {
					continue
				}
				frames = append(frames, *frame)
				if frame.Frame.(*wire.StreamFrame).FinBit {
					break
				}
			}
			Eventually(done).Should(BeClosed())

			// lose the first frame, acknowledge all others
			for _, f := range frames[1:] {
				f.OnAcked(f.Frame)
			}
			frames[0].OnLost(frames[0].Frame)

			// get the retransmission and acknowledge it
			ret, _ := str.popStreamFrame(protocol.MaxByteCount)
			Expect(ret).ToNot(BeNil())
			mockSender.EXPECT().onStreamCompleted(streamID)
			ret.OnAcked(ret.Frame)
		})
	})
})