From 58ea3665b0550c3d181693ee275c3f7cfe0c9bec Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Miroslav=20=C5=A0ediv=C3=BD?= <sedivy.miro@gmail.com>
Date: Fri, 5 Feb 2021 12:18:46 +0100
Subject: [PATCH] unify capture shutdown.

---
 internal/capture/broadcast.go  |  6 ++++
 internal/capture/manager.go    | 10 +++----
 internal/capture/screencast.go | 53 +++++++++++++++++++---------------
 internal/capture/stream.go     |  2 +-
 internal/types/capture.go      |  2 --
 5 files changed, 42 insertions(+), 31 deletions(-)

diff --git a/internal/capture/broadcast.go b/internal/capture/broadcast.go
index 50d6eb65..3bcd8ca3 100644
--- a/internal/capture/broadcast.go
+++ b/internal/capture/broadcast.go
@@ -25,6 +25,12 @@ func broadcastNew(config *config.Capture) *BroacastManagerCtx {
 	}
 }
 
+func (manager *BroacastManagerCtx) shutdown() {
+	manager.logger.Info().Msgf("shutting down")
+
+	manager.destroyPipeline()
+}
+
 func (manager *BroacastManagerCtx) Start(url string) error {
 	manager.url = url
 	manager.enabled = true
diff --git a/internal/capture/manager.go b/internal/capture/manager.go
index 2f2b660a..cf76adde 100644
--- a/internal/capture/manager.go
+++ b/internal/capture/manager.go
@@ -81,11 +81,11 @@ func (manager *CaptureManagerCtx) Start() {
 func (manager *CaptureManagerCtx) Shutdown() error {
 	manager.logger.Info().Msgf("capture shutting down")
 
-	manager.broadcast.destroyPipeline()
-	manager.screencast.destroyPipeline()
-	manager.screencast.shutdown <- true
-	manager.audio.Shutdown()
-	manager.video.Shutdown()
+	manager.broadcast.shutdown()
+	manager.screencast.shutdown()
+
+	manager.audio.shutdown()
+	manager.video.shutdown()
 	return nil
 }
 
diff --git a/internal/capture/screencast.go b/internal/capture/screencast.go
index 56d8ea72..8b210b7c 100644
--- a/internal/capture/screencast.go
+++ b/internal/capture/screencast.go
@@ -15,27 +15,27 @@ import (
 )
 
 type ScreencastManagerCtx struct {
-	logger    zerolog.Logger
-	mu        sync.Mutex
-	config    *config.Capture
-	pipeline  *gst.Pipeline
-	enabled   bool
-	started   bool
-	shutdown  chan bool
-	refresh   chan bool
-	expired   int32
-	sample    chan types.Sample
-	image     types.Sample
+	logger      zerolog.Logger
+	mu          sync.Mutex
+	config      *config.Capture
+	pipeline    *gst.Pipeline
+	enabled     bool
+	started     bool
+	emitStop    chan bool
+	emitUpdate  chan bool
+	expired     int32
+	sample      chan types.Sample
+	image       types.Sample
 }
 
 func screencastNew(config *config.Capture) *ScreencastManagerCtx {
 	manager := &ScreencastManagerCtx{
-		logger:    log.With().Str("module", "capture").Str("submodule", "screencast").Logger(),
-		config:    config,
-		enabled:   config.Screencast,
-		started:   false,
-		shutdown:  make(chan bool),
-		refresh:   make(chan bool),
+		logger:      log.With().Str("module", "capture").Str("submodule", "screencast").Logger(),
+		config:      config,
+		enabled:     config.Screencast,
+		started:     false,
+		emitStop:    make(chan bool),
+		emitUpdate:  make(chan bool),
 	}
 
 	if !manager.enabled {
@@ -44,16 +44,16 @@ func screencastNew(config *config.Capture) *ScreencastManagerCtx {
 
 	go func() {
 		ticker := time.NewTicker(5 * time.Second)
-		manager.logger.Debug().Msg("subroutine started")
+		manager.logger.Debug().Msg("started emitting samples")
 
 		for {
 			select {
-			case <-manager.shutdown:
-				manager.logger.Debug().Msg("shutting down")
+			case <-manager.emitStop:
+				manager.logger.Debug().Msg("stopped emitting samples")
 				ticker.Stop()
 				return
-			case <-manager.refresh:
-				manager.logger.Debug().Msg("subroutine updated")
+			case <-manager.emitUpdate:
+				manager.logger.Debug().Msg("update emitting samples")
 			case sample := <-manager.sample:
 				manager.image = sample
 			case <-ticker.C:
@@ -67,6 +67,13 @@ func screencastNew(config *config.Capture) *ScreencastManagerCtx {
 	return manager
 }
 
+func (manager *ScreencastManagerCtx) shutdown() {
+	manager.logger.Info().Msgf("shutting down")
+
+	manager.destroyPipeline()
+	manager.emitStop <- true
+}
+
 func (manager *ScreencastManagerCtx) Enabled() bool {
 	return manager.enabled
 }
@@ -144,7 +151,7 @@ func (manager *ScreencastManagerCtx) createPipeline() error {
 	manager.logger.Info().Msgf("starting pipeline")
 
 	manager.sample = manager.pipeline.Sample
-	manager.refresh <-true
+	manager.emitUpdate <-true
 	return nil
 }
 
diff --git a/internal/capture/stream.go b/internal/capture/stream.go
index e5494a2d..b98722a4 100644
--- a/internal/capture/stream.go
+++ b/internal/capture/stream.go
@@ -58,7 +58,7 @@ func streamNew(codec codec.RTPCodec, pipelineDevice string, pipelineSrc string)
 	return manager
 }
 
-func (manager *StreamManagerCtx) Shutdown() {
+func (manager *StreamManagerCtx) shutdown() {
 	manager.logger.Info().Msgf("shutting down")
 
 	manager.destroyPipeline()
diff --git a/internal/types/capture.go b/internal/types/capture.go
index bc61813d..0fa0737d 100644
--- a/internal/types/capture.go
+++ b/internal/types/capture.go
@@ -22,8 +22,6 @@ type ScreencastManager interface {
 }
 
 type StreamManager interface {
-	Shutdown()
-
 	Codec() codec.RTPCodec
 	OnSample(listener func(sample Sample))