includes/jobqueue/JobQueueMemory.php

   1 <?php
   2 /**
   3  * This program is free software; you can redistribute it and/or modify
   4  * it under the terms of the GNU General Public License as published by
   5  * the Free Software Foundation; either version 2 of the License, or
   6  * (at your option) any later version.
   7  *
   8  * This program is distributed in the hope that it will be useful,
   9  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  10  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  11  * GNU General Public License for more details.
  12  *
  13  * You should have received a copy of the GNU General Public License along
  14  * with this program; if not, write to the Free Software Foundation, Inc.,
  15  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
  16  * http://www.gnu.org/copyleft/gpl.html
  17  *
  18  * @file
  19  */
  20
  21 use Wikimedia\ObjectCache\HashBagOStuff;
  22 use Wikimedia\ObjectCache\WANObjectCache;
  23
  24 /**
  25  * PHP memory-backed job queue storage, for testing.
  26  *
  27  * JobQueueGroup does not remember every queue instance, so statically track it here.
  28  *
  29  * @since 1.27
  30  * @ingroup JobQueue
  31  */
  32 class JobQueueMemory extends JobQueue {
  33         /** @var array[] */
  34         protected static $data = [];
  35
  36         public function __construct( array $params ) {
  37                 $params['wanCache'] = new WANObjectCache( [ 'cache' => new HashBagOStuff() ] );
  38
  39                 parent::__construct( $params );
  40         }
  41
  42         /**
  43          * @see JobQueue::doBatchPush
  44          *
  45          * @param IJobSpecification[] $jobs
  46          * @param int $flags
  47          */
  48         protected function doBatchPush( array $jobs, $flags ) {
  49                 $unclaimed =& $this->getQueueData( 'unclaimed', [] );
  50
  51                 foreach ( $jobs as $job ) {
  52                         if ( $job->ignoreDuplicates() ) {
  53                                 $sha1 = sha1( serialize( $job->getDeduplicationInfo() ) );
  54                                 if ( !isset( $unclaimed[$sha1] ) ) {
  55                                         $unclaimed[$sha1] = $job;
  56                                 }
  57                         } else {
  58                                 $unclaimed[] = $job;
  59                         }
  60                 }
  61         }
  62
  63         /**
  64          * @see JobQueue::supportedOrders
  65          *
  66          * @return string[]
  67          */
  68         protected function supportedOrders() {
  69                 return [ 'random', 'timestamp', 'fifo' ];
  70         }
  71
  72         /**
  73          * @see JobQueue::optimalOrder
  74          *
  75          * @return string
  76          */
  77         protected function optimalOrder() {
  78                 return 'fifo';
  79         }
  80
  81         /**
  82          * @see JobQueue::doIsEmpty
  83          *
  84          * @return bool
  85          */
  86         protected function doIsEmpty() {
  87                 return ( $this->doGetSize() == 0 );
  88         }
  89
  90         /**
  91          * @see JobQueue::doGetSize
  92          *
  93          * @return int
  94          */
  95         protected function doGetSize() {
  96                 $unclaimed = $this->getQueueData( 'unclaimed' );
  97
  98                 return $unclaimed ? count( $unclaimed ) : 0;
  99         }
 100
 101         /**
 102          * @see JobQueue::doGetAcquiredCount
 103          *
 104          * @return int
 105          */
 106         protected function doGetAcquiredCount() {
 107                 $claimed = $this->getQueueData( 'claimed' );
 108
 109                 return $claimed ? count( $claimed ) : 0;
 110         }
 111
 112         /**
 113          * @see JobQueue::doPop
 114          *
 115          * @return RunnableJob|false
 116          */
 117         protected function doPop() {
 118                 if ( $this->doGetSize() == 0 ) {
 119                         return false;
 120                 }
 121
 122                 $unclaimed =& $this->getQueueData( 'unclaimed' );
 123                 $claimed =& $this->getQueueData( 'claimed', [] );
 124
 125                 if ( $this->order === 'random' ) {
 126                         $key = array_rand( $unclaimed );
 127                 } else {
 128                         $key = array_key_first( $unclaimed );
 129                 }
 130
 131                 $spec = $unclaimed[$key];
 132                 unset( $unclaimed[$key] );
 133                 $claimed[] = $spec;
 134
 135                 $job = $this->jobFromSpecInternal( $spec );
 136
 137                 $job->setMetadata( 'claimId', array_key_last( $claimed ) );
 138
 139                 return $job;
 140         }
 141
 142         /**
 143          * @see JobQueue::doAck
 144          *
 145          * @param RunnableJob $job
 146          */
 147         protected function doAck( RunnableJob $job ) {
 148                 if ( $this->getAcquiredCount() == 0 ) {
 149                         return;
 150                 }
 151
 152                 $claimed =& $this->getQueueData( 'claimed' );
 153                 unset( $claimed[$job->getMetadata( 'claimId' )] );
 154         }
 155
 156         /**
 157          * @inheritDoc
 158          */
 159         protected function doDelete() {
 160                 if ( isset( self::$data[$this->type][$this->domain] ) ) {
 161                         unset( self::$data[$this->type][$this->domain] );
 162                         if ( !self::$data[$this->type] ) {
 163                                 unset( self::$data[$this->type] );
 164                         }
 165                 }
 166         }
 167
 168         /**
 169          * @see JobQueue::getAllQueuedJobs
 170          *
 171          * @return Iterator<RunnableJob> of Job objects.
 172          */
 173         public function getAllQueuedJobs() {
 174                 $unclaimed = $this->getQueueData( 'unclaimed' );
 175                 if ( !$unclaimed ) {
 176                         return new ArrayIterator( [] );
 177                 }
 178
 179                 return new MappedIterator(
 180                         $unclaimed,
 181                         function ( $value ) {
 182                                 return $this->jobFromSpecInternal( $value );
 183                         }
 184                 );
 185         }
 186
 187         /**
 188          * @see JobQueue::getAllAcquiredJobs
 189          *
 190          * @return Iterator<RunnableJob> of Job objects.
 191          */
 192         public function getAllAcquiredJobs() {
 193                 $claimed = $this->getQueueData( 'claimed' );
 194                 if ( !$claimed ) {
 195                         return new ArrayIterator( [] );
 196                 }
 197
 198                 return new MappedIterator(
 199                         $claimed,
 200                         function ( $value ) {
 201                                 return $this->jobFromSpecInternal( $value );
 202                         }
 203                 );
 204         }
 205
 206         /**
 207          * @param IJobSpecification $spec
 208          * @return RunnableJob
 209          */
 210         public function jobFromSpecInternal( IJobSpecification $spec ) {
 211                 return $this->factoryJob( $spec->getType(), $spec->getParams() );
 212         }
 213
 214         /**
 215          * @param string $field
 216          * @param mixed|null $init
 217          *
 218          * @return mixed
 219          */
 220         private function &getQueueData( $field, $init = null ) {
 221                 if ( !isset( self::$data[$this->type][$this->domain][$field] ) ) {
 222                         if ( $init !== null ) {
 223                                 self::$data[$this->type][$this->domain][$field] = $init;
 224                         } else {
 225                                 return $init;
 226                         }
 227                 }
 228
 229                 return self::$data[$this->type][$this->domain][$field];
 230         }
 231 }