move change funcs to peer.

internal/webrtc/manager.go

@@ -340,76 +340,42 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int,
 	// let video stream bucket manager handle stream subscriptions
 	video.SetReceiver(videoTrack)
 
-	changeVideoFromBitrate := func(peerBitrate int) {
+	// data channel
-		// when switching from manual to auto bitrate estimation, in case the estimator is
-		// idle (lastBitrate > maxBitrate), we want to go back to the previous estimated bitrate
-		if peerBitrate == 0 && estimator != nil && !manager.config.EstimatorPassive {
-			peerBitrate = estimator.GetTargetBitrate()
-			manager.logger.Debug().
-				Int("peer_bitrate", peerBitrate).
-				Msg("evaluated bitrate")
-		}
 
-		ok, err := videoTrack.SetBitrate(peerBitrate)
+	dataChannel, err := connection.CreateDataChannel("data", nil)
 	if err != nil {
-			logger.Error().Err(err).
+		return nil, err
-				Int("peer_bitrate", peerBitrate).
-				Msg("unable to set video bitrate")
-			return
 	}
 
-		if !ok {
+	peer := &WebRTCPeerCtx{
-			return
+		logger:     logger,
+		session:    session,
+		metrics:    metrics,
+		connection: connection,
+		estimator:  estimator,
+		// tracks & channels
+		audioTrack:  audioTrack,
+		videoTrack:  videoTrack,
+		dataChannel: dataChannel,
+		rtcpChannel: videoRtcp,
+		// config
+		iceTrickle: manager.config.ICETrickle,
+		// deprecated functions
+		videoId: videoTrack.stream.ID,
+		setPaused: func(isPaused bool) {
+			videoTrack.SetPaused(isPaused)
+			audioTrack.SetPaused(isPaused)
+		},
+		setVideoAuto: func(videoAuto bool) {
+			// if estimator is enabled and not in passive mode, enable video auto bitrate
+			if manager.config.EstimatorEnabled && !manager.config.EstimatorPassive {
+				videoTrack.SetVideoAuto(videoAuto)
+			} else {
+				logger.Warn().Msg("estimator is disabled or in passive mode, cannot change video auto")
+				videoTrack.SetVideoAuto(false) // ensure video auto is disabled
 			}
-
+		},
-		videoID := videoTrack.stream.ID()
+		getVideoAuto: videoTrack.VideoAuto,
-		bitrate := videoTrack.stream.Bitrate()
-
-		metrics.SetVideoID(videoID)
-		manager.logger.Debug().
-			Int("peer_bitrate", peerBitrate).
-			Int("video_bitrate", bitrate).
-			Str("video_id", videoID).
-			Msg("peer bitrate triggered video stream change")
-
-		go session.Send(
-			event.SIGNAL_VIDEO,
-			message.SignalVideo{
-				Video:     videoID,
-				Bitrate:   bitrate,
-				VideoAuto: videoTrack.VideoAuto(),
-			})
-	}
-
-	changeVideoFromID := func(videoID string) (bitrate int) {
-		changed, err := videoTrack.SetVideoID(videoID)
-		if err != nil {
-			logger.Error().Err(err).
-				Str("video_id", videoID).
-				Msg("unable to set video stream")
-			return
-		}
-
-		if !changed {
-			return
-		}
-
-		bitrate = videoTrack.stream.Bitrate()
-
-		manager.logger.Debug().
-			Str("video_id", videoID).
-			Int("video_bitrate", bitrate).
-			Msg("peer video id triggered video stream change")
-
-		go session.Send(
-			event.SIGNAL_VIDEO,
-			message.SignalVideo{
-				Video:     videoID,
-				Bitrate:   bitrate,
-				VideoAuto: videoTrack.VideoAuto(),
-			})
-
-		return
 	}
 
 	manager.logger.Info().
@@ -417,7 +383,9 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int,
 		Msg("estimated initial peer bitrate")
 
 	// set initial video bitrate
-	changeVideoFromBitrate(bitrate)
+	if err := peer.SetVideoBitrate(bitrate); err != nil {
+		return nil, err
+	}
 
 	// if estimator is enabled, use it to change video stream
 	if estimator != nil {
@@ -437,51 +405,15 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int,
 					continue
 				}
 				if !manager.config.EstimatorPassive {
-					changeVideoFromBitrate(targetBitrate)
+					err := peer.SetVideoBitrate(targetBitrate)
+					if err != nil {
+						logger.Warn().Err(err).Msg("failed to set video bitrate")
+					}
 				}
 			}
 		}()
 	}
 
