diff --git a/pkg/agent/testutils/server.go b/pkg/agent/testutils/server.go index 5f045436d..ff3dcf725 100644 --- a/pkg/agent/testutils/server.go +++ b/pkg/agent/testutils/server.go @@ -4,14 +4,15 @@ import ( "context" "errors" "io" + "maps" "math" "math/rand/v2" + "slices" "sync" "time" "github.com/frostbyte73/core" "github.com/gammazero/deque" - "golang.org/x/exp/maps" "github.com/livekit/livekit-server/pkg/agent" "github.com/livekit/livekit-server/pkg/config" @@ -423,7 +424,7 @@ func (w *AgentWorker) SimulateRoomJob(roomName string) { func (w *AgentWorker) Jobs() []*AgentJob { w.mu.Lock() defer w.mu.Unlock() - return maps.Values(w.jobs) + return slices.Collect(maps.Values(w.jobs)) } type stableJobLoad struct { diff --git a/pkg/rtc/dynacast/dynacastmanagerbase.go b/pkg/rtc/dynacast/dynacastmanagerbase.go index a8e251ca2..dff3b9c23 100644 --- a/pkg/rtc/dynacast/dynacastmanagerbase.go +++ b/pkg/rtc/dynacast/dynacastmanagerbase.go @@ -15,10 +15,10 @@ package dynacast import ( + "maps" + "slices" "sync" - "golang.org/x/exp/maps" - "github.com/livekit/protocol/codecs/mime" "github.com/livekit/protocol/logger" @@ -161,5 +161,5 @@ func (d *dynacastManagerBase) getOrCreateDynacastQuality(mimeType mime.MimeType) } func (d *dynacastManagerBase) getDynacastQualitiesLocked() []dynacastQuality { - return maps.Values(d.dynacastQuality) + return slices.Collect(maps.Values(d.dynacastQuality)) } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index a3da1f2dc..5daed00a2 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -17,6 +17,7 @@ package rtc import ( "fmt" "io" + "maps" "math/rand" "os" "slices" @@ -33,7 +34,6 @@ import ( "github.com/pkg/errors" "go.uber.org/atomic" "go.uber.org/zap/zapcore" - "golang.org/x/exp/maps" "google.golang.org/protobuf/proto" "github.com/livekit/mediatransportutil/pkg/twcc" @@ -1443,7 +1443,7 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea // ensure this is synchronized p.CloseSignalConnection(types.SignallingCloseReasonParticipantClose) p.lock.RLock() - onClose := maps.Values(p.onClose) + onClose := slices.Collect(maps.Values(p.onClose)) p.lock.RUnlock() for _, cb := range onClose { cb(p) diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 54afdfad9..94e66fac0 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -17,6 +17,7 @@ package rtc import ( "context" "fmt" + "maps" "math" "slices" "sort" @@ -25,7 +26,6 @@ import ( "time" "go.uber.org/atomic" - "golang.org/x/exp/maps" "google.golang.org/protobuf/proto" "github.com/livekit/protocol/codecs/mime" @@ -196,7 +196,7 @@ func (ad *agentDispatch) jobsLaunching() (jobsLaunched func()) { func (ad *agentDispatch) waitForPendingJobs() { ad.lock.Lock() - cs := maps.Keys(ad.pending) + cs := slices.Collect(maps.Keys(ad.pending)) ad.lock.Unlock() for _, c := range cs { @@ -305,7 +305,7 @@ func NewRoom( r.createAgentDispatchesFromRoomAgent() - r.launchRoomAgents(maps.Values(r.agentDispatches)) + r.launchRoomAgents(slices.Collect(maps.Values(r.agentDispatches))) go r.audioUpdateWorker() go r.connectionQualityWorker() @@ -364,7 +364,7 @@ func (r *Room) GetParticipants() []types.LocalParticipant { r.lock.RLock() defer r.lock.RUnlock() - return maps.Values(r.participants) + return slices.Collect(maps.Values(r.participants)) } func (r *Room) GetLocalParticipants() []types.LocalParticipant { @@ -469,7 +469,7 @@ func (r *Room) Join( r.joinedAt.Store(time.Now().Unix()) } - r.launchTargetAgents(maps.Values(r.agentDispatches), participant, livekit.JobType_JT_PARTICIPANT) + r.launchTargetAgents(slices.Collect(maps.Values(r.agentDispatches)), participant, livekit.JobType_JT_PARTICIPANT) r.logger.Debugw( "new participant joined", @@ -1040,7 +1040,7 @@ func (r *Room) createJoinResponseLocked( OtherParticipants: GetOtherParticipantInfo( participant, false, // isMigratingIn - toParticipants(maps.Values(r.participants)), + toParticipants(slices.Collect(maps.Values(r.participants))), false, // skipSubscriberBroadcast ), IceServers: iceServers, @@ -1104,7 +1104,7 @@ func (r *Room) onTrackPublished(participant types.Participant, track types.Media if !hasPublished { r.lock.RLock() - r.launchTargetAgents(maps.Values(r.agentDispatches), participant, livekit.JobType_JT_PUBLISHER) + r.launchTargetAgents(slices.Collect(maps.Values(r.agentDispatches)), participant, livekit.JobType_JT_PUBLISHER) r.lock.RUnlock() if r.internal != nil && r.internal.ParticipantEgress != nil { go func() { @@ -1601,7 +1601,7 @@ func (r *Room) changeUpdateWorker() { r.batchedUpdates = make(map[livekit.ParticipantIdentity]*ParticipantUpdate) r.batchedUpdatesMu.Unlock() - SendParticipantUpdates(maps.Values(updatesMap), r.GetParticipants(), r.roomConfig.UpdateBatchTargetSize) + SendParticipantUpdates(slices.Collect(maps.Values(updatesMap)), r.GetParticipants(), r.roomConfig.UpdateBatchTargetSize) case <-cleanDataMessageTicker.C: r.dataMessageCache.Prune() diff --git a/pkg/rtc/types/ice.go b/pkg/rtc/types/ice.go index 9b6e36feb..1319a4e4c 100644 --- a/pkg/rtc/types/ice.go +++ b/pkg/rtc/types/ice.go @@ -18,12 +18,12 @@ package types import ( "fmt" + "slices" "strings" "sync" "github.com/pion/ice/v4" "github.com/pion/webrtc/v4" - "golang.org/x/exp/slices" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" diff --git a/pkg/rtc/updatatrackmanager.go b/pkg/rtc/updatatrackmanager.go index 12059731f..ce9387e4e 100644 --- a/pkg/rtc/updatatrackmanager.go +++ b/pkg/rtc/updatatrackmanager.go @@ -15,13 +15,14 @@ package rtc import ( + "maps" + "slices" "sync" "github.com/livekit/livekit-server/pkg/rtc/datatrack" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - "golang.org/x/exp/maps" ) type UpDataTrackManagerParams struct { @@ -94,7 +95,7 @@ func (u *UpDataTrackManager) GetPublishedDataTracks() []types.DataTrack { u.lock.RLock() defer u.lock.RUnlock() - return maps.Values(u.dataTracks) + return slices.Collect(maps.Values(u.dataTracks)) } func (u *UpDataTrackManager) GetPublishedDataTrack(handle uint16) types.DataTrack { diff --git a/pkg/rtc/uptrackmanager.go b/pkg/rtc/uptrackmanager.go index da7c9610e..1964d68f9 100644 --- a/pkg/rtc/uptrackmanager.go +++ b/pkg/rtc/uptrackmanager.go @@ -16,11 +16,12 @@ package rtc import ( "errors" + "maps" + "slices" "sync" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - "golang.org/x/exp/maps" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/protocol/utils" @@ -146,7 +147,7 @@ func (u *UpTrackManager) GetPublishedTracks() []types.MediaTrack { u.lock.RLock() defer u.lock.RUnlock() - return maps.Values(u.publishedTracks) + return slices.Collect(maps.Values(u.publishedTracks)) } func (u *UpTrackManager) UpdateSubscriptionPermission( diff --git a/pkg/rtc/wrappedreceiver.go b/pkg/rtc/wrappedreceiver.go index b6550d931..7b979eae9 100644 --- a/pkg/rtc/wrappedreceiver.go +++ b/pkg/rtc/wrappedreceiver.go @@ -16,12 +16,12 @@ package rtc import ( "errors" + "maps" + "slices" "sync" "github.com/pion/webrtc/v4" "go.uber.org/atomic" - "golang.org/x/exp/maps" - "golang.org/x/exp/slices" protoCodecs "github.com/livekit/protocol/codecs" "github.com/livekit/protocol/codecs/mime" @@ -394,7 +394,7 @@ func (d *DummyReceiver) GetDownTracks() []sfu.TrackSender { if receiver := d.getReceiver(); receiver != nil { return receiver.GetDownTracks() } - return maps.Values(d.downTracks) + return slices.Collect(maps.Values(d.downTracks)) } func (d *DummyReceiver) DebugInfo() map[string]any { @@ -567,7 +567,7 @@ func (d *DummyRedReceiver) GetDownTracks() []sfu.TrackSender { if r, ok := d.redReceiver.Load().(sfu.TrackReceiver); ok { return r.GetDownTracks() } - return maps.Values(d.downTracks) + return slices.Collect(maps.Values(d.downTracks)) } func (d *DummyRedReceiver) ReadRTP(buf []byte, layer uint8, esn uint64) (int, error) { diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index c9aeaed30..267324959 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -20,12 +20,13 @@ import ( "crypto/sha1" "encoding/base64" "fmt" + "maps" "os" + "slices" "sync" "time" "github.com/pkg/errors" - "golang.org/x/exp/maps" "github.com/livekit/mediatransportutil/pkg/rtcconfig" "github.com/livekit/protocol/auth" @@ -214,7 +215,7 @@ func (r *RoomManager) deleteRoom(ctx context.Context, roomName livekit.RoomName) func (r *RoomManager) CloseIdleRooms() { r.lock.RLock() - rooms := maps.Values(r.rooms) + rooms := slices.Collect(maps.Values(r.rooms)) r.lock.RUnlock() for _, room := range rooms { @@ -237,7 +238,7 @@ func (r *RoomManager) HasParticipants() bool { func (r *RoomManager) Stop() { // disconnect all clients r.lock.RLock() - rooms := maps.Values(r.rooms) + rooms := slices.Collect(maps.Values(r.rooms)) r.lock.RUnlock() for _, room := range rooms { diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 790df63b8..0747918ad 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -23,6 +23,7 @@ import ( "errors" "fmt" "io" + "maps" "math/rand" "net/http" "os" @@ -32,7 +33,6 @@ import ( "github.com/gorilla/websocket" "go.uber.org/atomic" - "golang.org/x/exp/maps" "google.golang.org/protobuf/proto" "github.com/livekit/protocol/livekit" diff --git a/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorwriter.go b/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorwriter.go index d90190233..3c6ba720d 100644 --- a/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorwriter.go +++ b/pkg/sfu/rtpextension/dependencydescriptor/dependencydescriptorwriter.go @@ -16,8 +16,7 @@ package dependencydescriptor import ( "fmt" - - "golang.org/x/exp/slices" + "slices" ) type TemplateMatch struct {