src/Charcoal/Queue/AbstractQueueManager.php
public function processQueue(callable $callback = null)
    {
        $queued = $this->loadQueueItems();

        if (!is_callable($callback)) {
            $callback = $this->processedCallback;
        }

        $success  = [];
        $failures = [];
        $skipped  = [];
        foreach ($queued as $q) {
            try {
                $res = $q->process($this->itemCallback, $this->itemSuccessCallback, $this->itemFailureCallback);
                if ($res === true) {
                    $success[] = $q;
                } elseif ($res === false) {
                    $failures[] = $q;
                } else {
                    $skipped[] = $q;
                }
            } catch (Exception $e) {
                $this->logger->error(
                    sprintf('Could not process a queue item: %s', $e->getMessage())
                );
                $failures[] = $q;
                continue;
            }

            // Throttle according to processing rate.
            if ($this->rate > 0) {
                usleep(1000000/$this->rate);
            }
        }

        if (is_callable($callback)) {
            $callback($success, $failures, $skipped);
        }

        return true;
    }
public function processQueue(callable $callback = null)
    {
        $queued = $this->loadQueueItems();

        if (!is_callable($callback)) {
            $callback = $this->processedCallback;
        }

        $success  = [];
        $failures = [];
        $skipped  = [];
        foreach ($queued as $q) {
            try {
                $res = $q->process($this->itemCallback, $this->itemSuccessCallback, $this->itemFailureCallback);
                if ($res === true) {
                    $success[] = $q;
                } elseif ($res === false) {
                    $failures[] = $q;
                } else {
                    $skipped[] = $q;
                }
            } catch (Exception $e) {
                $this->logger->error(
                    sprintf('Could not process a queue item: %s', $e->getMessage())
                );
                $failures[] = $q;
                continue;
            }

            // Throttle according to processing rate.
            if ($this->rate > 0) {
                usleep(1000000/$this->rate);
            }
        }

        if (is_callable($callback)) {
            $callback($success, $failures, $skipped);
        }

        return true;
    }
public function processQueue ( callable $ callback = null ) { $ queued = $ this -> loadQueueItems ( ) ; if ( ! is_callable ( $ callback ) ) { $ callback = $ this -> processedCallback ; } $ success = [ ] ; $ failures = [ ] ; $ skipped = [ ] ; foreach ( $ queued as $ q ) { try { $ res = $ q -> process ( $ this -> itemCallback , $ this -> itemSuccessCallback , $ this -> itemFailureCallback ) ; if ( $ res === true ) { $ success [ ] = $ q ; } elseif ( $ res === false ) { $ failures [ ] = $ q ; } else { $ skipped [ ] = $ q ; } } catch ( Exception $ e ) { $ this -> logger -> error ( sprintf ( 'Could not process a queue item: %s' , $ e -> getMessage ( ) ) ) ; $ failures [ ] = $ q ; continue ; } // Throttle according to processing rate. if ( $ this -> rate > 0 ) { usleep ( 1000000 / $ this -> rate ) ; } } if ( is_callable ( $ callback ) ) { $ callback ( $ success , $ failures , $ skipped ) ; } return true ; }
Process the items of the queue.

If no callback is passed and a self::$processedCallback is set, the latter is used.

@param  callable $callback An optional alternative callback routine executed
after all queue items are processed.
@return boolean  Success / Failure
Process the items of the queue .