// Copyright 2023 LiveKit, Inc. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package dynacast import ( "maps" "slices" "sync" "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/logger" "github.com/livekit/livekit-server/pkg/utils" ) type dynacastManagerBaseParams struct { Logger logger.Logger OpsQueueDepth uint OnRestart func() OnDynacastQualityCreate func(mimeType mime.MimeType) dynacastQuality OnRegressCodec func(fromMime, toMime mime.MimeType) OnUpdateNeeded func(force bool) } type dynacastManagerBase struct { params dynacastManagerBaseParams lock sync.RWMutex regressedCodec map[mime.MimeType]struct{} dynacastQuality map[mime.MimeType]dynacastQuality notifyOpsQueue *utils.OpsQueue isClosed bool dynacastManagerNull dynacastQualityListenerNull } func newDynacastManagerBase(params dynacastManagerBaseParams) *dynacastManagerBase { if params.OpsQueueDepth == 0 { params.OpsQueueDepth = 4 } d := &dynacastManagerBase{ params: params, regressedCodec: make(map[mime.MimeType]struct{}), dynacastQuality: make(map[mime.MimeType]dynacastQuality), notifyOpsQueue: utils.NewOpsQueue(utils.OpsQueueParams{ Name: "dynacast-notify", MinSize: params.OpsQueueDepth, FlushOnStop: true, Logger: params.Logger, }), } d.notifyOpsQueue.Start() return d } func (d *dynacastManagerBase) AddCodec(mime mime.MimeType) { d.getOrCreateDynacastQuality(mime) } func (d *dynacastManagerBase) HandleCodecRegression(fromMime, toMime mime.MimeType) { fromDq := d.getOrCreateDynacastQuality(fromMime) d.lock.Lock() if d.isClosed { d.lock.Unlock() return } if fromDq == nil { // should not happen as we have added the codec on setup receiver d.params.Logger.Warnw("regression from codec not found", nil, "mime", fromMime, "toMime", toMime) d.lock.Unlock() return } d.regressedCodec[fromMime] = struct{}{} d.params.OnRegressCodec(fromMime, toMime) d.lock.Unlock() d.params.OnUpdateNeeded(false) fromDq.Stop() fromDq.RegressTo(d.getOrCreateDynacastQuality(toMime)) } func (d *dynacastManagerBase) Restart() { d.lock.Lock() d.params.OnRestart() dqs := d.getDynacastQualitiesLocked() d.lock.Unlock() for _, dq := range dqs { dq.Restart() } } func (d *dynacastManagerBase) Close() { d.notifyOpsQueue.Stop() d.lock.Lock() dqs := d.getDynacastQualitiesLocked() d.dynacastQuality = make(map[mime.MimeType]dynacastQuality) d.isClosed = true d.lock.Unlock() for _, dq := range dqs { dq.Stop() } } // There are situations like track unmute or streaming from a different node // where subscription changes needs to sent to the provider immediately. // This bypasses any debouncing and forces a subscription change update // with immediate effect. func (d *dynacastManagerBase) ForceUpdate() { d.params.OnUpdateNeeded(true) } func (d *dynacastManagerBase) ClearSubscriberNodes() { d.lock.Lock() dqs := d.getDynacastQualitiesLocked() d.lock.Unlock() for _, dq := range dqs { dq.ClearSubscriberNodes() } } func (d *dynacastManagerBase) getOrCreateDynacastQuality(mimeType mime.MimeType) dynacastQuality { d.lock.Lock() defer d.lock.Unlock() if d.isClosed || mimeType == mime.MimeTypeUnknown { return nil } if dq := d.dynacastQuality[mimeType]; dq != nil { return dq } dq := d.params.OnDynacastQualityCreate(mimeType) dq.Start() d.dynacastQuality[mimeType] = dq return dq } func (d *dynacastManagerBase) getDynacastQualitiesLocked() []dynacastQuality { return slices.Collect(maps.Values(d.dynacastQuality)) }