package quic

import (
	"bytes"
	"errors"
	"io"
	"net"
	"reflect"
	"runtime"
	"sync/atomic"
	"time"
	"unsafe"

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

	"github.com/lucas-clemente/quic-go/ackhandler"
	"github.com/lucas-clemente/quic-go/crypto"
	"github.com/lucas-clemente/quic-go/frames"
	"github.com/lucas-clemente/quic-go/handshake"
	"github.com/lucas-clemente/quic-go/protocol"
	"github.com/lucas-clemente/quic-go/qerr"
	"github.com/lucas-clemente/quic-go/testdata"
	"github.com/lucas-clemente/quic-go/utils"
)

type mockConnection struct {
	remoteAddr net.Addr
	written    [][]byte
}

func (m *mockConnection) Write(p []byte) error {
	b := make([]byte, len(p))
	copy(b, p)
	m.written = append(m.written, b)
	return nil
}
func (m *mockConnection) Read([]byte) (int, net.Addr, error) { panic("not implemented") }

func (m *mockConnection) SetCurrentRemoteAddr(addr net.Addr) {
	m.remoteAddr = addr
}
func (*mockConnection) RemoteAddr() net.Addr { return &net.UDPAddr{} }
func (*mockConnection) Close() error         { panic("not implemented") }

type mockUnpacker struct {
	unpackErr error
}

func (m *mockUnpacker) Unpack(publicHeaderBinary []byte, hdr *PublicHeader, data []byte) (*unpackedPacket, error) {
	if m.unpackErr != nil {
		return nil, m.unpackErr
	}
	return &unpackedPacket{
		frames: nil,
	}, nil
}

type mockSentPacketHandler struct {
	retransmissionQueue  []*ackhandler.Packet
	sentPackets          []*ackhandler.Packet
	congestionLimited    bool
	maybeQueueRTOsCalled bool
	requestedStopWaiting bool
}

func (h *mockSentPacketHandler) SentPacket(packet *ackhandler.Packet) error {
	h.sentPackets = append(h.sentPackets, packet)
	return nil
}
func (h *mockSentPacketHandler) ReceivedAck(ackFrame *frames.AckFrame, withPacketNumber protocol.PacketNumber, recvTime time.Time) error {
	return nil
}
func (h *mockSentPacketHandler) BytesInFlight() protocol.ByteCount      { return 0 }
func (h *mockSentPacketHandler) GetLeastUnacked() protocol.PacketNumber { return 1 }
func (h *mockSentPacketHandler) GetStopWaitingFrame(force bool) *frames.StopWaitingFrame {
	h.requestedStopWaiting = true
	return &frames.StopWaitingFrame{LeastUnacked: 0x1337}
}
func (h *mockSentPacketHandler) SendingAllowed() bool      { return !h.congestionLimited }
func (h *mockSentPacketHandler) CheckForError() error      { return nil }
func (h *mockSentPacketHandler) TimeOfFirstRTO() time.Time { panic("not implemented") }

func (h *mockSentPacketHandler) MaybeQueueRTOs() {
	h.maybeQueueRTOsCalled = true
}

func (h *mockSentPacketHandler) DequeuePacketForRetransmission() *ackhandler.Packet {
	if len(h.retransmissionQueue) > 0 {
		packet := h.retransmissionQueue[0]
		h.retransmissionQueue = h.retransmissionQueue[1:]
		return packet
	}
	return nil
}

func newMockSentPacketHandler() ackhandler.SentPacketHandler {
	return &mockSentPacketHandler{}
}

var _ ackhandler.SentPacketHandler = &mockSentPacketHandler{}

type mockReceivedPacketHandler struct {
	nextAckFrame *frames.AckFrame
}

func (m *mockReceivedPacketHandler) GetAckFrame() *frames.AckFrame { return m.nextAckFrame }
func (m *mockReceivedPacketHandler) ReceivedPacket(packetNumber protocol.PacketNumber, shouldInstigateAck bool) error {
	panic("not implemented")
}
func (m *mockReceivedPacketHandler) ReceivedStopWaiting(*frames.StopWaitingFrame) error {
	panic("not implemented")
}

var _ ackhandler.ReceivedPacketHandler = &mockReceivedPacketHandler{}

