Aggregate::onAggregateClosed() - Code Metrics - Inspection of "update to php 7.4 version" - php-service-bus/event-sourcing - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — v4.0 ( 989be1...ff495d )

by Masiukevich

created 2019-12-01 21:04 UTC

Aggregate::onAggregateClosed() A

↳ Parent: Aggregate

Complexity

Conditions	1
Paths	1

Size

Total Lines	3
Code Lines	1

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
eloc	1
c	1
b	0
f	0
nc	1
nop	1
dl	0
loc	3
ccs	2
cts	2
cp	1
crap	1
rs	10

<?php

/**
 * Event Sourcing implementation.
 *
 * @author  Maksim Masiukevich <[email protected]>
 * @license MIT
 * @license https://opensource.org/licenses/MIT
 */

declare(strict_types = 1);

namespace ServiceBus\EventSourcing;

use function ServiceBus\Common\datetimeInstantiator;
use function ServiceBus\Common\uuid;
use ServiceBus\EventSourcing\Contract\AggregateClosed;
use ServiceBus\EventSourcing\Contract\AggregateCreated;
use ServiceBus\EventSourcing\EventStream\AggregateEvent;
use ServiceBus\EventSourcing\EventStream\AggregateEventStream;
use ServiceBus\EventSourcing\Exceptions\AttemptToChangeClosedStream;

/**
 * Aggregate base class.
 */
abstract class Aggregate
{
    public const   START_PLAYHEAD_INDEX = 0;

    private const  EVENT_APPLY_PREFIX = 'on';

    private const INTERNAL_EVENTS = [
        AggregateCreated::class,
        AggregateClosed::class,
    ];

    private const INCREASE_VERSION_STEP = 1;

    /**
     * Aggregate identifier.
     */
    private AggregateId $id;

    /**
     * Current version.
     */
    private int $version = self::START_PLAYHEAD_INDEX;

    /**
     * List of applied aggregate events.
     *
     * @psalm-var array<int, \ServiceBus\EventSourcing\EventStream\AggregateEvent>
     *
     * @var \ServiceBus\EventSourcing\EventStream\AggregateEvent[]
     */
    private array $events;

    /**
     * Created at datetime.
     */
    private \DateTimeImmutable $createdAt;

    /**
     * Closed at datetime.
     */
    private ?\DateTimeImmutable $closedAt = null;


    final public function __construct(AggregateId $id)
    {
        $this->id = $id;

        $this->clearEvents();

        $this->raise(
            new AggregateCreated($id, \get_class($this))
        );
    }

    /**
     * Receive id.
     */
    final public function id(): AggregateId
    {
        return $this->id;
    }

    /**
     * Receive created at datetime.
     */
    final public function getCreatedAt(): \DateTimeImmutable
    {
        return $this->createdAt;
    }

    /**
     * Raise (apply event).
     *
     * @throws \ServiceBus\EventSourcing\Exceptions\AttemptToChangeClosedStream
     */
    final protected function raise(object $event): void
    {
        if(null !== $this->closedAt)
        {
            throw new AttemptToChangeClosedStream($this->id);
        }

        $specifiedEvent = $event;

        $this->attachEvent($specifiedEvent);
        $this->applyEvent($specifiedEvent);
    }

    /**
     * Receive aggregate version.
     */
    final public function version(): int
    {
        return $this->version;
    }

    /**
     * Close aggregate (make it read-only).
     *
     * @throws \ServiceBus\EventSourcing\Exceptions\AttemptToChangeClosedStream
     */
    final protected function close(): void
    {
        /** @psalm-var class-string<\ServiceBus\EventSourcing\Aggregate> $aggregateClass */
        $aggregateClass = \get_class($this);

        $this->raise(
            new AggregateClosed($this->id, $aggregateClass)
        );
    }

    /**
     * On aggregate closed.
     *
     * @noinspection PhpUnusedPrivateMethodInspection
     */
    private function onAggregateClosed(AggregateClosed $event): void
    {
        $this->closedAt = $event->datetime;
    }

