Unverified Commit e0a2b899 authored by Andreas Braun's avatar Andreas Braun

Merge pull request #674

parents a800cdef 7f20ffb6
...@@ -54,6 +54,9 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber ...@@ -54,6 +54,9 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
/** @var boolean */ /** @var boolean */
private $isRewindNop; private $isRewindNop;
/** @var boolean */
private $isValid = false;
/** @var object|null */ /** @var object|null */
private $postBatchResumeToken; private $postBatchResumeToken;
...@@ -126,6 +129,15 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber ...@@ -126,6 +129,15 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
} }
} }
/**
* @see https://php.net/iteratoriterator.current
* @return mixed
*/
public function current()
{
return $this->isValid ? parent::current() : null;
}
/** /**
* Returns the resume token for the iterator's current position. * Returns the resume token for the iterator's current position.
* *
...@@ -140,6 +152,15 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber ...@@ -140,6 +152,15 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
return $this->resumeToken; return $this->resumeToken;
} }
/**
* @see https://php.net/iteratoriterator.key
* @return mixed
*/
public function key()
{
return $this->isValid ? parent::key() : null;
}
/** /**
* @see https://php.net/iteratoriterator.rewind * @see https://php.net/iteratoriterator.rewind
* @return void * @return void
...@@ -181,6 +202,15 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber ...@@ -181,6 +202,15 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
$this->onIteration(false); $this->onIteration(false);
} }
/**
* @see https://php.net/iteratoriterator.valid
* @return boolean
*/
public function valid()
{
return $this->isValid;
}
/** /**
* Extracts the resume token (i.e. "_id" field) from a change document. * Extracts the resume token (i.e. "_id" field) from a change document.
* *
...@@ -204,10 +234,12 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber ...@@ -204,10 +234,12 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
: (isset($document->_id) ? $document->_id : null); : (isset($document->_id) ? $document->_id : null);
if (! isset($resumeToken)) { if (! isset($resumeToken)) {
$this->isValid = false;
throw ResumeTokenException::notFound(); throw ResumeTokenException::notFound();
} }
if (! is_array($resumeToken) && ! is_object($resumeToken)) { if (! is_array($resumeToken) && ! is_object($resumeToken)) {
$this->isValid = false;
throw ResumeTokenException::invalidType($resumeToken); throw ResumeTokenException::invalidType($resumeToken);
} }
...@@ -232,16 +264,16 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber ...@@ -232,16 +264,16 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
*/ */
private function onIteration($incrementBatchPosition) private function onIteration($incrementBatchPosition)
{ {
$isValid = $this->valid(); $this->isValid = parent::valid();
/* Disable rewind()'s NOP behavior once we advance to a valid position. /* Disable rewind()'s NOP behavior once we advance to a valid position.
* This will allow the driver to throw a LogicException if rewind() is * This will allow the driver to throw a LogicException if rewind() is
* called after the cursor has advanced past its first element. */ * called after the cursor has advanced past its first element. */
if ($this->isRewindNop && $isValid) { if ($this->isRewindNop && $this->isValid) {
$this->isRewindNop = false; $this->isRewindNop = false;
} }
if ($incrementBatchPosition && $isValid) { if ($incrementBatchPosition && $this->isValid) {
$this->batchPosition++; $this->batchPosition++;
} }
...@@ -253,7 +285,7 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber ...@@ -253,7 +285,7 @@ class ChangeStreamIterator extends IteratorIterator implements CommandSubscriber
* from the current document if possible. */ * from the current document if possible. */
if ($this->isAtEndOfBatch() && $this->postBatchResumeToken !== null) { if ($this->isAtEndOfBatch() && $this->postBatchResumeToken !== null) {
$this->resumeToken = $this->postBatchResumeToken; $this->resumeToken = $this->postBatchResumeToken;
} elseif ($isValid) { } elseif ($this->isValid) {
$this->resumeToken = $this->extractResumeToken($this->current()); $this->resumeToken = $this->extractResumeToken($this->current());
} }
} }
......
...@@ -1125,34 +1125,23 @@ class WatchFunctionalTest extends FunctionalTestCase ...@@ -1125,34 +1125,23 @@ class WatchFunctionalTest extends FunctionalTestCase
$changeStream->next(); $changeStream->next();
$this->fail('Exception for missing resume token was not thrown'); $this->fail('Exception for missing resume token was not thrown');
} catch (ResumeTokenException $e) { } catch (ResumeTokenException $e) {
/* If a client-side error is thrown (server < 4.1.8), the tailable /* On server versions < 4.1.8, a client-side error is thrown. */
* cursor's position is still valid. This may change once PHPLIB-456
* is implemented. */
$expectedValid = true;
$expectedKey = 0;
} catch (ServerException $e) { } catch (ServerException $e) {
/* If a server-side error is thrown (server >= 4.1.8), the tailable /* On server versions >= 4.1.8, the error is thrown server-side. */
* cursor's position is not valid. */
$expectedValid = false;
$expectedKey = null;
} }
$this->assertSame($expectedValid, $changeStream->valid()); $this->assertFalse($changeStream->valid());
$this->assertSame($expectedKey, $changeStream->key()); $this->assertNull($changeStream->key());
try { try {
$changeStream->next(); $changeStream->next();
$this->fail('Exception for missing resume token was not thrown'); $this->fail('Exception for missing resume token was not thrown');
} catch (ResumeTokenException $e) { } catch (ResumeTokenException $e) {
$expectedValid = true;
$expectedKey = 0;
} catch (ServerException $e) { } catch (ServerException $e) {
$expectedValid = false;
$expectedKey = null;
} }
$this->assertSame($expectedValid, $changeStream->valid()); $this->assertFalse($changeStream->valid());
$this->assertSame($expectedKey, $changeStream->key()); $this->assertNull($changeStream->key());
} }
public function testSessionPersistsAfterResume() public function testSessionPersistsAfterResume()
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment