public function perform()
    {
        $this->getStat()->increment(
            MONGO_QUEUE_APPLY_OPERATION_JOB . '.' . SUBJECT_COUNT,
            count($this->args[self::SUBJECTS_KEY])
        );

        foreach ($this->args[self::SUBJECTS_KEY] as $subject) {
            $opTimer = new \Tripod\Timer();
            $opTimer->start();

            $impactedSubject = $this->createImpactedSubject($subject);
            $impactedSubject->update();

            $opTimer->stop();
            // stat time taken to perform operation for the given subject
            $this->getStat()->timer(MONGO_QUEUE_APPLY_OPERATION.'.'.$subject['operation'], $opTimer->result());

            /**
             * ApplyOperation jobs can either apply to a single resource (e.g. 'create composite for the given
             * resource uri) or for a specification id (i.e. regenerate all of the composites defined by the
             * specification).  For the latter, we need to keep track of how many jobs have run so we can clean
             * up any stale composite documents when completed.  The TRACKING_KEY value will be the JobGroup id.
             */
            if (isset($this->args[self::TRACKING_KEY])) {
                $jobGroup = $this->getJobGroup($subject['storeName'], $this->args[self::TRACKING_KEY]);
                $jobCount = $jobGroup->incrementJobCount(-1);
                if ($jobCount <= 0) {
                    // @todo Replace this with ObjectId->getTimestamp() if we upgrade Mongo driver to 1.2
                    $timestamp = new \MongoDB\BSON\UTCDateTime(hexdec(substr($jobGroup->getId(), 0, 8)) * 1000);
                    $tripod = $this->getTripod($subject['storeName'], $subject['podName']);
                    $count = 0;
                    foreach ($subject['specTypes'] as $specId) {
                        switch ($subject['operation']) {
                            case \OP_VIEWS:
                                $count += $tripod->getComposite(\OP_VIEWS)->deleteViewsByViewId($specId, $timestamp);
                                break;
                            case \OP_TABLES:
                                $count += $tripod->getComposite(\OP_TABLES)->deleteTableRowsByTableId($specId, $timestamp);
                                break;
                            case \OP_SEARCH:
                                $searchProvider = $this->getSearchProvider($tripod);
                                $count += $searchProvider->deleteSearchDocumentsByTypeId($specId, $timestamp);
                                break;
                        }
                    }
                    $this->infoLog(
                        '[JobGroupId ' . $jobGroup->getId()->__toString() . '] composite cleanup for ' .
                        $subject['operation'] . ' removed ' . $count . ' stale composite documents'
                    );
                }
            }
        }
    }
public function perform()
    {
        $this->getStat()->increment(
            MONGO_QUEUE_APPLY_OPERATION_JOB . '.' . SUBJECT_COUNT,
            count($this->args[self::SUBJECTS_KEY])
        );

        foreach ($this->args[self::SUBJECTS_KEY] as $subject) {
            $opTimer = new \Tripod\Timer();
            $opTimer->start();

            $impactedSubject = $this->createImpactedSubject($subject);
            $impactedSubject->update();

            $opTimer->stop();
            // stat time taken to perform operation for the given subject
            $this->getStat()->timer(MONGO_QUEUE_APPLY_OPERATION.'.'.$subject['operation'], $opTimer->result());

            /**
             * ApplyOperation jobs can either apply to a single resource (e.g. 'create composite for the given
             * resource uri) or for a specification id (i.e. regenerate all of the composites defined by the
             * specification).  For the latter, we need to keep track of how many jobs have run so we can clean
             * up any stale composite documents when completed.  The TRACKING_KEY value will be the JobGroup id.
             */
            if (isset($this->args[self::TRACKING_KEY])) {
                $jobGroup = $this->getJobGroup($subject['storeName'], $this->args[self::TRACKING_KEY]);
                $jobCount = $jobGroup->incrementJobCount(-1);
                if ($jobCount <= 0) {
                    // @todo Replace this with ObjectId->getTimestamp() if we upgrade Mongo driver to 1.2
                    $timestamp = new \MongoDB\BSON\UTCDateTime(hexdec(substr($jobGroup->getId(), 0, 8)) * 1000);
                    $tripod = $this->getTripod($subject['storeName'], $subject['podName']);
                    $count = 0;
                    foreach ($subject['specTypes'] as $specId) {
                        switch ($subject['operation']) {
                            case \OP_VIEWS:
                                $count += $tripod->getComposite(\OP_VIEWS)->deleteViewsByViewId($specId, $timestamp);
                                break;
                            case \OP_TABLES:
                                $count += $tripod->getComposite(\OP_TABLES)->deleteTableRowsByTableId($specId, $timestamp);
                                break;
                            case \OP_SEARCH:
                                $searchProvider = $this->getSearchProvider($tripod);
                                $count += $searchProvider->deleteSearchDocumentsByTypeId($specId, $timestamp);
                                break;
                        }
                    }
                    $this->infoLog(
                        '[JobGroupId ' . $jobGroup->getId()->__toString() . '] composite cleanup for ' .
                        $subject['operation'] . ' removed ' . $count . ' stale composite documents'
                    );
                }
            }
        }
    }
