lib/Resque.php
private static function removeItems($queue, $items = [])
    {
        $counter = 0;
        $originalQueue = 'queue:' . $queue;
        $tempQueue = $originalQueue . ':temp:' . time();
        $requeueQueue = $tempQueue . ':requeue';

        // move each item from original queue to temp queue and process it
        $finished = false;
        while (!$finished) {
            $string = self::redis()->rpoplpush($originalQueue, self::redis()->getPrefix() . $tempQueue);

            if (!empty($string)) {
                if (self::matchItem($string, $items)) {
                    self::redis()->rpop($tempQueue);
                    $counter++;
                } else {
                    self::redis()->rpoplpush($tempQueue, self::redis()->getPrefix() . $requeueQueue);
                }
            } else {
                $finished = true;
            }
        }

        // move back from temp queue to original queue
        $finished = false;
        while (!$finished) {
            $string = self::redis()->rpoplpush($requeueQueue, self::redis()->getPrefix() . $originalQueue);
            if (empty($string)) {
                $finished = true;
            }
        }

        // remove temp queue and requeue queue
        self::redis()->del($requeueQueue);
        self::redis()->del($tempQueue);

        return $counter;
    }
private static function removeItems($queue, $items = [])
    {
        $counter = 0;
        $originalQueue = 'queue:' . $queue;
        $tempQueue = $originalQueue . ':temp:' . time();
        $requeueQueue = $tempQueue . ':requeue';

        // move each item from original queue to temp queue and process it
        $finished = false;
        while (!$finished) {
            $string = self::redis()->rpoplpush($originalQueue, self::redis()->getPrefix() . $tempQueue);

            if (!empty($string)) {
                if (self::matchItem($string, $items)) {
                    self::redis()->rpop($tempQueue);
                    $counter++;
                } else {
                    self::redis()->rpoplpush($tempQueue, self::redis()->getPrefix() . $requeueQueue);
                }
            } else {
                $finished = true;
            }
        }

        // move back from temp queue to original queue
        $finished = false;
        while (!$finished) {
            $string = self::redis()->rpoplpush($requeueQueue, self::redis()->getPrefix() . $originalQueue);
            if (empty($string)) {
                $finished = true;
            }
        }

        // remove temp queue and requeue queue
        self::redis()->del($requeueQueue);
        self::redis()->del($tempQueue);

        return $counter;
    }
private static function removeItems ( $ queue , $ items = [ ] ) { $ counter = 0 ; $ originalQueue = 'queue:' . $ queue ; $ tempQueue = $ originalQueue . ':temp:' . time ( ) ; $ requeueQueue = $ tempQueue . ':requeue' ; // move each item from original queue to temp queue and process it $ finished = false ; while ( ! $ finished ) { $ string = self :: redis ( ) -> rpoplpush ( $ originalQueue , self :: redis ( ) -> getPrefix ( ) . $ tempQueue ) ; if ( ! empty ( $ string ) ) { if ( self :: matchItem ( $ string , $ items ) ) { self :: redis ( ) -> rpop ( $ tempQueue ) ; $ counter ++ ; } else { self :: redis ( ) -> rpoplpush ( $ tempQueue , self :: redis ( ) -> getPrefix ( ) . $ requeueQueue ) ; } } else { $ finished = true ; } } // move back from temp queue to original queue $ finished = false ; while ( ! $ finished ) { $ string = self :: redis ( ) -> rpoplpush ( $ requeueQueue , self :: redis ( ) -> getPrefix ( ) . $ originalQueue ) ; if ( empty ( $ string ) ) { $ finished = true ; } } // remove temp queue and requeue queue self :: redis ( ) -> del ( $ requeueQueue ) ; self :: redis ( ) -> del ( $ tempQueue ) ; return $ counter ; }
Remove Items from the queue
Safely moving each item to a temporary queue before processing it
If the Job matches, counts otherwise puts it in a requeue_queue
which at the end eventually be copied back into the original queue

@private

@param string $queue The name of the queue
@param array $items
@return integer number of deleted items
Remove Items from the queue Safely moving each item to a temporary queue before processing it If the Job matches counts otherwise puts it in a requeue_queue which at the end eventually be copied back into the original queue