CultuurNet\UDB3\EventSourcing\DBAL\AggregateAwareDBALEventStore

Complexity

Total Complexity

16

Size/Duplication

Total Lines	200
Duplicated Lines	5 %

Coupling/Cohesion

Components	1
Dependencies	10

Importance

Changes	1
Bugs	0	Features	0

9 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	13	1
A	load()	0	17	3
A	append()	0	17	3
A	insertMessage()	0	14	1
A	configureSchema()	0	8	2
A	configureTable()	0	20	1
A	prepareLoadStatement()	0	19	2
A	deserializeEvent()	10	10	1
A	guardStream()	0	7	2

<?php

namespace CultuurNet\UDB3\EventSourcing\DBAL;

use Broadway\Domain\DateTime as BroadwayDateTime;
use Broadway\Domain\DomainEventStream;
use Broadway\Domain\DomainEventStreamInterface;
use Broadway\Domain\DomainMessage;
use Broadway\EventStore\DBALEventStoreException;
use Broadway\EventStore\EventStoreInterface;
use Broadway\EventStore\EventStreamNotFoundException;
use Broadway\Serializer\SerializerInterface;
use Doctrine\DBAL\Connection;
use Doctrine\DBAL\DBALException;
use Doctrine\DBAL\Schema\Schema;
use Doctrine\DBAL\Schema\Table;

class AggregateAwareDBALEventStore implements EventStoreInterface
{
    /**
     * @var Connection
     */
    private $connection;

    /**
     * @var SerializerInterface
     */
    private $payloadSerializer;

    /**
     * @var SerializerInterface
     */
    private $metadataSerializer;

    /**
     * @var null
     */
    private $loadStatement = null;

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

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

    /**
     * @param Connection $connection
     * @param SerializerInterface $payloadSerializer
     * @param SerializerInterface $metadataSerializer
     * @param string $tableName
     * @param string $aggregateType
     */
    public function __construct(
        Connection $connection,
        SerializerInterface $payloadSerializer,
        SerializerInterface $metadataSerializer,
        $tableName,
        $aggregateType
    ) {
        $this->connection         = $connection;
        $this->payloadSerializer  = $payloadSerializer;
        $this->metadataSerializer = $metadataSerializer;
        $this->tableName          = $tableName;
        $this->aggregateType      = $aggregateType;
    }

    /**
     * {@inheritDoc}
     */
    public function load($id)
    {
        $statement = $this->prepareLoadStatement();
        $statement->bindValue('uuid', $id);
        $statement->execute();

        $events = array();
        while ($row = $statement->fetch()) {
            $events[] = $this->deserializeEvent($row);
        }

        if (empty($events)) {
            throw new EventStreamNotFoundException(sprintf('EventStream not found for aggregate with id %s', $id));
        }

        return new DomainEventStream($events);
    }

    /**
     * {@inheritDoc}
     */
    public function append($id, DomainEventStreamInterface $eventStream)
    {
        // The original Broadway implementation did only check the type of $id.
        // It is better to test all uuids inside the event stream.
        $this->guardStream($eventStream);

The call to the method CultuurNet\UDB3\EventSourcing\DBAL\AggregateAwareDBALEventStore::guardStream() seems un-needed as the method has no side-effects.

        // Make the transaction more robust by using the transactional statement.
        $this->connection->transactional(function (Connection $connection) use ($eventStream) {
            try {
                foreach ($eventStream as $domainMessage) {
                    $this->insertMessage($connection, $domainMessage);
                }
            } catch (DBALException $exception) {
                throw DBALEventStoreException::create($exception);
            }
        });
    }

    /**
     * @param Connection $connection
     * @param DomainMessage $domainMessage
     */
    private function insertMessage(Connection $connection, DomainMessage $domainMessage)
    {
        $data = array(
            'uuid'           => (string) $domainMessage->getId(),
            'playhead'       => $domainMessage->getPlayhead(),
            'metadata'       => json_encode($this->metadataSerializer->serialize($domainMessage->getMetadata())),
            'payload'        => json_encode($this->payloadSerializer->serialize($domainMessage->getPayload())),
            'recorded_on'    => $domainMessage->getRecordedOn()->toString(),
            'type'           => $domainMessage->getType(),
            'aggregate_type' => $this->aggregateType
        );

        $connection->insert($this->tableName, $data);
    }

    /**
     * @param Schema $schema
     * @return Table|null
     */
    public function configureSchema(Schema $schema)
    {
        if ($schema->hasTable($this->tableName)) {
            return null;
        }

        return $this->configureTable();
    }

    /**
     * @return mixed
     */
    public function configureTable()
    {
        $schema = new Schema();

        $table = $schema->createTable($this->tableName);

        $table->addColumn('id', 'integer', array('autoincrement' => true));
        $table->addColumn('uuid', 'guid', array('length' => 36,));
        $table->addColumn('playhead', 'integer', array('unsigned' => true));
        $table->addColumn('payload', 'text');
        $table->addColumn('metadata', 'text');
        $table->addColumn('recorded_on', 'string', array('length' => 32));
        $table->addColumn('type', 'string', array('length' => 128));
        $table->addColumn('aggregate_type', 'string', array('length' => 128));

        $table->setPrimaryKey(array('id'));
        $table->addUniqueIndex(array('uuid', 'playhead'));

        return $table;
    }

    /**
     * @return \Doctrine\DBAL\Driver\Statement|null
     */
    private function prepareLoadStatement()
    {
        if (null === $this->loadStatement) {
            $queryBuilder = $this->connection->createQueryBuilder();

            $queryBuilder->select(
                ['uuid', 'playhead', 'metadata', 'payload', 'recorded_on']
            )
                ->from($this->tableName)
                ->where('uuid = :uuid')
                ->orderBy('playhead', 'ASC');

            $this->loadStatement = $this->connection->prepare(

It seems like $this->connection->prepare($queryBuilder->getSQL()) of type object<Doctrine\DBAL\Driver\Statement> is incompatible with the declared type null of property $loadStatement.

                $queryBuilder->getSQL()
            );
        }

        return $this->loadStatement;
    }

    /**
     * @param $row
     * @return DomainMessage
     */
    private function deserializeEvent($row)

This method seems to be duplicated in your project.

    {
        return new DomainMessage(
            $row['uuid'],
            $row['playhead'],
            $this->metadataSerializer->deserialize(json_decode($row['metadata'], true)),
            $this->payloadSerializer->deserialize(json_decode($row['payload'], true)),
            BroadwayDateTime::fromString($row['recorded_on'])
        );
    }

    /**
     * @param DomainEventStreamInterface $eventStream
     */
    private function guardStream(DomainEventStreamInterface $eventStream)
    {
        foreach ($eventStream as $domainMessage) {
            /** @var DomainMessage $domainMessage */
            $id = (string) $domainMessage->getId();

$id is not used, you could remove the assignment.

        }
    }
}