From d38566850a86d99065b7a9cbd3be44b26fd04db6 Mon Sep 17 00:00:00 2001
From: Raja Subramanian <raja.gobi@tutanota.com>
Date: Sun, 24 Apr 2022 11:40:22 +0530
Subject: [PATCH] Do not post close callback in ops queue if not started.
 (#649)

* Do not post close callback in ops queue if not started.

Ops queue is started in `Bind()`. If `Close()` is called
when bind did not happen (because the underlying peer connection
closed before bind), the close callback does not run.

Check if ops queue is running before posting close callback
into the queue.

Not pretty, but covers this case. Need to think about it more.

* correct check
---
 pkg/sfu/downtrack.go  | 26 +++++++++++++++++---------
 pkg/utils/opsqueue.go | 17 +++++++++++++++++
 2 files changed, 34 insertions(+), 9 deletions(-)

diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go
index eae06e6d8..0a7baf472 100644
--- a/pkg/sfu/downtrack.go
+++ b/pkg/sfu/downtrack.go
@@ -595,17 +595,25 @@ func (d *DownTrack) CloseWithFlush(flush bool) {
 		d.rtpStats.Stop()
 		d.logger.Debugw("rtp stats", "stats", d.rtpStats.ToString())
 
-		if d.onMaxLayerChanged != nil && d.kind == webrtc.RTPCodecTypeVideo {
-			d.callbacksQueue.Enqueue(func() {
-				d.onMaxLayerChanged(d, InvalidLayerSpatial)
-			})
-		}
+		if d.callbacksQueue.IsStarted() {
+			if d.kind == webrtc.RTPCodecTypeVideo {
+				d.callbacksQueue.Enqueue(func() {
+					if d.onMaxLayerChanged != nil {
+						d.onMaxLayerChanged(d, InvalidLayerSpatial)
+					}
+				})
+			}
 
-		if d.onCloseHandler != nil {
-			d.callbacksQueue.Enqueue(d.onCloseHandler)
-		}
+			if d.onCloseHandler != nil {
+				d.callbacksQueue.Enqueue(d.onCloseHandler)
+			}
 
-		d.callbacksQueue.Stop()
+			d.callbacksQueue.Stop()
+		} else {
+			if d.onCloseHandler != nil {
+				d.onCloseHandler()
+			}
+		}
 		d.stopKeyFrameRequester()
 	})
 }
diff --git a/pkg/utils/opsqueue.go b/pkg/utils/opsqueue.go
index 63fd2c863..473430a3a 100644
--- a/pkg/utils/opsqueue.go
+++ b/pkg/utils/opsqueue.go
@@ -13,6 +13,7 @@ type OpsQueue struct {
 
 	lock      sync.RWMutex
 	ops       chan func()
+	isStarted bool
 	isStopped bool
 }
 
@@ -30,6 +31,15 @@ func (oq *OpsQueue) SetLogger(logger logger.Logger) {
 }
 
 func (oq *OpsQueue) Start() {
+	oq.lock.Lock()
+	if oq.isStarted {
+		oq.lock.Unlock()
+		return
+	}
+
+	oq.isStarted = true
+	oq.lock.Unlock()
+
 	go oq.process()
 }
 
@@ -45,6 +55,13 @@ func (oq *OpsQueue) Stop() {
 	oq.lock.Unlock()
 }
 
+func (oq *OpsQueue) IsStarted() bool {
+	oq.lock.RLock()
+	defer oq.lock.RUnlock()
+
+	return oq.isStarted
+}
+
 func (oq *OpsQueue) Enqueue(op func()) {
 	oq.lock.RLock()
 	if oq.isStopped {