Fix routine leak on close

Read from channel was blocked if closed when there is unread data.
Unblock channel read on close.

README.md

@@ -132,6 +132,7 @@ Check out the **[contributing wiki](https://github.com/pion/webrtc/wiki/Contribu
 * [Hayden James](https://github.com/hjames9)
 * [Jozef Kralik](https://github.com/jkralik)
 * [Robert Eperjesi](https://github.com/epes)
+* [Atsushi Watanabe](https://github.com/at-wat)
 
 ### License
 MIT License - see [LICENSE](LICENSE) for full text

conn.go

@@ -82,6 +82,8 @@ type Conn struct {
 
 	bufferedPackets []*packet
 
+	connectionClosed *Closer // Closed on connection close and unblock read
+
 	connErr atomic.Value
 	log     logging.LeveledLogger
 }
@@ -128,6 +130,7 @@ func createConn(nextConn net.Conn, flightHandler flightHandler, handshakeMessage
 	}
 
 	handshakeDoneSignal := NewCloser()
+	connectionClosed := NewCloser()
 
 	c := &Conn{
 		nextConn:                    nextConn,
@@ -156,6 +159,7 @@ func createConn(nextConn net.Conn, flightHandler flightHandler, handshakeMessage
 		decrypted:           make(chan []byte),
 		workerTicker:        time.NewTicker(workerInterval),
 		handshakeDoneSignal: handshakeDoneSignal,
+		connectionClosed:    connectionClosed,
 		log:                 logger,
 	}
 
@@ -624,7 +628,10 @@ func (c *Conn) handleIncomingPacket(buf []byte) (*alert, error) {
 			return &alert{alertLevelFatal, alertUnexpectedMessage}, fmt.Errorf("ApplicationData with epoch of 0")
 		}
 
-		c.decrypted <- content.data
+		select {
+		case c.decrypted <- content.data:
+		case <-c.connectionClosed.Done():
+		}
 	default:
 		return &alert{alertLevelFatal, alertUnexpectedMessage}, fmt.Errorf("unhandled contentType %d", content.contentType())
 	}
@@ -709,6 +716,7 @@ func (c *Conn) stopWithError(err error) {
 	c.workerTicker.Stop()
 
 	c.handshakeDoneSignal.Close()
+	c.connectionClosed.Close()
 }
 
 func (c *Conn) getConnErr() error {

conn_test.go

@@ -73,6 +73,33 @@ func stressDuplex(t *testing.T) {
 	}
 }
 
+func TestRoutineLeakOnClose(t *testing.T) {
+	// Limit runtime in case of deadlocks
+	lim := test.TimeOut(5 * time.Second)
+	defer lim.Stop()
+
+	// Check for leaking routines
+	report := test.CheckRoutines(t)
+	defer report()
+
+	ca, cb, err := pipeMemory()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if _, err := ca.Write(make([]byte, 100)); err != nil {
+		t.Fatal(err)
+	}
+	if err := cb.Close(); err != nil {
+		t.Fatal(err)
+	}
+	if err := ca.Close(); err != nil {
+		t.Fatal(err)
+	}
+	// Packet is sent, but not read.
+	// inboundLoop routine should not be leaked.
+}
+
 func pipeMemory() (*Conn, *Conn, error) {
 	// In memory pipe
 	ca, cb := net.Pipe()