diff --git a/h2quic/response_writer_test.go b/h2quic/response_writer_test.go index e3e3e27b..4ea0701c 100644 --- a/h2quic/response_writer_test.go +++ b/h2quic/response_writer_test.go @@ -11,6 +11,7 @@ import ( "golang.org/x/net/http2" "golang.org/x/net/http2/hpack" + quic "github.com/lucas-clemente/quic-go" "github.com/lucas-clemente/quic-go/internal/protocol" . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" @@ -29,6 +30,8 @@ type mockStream struct { ctxCancel context.CancelFunc } +var _ quic.Stream = &mockStream{} + func newMockStream(id protocol.StreamID) *mockStream { s := &mockStream{ id: id, @@ -39,7 +42,8 @@ func newMockStream(id protocol.StreamID) *mockStream { } func (s *mockStream) Close() error { s.closed = true; s.ctxCancel(); return nil } -func (s *mockStream) Reset(error) { s.reset = true } +func (s *mockStream) CancelRead(quic.ErrorCode) error { s.reset = true; return nil } +func (s *mockStream) CancelWrite(quic.ErrorCode) error { panic("not implemented") } func (s *mockStream) CloseRemote(offset protocol.ByteCount) { s.remoteClosed = true; s.ctxCancel() } func (s mockStream) StreamID() protocol.StreamID { return s.id } func (s *mockStream) Context() context.Context { return s.ctx } diff --git a/h2quic/server.go b/h2quic/server.go index 0d0cecf4..134ae9a8 100644 --- a/h2quic/server.go +++ b/h2quic/server.go @@ -223,7 +223,8 @@ func (s *Server) handleRequest(session streamCreator, headerStream quic.Stream, } if responseWriter.dataStream != nil { if !streamEnded && !reqBody.requestRead { - responseWriter.dataStream.Reset(nil) + // in gQUIC, the error code doesn't matter, so just use 0 here + responseWriter.dataStream.CancelRead(0) } responseWriter.dataStream.Close() } diff --git a/interface.go b/interface.go index 87bf9ea3..962f709b 100644 --- a/interface.go +++ b/interface.go @@ -19,20 +19,38 @@ type VersionNumber = protocol.VersionNumber // A Cookie can be used to verify the ownership of the client address. type Cookie = handshake.Cookie +// An ErrorCode is an application-defined error code. +type ErrorCode = protocol.ApplicationErrorCode + // Stream is the interface implemented by QUIC streams type Stream interface { // Read reads data from the stream. // Read can be made to time out and return a net.Error with Timeout() == true // after a fixed time limit; see SetDeadline and SetReadDeadline. + // If the stream was canceled by the peer, the error implements the StreamError + // interface, and Canceled() == true. io.Reader // Write writes data to the stream. // Write can be made to time out and return a net.Error with Timeout() == true // after a fixed time limit; see SetDeadline and SetWriteDeadline. + // If the stream was canceled by the peer, the error implements the StreamError + // interface, and Canceled() == true. io.Writer + // Close closes the write-direction of the stream. + // Future calls to Write are not permitted after calling Close. + // It must not be called concurrently with Write. + // It must not be called after calling CancelWrite. io.Closer StreamID() StreamID - // Reset closes the stream with an error. - Reset(error) + // CancelWrite aborts sending on this stream. + // It must not be called after Close. + // Data already written, but not yet delivered to the peer is not guaranteed to be delivered reliably. + // Write will unblock immediately, and future calls to Write will fail. + CancelWrite(ErrorCode) error + // CancelRead aborts receiving on this stream. + // It will ask the peer to stop transmitting stream data. + // Read will unblock immediately, and future Read calls will fail. + CancelRead(ErrorCode) error // The context is canceled as soon as the write-side of the stream is closed. // This happens when Close() is called, or when the stream is reset (either locally or remotely). // Warning: This API should not be considered stable and might change soon. @@ -53,6 +71,13 @@ type Stream interface { SetDeadline(t time.Time) error } +// StreamError is returned by Read and Write when the peer cancels the stream. +type StreamError interface { + error + Canceled() bool + ErrorCode() ErrorCode +} + // A Session is a QUIC connection between two peers. type Session interface { // AcceptStream returns the next stream opened by the peer, blocking until one is available. diff --git a/internal/mocks/stream.go b/internal/mocks/stream.go index e88ac1bd..8da42d9a 100644 --- a/internal/mocks/stream.go +++ b/internal/mocks/stream.go @@ -36,6 +36,30 @@ func (_m *MockStreamI) EXPECT() *MockStreamIMockRecorder { return _m.recorder } +// CancelRead mocks base method +func (_m *MockStreamI) CancelRead(_param0 protocol.ApplicationErrorCode) error { + ret := _m.ctrl.Call(_m, "CancelRead", _param0) + ret0, _ := ret[0].(error) + return ret0 +} + +// CancelRead indicates an expected call of CancelRead +func (_mr *MockStreamIMockRecorder) CancelRead(arg0 interface{}) *gomock.Call { + return _mr.mock.ctrl.RecordCallWithMethodType(_mr.mock, "CancelRead", reflect.TypeOf((*MockStreamI)(nil).CancelRead), arg0) +} + +// CancelWrite mocks base method +func (_m *MockStreamI) CancelWrite(_param0 protocol.ApplicationErrorCode) error { + ret := _m.ctrl.Call(_m, "CancelWrite", _param0) + ret0, _ := ret[0].(error) + return ret0 +} + +// CancelWrite indicates an expected call of CancelWrite +func (_mr *MockStreamIMockRecorder) CancelWrite(arg0 interface{}) *gomock.Call { + return _mr.mock.ctrl.RecordCallWithMethodType(_mr.mock, "CancelWrite", reflect.TypeOf((*MockStreamI)(nil).CancelWrite), arg0) +} + // Close mocks base method func (_m *MockStreamI) Close() error { ret := _m.ctrl.Call(_m, "Close") @@ -166,16 +190,6 @@ func (_mr *MockStreamIMockRecorder) Read(arg0 interface{}) *gomock.Call { return _mr.mock.ctrl.RecordCallWithMethodType(_mr.mock, "Read", reflect.TypeOf((*MockStreamI)(nil).Read), arg0) } -// Reset mocks base method -func (_m *MockStreamI) Reset(_param0 error) { - _m.ctrl.Call(_m, "Reset", _param0) -} - -// Reset indicates an expected call of Reset -func (_mr *MockStreamIMockRecorder) Reset(arg0 interface{}) *gomock.Call { - return _mr.mock.ctrl.RecordCallWithMethodType(_mr.mock, "Reset", reflect.TypeOf((*MockStreamI)(nil).Reset), arg0) -} - // SetDeadline mocks base method func (_m *MockStreamI) SetDeadline(_param0 time.Time) error { ret := _m.ctrl.Call(_m, "SetDeadline", _param0) diff --git a/stream.go b/stream.go index 72f17bdd..89c41e52 100644 --- a/stream.go +++ b/stream.go @@ -14,6 +14,19 @@ import ( "github.com/lucas-clemente/quic-go/internal/wire" ) +type streamCanceledError struct { + error + errorCode protocol.ApplicationErrorCode +} + +func (streamCanceledError) Canceled() bool { return true } +func (e streamCanceledError) ErrorCode() protocol.ApplicationErrorCode { return e.errorCode } + +var _ StreamError = &streamCanceledError{} +var _ error = &streamCanceledError{} + +const errorCodeStoppingGQUIC protocol.ApplicationErrorCode = 7 + type streamI interface { Stream @@ -48,27 +61,24 @@ type stream struct { writeOffset protocol.ByteCount readOffset protocol.ByteCount - // Once set, the errors must not be changed! - err error + closeForShutdownErr error + cancelWriteErr error + cancelReadErr error + resetRemotelyErr StreamError - // closedForShutdown is set when Cancel() is called - closedForShutdown utils.AtomicBool - // finishedReading is set once we read a frame with a FinBit - finishedReading utils.AtomicBool - // finisedWriting is set once Close() is called - finishedWriting utils.AtomicBool - // resetLocally is set if Reset() is called - resetLocally utils.AtomicBool - // resetRemotely is set if HandleRstStreamFrame() is called - resetRemotely utils.AtomicBool + closedForShutdown bool // set when CloseForShutdown() is called + finRead bool // set once we read a frame with a FinBit + finishedWriting bool // set once Close() is called + canceledWrite bool // set when CancelWrite() is called + canceledRead bool // set when CancelRead() is called + finSent bool // set when a STREAM_FRAME with FIN bit has b + resetRemotely bool // set when HandleRstStreamFrame() is called frameQueue *streamFrameSorter readChan chan struct{} readDeadline time.Time dataForWriting []byte - finSent utils.AtomicBool - rstSent utils.AtomicBool writeChan chan struct{} writeDeadline time.Time @@ -111,37 +121,43 @@ func newStream(StreamID protocol.StreamID, // Read implements io.Reader. It is not thread safe! func (s *stream) Read(p []byte) (int, error) { s.mutex.Lock() - err := s.err - s.mutex.Unlock() - if s.closedForShutdown.Get() || s.resetLocally.Get() { - return 0, err - } - if s.finishedReading.Get() { + defer s.mutex.Unlock() + + if s.finRead { return 0, io.EOF } + if s.canceledRead { + return 0, s.cancelReadErr + } + if s.resetRemotely { + return 0, s.resetRemotelyErr + } + if s.closedForShutdown { + return 0, s.closeForShutdownErr + } bytesRead := 0 for bytesRead < len(p) { - s.mutex.Lock() frame := s.frameQueue.Head() if frame == nil && bytesRead > 0 { - err = s.err - s.mutex.Unlock() - return bytesRead, err + return bytesRead, s.closeForShutdownErr } - var err error for { // Stop waiting on errors - if s.resetLocally.Get() || s.closedForShutdown.Get() { - err = s.err - break + if s.closedForShutdown { + return bytesRead, s.closeForShutdownErr + } + if s.canceledRead { + return bytesRead, s.cancelReadErr + } + if s.resetRemotely { + return bytesRead, s.resetRemotelyErr } deadline := s.readDeadline if !deadline.IsZero() && !time.Now().Before(deadline) { - err = errDeadline - break + return bytesRead, errDeadline } if frame != nil { @@ -161,13 +177,6 @@ func (s *stream) Read(p []byte) (int, error) { s.mutex.Lock() frame = s.frameQueue.Head() } - s.mutex.Unlock() - - if err != nil { - return bytesRead, err - } - - m := utils.Min(len(p)-bytesRead, int(frame.DataLen())-s.readPosInFrame) if bytesRead > len(p) { return bytesRead, fmt.Errorf("BUG: bytesRead (%d) > len(p) (%d) in stream.Read", bytesRead, len(p)) @@ -175,30 +184,30 @@ func (s *stream) Read(p []byte) (int, error) { if s.readPosInFrame > int(frame.DataLen()) { return bytesRead, fmt.Errorf("BUG: readPosInFrame (%d) > frame.DataLen (%d) in stream.Read", s.readPosInFrame, frame.DataLen()) } - copy(p[bytesRead:], frame.Data[s.readPosInFrame:]) + s.mutex.Unlock() + + copy(p[bytesRead:], frame.Data[s.readPosInFrame:]) + m := utils.Min(len(p)-bytesRead, int(frame.DataLen())-s.readPosInFrame) s.readPosInFrame += m bytesRead += m s.readOffset += protocol.ByteCount(m) + s.mutex.Lock() // when a RST_STREAM was received, the was already informed about the final byteOffset for this stream - if !s.resetRemotely.Get() { + if !s.resetRemotely { s.flowController.AddBytesRead(protocol.ByteCount(m)) } s.onData() // so that a possible WINDOW_UPDATE is sent if s.readPosInFrame >= int(frame.DataLen()) { - fin := frame.FinBit - s.mutex.Lock() s.frameQueue.Pop() - s.mutex.Unlock() - if fin { - s.finishedReading.Set(true) + s.finRead = frame.FinBit + if frame.FinBit { return bytesRead, io.EOF } } } - return bytesRead, nil } @@ -206,12 +215,15 @@ func (s *stream) Write(p []byte) (int, error) { s.mutex.Lock() defer s.mutex.Unlock() - if s.resetLocally.Get() || s.err != nil { - return 0, s.err - } - if s.finishedWriting.Get() { + if s.finishedWriting { return 0, fmt.Errorf("write on closed stream %d", s.streamID) } + if s.canceledWrite { + return 0, s.cancelWriteErr + } + if s.closeForShutdownErr != nil { + return 0, s.closeForShutdownErr + } if len(p) == 0 { return 0, nil } @@ -227,7 +239,7 @@ func (s *stream) Write(p []byte) (int, error) { err = errDeadline break } - if s.dataForWriting == nil || s.err != nil { + if s.dataForWriting == nil || s.canceledWrite || s.closedForShutdown { break } @@ -243,23 +255,21 @@ func (s *stream) Write(p []byte) (int, error) { s.mutex.Lock() } - if s.err != nil { - err = s.err + if s.closeForShutdownErr != nil { + err = s.closeForShutdownErr + } else if s.cancelWriteErr != nil { + err = s.cancelWriteErr } return len(p) - len(s.dataForWriting), err } -func (s *stream) GetWriteOffset() protocol.ByteCount { - return s.writeOffset -} - // PopStreamFrame returns the next STREAM frame that is supposed to be sent on this stream // maxBytes is the maximum length this frame (including frame header) will have. func (s *stream) PopStreamFrame(maxBytes protocol.ByteCount) *wire.StreamFrame { s.mutex.Lock() defer s.mutex.Unlock() - if s.err != nil { + if s.closeForShutdownErr != nil { return nil } @@ -277,14 +287,14 @@ func (s *stream) PopStreamFrame(maxBytes protocol.ByteCount) *wire.StreamFrame { return nil } if frame.FinBit { - s.finSent.Set(true) + s.finSent = true } return frame } func (s *stream) getDataForWriting(maxBytes protocol.ByteCount) ([]byte, bool /* should send FIN */) { if s.dataForWriting == nil { - return nil, s.finishedWriting.Get() && !s.finSent.Get() + return nil, s.finishedWriting && !s.finSent } // TODO(#657): Flow control for the crypto stream @@ -306,25 +316,29 @@ func (s *stream) getDataForWriting(maxBytes protocol.ByteCount) ([]byte, bool /* } s.writeOffset += protocol.ByteCount(len(ret)) s.flowController.AddBytesSent(protocol.ByteCount(len(ret))) - return ret, s.finishedWriting.Get() && s.dataForWriting == nil && !s.finSent.Get() + return ret, s.finishedWriting && s.dataForWriting == nil && !s.finSent } -// Close implements io.Closer func (s *stream) Close() error { - s.finishedWriting.Set(true) + s.mutex.Lock() + defer s.mutex.Unlock() + + if s.canceledWrite { + return fmt.Errorf("Close called for canceled stream %d", s.streamID) + } + if s.canceledRead && !s.version.UsesIETFFrameFormat() { + s.queueControlFrame(&wire.RstStreamFrame{ + StreamID: s.streamID, + ByteOffset: s.writeOffset, + ErrorCode: 0, + }) + } + s.finishedWriting = true s.ctxCancel() s.onData() return nil } -func (s *stream) shouldSendReset() bool { - if s.rstSent.Get() { - return false - } - return (s.resetLocally.Get() || s.resetRemotely.Get()) && !s.finishedWriteAndSentFin() -} - -// HandleStreamFrame adds a new stream frame func (s *stream) HandleStreamFrame(frame *wire.StreamFrame) error { maxOffset := frame.Offset + frame.DataLen() if err := s.flowController.UpdateHighestReceived(maxOffset, frame.FinBit); err != nil { @@ -348,7 +362,7 @@ func (s *stream) signalRead() { } } -// signalRead performs a non-blocking send on the writeChan +// signalWrite performs a non-blocking send on the writeChan func (s *stream) signalWrite() { select { case s.writeChan <- struct{}{}: @@ -395,79 +409,114 @@ func (s *stream) CloseRemote(offset protocol.ByteCount) { // The peer will NOT be informed about this: the stream is closed without sending a FIN or RST. func (s *stream) CloseForShutdown(err error) { s.mutex.Lock() - s.closedForShutdown.Set(true) - s.ctxCancel() - // errors must not be changed! - if s.err == nil { - s.err = err - s.signalRead() - s.signalWrite() - } + s.closedForShutdown = true + s.closeForShutdownErr = err s.mutex.Unlock() + s.signalRead() + s.signalWrite() + s.ctxCancel() } -// resets the stream locally -func (s *stream) Reset(err error) { - if s.resetLocally.Get() { - return - } +func (s *stream) CancelWrite(errorCode protocol.ApplicationErrorCode) error { s.mutex.Lock() - s.resetLocally.Set(true) + defer s.mutex.Unlock() + + return s.cancelWriteImpl(errorCode, fmt.Errorf("Write on stream %d canceled with error code %d", s.streamID, errorCode)) +} + +// must be called after locking the mutex +func (s *stream) cancelWriteImpl(errorCode protocol.ApplicationErrorCode, writeErr error) error { + if s.canceledWrite { + return nil + } + if s.finishedWriting { + return fmt.Errorf("CancelWrite for closed stream %d", s.streamID) + } + s.canceledWrite = true + s.cancelWriteErr = writeErr + s.signalWrite() + s.queueControlFrame(&wire.RstStreamFrame{ + StreamID: s.streamID, + ByteOffset: s.writeOffset, + ErrorCode: errorCode, + }) + // TODO(#991): cancel retransmissions for this stream + s.onData() s.ctxCancel() - // errors must not be changed! - if s.err == nil { - s.err = err - s.signalRead() - s.signalWrite() + return nil +} + +func (s *stream) CancelRead(errorCode protocol.ApplicationErrorCode) error { + s.mutex.Lock() + defer s.mutex.Unlock() + + if s.finRead { + return nil } - if s.shouldSendReset() { - s.queueControlFrame(&wire.RstStreamFrame{ - StreamID: s.streamID, - ByteOffset: s.writeOffset, - }) - s.onData() - s.rstSent.Set(true) + if s.canceledRead { + return nil } - s.mutex.Unlock() + s.canceledRead = true + s.cancelReadErr = fmt.Errorf("Read on stream %d canceled with error code %d", s.streamID, errorCode) + s.signalRead() + // TODO(#1034): queue a STOP_SENDING (in IETF QUIC) + return nil } func (s *stream) HandleRstStreamFrame(frame *wire.RstStreamFrame) error { - if s.resetRemotely.Get() { - return nil - } s.mutex.Lock() - s.resetRemotely.Set(true) - s.ctxCancel() - // errors must not be changed! - if s.err == nil { - s.err = fmt.Errorf("RST_STREAM received with code %d", frame.ErrorCode) - s.signalWrite() + defer s.mutex.Unlock() + + if s.closedForShutdown { + return nil } if err := s.flowController.UpdateHighestReceived(frame.ByteOffset, true); err != nil { return err } - if s.shouldSendReset() { - s.queueControlFrame(&wire.RstStreamFrame{ - StreamID: s.streamID, - ByteOffset: s.writeOffset, + if !s.version.UsesIETFFrameFormat() { + s.HandleStopSendingFrame(&wire.StopSendingFrame{ + StreamID: s.streamID, + ErrorCode: frame.ErrorCode, }) - s.onData() - s.rstSent.Set(true) + // In gQUIC, error code 0 has a special meaning. + // The peer will reliably continue transmitting, but is not interested in reading from the stream. + // We should therefore just continue reading from the stream, until we encounter the FIN bit. + if frame.ErrorCode == 0 { + return nil + } } - s.mutex.Unlock() + + // ignore duplicate RST_STREAM frames for this stream (after checking their final offset) + if s.resetRemotely { + return nil + } + s.resetRemotely = true + s.resetRemotelyErr = streamCanceledError{ + errorCode: frame.ErrorCode, + error: fmt.Errorf("Stream %d was reset with error code %d", s.streamID, frame.ErrorCode), + } + s.signalRead() return nil } -func (s *stream) finishedWriteAndSentFin() bool { - return s.finishedWriting.Get() && s.finSent.Get() +func (s *stream) HandleStopSendingFrame(frame *wire.StopSendingFrame) { + // send a RST_STREAM frame + writeErr := streamCanceledError{ + errorCode: frame.ErrorCode, + error: fmt.Errorf("Stream %d was reset with error code %d", s.streamID, frame.ErrorCode), + } + s.cancelWriteImpl(errorCodeStoppingGQUIC, writeErr) } func (s *stream) Finished() bool { - return s.closedForShutdown.Get() || - (s.finishedReading.Get() && s.finishedWriteAndSentFin()) || - (s.resetRemotely.Get() && s.rstSent.Get()) || - (s.finishedReading.Get() && s.rstSent.Get()) || - (s.finishedWriteAndSentFin() && s.resetRemotely.Get()) + s.mutex.Lock() + defer s.mutex.Unlock() + + sendSideClosed := s.finSent || s.canceledWrite + receiveSideClosed := s.finRead || s.resetRemotely + + return s.closedForShutdown || // if the stream was abruptly closed for shutting down + sendSideClosed && receiveSideClosed } func (s *stream) Context() context.Context { diff --git a/stream_test.go b/stream_test.go index fcbfbe7e..251a9d2b 100644 --- a/stream_test.go +++ b/stream_test.go @@ -74,7 +74,7 @@ var _ = Describe("Stream", func() { }) Context("reading", func() { - It("reads a single StreamFrame", func() { + It("reads a single STREAM frame", func() { mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false) mockFC.EXPECT().AddBytesRead(protocol.ByteCount(4)) frame := wire.StreamFrame{ @@ -90,7 +90,7 @@ var _ = Describe("Stream", func() { Expect(b).To(Equal([]byte{0xDE, 0xAD, 0xBE, 0xEF})) }) - It("reads a single StreamFrame in multiple goes", func() { + It("reads a single STREAM frame in multiple goes", func() { mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false) mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2)) mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2)) @@ -268,6 +268,7 @@ var _ = Describe("Stream", func() { It("the deadline error has the right net.Error properties", func() { Expect(errDeadline.Temporary()).To(BeTrue()) Expect(errDeadline.Timeout()).To(BeTrue()) + Expect(errDeadline).To(MatchError("deadline exceeded")) }) It("returns an error when Read is called after the deadline", func() { @@ -477,341 +478,6 @@ var _ = Describe("Stream", func() { }) }) - Context("resetting", func() { - Context("reset by the peer", func() { - It("continues reading after receiving a remote error", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false) - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(10), true) - frame := wire.StreamFrame{ - Offset: 0, - Data: []byte{0xDE, 0xAD, 0xBE, 0xEF}, - } - str.HandleStreamFrame(&frame) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 10, - }) - Expect(err).ToNot(HaveOccurred()) - b := make([]byte, 4) - n, err := strWithTimeout.Read(b) - Expect(err).ToNot(HaveOccurred()) - Expect(n).To(Equal(4)) - }) - - It("reads a delayed STREAM frame that arrives after receiving a remote error", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), true) - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 4, - }) - Expect(err).ToNot(HaveOccurred()) - frame := wire.StreamFrame{ - Offset: 0, - Data: []byte{0xDE, 0xAD, 0xBE, 0xEF}, - } - err = str.HandleStreamFrame(&frame) - Expect(err).ToNot(HaveOccurred()) - b := make([]byte, 4) - n, err := strWithTimeout.Read(b) - Expect(err).ToNot(HaveOccurred()) - Expect(n).To(Equal(4)) - }) - - It("returns the error if reading past the offset of the frame received", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false) - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(8), true) - frame := wire.StreamFrame{ - Offset: 0, - Data: []byte{0xDE, 0xAD, 0xBE, 0xEF}, - } - str.HandleStreamFrame(&frame) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 8, - ErrorCode: 1337, - }) - Expect(err).ToNot(HaveOccurred()) - b := make([]byte, 10) - n, err := strWithTimeout.Read(b) - Expect(b[0:4]).To(Equal(frame.Data)) - Expect(err).To(MatchError("RST_STREAM received with code 1337")) - Expect(n).To(Equal(4)) - }) - - It("returns an EOF when reading past the offset, if the stream received a FIN bit", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), true) - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(8), true) - frame := wire.StreamFrame{ - Offset: 0, - Data: []byte{0xDE, 0xAD, 0xBE, 0xEF}, - FinBit: true, - } - str.HandleStreamFrame(&frame) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 8, - }) - Expect(err).ToNot(HaveOccurred()) - b := make([]byte, 10) - n, err := strWithTimeout.Read(b) - Expect(b[:4]).To(Equal(frame.Data)) - Expect(err).To(MatchError(io.EOF)) - Expect(n).To(Equal(4)) - }) - - It("continues reading in small chunks after receiving a remote error", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), true) - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), true) - frame := wire.StreamFrame{ - Offset: 0, - Data: []byte{0xDE, 0xAD, 0xBE, 0xEF}, - FinBit: true, - } - str.HandleStreamFrame(&frame) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 4, - }) - b := make([]byte, 3) - _, err = strWithTimeout.Read(b) - Expect(err).ToNot(HaveOccurred()) - Expect(b).To(Equal([]byte{0xde, 0xad, 0xbe})) - b = make([]byte, 3) - n, err := strWithTimeout.Read(b) - Expect(err).To(MatchError(io.EOF)) - Expect(b[:1]).To(Equal([]byte{0xef})) - Expect(n).To(Equal(1)) - }) - - It("doesn't inform the flow controller about bytes read after receiving the remote error", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false) - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(10), true) - // No AddBytesRead() - frame := wire.StreamFrame{ - Offset: 0, - StreamID: 5, - Data: []byte{0xDE, 0xAD, 0xBE, 0xEF}, - } - str.HandleStreamFrame(&frame) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 10, - }) - Expect(err).ToNot(HaveOccurred()) - b := make([]byte, 3) - _, err = strWithTimeout.Read(b) - Expect(err).ToNot(HaveOccurred()) - }) - - It("stops writing after receiving a remote error", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(8), true) - done := make(chan struct{}) - go func() { - defer GinkgoRecover() - n, err := strWithTimeout.Write([]byte("foobar")) - Expect(n).To(BeZero()) - Expect(err).To(MatchError("RST_STREAM received with code 1337")) - close(done) - }() - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 8, - ErrorCode: 1337, - }) - Expect(err).ToNot(HaveOccurred()) - Eventually(done).Should(BeClosed()) - }) - - It("returns how much was written when receiving a remote error", func() { - frameHeaderSize := protocol.ByteCount(4) - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(10), true) - mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999)) - mockFC.EXPECT().AddBytesSent(protocol.ByteCount(4)) - done := make(chan struct{}) - go func() { - defer GinkgoRecover() - n, err := strWithTimeout.Write([]byte("foobar")) - Expect(err).To(MatchError("RST_STREAM received with code 1337")) - Expect(n).To(Equal(4)) - close(done) - }() - - var frame *wire.StreamFrame - Eventually(func() *wire.StreamFrame { frame = str.PopStreamFrame(4 + frameHeaderSize); return frame }).ShouldNot(BeNil()) - Expect(frame).ToNot(BeNil()) - Expect(frame.DataLen()).To(BeEquivalentTo(4)) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 10, - ErrorCode: 1337, - }) - Expect(err).ToNot(HaveOccurred()) - Eventually(done).Should(BeClosed()) - }) - - It("calls queues a RST_STREAM frame when receiving a remote error", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(10), true) - done := make(chan struct{}) - str.writeOffset = 0x1000 - go func() { - _, _ = strWithTimeout.Write([]byte("foobar")) - close(done) - }() - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 10, - }) - Expect(err).ToNot(HaveOccurred()) - Expect(queuedControlFrames).To(Equal([]wire.Frame{ - &wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 0x1000, - }, - })) - Eventually(done).Should(BeClosed()) - }) - - It("doesn't call onReset if it already sent a FIN", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(10), true) - str.Close() - f := str.PopStreamFrame(100) - Expect(f.FinBit).To(BeTrue()) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 10, - }) - Expect(err).ToNot(HaveOccurred()) - Expect(queuedControlFrames).To(BeEmpty()) - }) - - It("doesn't queue a RST_STREAM if the stream was reset locally before", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(10), true) - str.Reset(errors.New("reset")) - Expect(queuedControlFrames).To(HaveLen(1)) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 10, - }) - Expect(err).ToNot(HaveOccurred()) - Expect(queuedControlFrames).To(HaveLen(1)) // no additional queued frame - }) - - It("doesn't queue two RST_STREAMs twice, when it gets two remote errors", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(8), true) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 8, - }) - Expect(err).ToNot(HaveOccurred()) - Expect(queuedControlFrames).To(HaveLen(1)) - err = str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 9, - }) - Expect(err).ToNot(HaveOccurred()) - Expect(queuedControlFrames).To(HaveLen(1)) - }) - }) - - Context("reset locally", func() { - testErr := errors.New("test error") - - It("stops writing", func() { - done := make(chan struct{}) - go func() { - defer GinkgoRecover() - n, err := strWithTimeout.Write([]byte("foobar")) - Expect(n).To(BeZero()) - Expect(err).To(MatchError(testErr)) - close(done) - }() - Consistently(done).ShouldNot(BeClosed()) - str.Reset(testErr) - Expect(str.PopStreamFrame(1000)).To(BeNil()) - Eventually(done).Should(BeClosed()) - }) - - It("doesn't allow further writes", func() { - str.Reset(testErr) - n, err := strWithTimeout.Write([]byte("foobar")) - Expect(n).To(BeZero()) - Expect(err).To(MatchError(testErr)) - Expect(str.PopStreamFrame(1000)).To(BeNil()) - }) - - It("stops reading", func() { - done := make(chan struct{}) - go func() { - defer GinkgoRecover() - b := make([]byte, 4) - n, err := strWithTimeout.Read(b) - Expect(n).To(BeZero()) - Expect(err).To(MatchError(testErr)) - close(done) - }() - Consistently(done).ShouldNot(BeClosed()) - str.Reset(testErr) - Eventually(done).Should(BeClosed()) - }) - - It("doesn't allow further reads", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(6), false) - str.HandleStreamFrame(&wire.StreamFrame{ - Data: []byte("foobar"), - }) - str.Reset(testErr) - b := make([]byte, 6) - n, err := strWithTimeout.Read(b) - Expect(n).To(BeZero()) - Expect(err).To(MatchError(testErr)) - }) - - It("queues a RST_STREAM frame", func() { - str.writeOffset = 0x1000 - str.Reset(testErr) - Expect(queuedControlFrames).To(Equal([]wire.Frame{ - &wire.RstStreamFrame{ - StreamID: 1337, - ByteOffset: 0x1000, - }, - })) - }) - - It("doesn't queue a RST_STREAM if it already sent a FIN", func() { - str.Close() - f := str.PopStreamFrame(1000) - Expect(f.FinBit).To(BeTrue()) - str.Reset(testErr) - Expect(queuedControlFrames).To(BeEmpty()) - }) - - It("doesn't queue a new RST_STREAM, if the stream was reset remotely before", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(10), true) - err := str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 10, - }) - Expect(err).ToNot(HaveOccurred()) - str.Reset(testErr) - Expect(queuedControlFrames).To(HaveLen(1)) - }) - - It("doesn't call onReset twice", func() { - str.Reset(testErr) - Expect(queuedControlFrames).To(HaveLen(1)) - str.Reset(testErr) - Expect(queuedControlFrames).To(HaveLen(1)) // no additional queued frame - }) - - It("cancels the context", func() { - Expect(str.Context().Done()).ToNot(BeClosed()) - str.Reset(testErr) - Expect(str.Context().Done()).To(BeClosed()) - }) - }) - }) - Context("writing", func() { It("writes and gets all data at once", func() { mockFC.EXPECT().SendWindowSize().Return(protocol.ByteCount(9999)) @@ -910,6 +576,12 @@ var _ = Describe("Stream", func() { Expect(err).ToNot(HaveOccurred()) }) + It("cancels the context when Close is called", func() { + Expect(str.Context().Done()).ToNot(BeClosed()) + str.Close() + Expect(str.Context().Done()).To(BeClosed()) + }) + Context("deadlines", func() { It("returns an error when Write is called after the deadline", func() { str.SetWriteDeadline(time.Now().Add(-time.Second)) @@ -1069,18 +741,329 @@ var _ = Describe("Stream", func() { }) }) - It("errors when a StreamFrames causes a flow control violation", func() { - testErr := errors.New("flow control violation") - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(8), false).Return(testErr) - frame := wire.StreamFrame{ - Offset: 2, - Data: []byte("foobar"), - } - err := str.HandleStreamFrame(&frame) - Expect(err).To(MatchError(testErr)) + Context("stream cancelations", func() { + Context("canceling writing", func() { + It("queues a RST_STREAM frame", func() { + str.writeOffset = 1234 + err := str.CancelWrite(9876) + Expect(err).ToNot(HaveOccurred()) + Expect(queuedControlFrames).To(Equal([]wire.Frame{ + &wire.RstStreamFrame{ + StreamID: streamID, + ByteOffset: 1234, + ErrorCode: 9876, + }, + })) + }) + + It("unblocks Write", func() { + 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) + }() + var frame *wire.StreamFrame + Eventually(func() *wire.StreamFrame { + defer GinkgoRecover() + frame = str.PopStreamFrame(50) + return frame + }).ShouldNot(BeNil()) + err := str.CancelWrite(1234) + Expect(err).ToNot(HaveOccurred()) + Eventually(writeReturned).Should(BeClosed()) + Expect(n).To(BeEquivalentTo(frame.DataLen())) + }) + + It("cancels the context", func() { + Expect(str.Context().Done()).ToNot(BeClosed()) + str.CancelWrite(1234) + Expect(str.Context().Done()).To(BeClosed()) + }) + + It("doesn't allow further calls to Write", func() { + err := str.CancelWrite(1234) + Expect(err).ToNot(HaveOccurred()) + _, err = strWithTimeout.Write([]byte("foobar")) + Expect(err).To(MatchError("Write on stream 1337 canceled with error code 1234")) + }) + + It("only cancels once", func() { + err := str.CancelWrite(1234) + Expect(err).ToNot(HaveOccurred()) + Expect(queuedControlFrames).To(HaveLen(1)) + err = str.CancelWrite(4321) + Expect(err).ToNot(HaveOccurred()) + Expect(queuedControlFrames).To(HaveLen(1)) + }) + + It("doesn't cancel when the stream was already closed", func() { + err := str.Close() + Expect(err).ToNot(HaveOccurred()) + err = str.CancelWrite(123) + Expect(err).To(MatchError("CancelWrite for closed stream 1337")) + }) + }) + + Context("canceling read", func() { + It("unblocks Read", func() { + done := make(chan struct{}) + go func() { + defer GinkgoRecover() + _, err := strWithTimeout.Read([]byte{0}) + Expect(err).To(MatchError("Read on stream 1337 canceled with error code 1234")) + close(done) + }() + Consistently(done).ShouldNot(BeClosed()) + err := str.CancelRead(1234) + Expect(err).ToNot(HaveOccurred()) + Eventually(done).Should(BeClosed()) + }) + + It("doesn't allow further calls to Read", func() { + err := str.CancelRead(1234) + Expect(err).ToNot(HaveOccurred()) + _, err = strWithTimeout.Read([]byte{0}) + Expect(err).To(MatchError("Read on stream 1337 canceled with error code 1234")) + }) + + It("does nothing when CancelRead is called twice", func() { + err := str.CancelRead(1234) + Expect(err).ToNot(HaveOccurred()) + err = str.CancelRead(2345) + Expect(err).ToNot(HaveOccurred()) + _, err = strWithTimeout.Read([]byte{0}) + Expect(err).To(MatchError("Read on stream 1337 canceled with error code 1234")) + }) + + Context("for gQUIC", func() { + It("sends a RST_STREAM with error code 0, after the stream is closed", func() { + str.version = versionGQUICFrames + mockFC.EXPECT().SendWindowSize().Return(protocol.MaxByteCount).AnyTimes() + mockFC.EXPECT().AddBytesSent(protocol.ByteCount(6)) + err := str.CancelRead(1234) + Expect(err).ToNot(HaveOccurred()) + Expect(queuedControlFrames).To(BeEmpty()) // no RST_STREAM frame queued yet + writeReturned := make(chan struct{}) + go func() { + defer GinkgoRecover() + _, err := strWithTimeout.Write([]byte("foobar")) + Expect(err).ToNot(HaveOccurred()) + close(writeReturned) + }() + Eventually(func() *wire.StreamFrame { return str.PopStreamFrame(1000) }).ShouldNot(BeNil()) + Eventually(writeReturned).Should(BeClosed()) + Expect(queuedControlFrames).To(BeEmpty()) // no RST_STREAM frame queued yet + err = str.Close() + Expect(err).ToNot(HaveOccurred()) + Expect(queuedControlFrames).To(Equal([]wire.Frame{ + &wire.RstStreamFrame{ + StreamID: streamID, + ByteOffset: 6, + ErrorCode: 0, + }, + })) + }) + + It("doesn't send a RST_STREAM frame, if the FIN was already read", func() { + mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(6), true) + mockFC.EXPECT().AddBytesRead(protocol.ByteCount(6)) + err := str.HandleStreamFrame(&wire.StreamFrame{ + StreamID: streamID, + Data: []byte("foobar"), + FinBit: true, + }) + Expect(err).ToNot(HaveOccurred()) + _, err = strWithTimeout.Read(make([]byte, 100)) + Expect(err).To(MatchError(io.EOF)) + err = str.CancelRead(1234) + Expect(err).ToNot(HaveOccurred()) + Expect(queuedControlFrames).To(BeEmpty()) // no RST_STREAM frame queued yet + }) + }) + }) + + Context("receiving RST_STREAM frames", func() { + rst := &wire.RstStreamFrame{ + StreamID: streamID, + ByteOffset: 42, + ErrorCode: 1234, + } + + It("unblocks Read", func() { + mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(42), true) + done := make(chan struct{}) + go func() { + defer GinkgoRecover() + _, err := strWithTimeout.Read([]byte{0}) + Expect(err).To(MatchError("Stream 1337 was reset with error code 1234")) + Expect(err).To(BeAssignableToTypeOf(streamCanceledError{})) + Expect(err.(streamCanceledError).Canceled()).To(BeTrue()) + Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(1234))) + close(done) + }() + Consistently(done).ShouldNot(BeClosed()) + str.HandleRstStreamFrame(rst) + Eventually(done).Should(BeClosed()) + }) + + It("doesn't allow further calls to Read", func() { + mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(42), true) + err := str.HandleRstStreamFrame(rst) + Expect(err).ToNot(HaveOccurred()) + _, err = strWithTimeout.Read([]byte{0}) + Expect(err).To(MatchError("Stream 1337 was reset with error code 1234")) + Expect(err).To(BeAssignableToTypeOf(streamCanceledError{})) + Expect(err.(streamCanceledError).Canceled()).To(BeTrue()) + Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(1234))) + }) + + It("errors when receiving a RST_STREAM with an inconsistent offset", func() { + testErr := errors.New("already received a different final offset before") + mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(42), true).Return(testErr) + err := str.HandleRstStreamFrame(rst) + Expect(err).To(MatchError(testErr)) + }) + + It("ignores duplicate RST_STREAM frames", func() { + mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(42), true).Times(2) + err := str.HandleRstStreamFrame(rst) + Expect(err).ToNot(HaveOccurred()) + err = str.HandleRstStreamFrame(rst) + Expect(err).ToNot(HaveOccurred()) + }) + + It("doesn't do anyting when it was closed for shutdown", func() { + str.CloseForShutdown(nil) + err := str.HandleRstStreamFrame(rst) + Expect(err).ToNot(HaveOccurred()) + }) + + Context("for gQUIC", func() { + BeforeEach(func() { + str.version = versionGQUICFrames + }) + + It("unblocks Read when receiving a RST_STREAM frame with non-zero error code", func() { + mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(42), true) + readReturned := make(chan struct{}) + go func() { + defer GinkgoRecover() + _, err := strWithTimeout.Read([]byte{0}) + Expect(err).To(MatchError("Stream 1337 was reset with error code 1234")) + Expect(err).To(BeAssignableToTypeOf(streamCanceledError{})) + Expect(err.(streamCanceledError).Canceled()).To(BeTrue()) + Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(1234))) + close(readReturned) + }() + Consistently(readReturned).ShouldNot(BeClosed()) + err := str.HandleRstStreamFrame(rst) + Expect(err).ToNot(HaveOccurred()) + Eventually(readReturned).Should(BeClosed()) + }) + + It("unblocks Write when receiving a RST_STREAM frame with non-zero error code", func() { + mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(6), true) + str.writeOffset = 1000 + f := &wire.RstStreamFrame{ + StreamID: streamID, + ByteOffset: 6, + ErrorCode: 123, + } + writeReturned := make(chan struct{}) + go func() { + defer GinkgoRecover() + _, err := strWithTimeout.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(writeReturned) + }() + Consistently(writeReturned).ShouldNot(BeClosed()) + err := str.HandleRstStreamFrame(f) + Expect(err).ToNot(HaveOccurred()) + Expect(queuedControlFrames).To(Equal([]wire.Frame{ + &wire.RstStreamFrame{ + StreamID: streamID, + ByteOffset: 1000, + ErrorCode: errorCodeStoppingGQUIC, + }, + })) + Eventually(writeReturned).Should(BeClosed()) + }) + + It("sends a RST_STREAM and continues reading until the end when receiving a RST_STREAM frame with error code 0", func() { + mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(6), true).Times(2) + mockFC.EXPECT().AddBytesRead(protocol.ByteCount(4)) + mockFC.EXPECT().AddBytesRead(protocol.ByteCount(2)) + readReturned := make(chan struct{}) + go func() { + defer GinkgoRecover() + n, err := strWithTimeout.Read(make([]byte, 4)) + Expect(err).ToNot(HaveOccurred()) + Expect(n).To(Equal(4)) + n, err = strWithTimeout.Read(make([]byte, 4)) + Expect(err).To(MatchError(io.EOF)) + Expect(n).To(Equal(2)) + close(readReturned) + }() + Consistently(readReturned).ShouldNot(BeClosed()) + err := str.HandleStreamFrame(&wire.StreamFrame{ + StreamID: streamID, + Data: []byte("foobar"), + FinBit: true, + }) + Expect(err).ToNot(HaveOccurred()) + err = str.HandleRstStreamFrame(&wire.RstStreamFrame{ + StreamID: streamID, + ByteOffset: 6, + ErrorCode: 0, + }) + Expect(err).ToNot(HaveOccurred()) + Eventually(readReturned).Should(BeClosed()) + }) + + It("unblocks Write when receiving a RST_STREAM frame with error code 0", func() { + mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(6), true) + str.writeOffset = 1000 + f := &wire.RstStreamFrame{ + StreamID: streamID, + ByteOffset: 6, + ErrorCode: 0, + } + writeReturned := make(chan struct{}) + go func() { + defer GinkgoRecover() + _, err := strWithTimeout.Write([]byte("foobar")) + Expect(err).To(MatchError("Stream 1337 was reset with error code 0")) + Expect(err).To(BeAssignableToTypeOf(streamCanceledError{})) + Expect(err.(streamCanceledError).Canceled()).To(BeTrue()) + Expect(err.(streamCanceledError).ErrorCode()).To(Equal(protocol.ApplicationErrorCode(0))) + close(writeReturned) + }() + Consistently(writeReturned).ShouldNot(BeClosed()) + err := str.HandleRstStreamFrame(f) + Expect(err).ToNot(HaveOccurred()) + Expect(queuedControlFrames).To(Equal([]wire.Frame{ + &wire.RstStreamFrame{ + StreamID: streamID, + ByteOffset: 1000, + ErrorCode: errorCodeStoppingGQUIC, + }, + })) + Eventually(writeReturned).Should(BeClosed()) + }) + }) + }) }) - Context("closing", func() { + Context("saying if it is finished", func() { testErr := errors.New("testErr") finishReading := func() { @@ -1092,7 +1075,7 @@ var _ = Describe("Stream", func() { Expect(err).To(MatchError(io.EOF)) } - It("is finished after it is canceled", func() { + It("is finished after it is closed for shutdown", func() { str.CloseForShutdown(testErr) Expect(str.Finished()).To(BeTrue()) }) @@ -1116,38 +1099,6 @@ var _ = Describe("Stream", func() { Expect(str.Finished()).To(BeFalse()) }) - It("is finished after receiving a RST and sending one", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(0), true) - // this directly sends a rst - str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 0, - }) - Expect(str.rstSent.Get()).To(BeTrue()) - Expect(str.Finished()).To(BeTrue()) - }) - - It("cancels the context after receiving a RST", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(0), true) - Expect(str.Context().Done()).ToNot(BeClosed()) - str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 0, - }) - Expect(str.Context().Done()).To(BeClosed()) - }) - - It("is finished after being locally reset and receiving a RST in response", func() { - mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(13), true) - str.Reset(testErr) - Expect(str.Finished()).To(BeFalse()) - str.HandleRstStreamFrame(&wire.RstStreamFrame{ - StreamID: streamID, - ByteOffset: 13, - }) - Expect(str.Finished()).To(BeTrue()) - }) - It("is finished after finishing writing and receiving a RST", func() { mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(13), true) str.Close() @@ -1159,17 +1110,20 @@ var _ = Describe("Stream", func() { }) Expect(str.Finished()).To(BeTrue()) }) - - It("is finished after finishing reading and being locally reset", func() { - mockFC.EXPECT().AddBytesRead(protocol.ByteCount(0)) - finishReading() - Expect(str.Finished()).To(BeFalse()) - str.Reset(testErr) - Expect(str.Finished()).To(BeTrue()) - }) }) Context("flow control", func() { + It("errors when a STREAM frame causes a flow control violation", func() { + testErr := errors.New("flow control violation") + mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(8), false).Return(testErr) + frame := wire.StreamFrame{ + Offset: 2, + Data: []byte("foobar"), + } + err := str.HandleStreamFrame(&frame) + Expect(err).To(MatchError(testErr)) + }) + It("says when it's flow control blocked", func() { mockFC.EXPECT().IsBlocked().Return(false, protocol.ByteCount(0)) blocked, _ := str.IsFlowControlBlocked()