Blixit\EventSourcing\Store\EventStore

Complexity

Total Complexity

23

Size/Duplication

Total Lines	228
Duplicated Lines	0 %

Importance

Changes

0

13 Methods

Rating	Name	Duplication	Size	Complexity
A	beforeRead()	0	17	3
A	buildAggregate()	0	3	1
A	beforeWrite()	0	14	4
A	persist()	0	6	2
A	store()	0	23	3
A	afterRead()	0	2	1
A	writeLoopIteration()	0	18	1
A	getEmptyAggregate()	0	6	1
A	__construct()	0	14	1
A	load()	0	3	1
A	get()	0	30	3
A	getStreamNameForAggregateId()	0	4	1
A	afterWrite()	0	2	1

<?php

declare(strict_types=1);

namespace Blixit\EventSourcing\Store;

use Blixit\EventSourcing\Aggregate\AggregateAccessor;
use Blixit\EventSourcing\Aggregate\AggregateRoot;
use Blixit\EventSourcing\Aggregate\AggregateRootInterface;
use Blixit\EventSourcing\Event\EventAccessor;
use Blixit\EventSourcing\Event\EventInterface;
use Blixit\EventSourcing\Store\Exception\CorruptedReadEvent;
use Blixit\EventSourcing\Store\Exception\NonWritableEvent;
use Blixit\EventSourcing\Store\Persistence\EventPersisterException;
use Blixit\EventSourcing\Store\Persistence\EventPersisterInterface;
use Blixit\EventSourcing\Stream\ReadableStream;
use Blixit\EventSourcing\Stream\StorableStream;
use Blixit\EventSourcing\Stream\Strategy\StreamStrategy;
use Blixit\EventSourcing\Stream\Stream;
use Blixit\EventSourcing\Stream\StreamName;
use ReflectionClass;
use ReflectionException;
use Symfony\Component\Messenger\MessageBusInterface;
use Throwable;
use function sprintf;

class EventStore implements EventStoreInterface
{
    /** @var EventPersisterInterface $eventPersister */
    protected $eventPersister;

    /** @var string $aggregateClass */
    private $aggregateClass;

    /** @var StreamStrategy $streamStrategy */
    private $streamStrategy;

    /** @var MessageBusInterface $messageBus */
    private $messageBus;

    /** @var AggregateAccessor $aggregateAccessor */
    private $aggregateAccessor;

    /** @var EventAccessor $eventAccessor */
    private $eventAccessor;

    public function __construct(
        string $aggregateClass,
        EventPersisterInterface $eventPersister,
        string $streamStrategyClass,
        ?MessageBusInterface $messageBus = null
    ) {
        $this->eventPersister = $eventPersister;
        $this->aggregateClass = $aggregateClass;
        $this->streamStrategy = StreamStrategy::resolveStreamStrategy($streamStrategyClass);

        $this->messageBus = $messageBus;

        $this->aggregateAccessor = AggregateAccessor::getInstance();
        $this->eventAccessor     = EventAccessor::getInstance();
    }

    /**
     * @param mixed $aggregateId
     */
    public function getStreamNameForAggregateId($aggregateId = null) : StreamName
    {
        // TODO: add hook after stream computeName
        return $this->streamStrategy->computeName($this->aggregateClass, $aggregateId);
    }

    /**
     * @param mixed $aggregateId
     *
     * @throws ReflectionException
     * @throws CorruptedReadEvent
     */
    public function get($aggregateId) : ?AggregateRootInterface
    {
        /** @var AggregateRootInterface $aggregate */
        $aggregate = $this->buildAggregate($aggregateId);
        // compute streamName based on stream strategy
        $streamName = $this->getStreamNameForAggregateId($aggregateId);
        // get events from store
        $events = $this->eventPersister->getByStream($streamName, $aggregate->getSequence());
        // get beforeRead callback
        $beforeRead = [$this, 'beforeRead'];
        // build stream
        $stream = new ReadableStream(
            $streamName,
            $events,
            static function (EventInterface $event) use ($beforeRead, $aggregate) : void {
                $beforeRead($aggregate, $event);
            }
        );

        // replay events
        foreach ($stream->getIterator() as $event) {
            /** @var EventInterface $event */

            /** @var AggregateRoot $aggregate */
            $aggregate->apply($event);
            $this->aggregateAccessor->setVersionSequence($aggregate, $event->getSequence());
            $this->afterRead($aggregate, $event);
        }

        return $aggregate->getSequence() > 0 ? $aggregate : null;
    }

    /**
     * @param mixed $aggregateId
     *
     * @throws ReflectionException
     */
    protected function getEmptyAggregate($aggregateId) : AggregateRootInterface
    {
        /** @var AggregateRootInterface $aggregate */
        $aggregate = (new ReflectionClass($this->aggregateClass))->newInstanceWithoutConstructor();
        $this->aggregateAccessor->setAggregateId($aggregate, $aggregateId);
        return $aggregate;
    }

