make sure that all packets in the send queue are sent before closing

send_queue.go

@@ -1,16 +1,18 @@
 package quic
 
 type sendQueue struct {
-	queue     chan *packedPacket
+	queue       chan *packedPacket
-	closeChan chan struct{}
+	closeCalled chan struct{} // runStopped when Close() is called
-	conn      connection
+	runStopped  chan struct{} // runStopped when the run loop returns
+	conn        connection
 }
 
 func newSendQueue(conn connection) *sendQueue {
 	s := &sendQueue{
-		conn:      conn,
+		conn:        conn,
-		closeChan: make(chan struct{}),
+		runStopped:  make(chan struct{}),
-		queue:     make(chan *packedPacket, 1),
+		closeCalled: make(chan struct{}),
+		queue:       make(chan *packedPacket, 1),
 	}
 	return s
 }
@@ -20,20 +22,28 @@ func (h *sendQueue) Send(p *packedPacket) {
 }
 
 func (h *sendQueue) Run() error {
-	var p *packedPacket
+	defer close(h.runStopped)
+	var shouldClose bool
 	for {
-		select {
+		if shouldClose && len(h.queue) == 0 {
-		case <-h.closeChan:
 			return nil
-		case p = <-h.queue:
 		}
-		if err := h.conn.Write(p.raw); err != nil {
+		select {
-			return err
+		case <-h.closeCalled:
+			h.closeCalled = nil // prevent this case from being selected again
+			// make sure that all queued packets are actually sent out
+			shouldClose = true
+		case p := <-h.queue:
+			if err := h.conn.Write(p.raw); err != nil {
+				return err
+			}
+			p.buffer.Release()
 		}
-		p.buffer.Release()
 	}
 }
 
 func (h *sendQueue) Close() {
-	close(h.closeChan)
+	close(h.closeCalled)
+	// wait until the run loop returned
+	<-h.runStopped
 }

send_queue_test.go

@@ -70,4 +70,30 @@ var _ = Describe("Send Queue", func() {
 		q.Close()
 		Eventually(done).Should(BeClosed())
 	})
+
+	It("blocks Close() until the packet has been sent out", func() {
+		written := make(chan []byte)
+		c.EXPECT().Write(gomock.Any()).Do(func(p []byte) { written <- p })
+		done := make(chan struct{})
+		go func() {
+			defer GinkgoRecover()
+			q.Run()
+			close(done)
+		}()
+
+		q.Send(getPacket([]byte("foobar")))
+
+		closed := make(chan struct{})
+		go func() {
+			defer GinkgoRecover()
+			q.Close()
+			close(closed)
+		}()
+
+		Consistently(closed).ShouldNot(BeClosed())
+		// now write the packet
+		Expect(written).To(Receive())
+		Eventually(done).Should(BeClosed())
+		Eventually(closed).Should(BeClosed())
+	})
 })