JobQueue.php

Go to the documentation of this file.

<?php
abstract class JobQueue {
        protected $wiki; // string; wiki ID
        protected $type; // string; job type
        protected $order; // string; job priority for pop()
        protected $claimTTL; // integer; seconds
        protected $maxTries; // integer; maximum number of times to try a job

        const QoS_Atomic = 1; // integer; "all-or-nothing" job insertions

        protected function __construct( array $params ) {
                $this->wiki = $params['wiki'];
                $this->type = $params['type'];
                $this->claimTTL = isset( $params['claimTTL'] ) ? $params['claimTTL'] : 0;
                $this->maxTries = isset( $params['maxTries'] ) ? $params['maxTries'] : 3;
                if ( isset( $params['order'] ) && $params['order'] !== 'any' ) {
                        $this->order = $params['order'];
                } else {
                        $this->order = $this->optimalOrder();
                }
                if ( !in_array( $this->order, $this->supportedOrders() ) ) {
                        throw new MWException( __CLASS__ . " does not support '{$this->order}' order." );
                }
        }

        final public static function factory( array $params ) {
                $class = $params['class'];
                if ( !MWInit::classExists( $class ) ) {
                        throw new MWException( "Invalid job queue class '$class'." );
                }
                $obj = new $class( $params );
                if ( !( $obj instanceof self ) ) {
                        throw new MWException( "Class '$class' is not a " . __CLASS__ . " class." );
                }
                return $obj;
        }

        final public function getWiki() {
                return $this->wiki;
        }

        final public function getType() {
                return $this->type;
        }

        final public function getOrder() {
                return $this->order;
        }

        abstract protected function supportedOrders();

        abstract protected function optimalOrder();

        final public function isEmpty() {
                wfProfileIn( __METHOD__ );
                $res = $this->doIsEmpty();
                wfProfileOut( __METHOD__ );
                return $res;
        }

        abstract protected function doIsEmpty();

        final public function getSize() {
                wfProfileIn( __METHOD__ );
                $res = $this->doGetSize();
                wfProfileOut( __METHOD__ );
                return $res;
        }

        abstract protected function doGetSize();

        final public function getAcquiredCount() {
                wfProfileIn( __METHOD__ );
                $res = $this->doGetAcquiredCount();
                wfProfileOut( __METHOD__ );
                return $res;
        }

        abstract protected function doGetAcquiredCount();

        final public function push( $jobs, $flags = 0 ) {
                return $this->batchPush( is_array( $jobs ) ? $jobs : array( $jobs ), $flags );
        }

        final public function batchPush( array $jobs, $flags = 0 ) {
                if ( !count( $jobs ) ) {
                        return true; // nothing to do
                }

                foreach ( $jobs as $job ) {
                        if ( $job->getType() !== $this->type ) {
                                throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
                        }
                }

                wfProfileIn( __METHOD__ );
                $ok = $this->doBatchPush( $jobs, $flags );
                wfProfileOut( __METHOD__ );
                return $ok;
        }

        abstract protected function doBatchPush( array $jobs, $flags );

        final public function pop() {
                global $wgJobClasses;

                if ( $this->wiki !== wfWikiID() ) {
                        throw new MWException( "Cannot pop '{$this->type}' job off foreign wiki queue." );
                } elseif ( !isset( $wgJobClasses[$this->type] ) ) {
                        // Do not pop jobs if there is no class for the queue type
                        throw new MWException( "Unrecognized job type '{$this->type}'." );
                }

                wfProfileIn( __METHOD__ );
                $job = $this->doPop();
                wfProfileOut( __METHOD__ );
                return $job;
        }

        abstract protected function doPop();

        final public function ack( Job $job ) {
                if ( $job->getType() !== $this->type ) {
                        throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
                }
                wfProfileIn( __METHOD__ );
                $ok = $this->doAck( $job );
                wfProfileOut( __METHOD__ );
                return $ok;
        }

        abstract protected function doAck( Job $job );

        final public function deduplicateRootJob( Job $job ) {
                if ( $job->getType() !== $this->type ) {
                        throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
                }
                wfProfileIn( __METHOD__ );
                $ok = $this->doDeduplicateRootJob( $job );
                wfProfileOut( __METHOD__ );
                return $ok;
        }

        protected function doDeduplicateRootJob( Job $job ) {
                return true;
        }

        final public function waitForBackups() {
                wfProfileIn( __METHOD__ );
                $this->doWaitForBackups();
                wfProfileOut( __METHOD__ );
        }

        protected function doWaitForBackups() {}

        final public function getPeriodicTasks() {
                $tasks = $this->doGetPeriodicTasks();
                foreach ( $tasks as $name => &$def ) {
                        $def['name'] = $name;
                }
                return $tasks;
        }

        protected function doGetPeriodicTasks() {
                return array();
        }

        final public function flushCaches() {
                wfProfileIn( __METHOD__ );
                $this->doFlushCaches();
                wfProfileOut( __METHOD__ );
        }

        protected function doFlushCaches() {}

        abstract public function getAllQueuedJobs();

        public function setTestingPrefix( $key ) {
                throw new MWException( "Queue namespacing not supported for this queue type." );
        }
}