23use InvalidArgumentException;
53 private $localJobClasses;
55 private $jobTypeConfiguration;
57 private $jobTypesExcludedFromDefaultQueue;
59 private $statsFactory;
63 private $globalIdGenerator;
69 private const TYPE_ANY = 2;
73 private const PROC_CACHE_TTL = 15;
91 ?array $localJobClasses,
92 array $jobTypeConfiguration,
93 array $jobTypesExcludedFromDefaultQueue,
101 $this->localJobClasses = $localJobClasses;
102 $this->jobTypeConfiguration = $jobTypeConfiguration;
103 $this->jobTypesExcludedFromDefaultQueue = $jobTypesExcludedFromDefaultQueue;
104 $this->statsFactory = $statsFactory;
105 $this->wanCache = $wanCache;
106 $this->globalIdGenerator = $globalIdGenerator;
115 public function get( $type ) {
117 $conf += $this->jobTypeConfiguration[$type] ?? $this->jobTypeConfiguration[
'default'];
118 if ( !isset( $conf[
'readOnlyReason'] ) ) {
119 $conf[
'readOnlyReason'] = $this->readOnlyMode->getConfiguredReason();
122 $conf[
'stats'] = $this->statsFactory;
123 $conf[
'wanCache'] = $this->wanCache;
124 $conf[
'idGenerator'] = $this->globalIdGenerator;
138 public function push( $jobs ) {
139 $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
140 if ( $jobs === [] ) {
144 $this->assertValidJobs( $jobs );
147 foreach ( $jobs as
$job ) {
148 $type =
$job->getType();
149 if ( isset( $this->jobTypeConfiguration[$type] ) ) {
150 $jobsByType[$type][] =
$job;
153 isset( $this->jobTypeConfiguration[
'default'][
'typeAgnostic'] ) &&
154 $this->jobTypeConfiguration[
'default'][
'typeAgnostic']
156 $jobsByType[
'default'][] =
$job;
158 $jobsByType[$type][] =
$job;
163 foreach ( $jobsByType as $type => $jobs ) {
164 $this->
get( $type )->
push( $jobs );
167 if ( $this->cache->hasField(
'queues-ready',
'list' ) ) {
168 $list = $this->cache->getField(
'queues-ready',
'list' );
169 if ( count( array_diff( array_keys( $jobsByType ), $list ) ) ) {
170 $this->cache->clear(
'queues-ready' );
176 $cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs', self::TYPE_ANY ),
180 if ( array_diff( array_keys( $jobsByType ), $this->jobTypesExcludedFromDefaultQueue ) ) {
182 $cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs', self::TYPE_DEFAULT ),
197 if ( PHP_SAPI ===
'cli' || PHP_SAPI ===
'phpdbg' ) {
198 $this->
push( $jobs );
202 $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
205 $this->assertValidJobs( $jobs );
224 public function pop( $qtype = self::TYPE_DEFAULT, $flags = 0, array $ignored = [] ) {
227 if ( !$this->localJobClasses ) {
229 "Cannot pop '{$qtype}' job off foreign '{$this->domain}' wiki queue." );
231 if ( is_string( $qtype ) && !isset( $this->localJobClasses[$qtype] ) ) {
233 throw new JobQueueError(
"Unrecognized job type '$qtype'." );
236 if ( is_string( $qtype ) ) {
237 if ( !in_array( $qtype, $ignored ) ) {
238 $job = $this->
get( $qtype )->
pop();
241 if ( $flags & self::USE_CACHE ) {
242 if ( !$this->cache->hasField(
'queues-ready',
'list', self::PROC_CACHE_TTL ) ) {
245 $types = $this->cache->getField(
'queues-ready',
'list' );
250 if ( $qtype == self::TYPE_DEFAULT ) {
254 $types = array_diff( $types, $ignored );
257 foreach ( $types as $type ) {
258 $job = $this->
get( $type )->
pop();
262 $this->cache->clear(
'queues-ready' );
305 foreach ( $this->jobTypeConfiguration as $type => $conf ) {
318 if ( !$this->localJobClasses ) {
319 throw new JobQueueError(
'Cannot inspect job queue from foreign wiki' );
321 return array_keys( $this->localJobClasses );
332 return array_diff( $this->
getQueueTypes(), $this->jobTypesExcludedFromDefaultQueue );
346 $key =
$cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs', $type );
349 if ( $value ===
false ) {
351 if ( $type == self::TYPE_DEFAULT ) {
354 $value = count( $queues ) ?
'true' :
'false';
355 $cache->add( $key, $value, 15 );
358 return ( $value ===
'true' );
372 $queue = $info[
'queue'];
373 $nonEmpty = $queue->getSiblingQueuesWithJobs( $this->
getQueueTypes() );
374 if ( is_array( $nonEmpty ) ) {
375 $types = array_merge( $types, $nonEmpty );
377 foreach ( $info[
'types'] as $type ) {
378 if ( !$this->
get( $type )->isEmpty() ) {
399 $queue = $info[
'queue'];
400 $sizes = $queue->getSiblingQueueSizes( $this->
getQueueTypes() );
401 if ( is_array( $sizes ) ) {
404 foreach ( $info[
'types'] as $type ) {
405 $sizeMap[$type] = $this->
get( $type )->getSize();
418 if ( $this->coalescedQueues ===
null ) {
419 $this->coalescedQueues = [];
420 foreach ( $this->jobTypeConfiguration as $type => $conf ) {
422 $conf[
'type'] =
'null';
423 $conf[
'stats'] = $this->statsFactory;
424 $conf[
'wanCache'] = $this->wanCache;
425 $conf[
'idGenerator'] = $this->globalIdGenerator;
428 $loc = $queue->getCoalesceLocationInternal();
429 if ( !isset( $this->coalescedQueues[$loc] ) ) {
430 $this->coalescedQueues[$loc][
'queue'] = $queue;
431 $this->coalescedQueues[$loc][
'types'] = [];
433 if ( $type ===
'default' ) {
434 $this->coalescedQueues[$loc][
'types'] = array_merge(
435 $this->coalescedQueues[$loc][
'types'],
436 array_diff( $this->
getQueueTypes(), array_keys( $this->jobTypeConfiguration ) )
439 $this->coalescedQueues[$loc][
'types'][] = $type;
447 private function assertValidJobs( array $jobs ) {
448 foreach ( $jobs as
$job ) {
450 $type = get_debug_type(
$job );
451 throw new InvalidArgumentException(
"Expected IJobSpecification objects, got " . $type );
458class_alias( JobQueueGroup::class,
'JobQueueGroup' );
wfDeprecated( $function, $version=false, $component=false, $callerOffset=2)
Logs a warning that a deprecated feature was used.
if(count( $args)< 1) $job