public function perform ( ) { $ this -> getStat ( ) -> increment ( MONGO_QUEUE_APPLY_OPERATION_JOB . '.' . SUBJECT_COUNT , count ( $ this -> args [ self :: SUBJECTS_KEY ] ) ) ; foreach ( $ this -> args [ self :: SUBJECTS_KEY ] as $ subject ) { $ opTimer = new \ Tripod \ Timer ( ) ; $ opTimer -> start ( ) ; $ impactedSubject = $ this -> createImpactedSubject ( $ subject ) ; $ impactedSubject -> update ( ) ; $ opTimer -> stop ( ) ; // stat time taken to perform operation for the given subject $ this -> getStat ( ) -> timer ( MONGO_QUEUE_APPLY_OPERATION . '.' . $ subject [ 'operation' ] , $ opTimer -> result ( ) ) ; /**
             * ApplyOperation jobs can either apply to a single resource (e.g. 'create composite for the given
             * resource uri) or for a specification id (i.e. regenerate all of the composites defined by the
             * specification).  For the latter, we need to keep track of how many jobs have run so we can clean
             * up any stale composite documents when completed.  The TRACKING_KEY value will be the JobGroup id.
             */ if ( isset ( $ this -> args [ self :: TRACKING_KEY ] ) ) { $ jobGroup = $ this -> getJobGroup ( $ subject [ 'storeName' ] , $ this -> args [ self :: TRACKING_KEY ] ) ; $ jobCount = $ jobGroup -> incrementJobCount ( - 1 ) ; if ( $ jobCount <= 0 ) { // @todo Replace this with ObjectId->getTimestamp() if we upgrade Mongo driver to 1.2 $ timestamp = new \ MongoDB \ BSON \ UTCDateTime ( hexdec ( substr ( $ jobGroup -> getId ( ) , 0 , 8 ) ) * 1000 ) ; $ tripod = $ this -> getTripod ( $ subject [ 'storeName' ] , $ subject [ 'podName' ] ) ; $ count = 0 ; foreach ( $ subject [ 'specTypes' ] as $ specId ) { switch ( $ subject [ 'operation' ] ) { case \ OP_VIEWS : $ count += $ tripod -> getComposite ( \ OP_VIEWS ) -> deleteViewsByViewId ( $ specId , $ timestamp ) ; break ; case \ OP_TABLES : $ count += $ tripod -> getComposite ( \ OP_TABLES ) -> deleteTableRowsByTableId ( $ specId , $ timestamp ) ; break ; case \ OP_SEARCH : $ searchProvider = $ this -> getSearchProvider ( $ tripod ) ; $ count += $ searchProvider -> deleteSearchDocumentsByTypeId ( $ specId , $ timestamp ) ; break ; } } $ this -> infoLog ( '[JobGroupId ' . $ jobGroup -> getId ( ) -> __toString ( ) . '] composite cleanup for ' . $ subject [ 'operation' ] . ' removed ' . $ count . ' stale composite documents' ) ; } } } }
Run the ApplyOperation job
@throws \Exception
Run the ApplyOperation job