mirror of
https://we.phorge.it/source/phorge.git
synced 2024-11-23 07:12:41 +01:00
Add PhutilFactsUpateIterator
Summary: This iterator processes objects that have been updated. Test Plan: Ran this test script: $cursor = null; $table = new DifferentialRevision(); while (true) { $iterator = new PhabricatorFactsUpdateIterator($table, $cursor); foreach ($iterator as $new_cursor => $update) { echo "{$new_cursor} => D".$update->getID()."\n"; $cursor = $new_cursor; } echo "Zzz...\n"; sleep(5); } Verified it iterated over every object and then stopped. Made a comment on a differenial revision, verified it iterated over the object after 15 seconds. Reviewers: vrana, btrahan Reviewed By: vrana CC: aran Maniphest Tasks: T1562 Differential Revision: https://secure.phabricator.com/D3077
This commit is contained in:
parent
ff61dba7ac
commit
c614af53bf
2 changed files with 87 additions and 1 deletions
|
@ -624,6 +624,7 @@ phutil_register_library_map(array(
|
|||
'PhabricatorEventEngine' => 'infrastructure/events/PhabricatorEventEngine.php',
|
||||
'PhabricatorEventType' => 'infrastructure/events/constant/PhabricatorEventType.php',
|
||||
'PhabricatorExampleEventListener' => 'infrastructure/events/PhabricatorExampleEventListener.php',
|
||||
'PhabricatorFactsUpdateIterator' => 'applications/facts/extract/PhabricatorFactsUpdateIterator.php',
|
||||
'PhabricatorFeedBuilder' => 'applications/feed/builder/PhabricatorFeedBuilder.php',
|
||||
'PhabricatorFeedConstants' => 'applications/feed/constants/PhabricatorFeedConstants.php',
|
||||
'PhabricatorFeedController' => 'applications/feed/controller/PhabricatorFeedController.php',
|
||||
|
@ -1507,7 +1508,7 @@ phutil_register_library_map(array(
|
|||
'LiskIsolationTestCase' => 'PhabricatorTestCase',
|
||||
'LiskIsolationTestDAO' => 'LiskDAO',
|
||||
'LiskIsolationTestDAOException' => 'Exception',
|
||||
'LiskMigrationIterator' => 'Iterator',
|
||||
'LiskMigrationIterator' => 'PhutilBufferedIterator',
|
||||
'ManiphestAction' => 'ManiphestConstants',
|
||||
'ManiphestAuxiliaryFieldDefaultSpecification' => 'ManiphestAuxiliaryFieldSpecification',
|
||||
'ManiphestAuxiliaryFieldTypeException' => 'Exception',
|
||||
|
@ -1648,6 +1649,7 @@ phutil_register_library_map(array(
|
|||
'PhabricatorEvent' => 'PhutilEvent',
|
||||
'PhabricatorEventType' => 'PhutilEventType',
|
||||
'PhabricatorExampleEventListener' => 'PhutilEventListener',
|
||||
'PhabricatorFactsUpdateIterator' => 'PhutilBufferedIterator',
|
||||
'PhabricatorFeedController' => 'PhabricatorController',
|
||||
'PhabricatorFeedDAO' => 'PhabricatorLiskDAO',
|
||||
'PhabricatorFeedPublicStreamController' => 'PhabricatorFeedController',
|
||||
|
|
|
@ -0,0 +1,84 @@
|
|||
<?php
|
||||
|
||||
/*
|
||||
* Copyright 2012 Facebook, Inc.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
/**
|
||||
* Iterate over objects by update time in a stable way. This iterator only works
|
||||
* for "normal" Lisk objects: objects with an autoincrement ID and a
|
||||
* dateModified column.
|
||||
*/
|
||||
final class PhabricatorFactsUpdateIterator extends PhutilBufferedIterator {
|
||||
|
||||
private $cursor;
|
||||
private $object;
|
||||
private $start;
|
||||
private $ignoreUpdatesDuration = 15;
|
||||
|
||||
public function __construct(LiskDAO $object, $start = null) {
|
||||
if ($start === null) {
|
||||
$start = '0:0';
|
||||
}
|
||||
$this->object = $object;
|
||||
$this->start = $start;
|
||||
}
|
||||
|
||||
protected function didRewind() {
|
||||
$this->cursor = $this->start;
|
||||
}
|
||||
|
||||
protected function getCursorFromObject($object) {
|
||||
return $object->getDateModified().':'.$object->getID();
|
||||
}
|
||||
|
||||
public function key() {
|
||||
return $this->getCursorFromObject($this->current());
|
||||
}
|
||||
|
||||
protected function loadPage() {
|
||||
list($after_epoch, $after_id) = explode(':', $this->cursor);
|
||||
|
||||
// NOTE: We ignore recent updates because once we process an update we'll
|
||||
// never process rows behind it again. We need to read only rows which
|
||||
// we're sure no new rows will be inserted behind. If we read a row that
|
||||
// was updated on the current second, another update later on in this second
|
||||
// could affect an object with a lower ID, and we'd skip that update. To
|
||||
// avoid this, just ignore any rows which have been updated in the last few
|
||||
// seconds. This also reduces the amount of work we need to do if an object
|
||||
// is repeatedly updated; we will just look at the end state without
|
||||
// processing the intermediate states. Finally, this gives us reasonable
|
||||
// protections against clock skew between the machine the daemon is running
|
||||
// on and any machines performing writes.
|
||||
|
||||
$page = $this->object->loadAllWhere(
|
||||
'((dateModified > %d) OR (dateModified = %d AND id > %d))
|
||||
AND (dateModified < %d - %d)
|
||||
ORDER BY dateModified ASC, id ASC LIMIT %d',
|
||||
$after_epoch,
|
||||
$after_epoch,
|
||||
$after_id,
|
||||
time(),
|
||||
$this->ignoreUpdatesDuration,
|
||||
$this->getPageSize());
|
||||
|
||||
if ($page) {
|
||||
$this->cursor = $this->getCursorFromObject(end($page));
|
||||
}
|
||||
|
||||
return $page;
|
||||
}
|
||||
|
||||
}
|
Loading…
Reference in a new issue