-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathEventSourcedAggregateRoot.php
119 lines (101 loc) · 2.74 KB
/
EventSourcedAggregateRoot.php
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
<?php
/*
* This file is part of the broadway/broadway package.
*
* (c) Qandidate.com <[email protected]>
*
* For the full copyright and license information, please view the LICENSE
* file that was distributed with this source code.
*/
namespace Broadway\EventSourcing;
use Broadway\Domain\AggregateRoot as AggregateRootInterface;
use Broadway\Domain\DomainEventStream;
use Broadway\Domain\DomainEventStreamInterface;
use Broadway\Domain\DomainMessage;
use Broadway\Domain\Metadata;
/**
* Convenience base class for event sourced aggregate roots.
*/
abstract class EventSourcedAggregateRoot implements AggregateRootInterface
{
/**
* @var array
*/
private $uncommittedEvents = [];
private $playhead = -1; // 0-based playhead allows events[0] to contain playhead 0
/**
* Applies an event. The event is added to the AggregateRoot's list of uncommitted events.
*
* @param $event
*/
public function apply($event)
{
$this->handleRecursively($event);
$this->playhead++;
$this->uncommittedEvents[] = DomainMessage::recordNow(
$this->getAggregateRootId(),
$this->playhead,
new Metadata([]),
$event
);
}
/**
* {@inheritDoc}
*/
public function getUncommittedEvents()
{
$stream = new DomainEventStream($this->uncommittedEvents);
$this->uncommittedEvents = [];
return $stream;
}
/**
* Initializes the aggregate using the given "history" of events.
*/
public function initializeState(DomainEventStreamInterface $stream)
{
foreach ($stream as $message) {
$this->playhead++;
$this->handleRecursively($message->getPayload());
}
}
/**
* Handles event if capable.
*
* @param $event
*/
protected function handle($event)
{
$method = $this->getApplyMethod($event);
if (! method_exists($this, $method)) {
return;
}
$this->$method($event);
}
/**
* {@inheritDoc}
*/
protected function handleRecursively($event)
{
$this->handle($event);
foreach ($this->getChildEntities() as $entity) {
$entity->registerAggregateRoot($this);
$entity->handleRecursively($event);
}
}
/**
* Returns all child entities
*
* Override this method if your aggregate root contains child entities.
*
* @return EventSourcedEntityInterface[]
*/
protected function getChildEntities()
{
return [];
}
private function getApplyMethod($event)
{
$classParts = explode('\\', get_class($event));
return 'apply' . end($classParts);
}
}