Borobudur\EventSourcing\Repository\Repository

Complexity

Total Complexity

15

Size/Duplication

Total Lines	147
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	9

Importance

Changes	7
Bugs	0	Features	5

6 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	8	1
A	persist()	0	14	3
A	find()	0	18	4
B	publish()	0	24	3
A	broadcast()	0	6	2
A	assertInstanceOf()	0	10	2

<?php
/*
 * This file is part of the Borobudur-Event-Sourcing package.
 *
 * (c) Hexacodelabs <http://hexacodelabs.com>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Borobudur\EventSourcing\Repository;

use Borobudur\Broadcasting\Broadcaster;
use Borobudur\Broadcasting\BroadcastEventInterface;
use Borobudur\Bus\BusProxy;
use Borobudur\Cqrs\IdentifierInterface;
use Borobudur\Cqrs\Message\DomainEventInterface;
use Borobudur\EventSourcing\BroadcastEventMessage;
use Borobudur\EventSourcing\Domain\DomainEventStreamInterface;
use Borobudur\EventSourcing\Entity\AggregateRootInterface;
use Borobudur\EventSourcing\Exception\LogicException;
use Borobudur\EventSourcing\Snapshot\SnapshotInterface;
use Borobudur\EventSourcing\Storage\StorageInterface;

/**
 * @author      Iqbal Maulana <iq.bluejack@gmail.com>
 * @created     8/20/15
 */
class Repository implements RepositoryInterface
{
    /**
     * @var StorageInterface
     */
    private $storage;

    /**
     * @var SnapshotInterface
     */
    private $snapshot;

    /**
     * @var BusProxy
     */
    private $eventBus;

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

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

    /**
     * Constructor.
     *
     * @param StorageInterface $storage
     * @param BusProxy         $eventBus
     * @param string           $aggregateRootClass
     * @param string           $table
     */
    public function __construct(StorageInterface $storage, BusProxy $eventBus, $aggregateRootClass, $table)
    {
        $this->storage = $storage;
        $this->snapshot = $storage->getSnapshot();
        $this->eventBus = $eventBus;
        $this->aggregateRootClass = $aggregateRootClass;
        $this->table = $table;
    }

    /**
     * {@inheritdoc}
     */
    public function persist(AggregateRootInterface $aggregateRoot)
    {
        $this->assertInstanceOf($aggregateRoot, $this->aggregateRootClass);
        if (null !== $this->snapshot) {
            $this->snapshot->append($aggregateRoot, $this->table);
        }

        $stream = $aggregateRoot->getUncommittedEvents();
        if ($stream->count()) {
            return $this->publish($stream);
        }

        return null;
    }

    /**
     * {@inheritdoc}
     */
    public function find(IdentifierInterface $id)
    {
        $class = $this->aggregateRootClass;
        $aggregate = null;
        if (null !== $this->snapshot) {
            $aggregate = $this->snapshot->find($id, $class, $this->table);
        }

        if (null === $aggregate) {
            return $class::{'build'}($this->storage->findEventFromBeginning($this->table, $id));
        }

        if (null !== $events = $this->storage->findEventFromVersion($this->table, $id, $aggregate->getVersion())) {
            $aggregate->replay($events);
        }

        return $aggregate;
    }

    /**
     * Publish domain events in stream.
     *
     * @param DomainEventStreamInterface $stream
     *
     * @return array
     * @throws \Exception
     */
    private function publish(DomainEventStreamInterface $stream)
    {
        $results = array();
        foreach ($stream->all() as $domain) {
            try {
                $results[] = $this->eventBus->dispatch($domain->getPayload());
                $this->storage->append($this->table, $domain, 'OK');
                $this->broadcast($domain->getPayload(), array('status' => 'OK', 'exception' => null));
            } catch (\Exception $e) {
                $metadata = array(
                    'code'    => $e->getCode(),
                    'file'    => $e->getFile(),
                    'line'    => $e->getLine(),
                    'message' => $e->getMessage(),
                );
                $this->storage->append($this->table, $domain, 'ERROR', $metadata);
                $this->broadcast($domain->getPayload(), array('status' => 'ERROR', 'exception' => $metadata));

                throw $e;
            }
        }

        return $results;
    }

    /**
     * Broadcast event.
     *
     * @param DomainEventInterface $event
     * @param array                $metadata
     */
    private function broadcast(DomainEventInterface $event, array $metadata)
    {
        if ($event instanceof BroadcastEventInterface) {
            Broadcaster::getInstance()->broadcast(new BroadcastEventMessage($event, $metadata));
        }
    }

    /**
     * Assert that aggregate root should be instance of $className
     *
     * @param AggregateRootInterface $aggregateRoot
     * @param string                 $className
     */
    private function assertInstanceOf(AggregateRootInterface $aggregateRoot, $className)
    {
        if (!$aggregateRoot instanceof $className) {
            throw new LogicException(sprintf(
                'Class "%s" was expected to be instanceof of "%s" but is not.',
                get_class($aggregateRoot),
                $className
            ));
        }
    }
}