only close sinks if started

pkg/pipeline/controller.go

@@ -208,6 +208,13 @@ func (c *Controller) Run(ctx context.Context) *livekit.EgressInfo {
 		return c.Info
 	}
 
+	for _, s := range c.sinks {
+		if err := s.Close(); err != nil {
+			c.Info.Error = err.Error()
+			return c.Info
+		}
+	}
+
 	return c.Info
 }
 

pkg/pipeline/sink/file.go

@@ -43,7 +43,7 @@ func (s *FileSink) Start() error {
 	return nil
 }
 
-func (s *FileSink) OnStop() error {
+func (s *FileSink) Close() error {
 	location, size, err := s.Upload(s.LocalFilepath, s.StorageFilepath, s.OutputType, false)
 	if err != nil {
 		return err

pkg/pipeline/sink/segments.go

@@ -207,7 +207,7 @@ func (s *SegmentSink) endSegment(filename string, endTime int64) error {
 	return nil
 }
 
-func (s *SegmentSink) OnStop() error {
+func (s *SegmentSink) Close() error {
 	// wait for all pending upload jobs to finish
 	close(s.endedSegments)
 	<-s.done.Watch()

pkg/pipeline/sink/sink.go

@@ -23,7 +23,7 @@ import (
 
 type Sink interface {
 	Start() error
-	OnStop() error
+	Close() error
 	Cleanup()
 }
 
@@ -69,7 +69,6 @@ func CreateSinks(p *config.PipelineConfig, callbacks *gstreamer.Callbacks) (map[
 		}
 
 		if s != nil {
-			callbacks.AddOnStop(s.OnStop)
 			sinks[egressType] = s
 		}
 	}

pkg/pipeline/sink/websocket.go

@@ -196,10 +196,6 @@ func (s *WebsocketSink) writeMutedMessage(muted bool) error {
 	return s.conn.WriteMessage(websocket.TextMessage, data)
 }
 
-func (s *WebsocketSink) OnStop() error {
-	return nil
-}
-
 func (s *WebsocketSink) Close() error {
 	s.mu.Lock()
 	defer s.mu.Unlock()