Gica\Cqrs\EventStore\InMemory\InMemoryEventStore

Complexity

Total Complexity

21

Size/Duplication

Total Lines	140
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	7

Test Coverage

Coverage

100%

Importance

Changes

0

12 Methods

Rating	Name	Duplication	Size	Complexity
A	loadEventsForAggregate()	0	5	1
A	appendEventsForAggregate()	0	8	2
A	getEventsArrayForAggregate()	0	8	2
A	loadEventsByClassNames()	0	8	1
A	extractEventsFromCommits()	0	8	1
A	getAggregateVersion()	0	6	2
A	appendEventsForAggregateWithoutChecking()	0	19	2
A	addEventsToArrayForAggregate()	0	6	1
A	decorateEventsWithMetadata()	0	15	2
A	fetchLatestSequence()	0	4	1
A	constructKey()	0	4	1
B	findEventById()	0	21	5

<?php


namespace Gica\Cqrs\EventStore\InMemory;


use Gica\Cqrs\Event\EventWithMetaData;
use Gica\Cqrs\Event\MetaData;
use Gica\Cqrs\EventStore;
use Gica\Cqrs\EventStore\AggregateEventStream;
use Gica\Cqrs\EventStore\EventsCommit;
use Gica\Cqrs\EventStore\EventStreamGroupedByCommit;
use Gica\Cqrs\EventStore\Exception\ConcurrentModificationException;
use Gica\Iterator\IteratorTransformer\IteratorExpander;

class InMemoryEventStore implements EventStore
{
    /** @var EventsCommit[] */
    public $commitsByAggregate = [];
    private $versions = [];
    private $latestSequence = 0;

    public function loadEventsForAggregate(string $aggregateClass, $aggregateId): AggregateEventStream
    {
        return new InMemoryAggregateEventStream(
            $this->getEventsArrayForAggregate($aggregateClass, $aggregateId), $aggregateClass, $aggregateId, $this->latestSequence);
    }

    /**
     * @inheritdoc
     */
    public function appendEventsForAggregate($aggregateId, string $aggregateClass, $eventsWithMetaData, int $expectedVersion, int $expectedSequence)
    {
        if ($this->getAggregateVersion($aggregateClass, $aggregateId) != $expectedVersion) {
            throw new ConcurrentModificationException();
        }

        $this->appendEventsForAggregateWithoutChecking($aggregateId, $aggregateClass, $eventsWithMetaData, $expectedVersion, $expectedSequence);
    }

    public function appendEventsForAggregateWithoutChecking($aggregateId, $aggregateClass, $newEvents, int $expectedVersion, int $expectedSequence)
    {
        $this->addEventsToArrayForAggregate(
            $aggregateId,
            $aggregateClass,
            $this->decorateEventsWithMetadata($aggregateClass, $aggregateId, $newEvents),
            $expectedVersion,
            $expectedSequence
        );

        $constructKey = $this->constructKey($aggregateClass, $aggregateId);

        if (!isset($this->versions[$constructKey])) {
            $this->versions[$constructKey] = 0;
        }

        $this->versions[$constructKey]++;
        $this->latestSequence++;
    }

    private function getEventsArrayForAggregate(string $aggregateClass, $aggregateId)
    {
        $aggregateKey = $this->constructKey($aggregateClass, $aggregateId);

        return isset($this->commitsByAggregate[$aggregateKey])
            ? $this->extractEventsFromCommits($this->commitsByAggregate[$aggregateKey])

$this->commitsByAggregate[$aggregateKey] is of type object<Gica\Cqrs\EventStore\EventsCommit>, but the function expects a array.

            : [];
    }

    private function addEventsToArrayForAggregate($aggregateId, $aggregateClass, $newEvents, int $expectedVersion, int $expectedSequence)
    {
        $this->commitsByAggregate[$this->constructKey($aggregateClass, $aggregateId)][] = new EventsCommit(
            $expectedSequence, $expectedVersion, $newEvents
        );
    }

    public function loadEventsByClassNames(array $eventClasses): EventStreamGroupedByCommit
    {
        $commits = iterator_to_array((new IteratorExpander(function ($aggregateCommits) {
            yield from $aggregateCommits;
        }))($this->commitsByAggregate));

        return new FilteredRawEventStreamGroupedByCommit($commits, $eventClasses);
    }

    private function extractEventsFromCommits(array $commits = [])
    {
        $eventsExtracter = new IteratorExpander(function (EventsCommit $commit) {
            yield from $commit->getEventsWithMetadata();
        });

        return iterator_to_array($eventsExtracter($commits));
    }

    public function getAggregateVersion(string $aggregateClass, $aggregateId)
    {
        $key = $this->constructKey($aggregateClass, $aggregateId);

        return isset($this->versions[$key]) ? $this->versions[$key] : 0;
    }

    /**
     * @param $aggregateClass
     * @param $aggregateId
     * @param $priorEvents
     * @return EventWithMetaData[]
     */
    public function decorateEventsWithMetadata($aggregateClass, $aggregateId, array $priorEvents)
    {
        return array_map(function ($event) use ($aggregateClass, $aggregateId) {
            if ($event instanceof EventWithMetaData) {
                return $event;
            }

            return new EventWithMetaData($event, new MetaData(
                $aggregateId,
                    $aggregateClass,
                    new \DateTimeImmutable(),
                    null
            ));
        }, $priorEvents);
    }

    public function fetchLatestSequence(): int
    {
        return $this->latestSequence;
    }

    private function constructKey(string $aggregateClass, $aggregateId): string
    {
        return $aggregateClass . '_' . (string)$aggregateId;
    }

    public function findEventById(string $eventId): ?EventWithMetaData
    {
        foreach($this->commitsByAggregate as $commits)
        {
            foreach($commits as $commit)

The expression $commits of type object<Gica\Cqrs\EventStore\EventsCommit> is not traversable.

            {
                /** @var \Gica\Cqrs\EventStore\EventsCommit $commit */

                foreach($commit->getEventsWithMetadata() as $eventWithMetadata)
                {
                    if($eventWithMetadata->getMetaData()->getEventId() === $eventId)
                    {
                        return $eventWithMetadata;
                    }
                }
            }

        }

        return null;
    }
}