JobQueueMemory.php

Go to the documentation of this file.

<?php
use Wikimedia\ObjectCache\HashBagOStuff;
use Wikimedia\ObjectCache\WANObjectCache;
 
class JobQueueMemory extends JobQueue {
    protected static $data = [];
 
    public function __construct( array $params ) {
        $params['wanCache'] = new WANObjectCache( [ 'cache' => new HashBagOStuff() ] );
 
        parent::__construct( $params );
    }
 
    protected function doBatchPush( array $jobs, $flags ) {
        $unclaimed =& $this->getQueueData( 'unclaimed', [] );
 
        foreach ( $jobs as $job ) {
            if ( $job->ignoreDuplicates() ) {
                $sha1 = sha1( serialize( $job->getDeduplicationInfo() ) );
                if ( !isset( $unclaimed[$sha1] ) ) {
                    $unclaimed[$sha1] = $job;
                }
            } else {
                $unclaimed[] = $job;
            }
        }
    }
 
    protected function supportedOrders() {
        return [ 'random', 'timestamp', 'fifo' ];
    }
 
    protected function optimalOrder() {
        return 'fifo';
    }
 
    protected function doIsEmpty() {
        return ( $this->doGetSize() == 0 );
    }
 
    protected function doGetSize() {
        $unclaimed = $this->getQueueData( 'unclaimed' );
 
        return $unclaimed ? count( $unclaimed ) : 0;
    }
 
    protected function doGetAcquiredCount() {
        $claimed = $this->getQueueData( 'claimed' );
 
        return $claimed ? count( $claimed ) : 0;
    }
 
    protected function doPop() {
        if ( $this->doGetSize() == 0 ) {
            return false;
        }
 
        $unclaimed =& $this->getQueueData( 'unclaimed' );
        $claimed =& $this->getQueueData( 'claimed', [] );
 
        if ( $this->order === 'random' ) {
            $key = array_rand( $unclaimed );
        } else {
            $key = array_key_first( $unclaimed );
        }
 
        $spec = $unclaimed[$key];
        unset( $unclaimed[$key] );
        $claimed[] = $spec;
 
        $job = $this->jobFromSpecInternal( $spec );
 
        $job->setMetadata( 'claimId', array_key_last( $claimed ) );
 
        return $job;
    }
 
    protected function doAck( RunnableJob $job ) {
        if ( $this->getAcquiredCount() == 0 ) {
            return;
        }
 
        $claimed =& $this->getQueueData( 'claimed' );
        unset( $claimed[$job->getMetadata( 'claimId' )] );
    }
 
    protected function doDelete() {
        if ( isset( self::$data[$this->type][$this->domain] ) ) {
            unset( self::$data[$this->type][$this->domain] );
            if ( !self::$data[$this->type] ) {
                unset( self::$data[$this->type] );
            }
        }
    }
 
    public function getAllQueuedJobs() {
        $unclaimed = $this->getQueueData( 'unclaimed' );
        if ( !$unclaimed ) {
            return new ArrayIterator( [] );
        }
 
        return new MappedIterator(
            $unclaimed,
            function ( $value ) {
                return $this->jobFromSpecInternal( $value );
            }
        );
    }
 
    public function getAllAcquiredJobs() {
        $claimed = $this->getQueueData( 'claimed' );
        if ( !$claimed ) {
            return new ArrayIterator( [] );
        }
 
        return new MappedIterator(
            $claimed,
            function ( $value ) {
                return $this->jobFromSpecInternal( $value );
            }
        );
    }
 
    public function jobFromSpecInternal( IJobSpecification $spec ) {
        return $this->factoryJob( $spec->getType(), $spec->getParams() );
    }
 
    private function &getQueueData( $field, $init = null ) {
        if ( !isset( self::$data[$this->type][$this->domain][$field] ) ) {
            if ( $init !== null ) {
                self::$data[$this->type][$this->domain][$field] = $init;
            } else {
                return $init;
            }
        }
 
        return self::$data[$this->type][$this->domain][$field];
    }
}