stream use video_id in logs.

internal/capture/broadcast.go

@@ -76,7 +76,7 @@ func (manager *BroacastManagerCtx) createPipeline() error {
 	pipelineStr := strings.Replace(manager.pipelineStr, "{url}", manager.url, 1)
 
 	manager.logger.Info().
-		Str("str", pipelineStr).
+		Str("src", pipelineStr).
 		Msgf("starting pipeline")
 
 	manager.pipeline, err = gst.CreatePipeline(pipelineStr)

internal/capture/manager.go

@@ -57,7 +57,7 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
 	}
 
 	videos := map[string]*StreamManagerCtx{}
-	for key, cnf := range config.VideoPipelines {
+	for video_id, cnf := range config.VideoPipelines {
 		pipelineConf := cnf
 
 		createPipeline := func() string {
@@ -69,7 +69,7 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
 			pipeline, err := pipelineConf.GetPipeline(*screen)
 			if err != nil {
 				logger.Panic().Err(err).
-					Str("video_id", key).
+					Str("video_id", video_id).
 					Msg("unable to get video pipeline")
 			}
 
@@ -82,12 +82,12 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
 		// trigger function to catch evaluation errors at startup
 		pipeline := createPipeline()
 		logger.Info().
-			Str("video_id", key).
+			Str("video_id", video_id).
 			Str("pipeline", pipeline).
 			Msg("syntax check for video stream pipeline passed")
 
 		// append to videos
-		videos[key] = streamNew(config.VideoCodec, createPipeline)
+		videos[video_id] = streamNew(config.VideoCodec, createPipeline, video_id)
 	}
 
 	return &CaptureManagerCtx{
@@ -109,7 +109,7 @@ func New(desktop types.DesktopManager, config *config.Capture) *CaptureManagerCt
 					"! %s "+
 					"! appsink name=appsink", config.AudioDevice, config.AudioCodec.Pipeline,
 			)
-		}),
+		}, "audio"),
 		videos:   videos,
 		videoIDs: config.VideoIDs,
 	}

internal/capture/stream.go

@@ -27,9 +27,14 @@ type StreamManagerCtx struct {
 	started     bool
 }
 
-func streamNew(codec codec.RTPCodec, pipelineStr func() string) *StreamManagerCtx {
+func streamNew(codec codec.RTPCodec, pipelineStr func() string, video_id string) *StreamManagerCtx {
+	logger := log.With().
+		Str("module", "capture").
+		Str("submodule", "stream").
+		Str("video_id", video_id).Logger()
+
 	manager := &StreamManagerCtx{
-		logger:      log.With().Str("module", "capture").Str("submodule", "stream").Logger(),
+		logger:      logger,
 		codec:       codec,
 		pipelineStr: pipelineStr,
 		listeners:   map[uintptr]*func(sample types.Sample){},
@@ -85,6 +90,7 @@ func (manager *StreamManagerCtx) AddListener(listener *func(sample types.Sample)
 	if listener != nil {
 		ptr := reflect.ValueOf(listener).Pointer()
 		manager.listeners[ptr] = listener
+		manager.logger.Debug().Interface("ptr", ptr).Msgf("adding listener")
 	}
 }
 
@@ -95,6 +101,7 @@ func (manager *StreamManagerCtx) RemoveListener(listener *func(sample types.Samp
 	if listener != nil {
 		ptr := reflect.ValueOf(listener).Pointer()
 		delete(manager.listeners, ptr)
+		manager.logger.Debug().Interface("ptr", ptr).Msgf("removing listener")
 	}
 }
 
@@ -114,6 +121,7 @@ func (manager *StreamManagerCtx) Start() error {
 		return err
 	}
 
+	manager.logger.Info().Msgf("start")
 	manager.started = true
 	return nil
 }
@@ -122,6 +130,7 @@ func (manager *StreamManagerCtx) Stop() {
 	manager.mu.Lock()
 	defer manager.mu.Unlock()
 
+	manager.logger.Info().Msgf("stop")
 	manager.started = false
 	manager.destroyPipeline()
 }