    /**
     * On aggregate created.
     *
     * @noinspection PhpUnusedPrivateMethodInspection
     */
    private function onAggregateCreated(AggregateCreated $event): void
    {
        $this->createdAt = $event->datetime;
    }

    /**
     * Receive uncommitted events as stream.
     *
     * @noinspection PhpUnusedPrivateMethodInspection
     *
     * @see          EventSourcingProvider::save()
     */
    private function makeStream(): AggregateEventStream
    {
        $events = $this->events;

        /** @psalm-var class-string<\ServiceBus\EventSourcing\Aggregate> $aggregateClass */
        $aggregateClass = \get_class($this);

        $this->clearEvents();

        return new AggregateEventStream(
            $this->id,
            $aggregateClass,
            $events,
            $this->createdAt,
            $this->closedAt
        );
    }

    /**
     * Restore from event stream.
     *
     * @noinspection PhpUnusedPrivateMethodInspection
     *
     * @see          EventSourcingProvider::load()
     */
    private function appendStream(AggregateEventStream $aggregateEventsStream): void
    {
        $this->clearEvents();

        $this->id = $aggregateEventsStream->id;

        /** @var AggregateEvent $aggregateEvent */
        foreach($aggregateEventsStream->events as $aggregateEvent)
        {
            $this->applyEvent($aggregateEvent->event);

            $this->increaseVersion(self::INCREASE_VERSION_STEP);
        }
    }

    /**
     * Attach event to stream
     */
    private function attachEvent(object $event): void
    {
        $this->increaseVersion(self::INCREASE_VERSION_STEP);

        /** @var \DateTimeImmutable $currentDate */
        $currentDate = datetimeInstantiator('NOW');

        $this->events[] = AggregateEvent::create(uuid(), $event, $this->version, $currentDate);
    }

    /**
     * Apply event.
     */
    private function applyEvent(object $event): void
    {
        $eventListenerMethodName = self::createListenerName($event);

        true === self::isInternalEvent($event)
            ? $this->processInternalEvent($eventListenerMethodName, $event)
            : $this->processChildEvent($eventListenerMethodName, $event);
    }

    /**
     * Is internal event (for current class).
     */
    private static function isInternalEvent(object $event): bool
    {
        return true === \in_array(\get_class($event), self::INTERNAL_EVENTS, true);
    }

    private function processInternalEvent(string $listenerName, object $event): void
    {
        $this->{$listenerName}($event);
    }

    private function processChildEvent(string $listenerName, object $event): void
    {
        /**
         * Call child class method.
         *
         * @param object $event
         *
         * @return void
         */
        $closure = function(object $event) use ($listenerName): void
        {
            if(true === \method_exists($this, $listenerName))
            {
                $this->{$listenerName}($event);
            }
        };

        $closure->call($this, $event);
    }

    /**
     * Create event listener name.
     */
    private static function createListenerName(object $event): string
    {
        $eventListenerMethodNameParts = \explode('\\', \get_class($event));

        /** @var string $latestPart */
        $latestPart = \end($eventListenerMethodNameParts);

        return \sprintf(
            '%s%s',
            self::EVENT_APPLY_PREFIX,
            $latestPart
        );
    }

    /**
     * Increase aggregate version.
     */
    private function increaseVersion(int $step): void
    {
        $this->version += $step;
    }

    /**
     * Clear all aggregate events.
     */
    private function clearEvents(): void
    {
        $this->events = [];
    }
}