-	// data channel
-
-	dataChannel, err := connection.CreateDataChannel("data", nil)
-	if err != nil {
-		return nil, err
-	}
-
-	peer := &WebRTCPeerCtx{
-		logger:     logger,
-		session:    session,
-		metrics:    metrics,
-		connection: connection,
-		// tracks & channels
-		audioTrack:  audioTrack,
-		videoTrack:  videoTrack,
-		dataChannel: dataChannel,
-		rtcpChannel: videoRtcp,
-		// config
-		iceTrickle: manager.config.ICETrickle,
-		// deprecated functions
-		changeVideoFromBitrate: changeVideoFromBitrate,
-		changeVideoFromID:      changeVideoFromID,
-		videoId:                videoTrack.stream.ID,
-		setPaused: func(isPaused bool) {
-			videoTrack.SetPaused(isPaused)
-			audioTrack.SetPaused(isPaused)
-		},
-		setVideoAuto: func(videoAuto bool) {
-			// if estimator is enabled and not in passive mode, enable video auto bitrate
-			if manager.config.EstimatorEnabled && !manager.config.EstimatorPassive {
-				videoTrack.SetVideoAuto(videoAuto)
-			} else {
-				logger.Warn().Msg("estimator is disabled or in passive mode, cannot change video auto")
-				videoTrack.SetVideoAuto(false) // ensure video auto is disabled
-			}
-		},
-		getVideoAuto: videoTrack.VideoAuto,
-	}
-
 	connection.OnTrack(func(track *webrtc.TrackRemote, receiver *webrtc.RTPReceiver) {
 		logger := logger.With().
 			Str("kind", track.Kind().String()).

internal/webrtc/peer.go

@@ -5,12 +5,15 @@ import (
 	"encoding/binary"
 	"sync"
 
+	"github.com/pion/interceptor/pkg/cc"
 	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
 	"github.com/rs/zerolog"
 
 	"github.com/demodesk/neko/internal/webrtc/payload"
 	"github.com/demodesk/neko/pkg/types"
+	"github.com/demodesk/neko/pkg/types/event"
+	"github.com/demodesk/neko/pkg/types/message"
 )
 
 type WebRTCPeerCtx struct {
@@ -19,6 +22,7 @@ type WebRTCPeerCtx struct {
 	session    types.Session
 	metrics    *metrics
 	connection *webrtc.PeerConnection
+	estimator  cc.BandwidthEstimator
 	// tracks & channels
 	audioTrack  *Track
 	videoTrack  *Track
@@ -27,8 +31,6 @@ type WebRTCPeerCtx struct {
 	// config
 	iceTrickle bool
 	// deprecated functions
-	changeVideoFromBitrate func(bitrate int)
-	changeVideoFromID      func(id string) int
 	videoId      func() string
 	setPaused    func(isPaused bool)
 	setVideoAuto func(auto bool)
@@ -111,7 +113,43 @@ func (peer *WebRTCPeerCtx) SetVideoBitrate(peerBitrate int) error {
 	peer.mu.Lock()
 	defer peer.mu.Unlock()
 
-	peer.changeVideoFromBitrate(peerBitrate)
+	// when switching from manual to auto bitrate estimation, in case the estimator is
+	// idle (lastBitrate > maxBitrate), we want to go back to the previous estimated bitrate
+	if peerBitrate == 0 && peer.estimator != nil {
+		peerBitrate = peer.estimator.GetTargetBitrate()
+		peer.logger.Debug().
+			Int("peer_bitrate", peerBitrate).
+			Msg("evaluated bitrate")
+	}
+
+	changed, err := peer.videoTrack.SetBitrate(peerBitrate)
+	if err != nil {
+		return err
+	}
+
+	if !changed {
+		// TODO: return error?
+		return nil
+	}
+
+	videoID := peer.videoTrack.stream.ID()
+	bitrate := peer.videoTrack.stream.Bitrate()
+
+	peer.metrics.SetVideoID(videoID)
+	peer.logger.Debug().
+		Int("peer_bitrate", peerBitrate).
+		Int("video_bitrate", bitrate).
+		Str("video_id", videoID).
+		Msg("peer bitrate triggered video stream change")
+
+	go peer.session.Send(
+		event.SIGNAL_VIDEO,
+		message.SignalVideo{
+			Video:     videoID,
+			Bitrate:   bitrate,
+			VideoAuto: peer.videoTrack.VideoAuto(),
+		})
+
 	return nil
 }
 
@@ -119,7 +157,31 @@ func (peer *WebRTCPeerCtx) SetVideoID(videoID string) error {
 	peer.mu.Lock()
 	defer peer.mu.Unlock()
 
-	peer.changeVideoFromID(videoID)
+	changed, err := peer.videoTrack.SetVideoID(videoID)
+	if err != nil {
+		return err
+	}
+
+	if !changed {
+		// TODO: return error?
+		return nil
+	}
+
+	bitrate := peer.videoTrack.stream.Bitrate()
+
+	peer.logger.Debug().
+		Str("video_id", videoID).
+		Int("video_bitrate", bitrate).
+		Msg("peer video id triggered video stream change")
+
+	go peer.session.Send(
+		event.SIGNAL_VIDEO,
+		message.SignalVideo{
+			Video:     videoID,
+			Bitrate:   bitrate,
+			VideoAuto: peer.videoTrack.VideoAuto(),
+		})
+
 	return nil
 }