fix: Synchronize peer logging with a channel

kylecarbs · kylecarbs · commit f49dbce7503d · 2022-01-30T02:36:53.000Z
We were depending on the close mutex to properly
report connection state. This ensures the RTC
connection is properly closed before returning.
diff --git a/peer/conn.go b/peer/conn.go
@@ -69,6 +69,7 @@ func newWithClientOrServer(servers []webrtc.ICEServer, client bool, opts *ConnOp
 		rtc:                 rtc,
 		offerrer:            client,
 		closed:              make(chan struct{}),
+		closedRTC:           make(chan struct{}),
 		dcOpenChannel:       make(chan *webrtc.DataChannel),
 		dcDisconnectChannel: make(chan struct{}),
 		dcFailedChannel:     make(chan struct{}),
@@ -109,6 +110,7 @@ type Conn struct {
 	offerrer bool
 
 	closed     chan struct{}
+	closedRTC  chan struct{}
 	closeMutex sync.Mutex
 	closeError error
 
@@ -142,26 +144,14 @@ type Conn struct {
 func (c *Conn) init() error {
 	c.rtc.OnNegotiationNeeded(c.negotiate)
 	c.rtc.OnICEConnectionStateChange(func(iceConnectionState webrtc.ICEConnectionState) {
-		// Close must be locked here otherwise log output can appear
-		// after the connection has been closed.
-		c.closeMutex.Lock()
-		defer c.closeMutex.Unlock()
-		if c.isClosed() {
-			return
-		}
-
 		c.opts.Logger.Debug(context.Background(), "ice connection state updated",
 			slog.F("state", iceConnectionState))
 	})
+	c.rtc.OnSignalingStateChange(func(signalState webrtc.SignalingState) {
+		c.opts.Logger.Debug(context.Background(), "signal state updated",
+			slog.F("state", signalState))
+	})
 	c.rtc.OnICEGatheringStateChange(func(iceGatherState webrtc.ICEGathererState) {
-		// Close can't be locked here, because this is triggered
-		// when close is called. It doesn't appear this get's
-		// executed after close though, so it shouldn't cause
-		// problems.
-		if c.isClosed() {
-			return
-		}
-
 		c.opts.Logger.Debug(context.Background(), "ice gathering state updated",
 			slog.F("state", iceGatherState))
 	})
@@ -170,8 +160,9 @@ func (c *Conn) init() error {
 			return
 		}
 		json := iceCandidate.ToJSON()
+		hash := sha256.Sum224([]byte(json.Candidate))
 		c.opts.Logger.Debug(context.Background(), "writing candidate to channel",
-			slog.F("hash", sha256.Sum224([]byte(json.Candidate))),
+			slog.F("hash", string(hash[:])),
 			slog.F("length", len(json.Candidate)),
 		)
 		select {
@@ -189,14 +180,6 @@ func (c *Conn) init() error {
 		}
 	})
 	c.rtc.OnConnectionStateChange(func(pcs webrtc.PeerConnectionState) {
-		// Close must be locked here otherwise log output can appear
-		// after the connection has been closed.
-		c.closeMutex.Lock()
-		defer c.closeMutex.Unlock()
-		if c.isClosed() {
-			return
-		}
-
 		c.opts.Logger.Debug(context.Background(), "rtc connection updated",
 			slog.F("state", pcs))
 
@@ -216,6 +199,13 @@ func (c *Conn) init() error {
 				}
 			}
 		}
+
+		if pcs == webrtc.PeerConnectionStateClosed {
+			// Pion executes event handlers after close is called
+			// on the RTC connection. This ensures our Close()
+			// handler properly cleans up before returning.
+			close(c.closedRTC)
+		}
 	})
 	_, err := c.pingChannel()
 	if err != nil {
@@ -341,7 +331,7 @@ func (c *Conn) negotiate() {
 	for _, pendingCandidate := range c.pendingRemoteCandidates {
 		hash := sha256.Sum224([]byte(pendingCandidate.Candidate))
 		c.opts.Logger.Debug(context.Background(), "flushing buffered remote candidate",
-			slog.F("hash", hash),
+			slog.F("hash", string(hash[:])),
 			slog.F("length", len(pendingCandidate.Candidate)),
 		)
 		err := c.rtc.AddICECandidate(pendingCandidate)
@@ -367,8 +357,9 @@ func (c *Conn) LocalCandidate() <-chan webrtc.ICECandidateInit {
 func (c *Conn) AddRemoteCandidate(i webrtc.ICECandidateInit) error {
 	c.pendingCandidatesMutex.Lock()
 	defer c.pendingCandidatesMutex.Unlock()
+	hash := sha256.Sum224([]byte(i.Candidate))
 	fields := []slog.Field{
-		slog.F("hash", sha256.Sum224([]byte(i.Candidate))),
+		slog.F("hash", string(hash[:])),
 		slog.F("length", len(i.Candidate)),
 	}
 	if !c.pendingCandidatesFlushed {
@@ -542,5 +533,10 @@ func (c *Conn) CloseWithError(err error) error {
 	// All logging, goroutines, and async functionality is cleaned up after this.
 	c.dcClosedWaitGroup.Wait()
 
+	if c.rtc.ConnectionState() != webrtc.PeerConnectionStateNew {
+		c.opts.Logger.Debug(context.Background(), "waiting for connection close...")
+		<-c.closedRTC
+	}
+
 	return err
 }