1		<?php
2
3		/**
4		* Event Sourcing implementation.
5		*
6		* @author Maksim Masiukevich <[email protected]>
7		* @license MIT
8		* @license https://opensource.org/licenses/MIT
9		*/
10
11		declare(strict_types = 1);
12
13		namespace ServiceBus\EventSourcing;
14
15		use function ServiceBus\Common\datetimeInstantiator;
16		use function ServiceBus\Common\uuid;
17		use ServiceBus\EventSourcing\Contract\AggregateClosed;
18		use ServiceBus\EventSourcing\Contract\AggregateCreated;
19		use ServiceBus\EventSourcing\EventStream\AggregateEvent;
20		use ServiceBus\EventSourcing\EventStream\AggregateEventStream;
21		use ServiceBus\EventSourcing\Exceptions\AttemptToChangeClosedStream;
22
23		/**
24		* Aggregate base class.
25		*/
26		abstract class Aggregate
27		{
28		public const START_PLAYHEAD_INDEX = 0;
29
30		private const EVENT_APPLY_PREFIX = 'on';
31
32		private const INTERNAL_EVENTS = [
33		AggregateCreated::class,
34		AggregateClosed::class,
35		];
36
37		private const INCREASE_VERSION_STEP = 1;
38
39		/**
40		* Aggregate identifier.
41		*/
42		private AggregateId $id;
43
44		/**
45		* Current version.
46		*/
47		private int $version = self::START_PLAYHEAD_INDEX;
48
49		/**
50		* List of applied aggregate events.
51		*
52		* @psalm-var array<int, \ServiceBus\EventSourcing\EventStream\AggregateEvent>
53		*
54		* @var \ServiceBus\EventSourcing\EventStream\AggregateEvent[]
55		*/
56		private array $events;
57
58		/**
59		* Created at datetime.
60		*/
61		private \DateTimeImmutable $createdAt;
62
63		/**
64		* Closed at datetime.
65		*/
66		private ?\DateTimeImmutable $closedAt = null;
67
68
69	10	final public function __construct(AggregateId $id)
70		{
71	10	$this->id = $id;
72
73	10	$this->clearEvents();
74
75	10	$this->raise(
76	10	new AggregateCreated($id, \get_class($this))
77		);
78	10	}
79
80		/**
81		* Receive id.
82		*/
83	8	final public function id(): AggregateId
84		{
85	8	return $this->id;
86		}
87
88		/**
89		* Receive created at datetime.
90		*/
91	7	final public function getCreatedAt(): \DateTimeImmutable
92		{
93	7	return $this->createdAt;
94		}
95
96		/**
97		* Raise (apply event).
98		*
99		* @throws \ServiceBus\EventSourcing\Exceptions\AttemptToChangeClosedStream
100		*/
101	10	final protected function raise(object $event): void
102		{
103	10	if(null !== $this->closedAt)
104		{
105	1	throw new AttemptToChangeClosedStream($this->id);
106		}
107
108	10	$specifiedEvent = $event;
109
110	10	$this->attachEvent($specifiedEvent);
111	10	$this->applyEvent($specifiedEvent);
112	10	}
113
114		/**
115		* Receive aggregate version.
116		*/
117	7	final public function version(): int
118		{
119	7	return $this->version;
120		}
121
122		/**
123		* Close aggregate (make it read-only).
124		*
125		* @throws \ServiceBus\EventSourcing\Exceptions\AttemptToChangeClosedStream
126		*/
127	1	final protected function close(): void
128		{
129		/** @psalm-var class-string<\ServiceBus\EventSourcing\Aggregate> $aggregateClass */
130	1	$aggregateClass = \get_class($this);
131
132	1	$this->raise(
133	1	new AggregateClosed($this->id, $aggregateClass)
134		);
135	1	}
136
137		/**
138		* On aggregate closed.
139		*
140		* @noinspection PhpUnusedPrivateMethodInspection
141		*/
142	1	private function onAggregateClosed(AggregateClosed $event): void
143		{
144	1	$this->closedAt = $event->datetime;
145	1	}
146
147		/**
148		* On aggregate created.
149		*
150		* @noinspection PhpUnusedPrivateMethodInspection
151		*/
152	10	private function onAggregateCreated(AggregateCreated $event): void
153		{
154	10	$this->createdAt = $event->datetime;
155	10	}
156
157		/**
158		* Receive uncommitted events as stream.
159		*
160		* @noinspection PhpUnusedPrivateMethodInspection
161		*
162		* @see EventSourcingProvider::save()
163		*/
164	9	private function makeStream(): AggregateEventStream
165		{
166	9	$events = $this->events;
167
168		/** @psalm-var class-string<\ServiceBus\EventSourcing\Aggregate> $aggregateClass */
169	9	$aggregateClass = \get_class($this);
170
171	9	$this->clearEvents();
172
173	9	return new AggregateEventStream(
174	9	$this->id,
175		$aggregateClass,
176		$events,
177	9	$this->createdAt,
178	9	$this->closedAt
179		);
180		}
181
182		/**
183		* Restore from event stream.
184		*
185		* @noinspection PhpUnusedPrivateMethodInspection
186		*
187		* @see EventSourcingProvider::load()
188		*/
189	1	private function appendStream(AggregateEventStream $aggregateEventsStream): void
190		{
191	1	$this->clearEvents();
192
193	1	$this->id = $aggregateEventsStream->id;
194
195		/** @var AggregateEvent $aggregateEvent */
196	1	foreach($aggregateEventsStream->events as $aggregateEvent)
197		{
198		$this->applyEvent($aggregateEvent->event);
199
200		$this->increaseVersion(self::INCREASE_VERSION_STEP);
201		}
202	1	}
203
204		/**
205		* Attach event to stream
206		*/
207	10	private function attachEvent(object $event): void
208		{
209	10	$this->increaseVersion(self::INCREASE_VERSION_STEP);
210
211		/** @var \DateTimeImmutable $currentDate */
212	10	$currentDate = datetimeInstantiator('NOW');
213
214	10	$this->events[] = AggregateEvent::create(uuid(), $event, $this->version, $currentDate);
215	10	}
216
217		/**
218		* Apply event.
219		*/
220	10	private function applyEvent(object $event): void
221		{
222	10	$eventListenerMethodName = self::createListenerName($event);
223
224	10	true === self::isInternalEvent($event)
225	10	? $this->processInternalEvent($eventListenerMethodName, $event)
226	4	: $this->processChildEvent($eventListenerMethodName, $event);
227	10	}
228
229		/**
230		* Is internal event (for current class).
231		*/
232	10	private static function isInternalEvent(object $event): bool
233		{
234	10	return true === \in_array(\get_class($event), self::INTERNAL_EVENTS, true);
235		}
236
237	10	private function processInternalEvent(string $listenerName, object $event): void
238		{
239	10	$this->{$listenerName}($event);
240	10	}
241
242	4	private function processChildEvent(string $listenerName, object $event): void
243		{
244		/**
245		* Call child class method.
246		*
247		* @param object $event
248		*
249		* @return void
250		*/
251		$closure = function(object $event) use ($listenerName): void
252		{
253	4	if(true === \method_exists($this, $listenerName))
254		{
255	4	$this->{$listenerName}($event);
256		}
257	4	};
258
259	4	$closure->call($this, $event);
260	4	}
261
262		/**
263		* Create event listener name.
264		*/
265	10	private static function createListenerName(object $event): string
266		{
267	10	$eventListenerMethodNameParts = \explode('\\', \get_class($event));
268
269		/** @var string $latestPart */
270	10	$latestPart = \end($eventListenerMethodNameParts);
271
272	10	return \sprintf(
273	10	'%s%s',
274	10	self::EVENT_APPLY_PREFIX,
275	10	$latestPart
276		);
277		}
278
279		/**
280		* Increase aggregate version.
281		*/
282	10	private function increaseVersion(int $step): void
283		{
284	10	$this->version += $step;
285	10	}
286
287		/**
288		* Clear all aggregate events.
289		*/
290	10	private function clearEvents(): void
291		{
292	10	$this->events = [];
293	10	}
294		}
295

php-service-bus / event-sourcing

Push — v4.0 ( 989be1...ff495d )

Aggregate::onAggregateClosed() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like