var _ = Describe("Session", func() {
	var (
		sess                *session
		clientSess          *session
		closeCallbackCalled bool
		scfg                *handshake.ServerConfig
		mconn               *mockConnection
		cpm                 *mockConnectionParametersManager
	)

	BeforeEach(func() {
		mconn = &mockConnection{}
		closeCallbackCalled = false

		certChain := crypto.NewCertChain(testdata.GetTLSConfig())
		kex, err := crypto.NewCurve25519KEX()
		Expect(err).NotTo(HaveOccurred())
		scfg, err = handshake.NewServerConfig(kex, certChain)
		Expect(err).NotTo(HaveOccurred())
		pSess, err := newSession(
			mconn,
			protocol.Version35,
			0,
			scfg,
			func(protocol.ConnectionID) { closeCallbackCalled = true },
		)
		Expect(err).NotTo(HaveOccurred())
		sess = pSess.(*session)
		Expect(sess.streamsMap.openStreams).To(HaveLen(1)) // Crypto stream

		cpm = &mockConnectionParametersManager{idleTime: 60 * time.Second}
		sess.connectionParameters = cpm

		clientSess, err = newClientSession(
			&net.UDPConn{},
			&net.UDPAddr{},
			"hostname",
			protocol.Version35,
			0,
			nil,
			func(protocol.ConnectionID) { closeCallbackCalled = true },
			func(isForwardSecure bool) {},
			nil,
		)
		Expect(err).ToNot(HaveOccurred())
		Expect(clientSess.streamsMap.openStreams).To(HaveLen(1)) // Crypto stream
	})

	Context("source address", func() {
		It("uses the IP address if given an UDP connection", func() {
			conn := &conn{currentAddr: &net.UDPAddr{IP: net.IPv4(192, 168, 100, 200)[12:], Port: 1337}}
			sess, err := newSession(
				conn,
				protocol.VersionWhatever,
				0,
				scfg,
				func(protocol.ConnectionID) { closeCallbackCalled = true },
			)
			Expect(err).ToNot(HaveOccurred())
			Expect(*(*[]byte)(unsafe.Pointer(reflect.ValueOf(sess.(*session).cryptoSetup).Elem().FieldByName("sourceAddr").UnsafeAddr()))).To(Equal([]byte{192, 168, 100, 200}))
		})

		It("uses the string representation of the remote addresses if not given a UDP connection", func() {
			conn := &conn{
				currentAddr: &net.TCPAddr{IP: net.IPv4(192, 168, 100, 200)[12:], Port: 1337},
			}
			sess, err := newSession(
				conn,
				protocol.VersionWhatever,
				0,
				scfg,
				func(protocol.ConnectionID) { closeCallbackCalled = true },
			)
			Expect(err).ToNot(HaveOccurred())
			Expect(*(*[]byte)(unsafe.Pointer(reflect.ValueOf(sess.(*session).cryptoSetup).Elem().FieldByName("sourceAddr").UnsafeAddr()))).To(Equal([]byte("192.168.100.200:1337")))
		})
	})

	Context("when handling stream frames", func() {
		It("makes new streams", func() {
			sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 5,
				Data:     []byte{0xde, 0xca, 0xfb, 0xad},
			})
			p := make([]byte, 4)
			str, err := sess.streamsMap.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			Expect(str).ToNot(BeNil())
			_, err = str.Read(p)
			Expect(err).ToNot(HaveOccurred())
			Expect(p).To(Equal([]byte{0xde, 0xca, 0xfb, 0xad}))
		})

		It("does not reject existing streams with even StreamIDs", func() {
			_, err := sess.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			err = sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 5,
				Data:     []byte{0xde, 0xca, 0xfb, 0xad},
			})
			Expect(err).ToNot(HaveOccurred())
		})

		It("handles existing streams", func() {
			sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 5,
				Data:     []byte{0xde, 0xca},
			})
			numOpenStreams := len(sess.streamsMap.openStreams)
			sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 5,
				Offset:   2,
				Data:     []byte{0xfb, 0xad},
			})
			Expect(sess.streamsMap.openStreams).To(HaveLen(numOpenStreams))
			p := make([]byte, 4)
			str, _ := sess.streamsMap.GetOrOpenStream(5)
			Expect(str).ToNot(BeNil())
			_, err := str.Read(p)
			Expect(err).ToNot(HaveOccurred())
			Expect(p).To(Equal([]byte{0xde, 0xca, 0xfb, 0xad}))
		})

		It("does not delete streams with Close()", func() {
			str, err := sess.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			str.Close()
			sess.garbageCollectStreams()
			str, err = sess.streamsMap.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			Expect(str).ToNot(BeNil())
		})

		It("does not delete streams with FIN bit", func() {
			sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 5,
				Data:     []byte{0xde, 0xca, 0xfb, 0xad},
				FinBit:   true,
			})
			numOpenStreams := len(sess.streamsMap.openStreams)
			str, _ := sess.streamsMap.GetOrOpenStream(5)
			Expect(str).ToNot(BeNil())
			p := make([]byte, 4)
			_, err := str.Read(p)
			Expect(err).To(MatchError(io.EOF))
			Expect(p).To(Equal([]byte{0xde, 0xca, 0xfb, 0xad}))
			sess.garbageCollectStreams()
			Expect(sess.streamsMap.openStreams).To(HaveLen(numOpenStreams))
			str, _ = sess.streamsMap.GetOrOpenStream(5)
			Expect(str).ToNot(BeNil())
		})

		It("deletes streams with FIN bit & close", func() {
			sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 5,
				Data:     []byte{0xde, 0xca, 0xfb, 0xad},
				FinBit:   true,
			})
			numOpenStreams := len(sess.streamsMap.openStreams)
			str, _ := sess.streamsMap.GetOrOpenStream(5)
			Expect(str).ToNot(BeNil())
			p := make([]byte, 4)
			_, err := str.Read(p)
			Expect(err).To(MatchError(io.EOF))
			Expect(p).To(Equal([]byte{0xde, 0xca, 0xfb, 0xad}))
			sess.garbageCollectStreams()
			Expect(sess.streamsMap.openStreams).To(HaveLen(numOpenStreams))
			str, _ = sess.streamsMap.GetOrOpenStream(5)
			Expect(str).ToNot(BeNil())
			// We still need to close the stream locally
			str.Close()
			// ... and simulate that we actually the FIN
			str.sentFin()
			sess.garbageCollectStreams()
			Expect(len(sess.streamsMap.openStreams)).To(BeNumerically("<", numOpenStreams))
			str, err = sess.streamsMap.GetOrOpenStream(5)
			Expect(err).NotTo(HaveOccurred())
			Expect(str).To(BeNil())
			// flow controller should have been notified
			_, err = sess.flowControlManager.SendWindowSize(5)
			Expect(err).To(MatchError("Error accessing the flowController map."))
		})

		It("cancels streams with error", func() {
			sess.garbageCollectStreams()
			testErr := errors.New("test")
			sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 5,
				Data:     []byte{0xde, 0xca, 0xfb, 0xad},
			})
			str, err := sess.streamsMap.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			Expect(str).ToNot(BeNil())
			p := make([]byte, 4)
			_, err = str.Read(p)
			Expect(err).ToNot(HaveOccurred())
			sess.closeStreamsWithError(testErr)
			_, err = str.Read(p)
			Expect(err).To(MatchError(testErr))
			sess.garbageCollectStreams()
			str, err = sess.streamsMap.GetOrOpenStream(5)
			Expect(err).NotTo(HaveOccurred())
			Expect(str).To(BeNil())
		})

		It("cancels empty streams with error", func() {
			testErr := errors.New("test")
			sess.GetOrOpenStream(5)
			str, err := sess.streamsMap.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			Expect(str).ToNot(BeNil())
			sess.closeStreamsWithError(testErr)
			_, err = str.Read([]byte{0})
			Expect(err).To(MatchError(testErr))
			sess.garbageCollectStreams()
			str, err = sess.streamsMap.GetOrOpenStream(5)
			Expect(err).NotTo(HaveOccurred())
			Expect(str).To(BeNil())
		})

		It("informs the FlowControlManager about new streams", func() {
			// since the stream doesn't yet exist, this will throw an error
			err := sess.flowControlManager.UpdateHighestReceived(5, 1000)
			Expect(err).To(HaveOccurred())
			sess.GetOrOpenStream(5)
			err = sess.flowControlManager.UpdateHighestReceived(5, 2000)
			Expect(err).ToNot(HaveOccurred())
		})

		It("ignores streams that existed previously", func() {
			sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 5,
				Data:     []byte{},
				FinBit:   true,
			})
			str, _ := sess.streamsMap.GetOrOpenStream(5)
			Expect(str).ToNot(BeNil())
			_, err := str.Read([]byte{0})
			Expect(err).To(MatchError(io.EOF))
			str.Close()
			str.sentFin()
			sess.garbageCollectStreams()
			err = sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 5,
				Data:     []byte{},
			})
			Expect(err).To(BeNil())
		})
	})

	Context("handling RST_STREAM frames", func() {
		It("closes the streams for writing", func() {
			s, err := sess.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			err = sess.handleRstStreamFrame(&frames.RstStreamFrame{
				StreamID:  5,
				ErrorCode: 42,
			})
			Expect(err).ToNot(HaveOccurred())
			n, err := s.Write([]byte{0})
			Expect(n).To(BeZero())
			Expect(err).To(MatchError("RST_STREAM received with code 42"))
		})

		It("doesn't close the stream for reading", func() {
			s, err := sess.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 5,
				Data:     []byte("foobar"),
			})
			err = sess.handleRstStreamFrame(&frames.RstStreamFrame{
				StreamID:   5,
				ErrorCode:  42,
				ByteOffset: 6,
			})
			Expect(err).ToNot(HaveOccurred())
			b := make([]byte, 3)
			n, err := s.Read(b)
			Expect(n).To(Equal(3))
			Expect(err).ToNot(HaveOccurred())
		})

		It("queues a RST_STERAM frame with the correct offset", func() {
			str, err := sess.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			str.(*stream).writeOffset = 0x1337
			err = sess.handleRstStreamFrame(&frames.RstStreamFrame{
				StreamID: 5,
			})
			Expect(err).ToNot(HaveOccurred())
			Expect(sess.packer.controlFrames).To(HaveLen(1))
			Expect(sess.packer.controlFrames[0].(*frames.RstStreamFrame)).To(Equal(&frames.RstStreamFrame{
				StreamID:   5,
				ByteOffset: 0x1337,
			}))
			Expect(str.(*stream).finished()).To(BeTrue())
		})

		It("doesn't queue a RST_STREAM for a stream that it already sent a FIN on", func() {
			str, err := sess.GetOrOpenStream(5)
			str.(*stream).sentFin()
			str.Close()
			err = sess.handleRstStreamFrame(&frames.RstStreamFrame{
				StreamID: 5,
			})
			Expect(err).ToNot(HaveOccurred())
			Expect(sess.packer.controlFrames).To(BeEmpty())
			Expect(str.(*stream).finished()).To(BeTrue())
		})

		It("passes the byte offset to the flow controller", func() {
			sess.streamsMap.GetOrOpenStream(5)
			sess.flowControlManager = newMockFlowControlHandler()
			err := sess.handleRstStreamFrame(&frames.RstStreamFrame{
				StreamID:   5,
				ByteOffset: 0x1337,
			})
			Expect(err).ToNot(HaveOccurred())
			Expect(sess.flowControlManager.(*mockFlowControlHandler).highestReceivedForStream).To(Equal(protocol.StreamID(5)))
			Expect(sess.flowControlManager.(*mockFlowControlHandler).highestReceived).To(Equal(protocol.ByteCount(0x1337)))
		})

		It("returns errors from the flow controller", func() {
			sess.streamsMap.GetOrOpenStream(5)
			sess.flowControlManager = newMockFlowControlHandler()
			testErr := errors.New("flow control violation")
			sess.flowControlManager.(*mockFlowControlHandler).flowControlViolation = testErr
			err := sess.handleRstStreamFrame(&frames.RstStreamFrame{
				StreamID:   5,
				ByteOffset: 0x1337,
			})
			Expect(err).To(MatchError(testErr))
		})

		It("ignores the error when the stream is not known", func() {
			err := sess.handleFrames([]frames.Frame{&frames.RstStreamFrame{
				StreamID:  5,
				ErrorCode: 42,
			}})
			Expect(err).NotTo(HaveOccurred())
		})

		It("queues a RST_STREAM when a stream gets reset locally", func() {
			testErr := errors.New("testErr")
			str, err := sess.streamsMap.GetOrOpenStream(5)
			str.writeOffset = 0x1337
			Expect(err).ToNot(HaveOccurred())
			str.Reset(testErr)
			Expect(sess.packer.controlFrames).To(HaveLen(1))
			Expect(sess.packer.controlFrames[0]).To(Equal(&frames.RstStreamFrame{
				StreamID:   5,
				ByteOffset: 0x1337,
			}))
			Expect(str.finished()).To(BeFalse())
		})

		It("doesn't queue another RST_STREAM, when it receives an RST_STREAM as a response for the first", func() {
			testErr := errors.New("testErr")
			str, err := sess.streamsMap.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			str.Reset(testErr)
			Expect(sess.packer.controlFrames).To(HaveLen(1))
			err = sess.handleRstStreamFrame(&frames.RstStreamFrame{
				StreamID:   5,
				ByteOffset: 0x42,
			})
			Expect(err).ToNot(HaveOccurred())
			Expect(sess.packer.controlFrames).To(HaveLen(1))
		})
	})

	Context("handling WINDOW_UPDATE frames", func() {
		It("updates the Flow Control Window of a stream", func() {
			_, err := sess.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			err = sess.handleWindowUpdateFrame(&frames.WindowUpdateFrame{
				StreamID:   5,
				ByteOffset: 100,
			})
			Expect(err).ToNot(HaveOccurred())
			Expect(sess.flowControlManager.SendWindowSize(5)).To(Equal(protocol.ByteCount(100)))
		})

		It("updates the Flow Control Window of the connection", func() {
			err := sess.handleWindowUpdateFrame(&frames.WindowUpdateFrame{
				StreamID:   0,
				ByteOffset: 0x800000,
			})
			Expect(err).ToNot(HaveOccurred())
		})

		It("opens a new stream when receiving a WINDOW_UPDATE for an unknown stream", func() {
			err := sess.handleWindowUpdateFrame(&frames.WindowUpdateFrame{
				StreamID:   5,
				ByteOffset: 1337,
			})
			Expect(err).ToNot(HaveOccurred())
			str, err := sess.streamsMap.GetOrOpenStream(5)
			Expect(err).NotTo(HaveOccurred())
			Expect(str).ToNot(BeNil())
		})

		It("errors when receiving a WindowUpdateFrame for a closed stream", func() {
			sess.handleStreamFrame(&frames.StreamFrame{StreamID: 5})
			err := sess.streamsMap.RemoveStream(5)
			Expect(err).ToNot(HaveOccurred())
			sess.garbageCollectStreams()
			err = sess.handleWindowUpdateFrame(&frames.WindowUpdateFrame{
				StreamID:   5,
				ByteOffset: 1337,
			})
			Expect(err).To(MatchError(errWindowUpdateOnClosedStream))
		})

		It("ignores errors when receiving a WindowUpdateFrame for a closed stream", func() {
			sess.handleStreamFrame(&frames.StreamFrame{StreamID: 5})
			err := sess.streamsMap.RemoveStream(5)
			Expect(err).ToNot(HaveOccurred())
			sess.garbageCollectStreams()
			err = sess.handleFrames([]frames.Frame{&frames.WindowUpdateFrame{
				StreamID:   5,
				ByteOffset: 1337,
			}})
			Expect(err).NotTo(HaveOccurred())
		})
	})

	It("handles PING frames", func() {
		err := sess.handleFrames([]frames.Frame{&frames.PingFrame{}})
		Expect(err).NotTo(HaveOccurred())
	})

	It("handles BLOCKED frames", func() {
		err := sess.handleFrames([]frames.Frame{&frames.BlockedFrame{}})
		Expect(err).NotTo(HaveOccurred())
	})

	It("errors on GOAWAY frames", func() {
		err := sess.handleFrames([]frames.Frame{&frames.GoawayFrame{}})
		Expect(err).To(MatchError("unimplemented: handling GOAWAY frames"))
	})

	It("handles STOP_WAITING frames", func() {
		err := sess.handleFrames([]frames.Frame{&frames.StopWaitingFrame{LeastUnacked: 10}})
		Expect(err).NotTo(HaveOccurred())
	})

	It("handles CONNECTION_CLOSE frames", func() {
		str, _ := sess.GetOrOpenStream(5)
		err := sess.handleFrames([]frames.Frame{&frames.ConnectionCloseFrame{ErrorCode: 42, ReasonPhrase: "foobar"}})
		Expect(err).NotTo(HaveOccurred())
		_, err = str.Read([]byte{0})
		Expect(err).To(MatchError(qerr.Error(42, "foobar")))
	})

	Context("accepting streams", func() {
		It("waits for new streams", func() {
			var str utils.Stream
			go func() {
				defer GinkgoRecover()
				var err error
				str, err = sess.AcceptStream()
				Expect(err).ToNot(HaveOccurred())
			}()
			Consistently(func() utils.Stream { return str }).Should(BeNil())
			sess.handleStreamFrame(&frames.StreamFrame{
				StreamID: 3,
			})
			Eventually(func() utils.Stream { return str }).ShouldNot(BeNil())
			Expect(str.StreamID()).To(Equal(protocol.StreamID(3)))
		})

		It("stops accepting when the session is closed", func() {
			testErr := errors.New("testErr")
			var err error
			go func() {
				_, err = sess.AcceptStream()
			}()
			go sess.run()
			Consistently(func() error { return err }).ShouldNot(HaveOccurred())
			sess.Close(testErr)
			Eventually(func() error { return err }).Should(HaveOccurred())
			Expect(err).To(MatchError(qerr.ToQuicError(testErr)))
		})

		It("stops accepting when the session is closed after version negotiation", func() {
			testErr := errCloseSessionForNewVersion
			var err error
			go func() {
				_, err = sess.AcceptStream()
			}()
			go sess.run()
			Consistently(func() error { return err }).ShouldNot(HaveOccurred())
			sess.Close(testErr)
			Eventually(func() error { return err }).Should(HaveOccurred())
			Expect(err).To(MatchError(testErr))
		})
	})

	Context("closing", func() {
		var (
			nGoRoutinesBefore int
		)

		BeforeEach(func() {
			time.Sleep(10 * time.Millisecond) // Wait for old goroutines to finish
			nGoRoutinesBefore = runtime.NumGoroutine()
			go sess.run()
			Eventually(func() int { return runtime.NumGoroutine() }).Should(Equal(nGoRoutinesBefore + 2))
		})

		It("shuts down without error", func() {
			sess.Close(nil)
			Eventually(func() int { return runtime.NumGoroutine() }).Should(Equal(nGoRoutinesBefore))
			Expect(mconn.written).To(HaveLen(1))
			Expect(mconn.written[0][len(mconn.written[0])-7:]).To(Equal([]byte{0x02, byte(qerr.PeerGoingAway), 0, 0, 0, 0, 0}))
			Expect(closeCallbackCalled).To(BeTrue())
			Expect(sess.runClosed).ToNot(Receive()) // channel should be drained by Close()
		})

		It("only closes once", func() {
			sess.Close(nil)
			sess.Close(nil)
			Eventually(func() int { return runtime.NumGoroutine() }).Should(Equal(nGoRoutinesBefore))
			Expect(mconn.written).To(HaveLen(1))
			Expect(sess.runClosed).ToNot(Receive()) // channel should be drained by Close()
		})

		It("closes streams with proper error", func() {
			testErr := errors.New("test error")
			s, err := sess.GetOrOpenStream(5)
			Expect(err).NotTo(HaveOccurred())
			sess.Close(testErr)
			Eventually(func() int { return runtime.NumGoroutine() }).Should(Equal(nGoRoutinesBefore))
			Expect(closeCallbackCalled).To(BeTrue())
			n, err := s.Read([]byte{0})
			Expect(n).To(BeZero())
			Expect(err.Error()).To(ContainSubstring(testErr.Error()))
			n, err = s.Write([]byte{0})
			Expect(n).To(BeZero())
			Expect(err.Error()).To(ContainSubstring(testErr.Error()))
			Expect(sess.runClosed).ToNot(Receive()) // channel should be drained by Close()
		})

		It("closes the session in order to replace it with another QUIC version", func() {
			sess.Close(errCloseSessionForNewVersion)
			Expect(closeCallbackCalled).To(BeFalse())
			Eventually(func() int { return runtime.NumGoroutine() }).Should(Equal(nGoRoutinesBefore))
			Expect(atomic.LoadUint32(&sess.closed) != 0).To(BeTrue())
			Expect(mconn.written).To(BeEmpty()) // no CONNECTION_CLOSE or PUBLIC_RESET sent
		})
	})

	Context("receiving packets", func() {
		var hdr *PublicHeader

		BeforeEach(func() {
			sess.unpacker = &mockUnpacker{}
			clientSess.unpacker = &mockUnpacker{}
			hdr = &PublicHeader{PacketNumberLen: protocol.PacketNumberLen6}
		})

		It("sets the {last,largest}RcvdPacketNumber", func() {
			hdr.PacketNumber = 5
			err := sess.handlePacketImpl(&receivedPacket{publicHeader: hdr})
			Expect(err).ToNot(HaveOccurred())
			Expect(sess.lastRcvdPacketNumber).To(Equal(protocol.PacketNumber(5)))
			Expect(sess.largestRcvdPacketNumber).To(Equal(protocol.PacketNumber(5)))
		})

		It("sets the {last,largest}RcvdPacketNumber, for an out-of-order packet", func() {
			hdr.PacketNumber = 5
			err := sess.handlePacketImpl(&receivedPacket{publicHeader: hdr})
			Expect(err).ToNot(HaveOccurred())
			Expect(sess.lastRcvdPacketNumber).To(Equal(protocol.PacketNumber(5)))
			Expect(sess.largestRcvdPacketNumber).To(Equal(protocol.PacketNumber(5)))
			hdr.PacketNumber = 3
			err = sess.handlePacketImpl(&receivedPacket{publicHeader: hdr})
			Expect(err).ToNot(HaveOccurred())
			Expect(sess.lastRcvdPacketNumber).To(Equal(protocol.PacketNumber(3)))
			Expect(sess.largestRcvdPacketNumber).To(Equal(protocol.PacketNumber(5)))
		})

		It("ignores duplicate packets", func() {
			hdr.PacketNumber = 5
			err := sess.handlePacketImpl(&receivedPacket{publicHeader: hdr})
			Expect(err).ToNot(HaveOccurred())
			err = sess.handlePacketImpl(&receivedPacket{publicHeader: hdr})
			Expect(err).ToNot(HaveOccurred())
		})

		It("ignores packets smaller than the highest LeastUnacked of a StopWaiting", func() {
			err := sess.receivedPacketHandler.ReceivedStopWaiting(&frames.StopWaitingFrame{LeastUnacked: 10})
			Expect(err).ToNot(HaveOccurred())
			hdr.PacketNumber = 5
			err = sess.handlePacketImpl(&receivedPacket{publicHeader: hdr})
			Expect(err).ToNot(HaveOccurred())
		})

		It("passes the diversification nonce to the cryptoSetup, if it is a client", func() {
			hdr.PacketNumber = 5
			hdr.DiversificationNonce = []byte("foobar")
			err := clientSess.handlePacketImpl(&receivedPacket{publicHeader: hdr})
			Expect(err).ToNot(HaveOccurred())
			Expect((*[]byte)(unsafe.Pointer(reflect.ValueOf(clientSess.cryptoSetup).Elem().FieldByName("diversificationNonce").UnsafeAddr()))).To(Equal(&hdr.DiversificationNonce))
		})

		Context("updating the remote address", func() {
			It("sets the remote address", func() {
				remoteIP := &net.IPAddr{IP: net.IPv4(192, 168, 0, 100)}
				Expect(sess.conn.(*mockConnection).remoteAddr).ToNot(Equal(remoteIP))
				p := receivedPacket{
					remoteAddr:   remoteIP,
					publicHeader: &PublicHeader{PacketNumber: 1337},
				}
				err := sess.handlePacketImpl(&p)
				Expect(err).ToNot(HaveOccurred())
				Expect(sess.conn.(*mockConnection).remoteAddr).To(Equal(remoteIP))
			})

			It("doesn't change the remote address if authenticating the packet fails", func() {
				remoteIP := &net.IPAddr{IP: net.IPv4(192, 168, 0, 100)}
				attackerIP := &net.IPAddr{IP: net.IPv4(192, 168, 0, 102)}
				sess.conn.(*mockConnection).remoteAddr = remoteIP
				// use the real packetUnpacker here, to make sure this test fails if the error code for failed decryption changes
				sess.unpacker = &packetUnpacker{}
				sess.unpacker.(*packetUnpacker).aead = &crypto.NullAEAD{}
				p := receivedPacket{
					remoteAddr:   attackerIP,
					publicHeader: &PublicHeader{PacketNumber: 1337},
				}
				err := sess.handlePacketImpl(&p)
				quicErr := err.(*qerr.QuicError)
				Expect(quicErr.ErrorCode).To(Equal(qerr.DecryptionFailure))
				Expect(sess.conn.(*mockConnection).remoteAddr).To(Equal(remoteIP))
			})

			It("sets the remote address, if the packet is authenticated, but unpacking fails for another reason", func() {
				testErr := errors.New("testErr")
				remoteIP := &net.IPAddr{IP: net.IPv4(192, 168, 0, 100)}
				Expect(sess.conn.(*mockConnection).remoteAddr).ToNot(Equal(remoteIP))
				p := receivedPacket{
					remoteAddr:   remoteIP,
					publicHeader: &PublicHeader{PacketNumber: 1337},
				}
				sess.unpacker.(*mockUnpacker).unpackErr = testErr
				err := sess.handlePacketImpl(&p)
				Expect(err).To(MatchError(testErr))
				Expect(sess.conn.(*mockConnection).remoteAddr).To(Equal(remoteIP))
			})
		})
	})

	Context("sending packets", func() {
		It("sends ack frames", func() {
			packetNumber := protocol.PacketNumber(0x035E)
			sess.receivedPacketHandler.ReceivedPacket(packetNumber, true)
			err := sess.sendPacket()
			Expect(err).NotTo(HaveOccurred())
			Expect(mconn.written).To(HaveLen(1))
			Expect(mconn.written[0]).To(ContainSubstring(string([]byte{0x5E, 0x03})))
		})

		It("sends two WindowUpdate frames", func() {
			_, err := sess.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			sess.flowControlManager.AddBytesRead(5, protocol.ReceiveStreamFlowControlWindow)
			err = sess.sendPacket()
			Expect(err).NotTo(HaveOccurred())
			err = sess.sendPacket()
			Expect(err).NotTo(HaveOccurred())
			err = sess.sendPacket()
			Expect(err).NotTo(HaveOccurred())
			Expect(mconn.written).To(HaveLen(2))
			Expect(mconn.written[0]).To(ContainSubstring(string([]byte{0x04, 0x05, 0, 0, 0})))
			Expect(mconn.written[1]).To(ContainSubstring(string([]byte{0x04, 0x05, 0, 0, 0})))
		})

		It("sends public reset", func() {
			err := sess.sendPublicReset(1)
			Expect(err).NotTo(HaveOccurred())
			Expect(mconn.written).To(HaveLen(1))
			Expect(mconn.written[0]).To(ContainSubstring(string([]byte("PRST"))))
		})
	})

	Context("retransmissions", func() {
		It("sends a StreamFrame from a packet queued for retransmission", func() {
			// a StopWaitingFrame is added, so make sure the packet number of the new package is higher than the packet number of the retransmitted packet
			sess.packer.packetNumberGenerator.next = 0x1337 + 9

			f := frames.StreamFrame{
				StreamID: 0x5,
				Data:     []byte("foobar1234567"),
			}
			p := ackhandler.Packet{
				PacketNumber: 0x1337,
				Frames:       []frames.Frame{&f},
			}
			sph := newMockSentPacketHandler()
			sph.(*mockSentPacketHandler).retransmissionQueue = []*ackhandler.Packet{&p}
			sess.sentPacketHandler = sph

			err := sess.sendPacket()
			Expect(err).NotTo(HaveOccurred())
			Expect(mconn.written).To(HaveLen(1))
			Expect(sph.(*mockSentPacketHandler).requestedStopWaiting).To(BeTrue())
			Expect(mconn.written[0]).To(ContainSubstring("foobar1234567"))
		})

		It("sends a StreamFrame from a packet queued for retransmission", func() {
			// a StopWaitingFrame is added, so make sure the packet number of the new package is higher than the packet number of the retransmitted packet
			sess.packer.packetNumberGenerator.next = 0x1337 + 9

			f1 := frames.StreamFrame{
				StreamID: 0x5,
				Data:     []byte("foobar"),
			}
			f2 := frames.StreamFrame{
				StreamID: 0x7,
				Data:     []byte("loremipsum"),
			}
			p1 := ackhandler.Packet{
				PacketNumber: 0x1337,
				Frames:       []frames.Frame{&f1},
			}
			p2 := ackhandler.Packet{
				PacketNumber: 0x1338,
				Frames:       []frames.Frame{&f2},
			}
			sph := newMockSentPacketHandler()
			sph.(*mockSentPacketHandler).retransmissionQueue = []*ackhandler.Packet{&p1, &p2}
			sess.sentPacketHandler = sph

			err := sess.sendPacket()
			Expect(err).NotTo(HaveOccurred())
			Expect(mconn.written).To(HaveLen(1))
			Expect(mconn.written[0]).To(ContainSubstring("foobar"))
			Expect(mconn.written[0]).To(ContainSubstring("loremipsum"))
		})

		It("always attaches a StopWaiting to a packet that contains a retransmission", func() {
			// make sure the packet number of the new package is higher than the packet number of the retransmitted packet
			sess.packer.packetNumberGenerator.next = 0x1337 + 9

			f := &frames.StreamFrame{
				StreamID: 0x5,
				Data:     bytes.Repeat([]byte{'f'}, int(1.5*float32(protocol.MaxPacketSize))),
			}
			sess.streamFramer.AddFrameForRetransmission(f)

			sph := newMockSentPacketHandler()
			sess.sentPacketHandler = sph

			err := sess.sendPacket()
			Expect(err).NotTo(HaveOccurred())
			Expect(mconn.written).To(HaveLen(2))
			sentPackets := sph.(*mockSentPacketHandler).sentPackets
			Expect(sentPackets).To(HaveLen(2))
			_, ok := sentPackets[0].Frames[0].(*frames.StopWaitingFrame)
			Expect(ok).To(BeTrue())
			_, ok = sentPackets[1].Frames[0].(*frames.StopWaitingFrame)
			Expect(ok).To(BeTrue())
		})

		It("calls MaybeQueueRTOs even if congestion blocked, so that bytesInFlight is updated", func() {
			sph := newMockSentPacketHandler()
			sph.(*mockSentPacketHandler).congestionLimited = true
			sess.sentPacketHandler = sph
			err := sess.sendPacket()
			Expect(err).NotTo(HaveOccurred())
			Expect(sph.(*mockSentPacketHandler).maybeQueueRTOsCalled).To(BeTrue())
		})

		It("retransmits a WindowUpdates if it hasn't already sent a WindowUpdate with a higher ByteOffset", func() {
			_, err := sess.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			fc := newMockFlowControlHandler()
			fc.receiveWindow = 0x1000
			sess.flowControlManager = fc
			sph := newMockSentPacketHandler()
			sess.sentPacketHandler = sph
			wuf := &frames.WindowUpdateFrame{
				StreamID:   5,
				ByteOffset: 0x1000,
			}
			sph.(*mockSentPacketHandler).retransmissionQueue = []*ackhandler.Packet{{
				Frames: []frames.Frame{wuf},
			}}
			err = sess.sendPacket()
			Expect(err).ToNot(HaveOccurred())
			sentPackets := sph.(*mockSentPacketHandler).sentPackets
			Expect(sentPackets).To(HaveLen(1))
			Expect(sentPackets[0].Frames).To(ContainElement(wuf))
		})

		It("doesn't retransmit WindowUpdates if it already sent a WindowUpdate with a higher ByteOffset", func() {
			_, err := sess.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			fc := newMockFlowControlHandler()
			fc.receiveWindow = 0x2000
			sess.flowControlManager = fc
			sph := newMockSentPacketHandler()
			sess.sentPacketHandler = sph
			sph.(*mockSentPacketHandler).retransmissionQueue = []*ackhandler.Packet{{
				Frames: []frames.Frame{&frames.WindowUpdateFrame{
					StreamID:   5,
					ByteOffset: 0x1000,
				}},
			}}
			err = sess.sendPacket()
			Expect(err).ToNot(HaveOccurred())
			Expect(sph.(*mockSentPacketHandler).sentPackets).To(BeEmpty())
		})

		It("doesn't retransmit WindowUpdates for closed streams", func() {
			str, err := sess.GetOrOpenStream(5)
			Expect(err).ToNot(HaveOccurred())
			// close the stream
			str.(*stream).sentFin()
			str.Close()
			str.(*stream).RegisterRemoteError(nil)
			sess.garbageCollectStreams()
			_, err = sess.flowControlManager.SendWindowSize(5)
			Expect(err).To(MatchError("Error accessing the flowController map."))
			sph := newMockSentPacketHandler()
			sess.sentPacketHandler = sph
			sph.(*mockSentPacketHandler).retransmissionQueue = []*ackhandler.Packet{{
				Frames: []frames.Frame{&frames.WindowUpdateFrame{
					StreamID:   5,
					ByteOffset: 0x1337,
				}},
			}}
			err = sess.sendPacket()
			Expect(err).ToNot(HaveOccurred())
			sentPackets := sph.(*mockSentPacketHandler).sentPackets
			Expect(sentPackets).To(BeEmpty())
		})
	})

	Context("scheduling sending", func() {
		It("sends after writing to a stream", func(done Done) {
			Expect(sess.sendingScheduled).NotTo(Receive())
			s, err := sess.GetOrOpenStream(3)
			Expect(err).NotTo(HaveOccurred())
			go func() {
				s.Write([]byte("foobar"))
				close(done)
			}()
			Eventually(sess.sendingScheduled).Should(Receive())
			s.(*stream).getDataForWriting(1000) // unblock
		})

		It("sets the timer to the ack timer", func() {
			rph := &mockReceivedPacketHandler{}
			rph.nextAckFrame = &frames.AckFrame{LargestAcked: 0x1337}
			sess.receivedPacketHandler = rph
			go sess.run()
			sess.ackAlarmChanged(time.Now().Add(10 * time.Millisecond))
			time.Sleep(10 * time.Millisecond)
			Eventually(func() int { return len(mconn.written) }).ShouldNot(BeZero())
			Expect(mconn.written[0]).To(ContainSubstring(string([]byte{0x37, 0x13})))
		})

		Context("bundling of small packets", func() {
			It("bundles two small frames of different streams into one packet", func() {
				s1, err := sess.GetOrOpenStream(5)
				Expect(err).NotTo(HaveOccurred())
				s2, err := sess.GetOrOpenStream(7)
				Expect(err).NotTo(HaveOccurred())

				// Put data directly into the streams
				s1.(*stream).dataForWriting = []byte("foobar1")
				s2.(*stream).dataForWriting = []byte("foobar2")

				sess.scheduleSending()
				go sess.run()

				Eventually(func() [][]byte { return mconn.written }).Should(HaveLen(1))
				Expect(mconn.written[0]).To(ContainSubstring("foobar1"))
				Expect(mconn.written[0]).To(ContainSubstring("foobar2"))
			})

			It("sends out two big frames in two packets", func() {
				s1, err := sess.GetOrOpenStream(5)
				Expect(err).NotTo(HaveOccurred())
				s2, err := sess.GetOrOpenStream(7)
				Expect(err).NotTo(HaveOccurred())
				go sess.run()
				go func() {
					defer GinkgoRecover()
					_, err2 := s1.Write(bytes.Repeat([]byte{'e'}, 1000))
					Expect(err2).ToNot(HaveOccurred())
				}()
				_, err = s2.Write(bytes.Repeat([]byte{'e'}, 1000))
				Expect(err).ToNot(HaveOccurred())
				Eventually(func() [][]byte { return mconn.written }).Should(HaveLen(2))
			})

			It("sends out two small frames that are written to long after one another into two packets", func() {
				s, err := sess.GetOrOpenStream(5)
				Expect(err).NotTo(HaveOccurred())
				go sess.run()
				_, err = s.Write([]byte("foobar1"))
				Expect(err).NotTo(HaveOccurred())
				Eventually(func() [][]byte { return mconn.written }).Should(HaveLen(1))
				_, err = s.Write([]byte("foobar2"))
				Expect(err).NotTo(HaveOccurred())
				Eventually(func() [][]byte { return mconn.written }).Should(HaveLen(2))
			})

			It("sends a queued ACK frame only once", func() {
				packetNumber := protocol.PacketNumber(0x1337)
				sess.receivedPacketHandler.ReceivedPacket(packetNumber, true)

				s, err := sess.GetOrOpenStream(5)
				Expect(err).NotTo(HaveOccurred())
				go sess.run()
				_, err = s.Write([]byte("foobar1"))
				Expect(err).NotTo(HaveOccurred())
				Eventually(func() [][]byte { return mconn.written }).Should(HaveLen(1))
				_, err = s.Write([]byte("foobar2"))
				Expect(err).NotTo(HaveOccurred())

				Eventually(func() [][]byte { return mconn.written }).Should(HaveLen(2))
				Expect(mconn.written[0]).To(ContainSubstring(string([]byte{0x37, 0x13})))
				Expect(mconn.written[1]).ToNot(ContainSubstring(string([]byte{0x37, 0x13})))
			})
		})
	})

	It("closes when crypto stream errors", func() {
		go sess.run()
		s, err := sess.GetOrOpenStream(3)
		Expect(err).NotTo(HaveOccurred())
		err = sess.handleStreamFrame(&frames.StreamFrame{
			StreamID: 1,
			Data:     []byte("4242\x00\x00\x00\x00"),
		})
		Expect(err).NotTo(HaveOccurred())
		Eventually(func() bool { return atomic.LoadUint32(&sess.closed) != 0 }).Should(BeTrue())
		_, err = s.Write([]byte{})
		Expect(err.(*qerr.QuicError).ErrorCode).To(Equal(qerr.InvalidCryptoMessageType))
	})

	It("sends public reset after too many undecryptable packets", func() {
		// Write protocol.MaxUndecryptablePackets and expect a public reset to happen
		for i := 0; i < protocol.MaxUndecryptablePackets; i++ {
			hdr := &PublicHeader{
				PacketNumber: protocol.PacketNumber(i + 1),
			}
			sess.handlePacket(&receivedPacket{publicHeader: hdr, data: []byte("foobar")})
		}
		sess.run()

		Expect(mconn.written).To(HaveLen(1))
		Expect(mconn.written[0]).To(ContainSubstring(string([]byte("PRST"))))
		Expect(sess.runClosed).To(Receive())
	})

	It("ignores undecryptable packets after the handshake is complete", func() {
		*(*bool)(unsafe.Pointer(reflect.ValueOf(sess.cryptoSetup).Elem().FieldByName("receivedForwardSecurePacket").UnsafeAddr())) = true
		for i := 0; i < protocol.MaxUndecryptablePackets; i++ {
			hdr := &PublicHeader{
				PacketNumber: protocol.PacketNumber(i + 1),
			}
			sess.handlePacket(&receivedPacket{publicHeader: hdr, data: []byte("foobar")})
		}
		go sess.run()
		Consistently(sess.undecryptablePackets).Should(HaveLen(0))
		sess.closeImpl(nil, true)
		Eventually(sess.runClosed).Should(Receive())
	})

	It("unqueues undecryptable packets for later decryption", func() {
		sess.undecryptablePackets = []*receivedPacket{{
			publicHeader: &PublicHeader{PacketNumber: protocol.PacketNumber(42)},
		}}
		Expect(sess.receivedPackets).NotTo(Receive())
		sess.tryDecryptingQueuedPackets()
		Expect(sess.undecryptablePackets).To(BeEmpty())
		Expect(sess.receivedPackets).To(Receive())
	})

	It("calls the cryptoChangeCallback when the AEAD changes", func(done Done) {
		var callbackCalled bool
		var callbackCalledWith bool
		cb := func(p bool) {
			callbackCalled = true
			callbackCalledWith = p
		}
		sess.cryptoChangeCallback = cb
		sess.cryptoSetup = &mockCryptoSetup{handshakeComplete: false}
		sess.aeadChanged <- struct{}{}
		go sess.run()
		Eventually(func() bool { return callbackCalled }).Should(BeTrue())
		Expect(callbackCalledWith).To(BeFalse())
		close(done)
	})

	It("calls the cryptoChangeCallback when the AEAD changes to forward secure encryption", func(done Done) {
		var callbackCalledWith bool
		cb := func(p bool) {
			callbackCalledWith = p
		}
		sess.cryptoChangeCallback = cb
		sess.cryptoSetup = &mockCryptoSetup{handshakeComplete: true}
		sess.aeadChanged <- struct{}{}
		go sess.run()
		Eventually(func() bool { return callbackCalledWith }).Should(BeTrue())
		close(done)
	})

	Context("timeouts", func() {
		It("times out due to no network activity", func(done Done) {
			sess.lastNetworkActivityTime = time.Now().Add(-time.Hour)
			sess.run() // Would normally not return
			Expect(mconn.written[0]).To(ContainSubstring("No recent network activity."))
			Expect(closeCallbackCalled).To(BeTrue())
			Expect(sess.runClosed).To(Receive())
			close(done)
		})

		It("times out due to non-completed crypto handshake", func(done Done) {
			sess.sessionCreationTime = time.Now().Add(-time.Hour)
			sess.run() // Would normally not return
			Expect(mconn.written[0]).To(ContainSubstring("Crypto handshake did not complete in time."))
			Expect(closeCallbackCalled).To(BeTrue())
			Expect(sess.runClosed).To(Receive())
			close(done)
		})

		It("does not use ICSL before handshake", func(done Done) {
			sess.lastNetworkActivityTime = time.Now().Add(-time.Minute)
			cpm.idleTime = 99999 * time.Second
			sess.packer.connectionParameters = sess.connectionParameters
			sess.run() // Would normally not return
			Expect(mconn.written[0]).To(ContainSubstring("No recent network activity."))
			Expect(closeCallbackCalled).To(BeTrue())
			Expect(sess.runClosed).To(Receive())
			close(done)
		})

		It("uses ICSL after handshake", func(done Done) {
			// sess.lastNetworkActivityTime = time.Now().Add(-time.Minute)
			*(*bool)(unsafe.Pointer(reflect.ValueOf(sess.cryptoSetup).Elem().FieldByName("receivedForwardSecurePacket").UnsafeAddr())) = true
			*(*crypto.AEAD)(unsafe.Pointer(reflect.ValueOf(sess.cryptoSetup).Elem().FieldByName("forwardSecureAEAD").UnsafeAddr())) = &crypto.NullAEAD{}
			cpm.idleTime = 0 * time.Millisecond
			sess.packer.connectionParameters = sess.connectionParameters
			sess.run() // Would normally not return
			Expect(mconn.written[0]).To(ContainSubstring("No recent network activity."))
			Expect(closeCallbackCalled).To(BeTrue())
			Expect(sess.runClosed).To(Receive())
			close(done)
		})
	})

	It("errors when the SentPacketHandler has too many packets tracked", func() {
		streamFrame := frames.StreamFrame{StreamID: 5, Data: []byte("foobar")}
		for i := protocol.PacketNumber(1); i < protocol.MaxTrackedSentPackets+10; i++ {
			packet := ackhandler.Packet{PacketNumber: protocol.PacketNumber(i), Frames: []frames.Frame{&streamFrame}, Length: 1}
			err := sess.sentPacketHandler.SentPacket(&packet)
			Expect(err).ToNot(HaveOccurred())
		}
		// now sess.sentPacketHandler.CheckForError will return an error
		err := sess.sendPacket()
		Expect(err).To(MatchError(ackhandler.ErrTooManyTrackedSentPackets))
	})

	It("stores up to MaxSessionUnprocessedPackets packets", func(done Done) {
		// Nothing here should block
		for i := protocol.PacketNumber(0); i < protocol.MaxSessionUnprocessedPackets+10; i++ {
			sess.handlePacket(&receivedPacket{})
		}
		close(done)
	}, 0.5)

	It("retransmits RTO packets", func() {
		// We simulate consistently low RTTs, so that the test works faster
		n := protocol.PacketNumber(10)
		for p := protocol.PacketNumber(1); p < n; p++ {
			err := sess.sentPacketHandler.SentPacket(&ackhandler.Packet{PacketNumber: p, Length: 1})
			Expect(err).NotTo(HaveOccurred())
			time.Sleep(time.Microsecond)
			ack := &frames.AckFrame{}
			ack.LargestAcked = p
			err = sess.sentPacketHandler.ReceivedAck(ack, p, time.Now())
			Expect(err).NotTo(HaveOccurred())
		}
		sess.packer.packetNumberGenerator.next = n + 1
		// Now, we send a single packet, and expect that it was retransmitted later
		err := sess.sentPacketHandler.SentPacket(&ackhandler.Packet{
			PacketNumber: n,
			Length:       1,
			Frames: []frames.Frame{&frames.StreamFrame{
				Data: []byte("foobar"),
			}},
		})
		Expect(err).NotTo(HaveOccurred())
		go sess.run()
		sess.scheduleSending()
		Eventually(func() [][]byte { return mconn.written }).ShouldNot(BeEmpty())
		Expect(mconn.written[0]).To(ContainSubstring("foobar"))
	})

	Context("getting streams", func() {
		It("returns a new stream", func() {
			str, err := sess.GetOrOpenStream(11)
			Expect(err).ToNot(HaveOccurred())
			Expect(str).ToNot(BeNil())
			Expect(str.StreamID()).To(Equal(protocol.StreamID(11)))
		})

		It("returns a nil-value (not an interface with value nil) for closed streams", func() {
			_, err := sess.GetOrOpenStream(9)
			Expect(err).ToNot(HaveOccurred())
			sess.streamsMap.RemoveStream(9)
			sess.garbageCollectStreams()
			Expect(sess.streamsMap.GetOrOpenStream(9)).To(BeNil())
			str, err := sess.GetOrOpenStream(9)
			Expect(err).ToNot(HaveOccurred())
			Expect(str).To(BeNil())
			// make sure that the returned value is a plain nil, not an utils.Stream with value nil
			_, ok := str.(utils.Stream)
			Expect(ok).To(BeFalse())
		})
	})

	Context("counting streams", func() {
		It("errors when too many streams are opened", func() {
			for i := 0; i < 110; i++ {
				_, err := sess.GetOrOpenStream(protocol.StreamID(i*2 + 1))
				Expect(err).NotTo(HaveOccurred())
			}
			_, err := sess.GetOrOpenStream(protocol.StreamID(301))
			Expect(err).To(MatchError(qerr.TooManyOpenStreams))
		})

		It("does not error when many streams are opened and closed", func() {
			for i := 2; i <= 1000; i++ {
				s, err := sess.GetOrOpenStream(protocol.StreamID(i*2 + 1))
				Expect(err).NotTo(HaveOccurred())
				err = s.Close()
				Expect(err).NotTo(HaveOccurred())
				s.(*stream).sentFin()
				s.CloseRemote(0)
				_, err = s.Read([]byte("a"))
				Expect(err).To(MatchError(io.EOF))
				sess.garbageCollectStreams()
			}
		})
	})

	Context("ignoring errors", func() {
		It("ignores duplicate acks", func() {
			sess.sentPacketHandler.SentPacket(&ackhandler.Packet{
				PacketNumber: 1,
				Length:       1,
			})
			err := sess.handleFrames([]frames.Frame{&frames.AckFrame{
				LargestAcked: 1,
			}})
			Expect(err).NotTo(HaveOccurred())
			err = sess.handleFrames([]frames.Frame{&frames.AckFrame{
				LargestAcked: 1,
			}})
			Expect(err).NotTo(HaveOccurred())
		})
	})

	Context("window updates", func() {
		It("gets stream level window updates", func() {
			err := sess.flowControlManager.AddBytesRead(1, protocol.ReceiveStreamFlowControlWindow)
			Expect(err).NotTo(HaveOccurred())
			frames, err := sess.getWindowUpdateFrames()
			Expect(err).NotTo(HaveOccurred())
			Expect(frames).To(HaveLen(1))
			Expect(frames[0].StreamID).To(Equal(protocol.StreamID(1)))
			Expect(frames[0].ByteOffset).To(Equal(protocol.ReceiveStreamFlowControlWindow * 2))
		})

		It("gets connection level window updates", func() {
			_, err := sess.GetOrOpenStream(5)
			Expect(err).NotTo(HaveOccurred())
			err = sess.flowControlManager.AddBytesRead(5, protocol.ReceiveConnectionFlowControlWindow)
			Expect(err).NotTo(HaveOccurred())
			frames, err := sess.getWindowUpdateFrames()
			Expect(err).NotTo(HaveOccurred())
			Expect(frames).To(HaveLen(1))
			Expect(frames[0].StreamID).To(Equal(protocol.StreamID(0)))
			Expect(frames[0].ByteOffset).To(Equal(protocol.ReceiveConnectionFlowControlWindow * 2))
		})
	})
})