CultuurNet\UDB3\EventSourcing\DBAL\EventStream

Complexity

Total Complexity

17

Size/Duplication

Total Lines	198
Duplicated Lines	0 %

Coupling/Cohesion

Components	3
Dependencies	8

Importance

Changes	1
Bugs	0	Features	0

8 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	12	1
A	withCdbid()	0	14	3
A	withDomainEventStreamDecorator()	0	6	1
A	withStartId()	0	14	3
B	__invoke()	0	35	5
A	getLastProcessedId()	0	4	1
A	prepareLoadStatement()	0	18	2
A	deserializeEvent()	0	10	1

<?php

namespace CultuurNet\UDB3\EventSourcing\DBAL;

use Broadway\Domain\DateTime;
use Broadway\Domain\DomainEventStream;
use Broadway\Domain\DomainMessage;
use Broadway\EventSourcing\EventStreamDecoratorInterface;
use Broadway\Serializer\SerializerInterface;
use Doctrine\DBAL\Connection;
use Doctrine\DBAL\Driver\Statement;
use Doctrine\DBAL\DBALException;

class EventStream
{
    /**
     * @var Connection
     */
    protected $connection;

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

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

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

    /**
     * @var int
     */
    protected $startId;

    /**
     * @var int
     */
    protected $lastProcessedId;

    /**
     * @var string
     */
    protected $cdbid;

    /**
     * @var EventStreamDecoratorInterface
     */
    private $domainEventStreamDecorator;

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

    /**
     * @param int $startId
     * @return EventStream
     */
    public function withStartId($startId)
    {
        if (!is_int($startId)) {
            throw new \InvalidArgumentException('StartId should have type int.');
        }

        if ($startId <= 0) {
            throw new \InvalidArgumentException('StartId should be higher than 0.');
        }

        $c = clone $this;
        $c->startId = $startId;
        return $c;
    }

    /**
     * @param string $cdbid
     * @return EventStream
     */
    public function withCdbid($cdbid)
    {
        if (!is_string($cdbid)) {
            throw new \InvalidArgumentException('Cdbid should have type string.');
        }

        if (empty($cdbid)) {
            throw new \InvalidArgumentException('Cdbid can\'t be empty.');
        }

        $c = clone $this;
        $c->cdbid = $cdbid;
        return $c;
    }

    /**
     * @param EventStreamDecoratorInterface $domainEventStreamDecorator
     * @return EventStream
     */
    public function withDomainEventStreamDecorator(EventStreamDecoratorInterface $domainEventStreamDecorator)
    {
        $c = clone $this;
        $c->domainEventStreamDecorator = $domainEventStreamDecorator;
        return $c;
    }

    public function __invoke()
    {
        do {
            $statement = $this->prepareLoadStatement();

            $events = [];
            while ($row = $statement->fetch()) {
                $events[] = $this->deserializeEvent($row);
                $this->lastProcessedId = $row['id'];
                // Make sure to increment to prevent endless loop.
                $this->startId = $row['id'] + 1;
            }

            /* @var DomainMessage[] $events */
            if (!empty($events)) {
                $event = $events[0];
                $domainEventStream = new DomainEventStream($events);

                if (!is_null($this->domainEventStreamDecorator)) {
                    // Because the load statement always returns one row at a
                    // time, and we always wrap a single domain message in a
                    // stream as a result, we can simply get the aggregate type
                    // and aggregate id from the first domain message in the
                    // stream.
                    $domainEventStream = $this->domainEventStreamDecorator->decorateForWrite(
                        get_class($event->getPayload()),
                        $event->getId(),
                        $domainEventStream
                    );
                }

                yield $domainEventStream;
            }
        } while (!empty($events));
    }

    /**
     * @return int
     */
    public function getLastProcessedId()
    {
        return $this->lastProcessedId;
    }

    /**
     * The load statement can no longer be 'cached' because of using the query
     * builder. The query builder requires all parameters to be set before
     * using the execute command. The previous solution used the prepare
     * statement on the connection, this did not require all parameters to be
     * set up front.
     *
     * @return Statement
     * @throws DBALException
     */
    protected function prepareLoadStatement()
    {
        $queryBuilder = $this->connection->createQueryBuilder();

        $queryBuilder->select('id', 'uuid', 'playhead', 'metadata', 'payload', 'recorded_on')
            ->from($this->tableName)
            ->where('id >= :startid')
            ->setParameter('startid', $this->startId)
            ->orderBy('id', 'ASC')
            ->setMaxResults(1);

        if ($this->cdbid) {
            $queryBuilder->andWhere('uuid = :uuid')
                ->setParameter('uuid', $this->cdbid);
        }

        return $queryBuilder->execute();

The expression $queryBuilder->execute(); of type Doctrine\DBAL\Driver\Statement|integer adds the type integer to the return on line 194 which is incompatible with the return type documented by CultuurNet\UDB3\EventSourcing\DBAL\EventStream::prepareLoadStatement of type Doctrine\DBAL\Driver\Statement.

    }

    /**
     * @param $row
     * @return DomainMessage
     */
    private function deserializeEvent($row)
    {
        return new DomainMessage(
            $row['uuid'],
            $row['playhead'],
            $this->metadataSerializer->deserialize(json_decode($row['metadata'], true)),
            $this->payloadSerializer->deserialize(json_decode($row['payload'], true)),
            DateTime::fromString($row['recorded_on'])
        );
    }
}