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

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

        protected function __construct( array $params ) {
                $this->wiki     = $params['wiki'];
                $this->type     = $params['type'];
                $this->order    = isset( $params['order'] ) ? $params['order'] : 'random';
                $this->claimTTL = isset( $params['claimTTL'] ) ? $params['claimTTL'] : 0;
        }

        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 isEmpty() {
                wfProfileIn( __METHOD__ );
                $res = $this->doIsEmpty();
                wfProfileOut( __METHOD__ );
                return $res;
        }

        abstract protected function doIsEmpty();

        final public function batchPush( array $jobs, $flags = 0 ) {
                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 );
                if ( $ok ) {
                        wfIncrStats( 'job-insert', count( $jobs ) );
                }
                wfProfileOut( __METHOD__ );
                return $ok;
        }

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

        final public function pop() {
                wfProfileIn( __METHOD__ );
                $job = $this->doPop();
                if ( $job ) {
                        wfIncrStats( 'job-pop' );
                }
                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 waitForBackups() {
                wfProfileIn( __METHOD__ );
                $this->doWaitForBackups();
                wfProfileOut( __METHOD__ );
        }

        protected function doWaitForBackups() {}
}