remove queued StreamFrames when a Stream is closed with an error

fixes #149

session_test.go

@@ -187,6 +187,22 @@ var _ = Describe("Session", func() {
 			Expect(session.streams[5]).To(BeNil())
 		})
 
+		It("removes queued StreamFrames from StreamFrameQueue when closing with an error", func() {
+			testErr := errors.New("test")
+			session.handleStreamFrame(&frames.StreamFrame{
+				StreamID: 5,
+				Data:     []byte{0xde, 0xca, 0xfb, 0xad},
+			})
+			f := frames.StreamFrame{
+				StreamID: 5,
+				Data:     []byte("foobar"),
+			}
+			session.streamFrameQueue.Push(&f, true)
+			Expect(session.streams[5]).ToNot(BeNil())
+			session.closeStreamsWithError(testErr)
+			Expect(session.streamFrameQueue.Pop(1000)).To(BeNil())
+		})
+
 		It("removes closed streams from BlockedManager", func() {
 			session.handleStreamFrame(&frames.StreamFrame{
 				StreamID: 5,
@@ -260,6 +276,22 @@ var _ = Describe("Session", func() {
 			Expect(err).To(MatchError("RST_STREAM received with code 42"))
 		})
 
+		It("deletes queued StreamFrames from the StreamFrameQueue", func() {
+			_, err := session.OpenStream(5)
+			Expect(err).ToNot(HaveOccurred())
+			f := frames.StreamFrame{
+				StreamID: 5,
+				Data:     []byte("foobar"),
+			}
+			session.streamFrameQueue.Push(&f, false)
+			err = session.handleRstStreamFrame(&frames.RstStreamFrame{
+				StreamID:  5,
+				ErrorCode: 42,
+			})
+			Expect(err).ToNot(HaveOccurred())
+			Expect(session.streamFrameQueue.Pop(1000)).To(BeNil())
+		})
+
 		It("errors when the stream is not known", func() {
 			err := session.handleRstStreamFrame(&frames.RstStreamFrame{
 				StreamID:  5,