dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-05-13 20:53:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Taking interface{} -> any modernize bits (#4204)

Browse Source
This commit is contained in:
Raja Subramanian
2025-12-28 05:22:12 +05:30
committed by GitHub
parent b91cd2e4ea
commit b8ddd0f98c
24 changed files with 135 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/config/config.go
+1 -1
View File
@@ -762,7 +762,7 @@ func (conf *Config) updateFromCLI(c *cli.Command, baseFlags []cli.Flag) error {
}
func (conf *Config) unmarshalKeys(keys string) error {
temp := make(map[string]interface{})
temp := make(map[string]any)
if err := yaml.Unmarshal([]byte(keys), temp); err != nil {
return err
}
pkg/rtc/mediatrackreceiver.go
+2 -2
View File
@@ -1093,8 +1093,8 @@ func (t *MediaTrackReceiver) onDownTrackCreated(downTrack *sfu.DownTrack) {
}
}
func (t *MediaTrackReceiver) DebugInfo() map[string]interface{} {
info := map[string]interface{}{
func (t *MediaTrackReceiver) DebugInfo() map[string]any {
info := map[string]any{
"ID": t.ID(),
"Kind": t.Kind().String(),
"PubMuted": t.IsMuted(),
pkg/rtc/mediatracksubscriptions.go
+2 -2
View File
@@ -355,8 +355,8 @@ func (t *MediaTrackSubscriptions) getAllSubscribedTracksLocked() []types.Subscri
return subTracks
}
func (t *MediaTrackSubscriptions) DebugInfo() []map[string]interface{} {
subscribedTrackInfo := make([]map[string]interface{}, 0)
func (t *MediaTrackSubscriptions) DebugInfo() []map[string]any {
subscribedTrackInfo := make([]map[string]any, 0)
for _, val := range t.getAllSubscribedTracks() {
if st, ok := val.(*SubscribedTrack); ok {
subscribedTrackInfo = append(subscribedTrackInfo, st.DownTrack().DebugInfo())
pkg/rtc/participant.go
+4 -4
View File
@@ -3496,13 +3496,13 @@ func (p *ParticipantImpl) getPublishedTrackBySdpCid(clientId string) types.Media
return nil
}
func (p *ParticipantImpl) DebugInfo() map[string]interface{} {
info := map[string]interface{}{
func (p *ParticipantImpl) DebugInfo() map[string]any {
info := map[string]any{
"ID": p.ID(),
"State": p.State().String(),
}
pendingTrackInfo := make(map[string]interface{})
pendingTrackInfo := make(map[string]any)
p.pendingTracksLock.RLock()
for clientID, pti := range p.pendingTracks {
var trackInfos []string
@@ -3510,7 +3510,7 @@ func (p *ParticipantImpl) DebugInfo() map[string]interface{} {
trackInfos = append(trackInfos, ti.String())
}
pendingTrackInfo[clientID] = map[string]interface{}{
pendingTrackInfo[clientID] = map[string]any{
"TrackInfos": trackInfos,
"Migrated": pti.migrated,
}
pkg/rtc/room.go
+3 -3
View File
@@ -1795,15 +1795,15 @@ func (r *Room) handleNewJobs(ad *livekit.AgentDispatch, inc *sutils.IncrementalD
})
}
func (r *Room) DebugInfo() map[string]interface{} {
info := map[string]interface{}{
func (r *Room) DebugInfo() map[string]any {
info := map[string]any{
"Name": r.protoRoom.Name,
"Sid": r.protoRoom.Sid,
"CreatedAt": r.protoRoom.CreationTime,
}
participants := r.GetParticipants()
participantInfo := make(map[string]interface{})
participantInfo := make(map[string]any)
for _, p := range participants {
participantInfo[string(p.Identity())] = p.DebugInfo()
}
pkg/rtc/uptrackmanager.go
+4 -4
View File
@@ -398,16 +398,16 @@ func (u *UpTrackManager) maybeRevokeSubscriptions() {
}
}
func (u *UpTrackManager) DebugInfo() map[string]interface{} {
info := map[string]interface{}{}
publishedTrackInfo := make(map[livekit.TrackID]interface{})
func (u *UpTrackManager) DebugInfo() map[string]any {
info := map[string]any{}
publishedTrackInfo := make(map[livekit.TrackID]any)
u.lock.RLock()
for trackID, track := range u.publishedTracks {
if mt, ok := track.(*MediaTrack); ok {
publishedTrackInfo[trackID] = mt.DebugInfo()
} else {
publishedTrackInfo[trackID] = map[string]interface{}{
publishedTrackInfo[trackID] = map[string]any{
"ID": track.ID(),
"Kind": track.Kind().String(),
"PubMuted": track.IsMuted(),
pkg/rtc/utils.go
+2 -2
View File
@@ -117,7 +117,7 @@ func Recover(l logger.Logger) any {
// logger helpers
func LoggerWithParticipant(l logger.Logger, identity livekit.ParticipantIdentity, sid livekit.ParticipantID, isRemote bool) logger.Logger {
values := make([]interface{}, 0, 4)
values := make([]any, 0, 4)
if identity != "" {
values = append(values, "participant", identity)
}
@@ -130,7 +130,7 @@ func LoggerWithParticipant(l logger.Logger, identity livekit.ParticipantIdentity
}
func LoggerWithRoom(l logger.Logger, name livekit.RoomName, roomID livekit.RoomID) logger.Logger {
values := make([]interface{}, 0, 2)
values := make([]any, 0, 2)
if name != "" {
values = append(values, "room", name)
}
pkg/service/egress.go
+5 -5
View File
@@ -73,7 +73,7 @@ func NewEgressLauncher(client rpc.EgressClient, io IOClient, store ServiceStore)
}
func (s *EgressService) StartRoomCompositeEgress(ctx context.Context, req *livekit.RoomCompositeEgressRequest) (*livekit.EgressInfo, error) {
fields := []interface{}{
fields := []any{
"room", req.RoomName,
"baseUrl", req.CustomBaseUrl,
"outputType", egress.GetOutputType(req),
@@ -94,7 +94,7 @@ func (s *EgressService) StartRoomCompositeEgress(ctx context.Context, req *livek
}
func (s *EgressService) StartWebEgress(ctx context.Context, req *livekit.WebEgressRequest) (*livekit.EgressInfo, error) {
fields := []interface{}{
fields := []any{
"url", req.Url,
"outputType", egress.GetOutputType(req),
}
@@ -114,7 +114,7 @@ func (s *EgressService) StartWebEgress(ctx context.Context, req *livekit.WebEgre
}
func (s *EgressService) StartParticipantEgress(ctx context.Context, req *livekit.ParticipantEgressRequest) (*livekit.EgressInfo, error) {
fields := []interface{}{
fields := []any{
"room", req.RoomName,
"identity", req.Identity,
"outputType", egress.GetOutputType(req),
@@ -135,7 +135,7 @@ func (s *EgressService) StartParticipantEgress(ctx context.Context, req *livekit
}
func (s *EgressService) StartTrackCompositeEgress(ctx context.Context, req *livekit.TrackCompositeEgressRequest) (*livekit.EgressInfo, error) {
fields := []interface{}{
fields := []any{
"room", req.RoomName,
"audioTrackID", req.AudioTrackId,
"videoTrackID", req.VideoTrackId,
@@ -157,7 +157,7 @@ func (s *EgressService) StartTrackCompositeEgress(ctx context.Context, req *live
}
func (s *EgressService) StartTrackEgress(ctx context.Context, req *livekit.TrackEgressRequest) (*livekit.EgressInfo, error) {
fields := []interface{}{"room", req.RoomName, "trackID", req.TrackId}
fields := []any{"room", req.RoomName, "trackID", req.TrackId}
if t := reflect.TypeOf(req.Output); t != nil {
fields = append(fields, "outputType", t.String())
}
pkg/service/ingress.go
+2 -2
View File
@@ -85,7 +85,7 @@ func NewIngressService(
}
func (s *IngressService) CreateIngress(ctx context.Context, req *livekit.CreateIngressRequest) (*livekit.IngressInfo, error) {
fields := []interface{}{
fields := []any{
"inputType", req.InputType,
"name", req.Name,
}
@@ -275,7 +275,7 @@ func updateInfoUsingRequest(req *livekit.UpdateIngressRequest, info *livekit.Ing
}
func (s *IngressService) UpdateIngress(ctx context.Context, req *livekit.UpdateIngressRequest) (*livekit.IngressInfo, error) {
fields := []interface{}{
fields := []any{
"ingress", req.IngressId,
"name", req.Name,
}
pkg/service/redisstore.go
+1 -1
View File
@@ -215,7 +215,7 @@ func (s *RedisStore) ListRooms(_ context.Context, roomNames []livekit.RoomName)
}
} else {
names := livekit.IDsAsStrings(roomNames)
var results []interface{}
var results []any
results, err = s.rc.HMGet(s.ctx, RoomsKey, names...).Result()
if err != nil && err != redis.Nil {
return nil, errors.Wrap(err, "could not get rooms by names")
pkg/service/server.go
+3 -3
View File
@@ -115,7 +115,7 @@ func NewLivekitServer(conf *config.Config,
middlewares = append(middlewares, NewAPIKeyAuthMiddleware(keyProvider))
}
serverOptions := []interface{}{
serverOptions := []any{
twirp.WithServerHooks(twirp.ChainHooks(
TwirpLogger(),
TwirpRequestStatusReporter(),
@@ -236,7 +236,7 @@ func (s *LivekitServer) Start() error {
}
}
values := []interface{}{
values := []any{
"portHttp", s.config.Port,
"nodeID", s.currentNode.NodeID(),
"nodeIP", s.currentNode.NodeIP(),
@@ -347,7 +347,7 @@ func (s *LivekitServer) debugGoroutines(w http.ResponseWriter, _ *http.Request)
func (s *LivekitServer) debugInfo(w http.ResponseWriter, _ *http.Request) {
s.roomManager.lock.RLock()
info := make([]map[string]interface{}, 0, len(s.roomManager.rooms))
info := make([]map[string]any, 0, len(s.roomManager.rooms))
for _, room := range s.roomManager.rooms {
info = append(info, room.DebugInfo())
}
pkg/service/turn.go
+1 -1
View File
@@ -69,7 +69,7 @@ func NewTurnServer(conf *config.Config, authHandler turn.AuthHandler, standalone
if standalone {
relayAddrGen = telemetry.NewRelayAddressGenerator(relayAddrGen)
}
var logValues []interface{}
var logValues []any
logValues = append(logValues, "turn.relay_range_start", turnConf.RelayPortRangeStart)
logValues = append(logValues, "turn.relay_range_end", turnConf.RelayPortRangeEnd)
pkg/service/twirp.go
+5 -5
View File
@@ -46,9 +46,9 @@ type twirpLoggerKey struct{}
// License: Apache-2.0
func TwirpLogger() *twirp.ServerHooks {
loggerPool := &sync.Pool{
New: func() interface{} {
New: func() any {
return &twirpLogger{
fieldsOrig: make([]interface{}, 0, 30),
fieldsOrig: make([]any, 0, 30),
}
},
}
@@ -67,12 +67,12 @@ func TwirpLogger() *twirp.ServerHooks {
type twirpLogger struct {
twirpRequestFields
fieldsOrig []interface{}
fields []interface{}
fieldsOrig []any
fields []any
startedAt time.Time
}
func AppendLogFields(ctx context.Context, fields ...interface{}) {
func AppendLogFields(ctx context.Context, fields ...any) {
r, ok := ctx.Value(twirpLoggerKey{}).(*twirpLogger)
if !ok || r == nil {
return
pkg/service/utils.go
+5 -5
View File
@@ -39,7 +39,7 @@ import (
"github.com/livekit/protocol/logger"
)
func handleError(w http.ResponseWriter, r *http.Request, status int, err error, keysAndValues ...interface{}) {
func handleError(w http.ResponseWriter, r *http.Request, status int, err error, keysAndValues ...any) {
keysAndValues = append(keysAndValues, "status", status)
if r != nil && r.URL != nil {
keysAndValues = append(keysAndValues, "method", r.Method, "path", r.URL.Path)
@@ -50,12 +50,12 @@ func handleError(w http.ResponseWriter, r *http.Request, status int, err error,
w.WriteHeader(status)
}
func HandleError(w http.ResponseWriter, r *http.Request, status int, err error, keysAndValues ...interface{}) {
func HandleError(w http.ResponseWriter, r *http.Request, status int, err error, keysAndValues ...any) {
handleError(w, r, status, err, keysAndValues...)
_, _ = w.Write([]byte(err.Error()))
}
func HandleErrorJson(w http.ResponseWriter, r *http.Request, status int, err error, keysAndValues ...interface{}) {
func HandleErrorJson(w http.ResponseWriter, r *http.Request, status int, err error, keysAndValues ...any) {
handleError(w, r, status, err, keysAndValues...)
json.NewEncoder(w).Encode(struct {
Error string `json:"error"`
@@ -166,13 +166,13 @@ var (
func createUserAgentParserWithCustomRules() (*uaparser.Parser, error) {
defaultYaml := uaparser.DefinitionYaml
rules := make(map[string]interface{})
rules := make(map[string]any)
err := yaml.Unmarshal(defaultYaml, rules)
if err != nil {
return nil, err
}
rules["user_agent_parsers"] = append(rules["user_agent_parsers"].([]interface{}), map[string]interface{}{
rules["user_agent_parsers"] = append(rules["user_agent_parsers"].([]any), map[string]any{
"regex": "OBS-Studio\\/([0-9\\.]+)",
"family_replacement": "OBS Studio",
"v1_replacement": "$1",
pkg/sfu/codecmunger/codecmunger.go
+2 -2
View File
@@ -27,8 +27,8 @@ var (
)
type CodecMunger interface {
GetState() interface{}
SeedState(state interface{})
GetState() any
SeedState(state any)
SetLast(extPkt *buffer.ExtPacket)
UpdateOffsets(extPkt *buffer.ExtPacket)
pkg/sfu/codecmunger/null.go
+4 -4
View File
@@ -20,22 +20,22 @@ import (
)
type Null struct {
seededState interface{}
seededState any
}
func NewNull(_logger logger.Logger) *Null {
return &Null{}
}
func (n *Null) GetState() interface{} {
func (n *Null) GetState() any {
return nil
}
func (n *Null) SeedState(state interface{}) {
func (n *Null) SeedState(state any) {
n.seededState = state
}
func (n *Null) GetSeededState() interface{} {
func (n *Null) GetSeededState() any {
return n.seededState
}
pkg/sfu/codecmunger/vp8.go
+2 -2
View File
@@ -66,7 +66,7 @@ func NewVP8FromNull(cm CodecMunger, logger logger.Logger) *VP8 {
return v
}
func (v *VP8) GetState() interface{} {
func (v *VP8) GetState() any {
return &livekit.VP8MungerState{
ExtLastPictureId: v.extLastPictureId,
PictureIdUsed: v.pictureIdUsed,
@@ -78,7 +78,7 @@ func (v *VP8) GetState() interface{} {
}
}
func (v *VP8) SeedState(seed interface{}) {
func (v *VP8) SeedState(seed any) {
switch cm := seed.(type) {
case *livekit.RTPForwarderState_Vp8Munger:
state := cm.Vp8Munger
pkg/sfu/downtrack.go
+4 -4
View File
@@ -552,7 +552,7 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
isFECEnabled = strings.Contains(strings.ToLower(matchedUpstreamCodec.SDPFmtpLine), "fec")
}
logFields := []interface{}{
logFields := []any{
"codecs", d.upstreamCodecs,
"matchCodec", codec,
"ssrc", t.SSRC(),
@@ -2341,8 +2341,8 @@ func (d *DownTrack) getTranslatedPayloadType(srcPT uint8) uint8 {
return uint8(d.payloadType.Load())
}
func (d *DownTrack) DebugInfo() map[string]interface{} {
stats := map[string]interface{}{
func (d *DownTrack) DebugInfo() map[string]any {
stats := map[string]any{
"LastPli": d.rtpStats.LastPli(),
}
stats["RTPMunger"] = d.forwarder.RTPMungerDebugInfo()
@@ -2354,7 +2354,7 @@ func (d *DownTrack) DebugInfo() map[string]interface{} {
stats["PacketCount"] = senderReport.PacketCount
}
return map[string]interface{}{
return map[string]any{
"SubscriberID": d.params.SubID,
"TrackID": d.id,
"StreamID": d.params.StreamID,
pkg/sfu/forwarder.go
+1 -1
View File
@@ -2263,7 +2263,7 @@ func (f *Forwarder) GetPadding(frameEndNeeded bool) ([]byte, error) {
return f.codecMunger.UpdateAndGetPadding(!frameEndNeeded)
}
func (f *Forwarder) RTPMungerDebugInfo() map[string]interface{} {
func (f *Forwarder) RTPMungerDebugInfo() map[string]any {
f.lock.RLock()
defer f.lock.RUnlock()
pkg/sfu/rtpmunger.go
+2 -2
View File
@@ -77,8 +77,8 @@ func NewRTPMunger(logger logger.Logger) *RTPMunger {
}
}
func (r *RTPMunger) DebugInfo() map[string]interface{} {
return map[string]interface{}{
func (r *RTPMunger) DebugInfo() map[string]any {
return map[string]any{
"ExtHighestIncomingSN": r.extHighestIncomingSN,
"ExtLastSN": r.extLastSN,
"ExtSecondLastSN": r.extSecondLastSN,
pkg/sfu/rtpstats/rtpstats_base_lite.go
+1 -1
View File
@@ -305,7 +305,7 @@ func (r *rtpStatsBaseLite) deltaInfoLite(
snapshotLiteID uint32,
extStartSN uint64,
extHighestSN uint64,
) (deltaInfoLite *RTPDeltaInfoLite, err error, loggingFields []interface{}) {
) (deltaInfoLite *RTPDeltaInfoLite, err error, loggingFields []any) {
then, now := r.getAndResetSnapshotLite(snapshotLiteID, extStartSN, extHighestSN)
if now == nil || then == nil {
return
pkg/sfu/streamallocator/streamallocator.go
+1 -1
View File
@@ -136,7 +136,7 @@ type Event struct {
*StreamAllocator
Signal streamAllocatorSignal
TrackID livekit.TrackID
Data interface{}
Data any
}
func (e Event) String() string {
pkg/telemetry/analyticsservice.go
+15
View File
@@ -38,6 +38,21 @@ type AnalyticsService interface {
RoomProjectReporter(ctx context.Context) roomobs.ProjectReporter
}
// ----------------------------
var _ AnalyticsService = &NullAnalyticService{}
type NullAnalyticService struct{}
func (n NullAnalyticService) SendStats(_ context.Context, _ []*livekit.AnalyticsStat) {}
func (n NullAnalyticService) SendEvent(_ context.Context, _ *livekit.AnalyticsEvent) {}
func (n NullAnalyticService) SendNodeRoomStates(_ context.Context, _ *livekit.AnalyticsNodeRooms) {}
func (n NullAnalyticService) RoomProjectReporter(_ctx context.Context) roomobs.ProjectReporter {
return nil
}
// ----------------------------
type analyticsService struct {
analyticsKey string
nodeID string
pkg/telemetry/telemetryservice.go
+63
View File
@@ -87,6 +87,69 @@ type TelemetryService interface {
FlushStats()
}
// -----------------------------
var _ TelemetryService = (*NullTelemetryService)(nil)
type NullTelemetryService struct {
NullAnalyticService
}
func (n NullTelemetryService) TrackStats(key StatsKey, stat *livekit.AnalyticsStat) {}
func (n NullTelemetryService) RoomStarted(ctx context.Context, room *livekit.Room) {}
func (n NullTelemetryService) RoomEnded(ctx context.Context, room *livekit.Room) {}
func (n NullTelemetryService) ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta, shouldSendEvent bool, guard *ReferenceGuard) {
}
func (n NullTelemetryService) ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, guard *ReferenceGuard) {
}
func (n NullTelemetryService) ParticipantResumed(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, nodeID livekit.NodeID, reason livekit.ReconnectReason) {
}
func (n NullTelemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard) {
}
func (n NullTelemetryService) TrackPublishRequested(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo) {
}
func (n NullTelemetryService) TrackPublished(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) {
}
func (n NullTelemetryService) TrackUnpublished(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) {
}
func (n NullTelemetryService) TrackSubscribeRequested(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo) {
}
func (n NullTelemetryService) TrackSubscribed(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, publisher *livekit.ParticipantInfo, shouldSendEvent bool) {
}
func (n NullTelemetryService) TrackUnsubscribed(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, shouldSendEvent bool) {
}
func (n NullTelemetryService) TrackSubscribeFailed(ctx context.Context, participantID livekit.ParticipantID, trackID livekit.TrackID, err error, isUserError bool) {
}
func (n NullTelemetryService) TrackMuted(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo) {
}
func (n NullTelemetryService) TrackUnmuted(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo) {
}
func (n NullTelemetryService) TrackPublishedUpdate(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo) {
}
func (n NullTelemetryService) TrackMaxSubscribedVideoQuality(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, mime mime.MimeType, maxQuality livekit.VideoQuality) {
}
func (n NullTelemetryService) TrackPublishRTPStats(ctx context.Context, participantID livekit.ParticipantID, trackID livekit.TrackID, mimeType mime.MimeType, layer int, stats *livekit.RTPStats) {
}
func (n NullTelemetryService) TrackSubscribeRTPStats(ctx context.Context, participantID livekit.ParticipantID, trackID livekit.TrackID, mimeType mime.MimeType, stats *livekit.RTPStats) {
}
func (n NullTelemetryService) EgressStarted(ctx context.Context, info *livekit.EgressInfo) {}
func (n NullTelemetryService) EgressUpdated(ctx context.Context, info *livekit.EgressInfo) {}
func (n NullTelemetryService) EgressEnded(ctx context.Context, info *livekit.EgressInfo) {}
func (n NullTelemetryService) IngressCreated(ctx context.Context, info *livekit.IngressInfo) {}
func (n NullTelemetryService) IngressDeleted(ctx context.Context, info *livekit.IngressInfo) {}
func (n NullTelemetryService) IngressStarted(ctx context.Context, info *livekit.IngressInfo) {}
func (n NullTelemetryService) IngressUpdated(ctx context.Context, info *livekit.IngressInfo) {}
func (n NullTelemetryService) IngressEnded(ctx context.Context, info *livekit.IngressInfo) {}
func (n NullTelemetryService) LocalRoomState(ctx context.Context, info *livekit.AnalyticsNodeRooms) {}
func (n NullTelemetryService) Report(ctx context.Context, reportInfo *livekit.ReportInfo) {}
func (n NullTelemetryService) APICall(ctx context.Context, apiCallInfo *livekit.APICallInfo) {}
func (n NullTelemetryService) Webhook(ctx context.Context, webhookInfo *livekit.WebhookInfo) {}
func (n NullTelemetryService) NotifyEgressEvent(ctx context.Context, event string, info *livekit.EgressInfo) {
}
func (n NullTelemetryService) FlushStats() {}
// -----------------------------
const (
workerCleanupWait = 3 * time.Minute
jobsQueueMinSize = 2048