Skip to content

Commit 7aa2150

Browse files
author
Alexander Miertsch
committed
Merge pull request #113 from prolic/snapshot
Snapshots
2 parents be6d688 + 4ddf5af commit 7aa2150

11 files changed

Lines changed: 506 additions & 16 deletions

‎src/Adapter/InMemoryAdapter.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ public function load(StreamName $streamName, $minVersion = null)
7979

8080
$streamEvents = $this->streams[$streamName->toString()];
8181

82-
if (!is_null($minVersion)) {
82+
if (null !== $minVersion) {
8383
$filteredEvents = [];
8484

8585
foreach ($streamEvents as $streamEvent) {
@@ -111,7 +111,7 @@ public function loadEvents(StreamName $streamName, array $metadata = [], $minVer
111111

112112
foreach ($this->streams[$streamName->toString()] as $index => $streamEvent) {
113113
if ($this->matchMetadataWith($streamEvent, $metadata)) {
114-
if (is_null($minVersion) || $streamEvent->version() >= $minVersion) {
114+
if (null === $minVersion || $streamEvent->version() >= $minVersion) {
115115
$streamEvents[] = $streamEvent;
116116
}
117117
}

‎src/Aggregate/AggregateRepository.php‎

Lines changed: 54 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414
use ArrayIterator;
1515
use Assert\Assertion;
1616
use Prooph\EventStore\EventStore;
17+
use Prooph\EventStore\Snapshot\SnapshotStore;
1718
use Prooph\EventStore\Stream\SingleStreamStrategy;
1819
use Prooph\EventStore\Stream\StreamStrategy;
1920

@@ -50,6 +51,11 @@ class AggregateRepository
5051
*/
5152
protected $identityMap;
5253

54+
/**
55+
* @var SnapshotStore|null
56+
*/
57+
protected $snapshotStore;
58+
5359
/**
5460
* @var array
5561
*/
@@ -60,21 +66,24 @@ class AggregateRepository
6066
* @param AggregateType $aggregateType
6167
* @param AggregateTranslator $aggregateTranslator
6268
* @param StreamStrategy|null $streamStrategy
63-
* @param IdentityMap $identityMap
69+
* @param IdentityMap|null $identityMap
70+
* @param SnapshotStore|null $snapshotStore
6471
*/
6572
public function __construct(
6673
EventStore $eventStore,
6774
AggregateType $aggregateType,
6875
AggregateTranslator $aggregateTranslator,
6976
StreamStrategy $streamStrategy = null,
70-
IdentityMap $identityMap = null
77+
IdentityMap $identityMap = null,
78+
SnapshotStore $snapshotStore = null
7179
) {
7280
$this->eventStore = $eventStore;
7381
$this->eventStore->getActionEventEmitter()->attachListener('commit.pre', [$this, 'addPendingEventsToStream']);
7482
$this->eventStore->getActionEventEmitter()->attachListener('commit.post', [$this, 'applyPendingStreamEvents']);
7583

7684
$this->aggregateType = $aggregateType;
7785
$this->aggregateTranslator = $aggregateTranslator;
86+
$this->snapshotStore = $snapshotStore;
7887

7988
if (null === $streamStrategy) {
8089
$streamStrategy = new SingleStreamStrategy($this->eventStore);
@@ -184,15 +193,25 @@ public function getAggregateRoot($aggregateId)
184193
{
185194
Assertion::string($aggregateId, 'AggregateId needs to be string');
186195

187-
$aggregateRoot = $this->identityMap->get($this->aggregateType, $aggregateId);
196+
$eventSourcedAggregateRoot = $this->identityMap->get($this->aggregateType, $aggregateId);
188197

189-
if ($aggregateRoot) {
190-
return $aggregateRoot;
198+
if ($eventSourcedAggregateRoot) {
199+
return $eventSourcedAggregateRoot;
200+
}
201+
202+
if ($this->snapshotStore) {
203+
$eventSourcedAggregateRoot = $this->loadFromSnapshotStore($aggregateId);
204+
205+
if ($eventSourcedAggregateRoot) {
206+
$this->identityMap->add($this->aggregateType, $aggregateId, $eventSourcedAggregateRoot);
207+
208+
return $eventSourcedAggregateRoot;
209+
}
191210
}
192211

193212
$streamEvents = $this->streamStrategy->read($this->aggregateType, $aggregateId);
194213

195-
if (count($streamEvents) === 0) {
214+
if (!$streamEvents->valid()) {
196215
return;
197216
}
198217

@@ -207,4 +226,33 @@ public function getAggregateRoot($aggregateId)
207226

208227
return $eventSourcedAggregateRoot;
209228
}
229+
230+
/**
231+
* @param string $aggregateId
232+
* @return null|object
233+
*/
234+
protected function loadFromSnapshotStore($aggregateId)
235+
{
236+
$snapshot = $this->snapshotStore->get($this->aggregateType, $aggregateId);
237+
238+
if (!$snapshot) {
239+
return;
240+
}
241+
242+
$aggregateRoot = $snapshot->aggregateRoot();
243+
244+
$streamEvents = $this->streamStrategy->read(
245+
$this->aggregateType,
246+
$aggregateId,
247+
$snapshot->lastVersion() + 1
248+
);
249+
250+
if (!$streamEvents->valid()) {
251+
return $aggregateRoot;
252+
}
253+
254+
$this->aggregateTranslator->applyPendingStreamEvents($aggregateRoot, $streamEvents);
255+
256+
return $aggregateRoot;
257+
}
210258
}

‎src/Aggregate/ConfigurableAggregateTranslator.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -125,11 +125,11 @@ public function extractAggregateId($eventSourcedAggregateRoot)
125125

126126
/**
127127
* @param AggregateType $aggregateType
128-
* @param \Iterator $historyEvents
128+
* @param Iterator $historyEvents
129129
* @throws Exception\AggregateTranslationFailedException
130130
* @return object reconstructed EventSourcedAggregateRoot
131131
*/
132-
public function reconstituteAggregateFromHistory(AggregateType $aggregateType, \Iterator $historyEvents)
132+
public function reconstituteAggregateFromHistory(AggregateType $aggregateType, Iterator $historyEvents)
133133
{
134134
if ($this->messageToEventCallback) {
135135
$historyEvents = new MapIterator($historyEvents, $this->messageToEventCallback);

‎src/Snapshot/Adapter/Adapter.php‎

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
<?php
2+
/*
3+
* This file is part of the prooph/event-store.
4+
* (c) 2014 - 2015 prooph software GmbH <contact@prooph.de>
5+
*
6+
* For the full copyright and license information, please view the LICENSE
7+
* file that was distributed with this source code.
8+
*
9+
* Date: 10/09/15 - 07:20 PM
10+
*/
11+
12+
namespace Prooph\EventStore\Snapshot\Adapter;
13+
14+
use Prooph\EventStore\Aggregate\AggregateType;
15+
use Prooph\EventStore\Snapshot\Snapshot;
16+
17+
/**
18+
* Interface Adapter
19+
* @package Prooph\EventStore\Snapshot\Adapter
20+
*/
21+
interface Adapter
22+
{
23+
/**
24+
* Get the aggregate root if it exists otherwise null
25+
*
26+
* @param AggregateType $aggregateType
27+
* @param string $aggregateId
28+
* @return Snapshot
29+
*/
30+
public function get(AggregateType $aggregateType, $aggregateId);
31+
32+
/**
33+
* Add a snapshot
34+
*
35+
* @param Snapshot $snapshot
36+
* @return void
37+
*/
38+
public function add(Snapshot $snapshot);
39+
}

‎src/Snapshot/Adapter/InMemoryAdapter.php‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
<?php
2+
/*
3+
* This file is part of the prooph/event-store.
4+
* (c) 2014 - 2015 prooph software GmbH <contact@prooph.de>
5+
*
6+
* For the full copyright and license information, please view the LICENSE
7+
* file that was distributed with this source code.
8+
*
9+
* Date: 10/09/15 - 07:20 PM
10+
*/
11+
12+
namespace Prooph\EventStore\Snapshot\Adapter;
13+
14+
use Prooph\EventStore\Aggregate\AggregateType;
15+
use Prooph\EventStore\Snapshot\Snapshot;
16+
17+
/**
18+
* Class InMemoryAdapter
19+
* @package Prooph\EventStore\Snapshot\Adapter
20+
*/
21+
final class InMemoryAdapter implements Adapter
22+
{
23+
/**
24+
* @var array
25+
*/
26+
private $map = [];
27+
28+
/**
29+
* Get the aggregate root if it exists otherwise null
30+
*
31+
* @param AggregateType $aggregateType
32+
* @param string $aggregateId
33+
* @return null|object
34+
*/
35+
public function get(AggregateType $aggregateType, $aggregateId)
36+
{
37+
if (! isset($this->map[$aggregateType->toString()][$aggregateId])) {
38+
return;
39+
}
40+
41+
return $this->map[$aggregateType->toString()][$aggregateId];
42+
}
43+
44+
/**
45+
* Add a snapshot
46+
*
47+
* @param Snapshot $snapshot
48+
* @return void
49+
*/
50+
public function add(Snapshot $snapshot)
51+
{
52+
$this->map[$snapshot->aggregateType()->toString()][$snapshot->aggregateId()] = $snapshot;
53+
}
54+
}

‎src/Snapshot/Snapshot.php‎

Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
<?php
2+
/*
3+
* This file is part of the prooph/event-store.
4+
* (c) 2014 - 2015 prooph software GmbH <contact@prooph.de>
5+
*
6+
* For the full copyright and license information, please view the LICENSE
7+
* file that was distributed with this source code.
8+
*
9+
* Date: 10/09/15 - 07:20 PM
10+
*/
11+
12+
namespace Prooph\EventStore\Snapshot;
13+
14+
use Assert\Assertion;
15+
use DateTimeImmutable;
16+
use Prooph\EventStore\Aggregate\AggregateType;
17+
18+
/**
19+
* Class Snapshot
20+
* @package Prooph\EventStore\Snapshot
21+
*/
22+
final class Snapshot
23+
{
24+
/**
25+
* @var AggregateType
26+
*/
27+
private $aggregateType;
28+
29+
/**
30+
* @var string
31+
*/
32+
private $aggregateId;
33+
34+
/**
35+
* @var object
36+
*/
37+
private $aggregateRoot;
38+
39+
/**
40+
* @var int
41+
*/
42+
private $lastVersion;
43+
44+
/**
45+
* @var DateTimeImmutable
46+
*/
47+
private $createdAt;
48+
49+
/**
50+
* @param AggregateType $aggregateType
51+
* @param string $aggregateId
52+
* @param object $aggregateRoot
53+
* @param int $lastVersion
54+
* @param DateTimeImmutable $createdAt
55+
*/
56+
public function __construct(
57+
AggregateType $aggregateType,
58+
$aggregateId,
59+
$aggregateRoot,
60+
$lastVersion,
61+
DateTimeImmutable $createdAt
62+
) {
63+
Assertion::minLength($aggregateId, 1);
64+
Assertion::isObject($aggregateRoot);
65+
Assertion::min($lastVersion, 1);
66+
67+
$this->aggregateType = $aggregateType;
68+
$this->aggregateId = $aggregateId;
69+
$this->aggregateRoot = $aggregateRoot;
70+
$this->lastVersion = $lastVersion;
71+
$this->createdAt = $createdAt;
72+
}
73+
74+
/**
75+
* @return AggregateType
76+
*/
77+
public function aggregateType()
78+
{
79+
return $this->aggregateType;
80+
}
81+
82+
/**
83+
* @return string
84+
*/
85+
public function aggregateId()
86+
{
87+
return $this->aggregateId;
88+
}
89+
90+
/**
91+
* @return object
92+
*/
93+
public function aggregateRoot()
94+
{
95+
return $this->aggregateRoot;
96+
}
97+
98+
/**
99+
* @return int
100+
*/
101+
public function lastVersion()
102+
{
103+
return $this->lastVersion;
104+
}
105+
106+
/**
107+
* @return DateTimeImmutable
108+
*/
109+
public function createdAt()
110+
{
111+
return $this->createdAt;
112+
}
113+
}

0 commit comments

Comments
 (0)