ServiceBus\EventSourcing\Aggregate

Complexity

Total Complexity

22

Size/Duplication

Total Lines	271
Duplicated Lines	0 %

Test Coverage

Coverage

100%

Importance

Changes	1
Bugs	0	Features	0

18 Methods

Rating	Name	Duplication	Size	Complexity
A	createListenerName()	0	11	1
A	raise()	0	11	2
A	onAggregateClosed()	0	3	1
A	makeStream()	0	15	1
A	processInternalEvent()	0	3	1
A	version()	0	3	1
A	processChildEvent()	0	18	2
A	id()	0	3	1
A	getCreatedAt()	0	3	1
A	clearEvents()	0	3	1
A	__construct()	0	8	1
A	close()	0	7	1
A	attachEvent()	0	5	1
A	onAggregateCreated()	0	3	1
A	isInternalEvent()	0	3	1
A	applyEvent()	0	7	2
A	increaseVersion()	0	3	1
A	appendStream()	0	12	2

<?php

/**
 * Event Sourcing implementation.
 *
 * @author  Maksim Masiukevich <dev@async-php.com>
 * @license MIT
 * @license https://opensource.org/licenses/MIT
 */

declare(strict_types = 1);

namespace ServiceBus\EventSourcing;

use function ServiceBus\Common\now;
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.
     *
     * @var AggregateId
     */
    private $id;

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

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

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

    /**
     * Closed at datetime.
     *
     * @var \DateTimeImmutable|null
     */
    private $closedAt = null;

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

        $this->clearEvents();

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

    /**
     * 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, now())
        );
    }

    /**
     * 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);

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

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

        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 \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 (\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 = [];
    }
}