|
1
|
|
|
<?php |
|
2
|
|
|
|
|
3
|
|
|
/* |
|
4
|
|
|
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS |
|
5
|
|
|
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT |
|
6
|
|
|
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR |
|
7
|
|
|
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT |
|
8
|
|
|
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, |
|
9
|
|
|
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT |
|
10
|
|
|
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, |
|
11
|
|
|
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY |
|
12
|
|
|
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT |
|
13
|
|
|
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE |
|
14
|
|
|
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. |
|
15
|
|
|
* |
|
16
|
|
|
* The software is based on the Axon Framework project which is |
|
17
|
|
|
* licensed under the Apache 2.0 license. For more information on the Axon Framework |
|
18
|
|
|
* see <http://www.axonframework.org/>. |
|
19
|
|
|
* |
|
20
|
|
|
* This software consists of voluntary contributions made by many individuals |
|
21
|
|
|
* and is licensed under the MIT license. For more information, see |
|
22
|
|
|
* <http://www.governor-framework.org/>. |
|
23
|
|
|
*/ |
|
24
|
|
|
|
|
25
|
|
|
namespace Governor\Framework\EventHandling\Replay; |
|
26
|
|
|
|
|
27
|
|
|
use Governor\Framework\Common\Logging\NullLogger; |
|
28
|
|
|
use Governor\Framework\EventHandling\EventBusInterface; |
|
29
|
|
|
use Governor\Framework\EventHandling\EventListenerRegistryInterface; |
|
30
|
|
|
use Psr\Log\LoggerInterface; |
|
31
|
|
|
use Psr\Log\LoggerAwareInterface; |
|
32
|
|
|
use Governor\Framework\EventHandling\EventProcessingMonitorInterface; |
|
33
|
|
|
use Governor\Framework\EventStore\Management\EventStoreManagementInterface; |
|
34
|
|
|
use Governor\Framework\EventStore\Management\CriteriaBuilderInterface; |
|
35
|
|
|
use Governor\Framework\EventStore\Management\CriteriaInterface; |
|
36
|
|
|
|
|
37
|
|
|
/** |
|
38
|
|
|
* Description of ReplayingEventBus |
|
39
|
|
|
* |
|
40
|
|
|
* @author "David Kalosi" <[email protected]> |
|
41
|
|
|
* @license <a href="http://www.opensource.org/licenses/mit-license.php">MIT License</a> |
|
42
|
|
|
*/ |
|
43
|
|
|
class ReplayingEventBus implements EventBusInterface, LoggerAwareInterface |
|
44
|
|
|
{ |
|
45
|
|
|
|
|
46
|
|
|
const STATUS_LIVE = 0; |
|
47
|
|
|
const STATUS_REPLAYING = 1; |
|
48
|
|
|
const STATUS_PROCESSING_BACKLOG = 2; |
|
49
|
|
|
|
|
50
|
|
|
/** |
|
51
|
|
|
* @var LoggerInterface |
|
52
|
|
|
*/ |
|
53
|
|
|
private $logger; |
|
54
|
|
|
|
|
55
|
|
|
/** |
|
56
|
|
|
* @var EventBusInterface |
|
57
|
|
|
*/ |
|
58
|
|
|
private $delegate; |
|
59
|
|
|
|
|
60
|
|
|
/** |
|
61
|
|
|
* @var EventStoreManagementInterface |
|
62
|
|
|
*/ |
|
63
|
|
|
private $replayingEventStore; |
|
64
|
|
|
//private final int commitThreshold; |
|
|
|
|
|
|
65
|
|
|
private $incomingMessageHandler; |
|
66
|
|
|
|
|
67
|
|
|
private $status = self::STATUS_LIVE; |
|
68
|
|
|
|
|
69
|
|
|
private $eventHandlingListeners; |
|
70
|
|
|
|
|
71
|
6 |
|
public function __construct( |
|
72
|
|
|
EventBusInterface $delegate, |
|
73
|
|
|
EventStoreManagementInterface $eventStore, |
|
74
|
|
|
IncomingMessageHandlerInterface $incomingMessageHandler |
|
75
|
|
|
) { |
|
76
|
6 |
|
$this->delegate = $delegate; |
|
77
|
6 |
|
$this->replayingEventStore = $eventStore; |
|
78
|
6 |
|
$this->incomingMessageHandler = $incomingMessageHandler; |
|
79
|
|
|
|
|
80
|
6 |
|
$this->eventHandlingListeners = new EventProcessingListeners(); |
|
81
|
|
|
|
|
82
|
6 |
|
$this->logger = new NullLogger(); |
|
83
|
|
|
//this.delegate.subscribeEventProcessingMonitor(eventHandlingListeners); |
|
84
|
6 |
|
} |
|
85
|
|
|
|
|
86
|
|
|
/** |
|
87
|
|
|
* Returns a CriteriaBuilder that allows the construction of criteria for this EventStore implementation |
|
88
|
|
|
* |
|
89
|
|
|
* @return CriteriaBuilderInterface a builder to create Criteria for this Event Store. |
|
90
|
|
|
*/ |
|
91
|
1 |
|
public function newCriteriaBuilder() |
|
92
|
|
|
{ |
|
93
|
1 |
|
return $this->replayingEventStore->newCriteriaBuilder(); |
|
94
|
|
|
} |
|
95
|
|
|
|
|
96
|
|
|
/** |
|
97
|
|
|
* @return ReplayAwareInterface[] |
|
98
|
|
|
*/ |
|
99
|
5 |
|
private function getReplayAwareListeners() |
|
100
|
|
|
{ |
|
101
|
5 |
|
$list = []; |
|
102
|
|
|
|
|
103
|
5 |
|
foreach ($this->delegate->getEventListenerRegistry()->getListeners() as $listener) { |
|
104
|
3 |
|
if ($listener instanceof ReplayAwareInterface) { |
|
105
|
3 |
|
$list[] = $listener; |
|
106
|
3 |
|
} |
|
107
|
5 |
|
} |
|
108
|
|
|
|
|
109
|
5 |
|
return $list; |
|
110
|
|
|
} |
|
111
|
|
|
|
|
112
|
5 |
|
public function startReplay(CriteriaInterface $criteria = null) |
|
113
|
|
|
{ |
|
114
|
5 |
|
$this->incomingMessageHandler->prepareForReplay($this->delegate); |
|
115
|
5 |
|
$this->status = self::STATUS_REPLAYING; |
|
116
|
|
|
|
|
117
|
5 |
|
$visitor = new ReplayingEventVisitor($this->delegate); |
|
118
|
5 |
|
$visitor->setLogger($this->logger); |
|
119
|
|
|
|
|
120
|
5 |
|
foreach ($this->getReplayAwareListeners() as $replayAwareListener) { |
|
121
|
3 |
|
$replayAwareListener->beforeReplay(); |
|
122
|
5 |
|
} |
|
123
|
|
|
|
|
124
|
5 |
|
$this->replayingEventStore->visitEvents($visitor, $criteria); |
|
125
|
|
|
|
|
126
|
5 |
|
foreach ($this->getReplayAwareListeners() as $replayAwareListener) { |
|
127
|
3 |
|
$replayAwareListener->afterReplay(); |
|
128
|
5 |
|
} |
|
129
|
|
|
|
|
130
|
5 |
|
$this->status = self::STATUS_PROCESSING_BACKLOG; |
|
131
|
5 |
|
$this->incomingMessageHandler->processBacklog($this->delegate); |
|
132
|
|
|
|
|
133
|
5 |
|
$this->status = self::STATUS_LIVE; |
|
134
|
5 |
|
} |
|
135
|
|
|
|
|
136
|
|
|
/** |
|
137
|
|
|
* Indicates whether this cluster is in replay mode. While in replay mode, EventMessages published to this cluster |
|
138
|
|
|
* are forwarded to the IncomingMessageHandler. |
|
139
|
|
|
* |
|
140
|
|
|
* @return boolean <code>true</code> if this cluster is in replay mode, <code>false</code> otherwise. |
|
141
|
|
|
*/ |
|
142
|
1 |
|
public function isInReplayMode() |
|
143
|
|
|
{ |
|
144
|
1 |
|
return $this->status !== self::STATUS_LIVE; |
|
145
|
|
|
} |
|
146
|
|
|
|
|
147
|
1 |
|
public function publish(array $events) |
|
148
|
|
|
{ |
|
149
|
1 |
|
if ($this->status === self::STATUS_LIVE) { |
|
150
|
|
|
$this->delegate->publish($events); |
|
151
|
|
|
} else { |
|
152
|
1 |
|
$this->logger->debug("Cluster is in replaying: sending message to process backlog"); |
|
153
|
1 |
|
$acknowledgedMessages = $this->incomingMessageHandler->onIncomingMessages( |
|
154
|
1 |
|
$this->delegate, |
|
155
|
|
|
$events |
|
156
|
1 |
|
); |
|
157
|
1 |
|
if (null !== $acknowledgedMessages && !empty($acknowledgedMessages)) { |
|
158
|
|
|
$this->eventHandlingListeners->onEventProcessingCompleted($acknowledgedMessages); |
|
159
|
|
|
} |
|
160
|
|
|
} |
|
161
|
1 |
|
} |
|
162
|
|
|
|
|
163
|
|
|
/** |
|
164
|
|
|
* Returns the EventListenerRegistryInterface of this EventBus. |
|
165
|
|
|
* |
|
166
|
|
|
* @return EventListenerRegistryInterface |
|
167
|
|
|
*/ |
|
168
|
3 |
|
public function getEventListenerRegistry() |
|
169
|
|
|
{ |
|
170
|
3 |
|
return $this->delegate->getEventListenerRegistry(); |
|
171
|
|
|
} |
|
172
|
|
|
|
|
173
|
|
|
|
|
174
|
|
|
/** |
|
175
|
|
|
* @param LoggerInterface $logger |
|
176
|
|
|
* @return null |
|
177
|
|
|
*/ |
|
178
|
|
|
public function setLogger(LoggerInterface $logger) |
|
179
|
|
|
{ |
|
180
|
|
|
$this->logger = $logger; |
|
181
|
|
|
} |
|
182
|
|
|
|
|
183
|
|
|
public function subscribeEventProcessingMonitor(EventProcessingMonitorInterface $monitor) |
|
|
|
|
|
|
184
|
|
|
{ |
|
185
|
|
|
//$this->eventHandlingListeners- |
|
186
|
|
|
} |
|
187
|
|
|
|
|
188
|
|
|
public function unsubscribeEventProcessingMonitor(EventProcessingMonitorInterface $monitor) |
|
|
|
|
|
|
189
|
|
|
{ |
|
190
|
|
|
|
|
191
|
|
|
} |
|
192
|
|
|
|
|
193
|
|
|
} |
|
194
|
|
|
|
SmartGecko /
GovernorFramework
Loading history...
Sometimes obsolete code just ends up commented out instead of removed. In this case it is better to remove the code once you have checked you do not need it.
The code might also have been commented out for debugging purposes. In this case it is vital that someone uncomments it again or your project may behave in very unexpected ways in production.
This check looks for comments that seem to be mostly valid code and reports them.