diff --git a/packet_packer.go b/packet_packer.go index dcfb18ce..1f832e9c 100644 --- a/packet_packer.go +++ b/packet_packer.go @@ -182,53 +182,23 @@ func (p *packetPacker) composeNextPacket(stopWaitingFrame *frames.StopWaitingFra return nil, fmt.Errorf("Packet Packer BUG: packet payload (%d) too large (%d)", payloadLength, maxFrameSize) } - hasStreamFrames := false - // temporarily increase the maxFrameSize by 2 bytes // this leads to a properly sized packet in all cases, since we do all the packet length calculations with StreamFrames that have the DataLen set // however, for the last StreamFrame in the packet, we can omit the DataLen, thus saving 2 bytes and yielding a packet of exactly the correct size maxFrameSize += 2 - for { - if payloadLength > maxFrameSize { - return nil, fmt.Errorf("Packet Packer BUG: packet payload (%d) too large (%d)", payloadLength, maxFrameSize) - } - - frame, err := p.streamFramer.PopStreamFrame(maxFrameSize - payloadLength) - if err != nil { - return nil, err - } - if frame == nil { - break - } - frame.DataLenPresent = true // set the dataLen by default. Remove them later if applicable - - frameHeaderLen, _ := frame.MinLength(p.version) // StreamFrame.MinLength *never* returns an error - payloadLength += frameHeaderLen + frame.DataLen() - - blockedFrame := p.streamFramer.PopBlockedFrame() - if blockedFrame != nil { - blockedLength, _ := blockedFrame.MinLength(p.version) // BlockedFrame.MinLength *never* returns an error - if payloadLength+blockedLength <= maxFrameSize { - payloadFrames = append(payloadFrames, blockedFrame) - payloadLength += blockedLength - } else { - p.controlFrames = append(p.controlFrames, blockedFrame) - } - } - - payloadFrames = append(payloadFrames, frame) - hasStreamFrames = true + fs := p.streamFramer.PopStreamFrames(maxFrameSize - payloadLength) + if len(fs) != 0 { + fs[len(fs)-1].DataLenPresent = false } - // remove the dataLen for the last StreamFrame in the packet - if hasStreamFrames { - lastStreamFrame, ok := payloadFrames[len(payloadFrames)-1].(*frames.StreamFrame) - if !ok { - return nil, errors.New("PacketPacker BUG: StreamFrame type assertion failed") - } - lastStreamFrame.DataLenPresent = false - // payloadLength -= 2 + // TODO: Simplify + for _, f := range fs { + payloadFrames = append(payloadFrames, f) + } + + for b := p.streamFramer.PopBlockedFrame(); b != nil; b = p.streamFramer.PopBlockedFrame() { + p.controlFrames = append(p.controlFrames, b) } return payloadFrames, nil diff --git a/packet_packer_test.go b/packet_packer_test.go index d2167b0a..5c4cd99e 100644 --- a/packet_packer_test.go +++ b/packet_packer_test.go @@ -385,7 +385,20 @@ var _ = Describe("Packet packer", func() { }) Context("Blocked frames", func() { - It("adds a blocked frame to a packet if there is enough space", func() { + It("queues a BLOCKED frame", func() { + length := 100 + streamFramer.blockedFrameQueue = []*frames.BlockedFrame{{StreamID: 5}} + f := &frames.StreamFrame{ + StreamID: 5, + Data: bytes.Repeat([]byte{'f'}, length), + } + streamFramer.AddFrameForRetransmission(f) + _, err := packer.composeNextPacket(nil, publicHeaderLen) + Expect(err).ToNot(HaveOccurred()) + Expect(packer.controlFrames[0]).To(Equal(&frames.BlockedFrame{StreamID: 5})) + }) + + It("removes the dataLen attribute from the last StreamFrame, even if it queued a BLOCKED frame", func() { length := 100 streamFramer.blockedFrameQueue = []*frames.BlockedFrame{{StreamID: 5}} f := &frames.StreamFrame{ @@ -395,61 +408,8 @@ var _ = Describe("Packet packer", func() { streamFramer.AddFrameForRetransmission(f) p, err := packer.composeNextPacket(nil, publicHeaderLen) Expect(err).ToNot(HaveOccurred()) - Expect(p).To(HaveLen(2)) - Expect(p[0]).To(Equal(&frames.BlockedFrame{StreamID: 5})) - }) - - It("removes the dataLen attribute from the last StreamFrame, even if it inserted a BlockedFrame before", func() { - length := 100 - streamFramer.blockedFrameQueue = []*frames.BlockedFrame{{StreamID: 5}} - f := &frames.StreamFrame{ - StreamID: 5, - Data: bytes.Repeat([]byte{'f'}, length), - } - streamFramer.AddFrameForRetransmission(f) - p, err := packer.composeNextPacket(nil, publicHeaderLen) - Expect(err).ToNot(HaveOccurred()) - Expect(p).To(HaveLen(2)) - Expect(p[1].(*frames.StreamFrame).DataLenPresent).To(BeFalse()) - }) - - It("packs a BlockedFrame in the next packet if the current packet doesn't have enough space", func() { - dataLen := int(protocol.MaxFrameAndPublicHeaderSize-publicHeaderLen) - (1 + 1 + 2) + 1 - streamFramer.blockedFrameQueue = []*frames.BlockedFrame{{StreamID: 5}} - f := &frames.StreamFrame{ - StreamID: 5, - Data: bytes.Repeat([]byte{'f'}, dataLen), - } - streamFramer.AddFrameForRetransmission(f) - p, err := packer.composeNextPacket(nil, publicHeaderLen) - Expect(err).ToNot(HaveOccurred()) Expect(p).To(HaveLen(1)) - p, err = packer.composeNextPacket(nil, publicHeaderLen) - Expect(err).ToNot(HaveOccurred()) - Expect(p).To(HaveLen(1)) - Expect(p[0]).To(Equal(&frames.BlockedFrame{StreamID: 5})) - }) - - It("packs a packet with the maximum size with a BlockedFrame", func() { - blockedFrame := &frames.BlockedFrame{StreamID: 0x1337} - blockedFrameLen, _ := blockedFrame.MinLength(0) - f1 := &frames.StreamFrame{ - StreamID: 5, - Offset: 1, - } - streamFrameHeaderLen, _ := f1.MinLength(0) - // this is the maximum dataLen of a StreamFrames that fits into one packet - dataLen := int(protocol.MaxFrameAndPublicHeaderSize - publicHeaderLen - streamFrameHeaderLen - blockedFrameLen) - streamFramer.blockedFrameQueue = []*frames.BlockedFrame{{StreamID: 5}} - f1.Data = bytes.Repeat([]byte{'f'}, dataLen) - streamFramer.AddFrameForRetransmission(f1) - p, err := packer.PackPacket(nil, []frames.Frame{}, 0) - Expect(err).ToNot(HaveOccurred()) - Expect(p).ToNot(BeNil()) - Expect(p.raw).To(HaveLen(int(protocol.MaxPacketSize))) - p, err = packer.PackPacket(nil, []frames.Frame{}, 0) - Expect(err).ToNot(HaveOccurred()) - Expect(p).To(BeNil()) + Expect(p[0].(*frames.StreamFrame).DataLenPresent).To(BeFalse()) }) It("packs a connection-level BlockedFrame", func() { @@ -459,10 +419,9 @@ var _ = Describe("Packet packer", func() { Data: []byte("foobar"), } streamFramer.AddFrameForRetransmission(f) - p, err := packer.composeNextPacket(nil, publicHeaderLen) + _, err := packer.composeNextPacket(nil, publicHeaderLen) Expect(err).ToNot(HaveOccurred()) - Expect(p).To(HaveLen(2)) - Expect(p[0]).To(Equal(&frames.BlockedFrame{StreamID: 0})) + Expect(packer.controlFrames[0]).To(Equal(&frames.BlockedFrame{StreamID: 0})) }) }) }) diff --git a/stream_framer.go b/stream_framer.go index 0b277af0..b6db27df 100644 --- a/stream_framer.go +++ b/stream_framer.go @@ -89,11 +89,9 @@ func (f *streamFramer) EstimatedDataLen() protocol.ByteCount { return l } -func (f *streamFramer) PopStreamFrame(maxLen protocol.ByteCount) (*frames.StreamFrame, error) { - if frame := f.maybePopFrameForRetransmission(maxLen); frame != nil { - return frame, nil - } - return f.maybePopNormalFrame(maxLen) +func (f *streamFramer) PopStreamFrames(maxLen protocol.ByteCount) []*frames.StreamFrame { + fs, currentLen := f.maybePopFramesForRetransmission(maxLen) + return append(fs, f.maybePopNormalFrames(maxLen-currentLen)...) } func (f *streamFramer) PopBlockedFrame() *frames.BlockedFrame { @@ -105,32 +103,39 @@ func (f *streamFramer) PopBlockedFrame() *frames.BlockedFrame { return frame } -func (f *streamFramer) maybePopFrameForRetransmission(maxLen protocol.ByteCount) *frames.StreamFrame { - if len(f.retransmissionQueue) == 0 { - return nil +func (f *streamFramer) maybePopFramesForRetransmission(maxLen protocol.ByteCount) (res []*frames.StreamFrame, currentLen protocol.ByteCount) { + for len(f.retransmissionQueue) > 0 { + frame := f.retransmissionQueue[0] + frame.DataLenPresent = true + + frameHeaderLen, _ := frame.MinLength(protocol.VersionWhatever) // can never error + if currentLen+frameHeaderLen > maxLen { + break + } + + currentLen += frameHeaderLen + + splitFrame := maybeSplitOffFrame(frame, maxLen-currentLen) + if splitFrame != nil { // StreamFrame was split + res = append(res, splitFrame) + currentLen += splitFrame.DataLen() + break + } + + f.retransmissionQueue = f.retransmissionQueue[1:] + res = append(res, frame) + currentLen += frame.DataLen() } - - frame := f.retransmissionQueue[0] - frame.DataLenPresent = true - - frameHeaderLen, _ := frame.MinLength(protocol.VersionWhatever) // can never error - if maxLen < frameHeaderLen { - return nil - } - - splitFrame := maybeSplitOffFrame(frame, maxLen-frameHeaderLen) - if splitFrame != nil { // StreamFrame was split - return splitFrame - } - - f.retransmissionQueue = f.retransmissionQueue[1:] - return frame + return } -func (f *streamFramer) maybePopNormalFrame(maxBytes protocol.ByteCount) (*frames.StreamFrame, error) { - frame := &frames.StreamFrame{DataLenPresent: true} +func (f *streamFramer) maybePopNormalFrames(maxBytes protocol.ByteCount) (res []*frames.StreamFrame) { f.streamsMutex.RLock() defer f.streamsMutex.RUnlock() + + frame := &frames.StreamFrame{DataLenPresent: true} + var currentLen protocol.ByteCount + for _, s := range *f.streams { if s == nil { continue @@ -140,16 +145,13 @@ func (f *streamFramer) maybePopNormalFrame(maxBytes protocol.ByteCount) (*frames // not perfect, but thread-safe since writeOffset is only written when getting data frame.Offset = s.writeOffset frameHeaderBytes, _ := frame.MinLength(protocol.VersionWhatever) // can never error - if maxBytes < frameHeaderBytes { - continue + if currentLen+frameHeaderBytes > maxBytes { + return // theoretically, we could find another stream that fits, but this is quite unlikely, so we stop here } - maxLen := maxBytes - frameHeaderBytes + maxLen := maxBytes - currentLen - frameHeaderBytes if s.lenOfDataForWriting() != 0 { - fcAllowance, err := f.getFCAllowanceForStream(s) - if err != nil { - return nil, err - } + fcAllowance, _ := f.getFCAllowanceForStream(s) // can never error maxLen = utils.MinByteCount(maxLen, fcAllowance) } @@ -162,15 +164,15 @@ func (f *streamFramer) maybePopNormalFrame(maxBytes protocol.ByteCount) (*frames if s.shouldSendFin() { frame.FinBit = true s.sentFin() - return frame, nil + res = append(res, frame) + currentLen += frameHeaderBytes + frame.DataLen() + frame = &frames.StreamFrame{DataLenPresent: true} } continue } frame.Data = data - if err := f.flowControlManager.AddBytesSent(s.streamID, protocol.ByteCount(len(data))); err != nil { - return nil, err - } + f.flowControlManager.AddBytesSent(s.streamID, protocol.ByteCount(len(data))) // Finally, check if we are now FC blocked and should queue a BLOCKED frame individualFcOffset, _ := f.flowControlManager.SendWindowSize(s.streamID) // can never error @@ -183,9 +185,11 @@ func (f *streamFramer) maybePopNormalFrame(maxBytes protocol.ByteCount) (*frames f.blockedFrameQueue = append(f.blockedFrameQueue, &frames.BlockedFrame{StreamID: 0}) } - return frame, nil + res = append(res, frame) + currentLen += frameHeaderBytes + frame.DataLen() + frame = &frames.StreamFrame{DataLenPresent: true} } - return nil, nil + return } func (f *streamFramer) getFCAllowanceForStream(s *stream) (protocol.ByteCount, error) { diff --git a/stream_framer_test.go b/stream_framer_test.go index 2c34dc3d..f7cd212e 100644 --- a/stream_framer_test.go +++ b/stream_framer_test.go @@ -47,16 +47,16 @@ var _ = Describe("Stream Framer", func() { It("sets the DataLenPresent for dequeued retransmitted frames", func() { framer.AddFrameForRetransmission(retransmittedFrame1) - f, err := framer.PopStreamFrame(protocol.MaxByteCount) - Expect(err).NotTo(HaveOccurred()) - Expect(f.DataLenPresent).To(BeTrue()) + fs := framer.PopStreamFrames(protocol.MaxByteCount) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].DataLenPresent).To(BeTrue()) }) It("sets the DataLenPresent for dequeued normal frames", func() { stream1.dataForWriting = []byte("foobar") - f, err := framer.PopStreamFrame(protocol.MaxByteCount) - Expect(err).NotTo(HaveOccurred()) - Expect(f.DataLenPresent).To(BeTrue()) + fs := framer.PopStreamFrames(protocol.MaxByteCount) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].DataLenPresent).To(BeTrue()) }) Context("HasData", func() { @@ -103,9 +103,9 @@ var _ = Describe("Stream Framer", func() { framer.AddFrameForRetransmission(retransmittedFrame1) stream1.dataForWriting = []byte("foobar") Expect(framer.EstimatedDataLen()).To(Equal(protocol.ByteCount(2 + 6))) - framer.PopStreamFrame(1000) + framer.PopStreamFrames(8) Expect(framer.EstimatedDataLen()).To(Equal(protocol.ByteCount(6))) - framer.PopStreamFrame(1000) + framer.PopStreamFrames(1000) Expect(framer.EstimatedDataLen()).To(BeZero()) }) @@ -130,66 +130,52 @@ var _ = Describe("Stream Framer", func() { Context("Popping", func() { It("returns nil when popping an empty framer", func() { - Expect(framer.PopStreamFrame(1000)).To(BeNil()) + Expect(framer.PopStreamFrames(1000)).To(BeEmpty()) }) It("pops frames for retransmission", func() { framer.AddFrameForRetransmission(retransmittedFrame1) framer.AddFrameForRetransmission(retransmittedFrame2) - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).To(Equal(retransmittedFrame1)) - frame, err = framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).To(Equal(retransmittedFrame2)) - frame, err = framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).To(BeNil()) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(2)) + Expect(fs[0]).To(Equal(retransmittedFrame1)) + Expect(fs[1]).To(Equal(retransmittedFrame2)) + Expect(framer.PopStreamFrames(1000)).To(BeEmpty()) }) It("returns normal frames", func() { stream1.dataForWriting = []byte("foobar") - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.StreamID).To(Equal(stream1.streamID)) - Expect(frame.Data).To(Equal([]byte("foobar"))) - frame, err = framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).To(BeNil()) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].StreamID).To(Equal(stream1.streamID)) + Expect(fs[0].Data).To(Equal([]byte("foobar"))) + Expect(framer.PopStreamFrames(1000)).To(BeEmpty()) }) It("returns multiple normal frames", func() { stream1.dataForWriting = []byte("foobar") stream2.dataForWriting = []byte("foobaz") - frame1, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - frame2, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(2)) // Swap if we dequeued in other order - if frame1.StreamID != stream1.streamID { - frame1, frame2 = frame2, frame1 + if fs[0].StreamID != stream1.streamID { + fs[0], fs[1] = fs[1], fs[0] } - Expect(frame1.StreamID).To(Equal(stream1.streamID)) - Expect(frame1.Data).To(Equal([]byte("foobar"))) - Expect(frame2.StreamID).To(Equal(stream2.streamID)) - Expect(frame2.Data).To(Equal([]byte("foobaz"))) - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).To(BeNil()) + Expect(fs[0].StreamID).To(Equal(stream1.streamID)) + Expect(fs[0].Data).To(Equal([]byte("foobar"))) + Expect(fs[1].StreamID).To(Equal(stream2.streamID)) + Expect(fs[1].Data).To(Equal([]byte("foobaz"))) + Expect(framer.PopStreamFrames(1000)).To(BeEmpty()) }) It("returns retransmission frames before normal frames", func() { framer.AddFrameForRetransmission(retransmittedFrame1) stream1.dataForWriting = []byte("foobar") - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).To(Equal(retransmittedFrame1)) - frame, err = framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.StreamID).To(Equal(stream1.streamID)) - frame, err = framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).To(BeNil()) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(2)) + Expect(fs[0]).To(Equal(retransmittedFrame1)) + Expect(fs[1].StreamID).To(Equal(stream1.streamID)) + Expect(framer.PopStreamFrames(1000)).To(BeEmpty()) }) Context("splitting of frames", func() { @@ -229,48 +215,46 @@ var _ = Describe("Stream Framer", func() { It("splits a frame", func() { framer.AddFrameForRetransmission(retransmittedFrame2) origlen := retransmittedFrame2.DataLen() - frame, err := framer.PopStreamFrame(6) - Expect(err).ToNot(HaveOccurred()) - minLength, _ := frame.MinLength(0) - Expect(minLength + frame.DataLen()).To(Equal(protocol.ByteCount(6))) - Expect(framer.retransmissionQueue[0].Data).To(HaveLen(int(origlen - frame.DataLen()))) - Expect(framer.retransmissionQueue[0].Offset).To(Equal(frame.DataLen())) + fs := framer.PopStreamFrames(6) + Expect(fs).To(HaveLen(1)) + minLength, _ := fs[0].MinLength(0) + Expect(minLength + fs[0].DataLen()).To(Equal(protocol.ByteCount(6))) + Expect(framer.retransmissionQueue[0].Data).To(HaveLen(int(origlen - fs[0].DataLen()))) + Expect(framer.retransmissionQueue[0].Offset).To(Equal(fs[0].DataLen())) }) It("only removes a frame from the framer after returning all split parts", func() { framer.AddFrameForRetransmission(retransmittedFrame2) - frame, err := framer.PopStreamFrame(6) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).ToNot(BeNil()) + fs := framer.PopStreamFrames(6) + Expect(fs).To(HaveLen(1)) Expect(framer.HasData()).To(BeTrue()) - frame, err = framer.PopStreamFrame(100) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).ToNot(BeNil()) + fs = framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) Expect(framer.HasData()).To(BeFalse()) }) It("gets the whole data of a frame if it was split", func() { origdata := []byte("foobar") stream1.dataForWriting = origdata - frame, err := framer.PopStreamFrame(7) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.Data).To(Equal([]byte("foo"))) + fs := framer.PopStreamFrames(7) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].Data).To(Equal([]byte("foo"))) var b bytes.Buffer - frame.Write(&b, 0) + fs[0].Write(&b, 0) Expect(b.Len()).To(Equal(7)) - frame, err = framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.Data).To(Equal([]byte("bar"))) + fs = framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].Data).To(Equal([]byte("bar"))) }) It("correctly calculates the byte length when returning a split frame", func() { framer.AddFrameForRetransmission(retransmittedFrame1) framer.AddFrameForRetransmission(retransmittedFrame2) startByteLength := framer.EstimatedDataLen() - frame, err := framer.PopStreamFrame(6) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.StreamID).To(Equal(retransmittedFrame1.StreamID)) // make sure the right frame was popped - Expect(framer.EstimatedDataLen()).To(Equal(startByteLength - frame.DataLen())) + fs := framer.PopStreamFrames(6) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].StreamID).To(Equal(retransmittedFrame1.StreamID)) // make sure the right frame was popped + Expect(framer.EstimatedDataLen()).To(Equal(startByteLength - fs[0].DataLen())) }) }) @@ -278,12 +262,12 @@ var _ = Describe("Stream Framer", func() { It("sends FINs when streams are closed", func() { stream1.writeOffset = 42 stream1.closed = 1 - frame, err := framer.PopStreamFrame(1000) - Expect(err).NotTo(HaveOccurred()) - Expect(frame.StreamID).To(Equal(stream1.streamID)) - Expect(frame.Offset).To(Equal(stream1.writeOffset)) - Expect(frame.FinBit).To(BeTrue()) - Expect(frame.Data).To(BeEmpty()) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].StreamID).To(Equal(stream1.streamID)) + Expect(fs[0].Offset).To(Equal(stream1.writeOffset)) + Expect(fs[0].FinBit).To(BeTrue()) + Expect(fs[0].Data).To(BeEmpty()) }) }) }) @@ -291,15 +275,13 @@ var _ = Describe("Stream Framer", func() { Context("flow control", func() { It("tells the FlowControlManager how many bytes it sent", func() { stream1.dataForWriting = []byte("foobar") - _, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) + framer.PopStreamFrames(1000) Expect(fcm.bytesSent).To(Equal(protocol.ByteCount(6))) }) It("does not count retransmitted frames as sent bytes", func() { framer.AddFrameForRetransmission(retransmittedFrame1) - _, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) + framer.PopStreamFrames(1000) Expect(fcm.bytesSent).To(BeZero()) }) @@ -307,62 +289,61 @@ var _ = Describe("Stream Framer", func() { stream1.writeOffset = 10 stream1.dataForWriting = []byte("foobar") fcm.sendWindowSizes[stream1.streamID] = 10 + 6 - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.DataLen()).To(Equal(protocol.ByteCount(6))) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].DataLen()).To(Equal(protocol.ByteCount(6))) }) It("returns a smaller frame if the whole frame doesn't fit", func() { stream1.dataForWriting = []byte("foobar") fcm.sendWindowSizes[stream1.streamID] = 3 - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.Data).To(Equal([]byte("foo"))) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].Data).To(Equal([]byte("foo"))) }) It("returns a smaller frame if the whole frame doesn't fit in the stream flow control window, for non-zero StreamFrame offset", func() { stream1.writeOffset = 1 stream1.dataForWriting = []byte("foobar") fcm.sendWindowSizes[stream1.StreamID()] = 4 - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.Data).To(Equal([]byte("foo"))) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].Data).To(Equal([]byte("foo"))) }) It("returns a smaller frame if the whole frame doesn't fit in the connection flow control window", func() { stream1.dataForWriting = []byte("foobar") fcm.streamsContributing = []protocol.StreamID{stream1.StreamID()} fcm.remainingConnectionWindowSize = 3 - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.Data).To(Equal([]byte("foo"))) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].Data).To(Equal([]byte("foo"))) }) It("ignores the connection flow control window for non-contributing streams", func() { stream1.dataForWriting = []byte("foobar") fcm.remainingConnectionWindowSize = 0 - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.Data).To(Equal([]byte("foobar"))) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].Data).To(Equal([]byte("foobar"))) }) It("respects the connection flow control window for contributing streams", func() { stream1.dataForWriting = []byte("foobar") fcm.remainingConnectionWindowSize = 0 fcm.streamsContributing = []protocol.StreamID{stream1.StreamID()} - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).To(BeNil()) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(BeEmpty()) }) It("selects a stream that is not flow control blocked", func() { fcm.sendWindowSizes[stream1.StreamID()] = 0 stream1.dataForWriting = []byte("foobar") stream2.dataForWriting = []byte("foobaz") - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.StreamID).To(Equal(stream2.StreamID())) - Expect(frame.Data).To(Equal([]byte("foobaz"))) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].StreamID).To(Equal(stream2.StreamID())) + Expect(fs[0].Data).To(Equal([]byte("foobaz"))) }) It("chooses a non-contributing stream if the connection is flow control blocked", func() { @@ -370,9 +351,9 @@ var _ = Describe("Stream Framer", func() { stream2.dataForWriting = []byte("foobaz") fcm.streamsContributing = []protocol.StreamID{stream1.StreamID()} fcm.remainingConnectionWindowSize = 0 - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame.StreamID).To(Equal(stream2.StreamID())) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(HaveLen(1)) + Expect(fs[0].StreamID).To(Equal(stream2.StreamID())) }) It("returns nil if every stream is individually flow control blocked", func() { @@ -380,9 +361,8 @@ var _ = Describe("Stream Framer", func() { fcm.sendWindowSizes[stream2.StreamID()] = 0 stream1.dataForWriting = []byte("foobar") stream2.dataForWriting = []byte("foobaz") - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).To(BeNil()) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(BeEmpty()) }) It("returns nil if every stream is connection flow control blocked", func() { @@ -390,9 +370,8 @@ var _ = Describe("Stream Framer", func() { stream1.dataForWriting = []byte("foobar") stream2.dataForWriting = []byte("foobaz") fcm.streamsContributing = []protocol.StreamID{stream1.StreamID(), stream2.StreamID()} - frame, err := framer.PopStreamFrame(1000) - Expect(err).ToNot(HaveOccurred()) - Expect(frame).To(BeNil()) + fs := framer.PopStreamFrames(1000) + Expect(fs).To(BeEmpty()) }) }) @@ -408,8 +387,7 @@ var _ = Describe("Stream Framer", func() { It("queues and pops BLOCKED frames for individually blocked streams", func() { fcm.sendWindowSizes[stream1.StreamID()] = 3 stream1.dataForWriting = []byte("foo") - _, err := framer.PopStreamFrame(1000) - Expect(err).NotTo(HaveOccurred()) + framer.PopStreamFrames(1000) blockedFrame := framer.PopBlockedFrame() Expect(blockedFrame).ToNot(BeNil()) Expect(blockedFrame.StreamID).To(Equal(stream1.StreamID())) @@ -420,8 +398,7 @@ var _ = Describe("Stream Framer", func() { fcm.remainingConnectionWindowSize = 3 fcm.streamsContributing = []protocol.StreamID{stream1.StreamID()} stream1.dataForWriting = []byte("foo") - _, err := framer.PopStreamFrame(1000) - Expect(err).NotTo(HaveOccurred()) + framer.PopStreamFrames(1000) blockedFrame := framer.PopBlockedFrame() Expect(blockedFrame).ToNot(BeNil()) Expect(blockedFrame.StreamID).To(BeZero()) @@ -431,19 +408,14 @@ var _ = Describe("Stream Framer", func() { It("does not queue BLOCKED frames for non-contributing streams", func() { fcm.remainingConnectionWindowSize = 3 stream1.dataForWriting = []byte("foo") - _, err := framer.PopStreamFrame(1000) - Expect(err).NotTo(HaveOccurred()) + framer.PopStreamFrames(1000) Expect(framer.PopBlockedFrame()).To(BeNil()) }) It("does not queue BLOCKED frames twice", func() { fcm.sendWindowSizes[stream1.StreamID()] = 3 stream1.dataForWriting = []byte("foobar") - _, err := framer.PopStreamFrame(1000) - Expect(err).NotTo(HaveOccurred()) - frame, err := framer.PopStreamFrame(1000) - Expect(err).NotTo(HaveOccurred()) - Expect(frame).To(BeNil()) + framer.PopStreamFrames(1000) blockedFrame := framer.PopBlockedFrame() Expect(blockedFrame).ToNot(BeNil()) Expect(blockedFrame.StreamID).To(Equal(stream1.StreamID()))