-
-
Notifications
You must be signed in to change notification settings - Fork 30
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #269 from neos/feature/replay-until-sequence-number
FEATURE: Replay / catch up until sequence number
- Loading branch information
Showing
6 changed files
with
390 additions
and
43 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
111 changes: 111 additions & 0 deletions
111
Classes/EventStore/Storage/InMemory/InMemoryStreamIterator.php
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| <?php | ||
| declare(strict_types=1); | ||
| namespace Neos\EventSourcing\EventStore\Storage\InMemory; | ||
|
|
||
| /* | ||
| * This file is part of the Neos.EventSourcing package. | ||
| * | ||
| * (c) Contributors of the Neos Project - www.neos.io | ||
| * | ||
| * This package is Open Source Software. For the full copyright and license | ||
| * information, please view the LICENSE file which was distributed with this | ||
| * source code. | ||
| */ | ||
|
|
||
| use ArrayIterator; | ||
| use Neos\EventSourcing\EventStore\EventStreamIteratorInterface; | ||
| use Neos\EventSourcing\EventStore\RawEvent; | ||
| use Neos\EventSourcing\EventStore\StreamName; | ||
|
|
||
| /** | ||
| * Stream Iterator for an in-memory based EventStore – intended for testing | ||
| */ | ||
| final class InMemoryStreamIterator implements EventStreamIteratorInterface | ||
| { | ||
| /** | ||
| * @var int | ||
| */ | ||
| private $currentOffset = 0; | ||
|
|
||
| /** | ||
| * @var ArrayIterator | ||
| */ | ||
| private $innerIterator; | ||
|
|
||
| /** | ||
| * @var array | ||
| */ | ||
| private $eventRecords = []; | ||
|
|
||
| /** | ||
| * @param array $eventRecords | ||
| */ | ||
| public function setEventRecords(array $eventRecords): void | ||
| { | ||
| $this->eventRecords = $eventRecords; | ||
| $this->innerIterator = new ArrayIterator($this->eventRecords); | ||
| } | ||
|
|
||
| /** | ||
| * @return RawEvent | ||
| * @throws \JsonException | ||
| */ | ||
| public function current(): RawEvent | ||
| { | ||
| $currentEventData = $this->innerIterator->current(); | ||
| $payload = json_decode($currentEventData['payload'], true, 512, JSON_THROW_ON_ERROR); | ||
| $metadata = json_decode($currentEventData['metadata'], true, 512, JSON_THROW_ON_ERROR); | ||
| try { | ||
| $recordedAt = new \DateTimeImmutable($currentEventData['recordedat']); | ||
| } catch (\Exception $exception) { | ||
| throw new \RuntimeException(sprintf('Could not parse recordedat timestamp "%s" as date.', $currentEventData['recordedat']), 1597843669, $exception); | ||
| } | ||
| return new RawEvent( | ||
| (int)$currentEventData['sequencenumber'], | ||
| $currentEventData['type'], | ||
| $payload, | ||
| $metadata, | ||
| StreamName::fromString($currentEventData['stream']), | ||
| (int)$currentEventData['version'], | ||
| $currentEventData['id'], | ||
| $recordedAt | ||
| ); | ||
| } | ||
|
|
||
| /** | ||
| * @return void | ||
| */ | ||
| public function next(): void | ||
| { | ||
| $this->currentOffset = $this->innerIterator->current()['sequencenumber']; | ||
| $this->innerIterator->next(); | ||
| } | ||
|
|
||
| /** | ||
| * @return bool|int | ||
| */ | ||
| public function key() | ||
| { | ||
| return $this->innerIterator->valid() ? $this->innerIterator->current()['sequencenumber'] : null; | ||
| } | ||
|
|
||
| /** | ||
| * @return bool | ||
| */ | ||
| public function valid(): bool | ||
| { | ||
| return $this->innerIterator->valid(); | ||
| } | ||
|
|
||
| /** | ||
| * @return void | ||
| */ | ||
| public function rewind(): void | ||
| { | ||
| if ($this->currentOffset === 0) { | ||
| return; | ||
| } | ||
| $this->innerIterator->rewind(); | ||
| $this->currentOffset = 0; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.