send a RstStreamFrame when receiving a RstStreamFrame on an open stream

fixes #378

session.go

@@ -317,7 +317,6 @@ func (s *Session) handleFrames(fs []frames.Frame) error {
 		switch frame := ff.(type) {
 		case *frames.StreamFrame:
 			err = s.handleStreamFrame(frame)
-			// TODO: send RstStreamFrame
 		case *frames.AckFrame:
 			err = s.handleAckFrame(frame)
 		case *frames.ConnectionCloseFrame:
@@ -369,7 +368,8 @@ func (s *Session) handleStreamFrame(frame *frames.StreamFrame) error {
 		return err
 	}
 	if str == nil {
-		// Stream is closed, ignore
+		// Stream is closed and already garbage collected
+		// ignore this StreamFrame
 		return nil
 	}
 	err = str.AddStreamFrame(frame)
@@ -401,11 +401,20 @@ func (s *Session) handleRstStreamFrame(frame *frames.RstStreamFrame) error {
 	if str == nil {
 		return errRstStreamOnInvalidStream
 	}
+
+	shouldSendRst := !str.finishedWriting()
 	s.closeStreamWithError(str, fmt.Errorf("RST_STREAM received with code %d", frame.ErrorCode))
-	_, err = s.flowControlManager.ResetStream(frame.StreamID, frame.ByteOffset)
+	bytesSent, err := s.flowControlManager.ResetStream(frame.StreamID, frame.ByteOffset)
 	if err != nil {
 		return err
 	}
+
+	if shouldSendRst {
+		s.packer.QueueControlFrameForNextPacket(&frames.RstStreamFrame{
+			StreamID:   frame.StreamID,
+			ByteOffset: bytesSent,
+		})
+	}
 	return nil
 }
 

session_test.go

@@ -317,6 +317,33 @@ var _ = Describe("Session", func() {
 			Expect(err).To(MatchError("RST_STREAM received with code 42"))
 		})
 
+		It("queues a RST_STERAM frame with the correct offset", func() {
+			_, err := session.GetOrOpenStream(5)
+			Expect(err).ToNot(HaveOccurred())
+			session.flowControlManager = newMockFlowControlHandler()
+			session.flowControlManager.(*mockFlowControlHandler).bytesSent = 0x1337
+			err = session.handleRstStreamFrame(&frames.RstStreamFrame{
+				StreamID: 5,
+			})
+			Expect(err).ToNot(HaveOccurred())
+			Expect(session.packer.controlFrames).To(HaveLen(1))
+			Expect(session.packer.controlFrames[0].(*frames.RstStreamFrame)).To(Equal(&frames.RstStreamFrame{
+				StreamID:   5,
+				ByteOffset: 0x1337,
+			}))
+		})
+
+		It("doesn't queue a RST_STREAM for a stream that it already sent a FIN on", func() {
+			str, err := session.GetOrOpenStream(5)
+			str.(*stream).sentFin()
+			str.Close()
+			err = session.handleRstStreamFrame(&frames.RstStreamFrame{
+				StreamID: 5,
+			})
+			Expect(err).ToNot(HaveOccurred())
+			Expect(session.packer.controlFrames).To(BeEmpty())
+		})
+
 		It("passes the byte offset to the flow controller", func() {
 			session.streamsMap.GetOrOpenStream(5)
 			session.flowControlManager = newMockFlowControlHandler()

stream_test.go

@@ -62,7 +62,7 @@ func (m *mockFlowControlHandler) AddBytesRead(streamID protocol.StreamID, n prot
 }
 
 func (m *mockFlowControlHandler) ResetStream(streamID protocol.StreamID, byteOffset protocol.ByteCount) (protocol.ByteCount, error) {
-	return 0, m.UpdateHighestReceived(streamID, byteOffset)
+	return m.bytesSent, m.UpdateHighestReceived(streamID, byteOffset)
 }
 
 func (m *mockFlowControlHandler) UpdateHighestReceived(streamID protocol.StreamID, byteOffset protocol.ByteCount) error {