918
919
// check for default priority and add if not present (this may be initializing subqueue 0)
920
>
defKey := &persistencespb.SubqueueKey{
db.go
921
>
Priority: int32(db.config.DefaultPriorityKey),
922
>
}
923
>
hasDefault := slices.ContainsFunc(subqueues, func(s *dbSubqueue) bool {
924
return proto.Equal(s.Key, defKey)
925
})
926
>
if !hasDefault {
db.go
927
>
subqueues = append(subqueues, db.newSubqueueLocked(defKey))
928
>
// If we are transitioning from no-subqueues to subqueues, initialize subqueue 0 with
929
>
// the ack level and approx count from TaskQueueInfo.
930
>
if len(subqueues) == 1 {
931
>
subqueues[subqueueZero].AckLevel = initAckLevel
932
>
subqueues[subqueueZero].ApproximateBacklogCount = initApproxCount
933
>
}
934
}
935
>
return subqueues
db.go
936
}
937
938
>
func (db *taskQueueDB) newSubqueueLocked(key *persistencespb.SubqueueKey) *dbSubqueue {
db.go
939
>
// For fifo queues: start ack level + max read level just before the current block.
940
>
// For fair queues: ack level and max read level don't matter here.
941
>
initAckLevel := rangeIDToTaskIDBlock(db.rangeID, db.config.RangeSize).start - 1
942
>
softassert.That(db.logger, initAckLevel >= 0, "initAckLevel should not be negative")
943
>
944
>
s := &dbSubqueue{maxReadLevel: initAckLevel}
945
>
s.Key = key
946
>
s.AckLevel = initAckLevel
947
>
return s
948
>
}
949
950
// clone db.subqueues so we can return it outside our lock
951
>
func (db *taskQueueDB) cloneSubqueues() []persistencespb.SubqueueInfo {
db.go
952
>
infos := make([]persistencespb.SubqueueInfo, len(db.subqueues))
953
>
for i := range db.subqueues {
954
>
proto.Merge(&infos[i], &db.subqueues[i].SubqueueInfo)
955
>
}
956
>
return infos
957
}
958