    /**
     * @param mixed $aggregateId
     *
     * @throws ReflectionException
     */
    protected function buildAggregate($aggregateId) : AggregateRootInterface
    {
        return $this->getEmptyAggregate($aggregateId);
    }

    /**
     * @throws EventPersisterException
     * @throws NonWritableEvent
     */
    public function store(AggregateRootInterface &$aggregateRoot) : void
    {
        // compute streamName based on stream strategy
        $streamName = $this->getStreamNameForAggregateId($aggregateRoot->getAggregateId());
        // get beforeWrite callback
        $beforeWrite = [$this, 'beforeWrite'];

        /** @var AggregateRoot $aggregateRoot */
        $stream = new StorableStream(
            $streamName,
            $aggregateRoot->getRecordedEvents(),
            static function (EventInterface $event) use ($beforeWrite, $aggregateRoot) : void {
                $beforeWrite($aggregateRoot, $event);
            }
        );

        foreach ($stream->getIterator() as $event) {
            $this->writeLoopIteration($aggregateRoot, $event);
            // dispatch event
            if (empty($this->messageBus)) {
                continue;
            }
            $this->messageBus->dispatch($event);
        }
    }

    /**
     * @throws EventPersisterException
     */
    protected function writeLoopIteration(AggregateRootInterface &$aggregateRoot, EventInterface &$event) : void
    {
        // next sequence number
        $nextSequence = $aggregateRoot->getSequence() + 1;
        // set event sequence
        $this->eventAccessor->setSequence($event, $nextSequence);
        // set event aggregate class
        $this->eventAccessor->setAggregateClass($event, $this->aggregateClass);
        // save event
        $committedEvent = $this->persist($event);
        // remove from recorded events
        $this->aggregateAccessor->shiftEvent($aggregateRoot);
        // if persistence works, then increment aggregate sequence
        $this->aggregateAccessor->setVersionSequence($aggregateRoot, $nextSequence);
        // apply event
        $aggregateRoot->apply($committedEvent);
        // do something with the updated aggregate like snapShotting
        $this->afterWrite($aggregateRoot, $committedEvent);
    }

    /**
     * @throws EventPersisterException
     */
    protected function persist(EventInterface $event) : EventInterface
    {
        try {
            return clone $this->eventPersister->persist($event);
        } catch (Throwable $exception) {
            throw new EventPersisterException($exception->getMessage());
        }
    }

    /**
     * @param mixed $aggregateId
     *
     * @throws NonWritableEvent
     */
    protected function beforeWrite(AggregateRootInterface $aggregateRoot, EventInterface $event) : void
    {
        if ($event->getAggregateId() !== $aggregateRoot->getAggregateId()) {
            throw new NonWritableEvent(sprintf(
                'Event aggregate Id not expected. Expected: %s . Found: %s',
                $aggregateRoot->getAggregateId(),
                $event->getAggregateId()
            ));
        }
        if (! empty($event->getSequence())) {
            throw new NonWritableEvent('Sequence number should be empty. Found: ' . $event->getSequence());
        }
        if (! empty($event->getStreamName())) {
            throw new NonWritableEvent('Stream name should be empty. Found: ' . $event->getStreamName());
        }
    }

    protected function afterWrite(AggregateRootInterface $aggregateRoot, EventInterface $event) : void

The parameter $aggregateRoot is not used and could be removed.

The parameter $event is not used and could be removed.

    {
    }

    /**
     * @throws CorruptedReadEvent
     */
    protected function beforeRead(AggregateRootInterface $aggregateRoot, EventInterface $event) : void
    {
        // ignores not relevant events
        if ($event->getAggregateId() !== $aggregateRoot->getAggregateId()) {
            throw new CorruptedReadEvent(sprintf(
                'Event aggregate Id not expected. Expected: %s . Found: %s',
                $aggregateRoot->getAggregateId(),
                $event->getAggregateId()
            ));
        }

        // ignore event with bad type
        if ($this->aggregateClass !== $event->getAggregateClass()) {
            throw new CorruptedReadEvent(sprintf(
                'Event aggregate Class not expected. Expected: %s . Found: %s',
                $this->aggregateClass,
                $event->getAggregateClass()
            ));
        }
    }

    protected function afterRead(AggregateRootInterface $aggregateRoot, EventInterface $event) : void

The parameter $event is not used and could be removed.

The parameter $aggregateRoot is not used and could be removed.

    {
    }

    public function load() : Stream
    {
        return null;

The expression return null returns the type null which is incompatible with the type-hinted return Blixit\EventSourcing\Stream\Stream.

    }
}