EventStream::withCdbids() - Code Metrics - Inspection of "III-2314 Streamline replay for copied events." - cultuurnet/udb3-php - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#332)

by Luc

created 2017-10-13 12:00 UTC

EventStream::withCdbids() A

↳ Parent: EventStream

Complexity

Conditions	3
Paths	3

Size

Total Lines	14
Code Lines	8

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	14
rs	9.4285
c	0
b	0
f	0
cc	3
eloc	8
nc	3
nop	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 $cdbids;

    /**
     * @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[] $cdbids
     * @return EventStream
     */
    public function withCdbids($cdbids)
    {
        if (!is_array($cdbids)) {
            throw new \InvalidArgumentException('Cdbids should have type array.');
        }

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

        $c = clone $this;
        $c->cdbids = $cdbids;
        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->cdbids) {

            $queryBuilder->andWhere('uuid IN (:uuids)')
                ->setParameter('uuids', $this->cdbids, Connection::PARAM_STR_ARRAY);
        }

        return $queryBuilder->execute();

    }

    /**
     * @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'])
        );
    }
}


1			<?php
2
3			namespace CultuurNet\UDB3\EventSourcing\DBAL;
4
5			use Broadway\Domain\DateTime;
6			use Broadway\Domain\DomainEventStream;
7			use Broadway\Domain\DomainMessage;
8			use Broadway\EventSourcing\EventStreamDecoratorInterface;
9			use Broadway\Serializer\SerializerInterface;
10			use Doctrine\DBAL\Connection;
11			use Doctrine\DBAL\Driver\Statement;
12			use Doctrine\DBAL\DBALException;
13
14			class EventStream
15			{
16			/**
17			* @var Connection
18			*/
19			protected $connection;
20
21			/**
22			* @var SerializerInterface
23			*/
24			protected $payloadSerializer;
25
26			/**
27			* @var SerializerInterface
28			*/
29			protected $metadataSerializer;
30
31			/**
32			* @var string
33			*/
34			protected $tableName;
35
36			/**
37			* @var int
38			*/
39			protected $startId;
40
41			/**
42			* @var int
43			*/
44			protected $lastProcessedId;
45
46			/**
47			* @var string[]
48			*/
49			protected $cdbids;
50
51			/**
52			* @var EventStreamDecoratorInterface
53			*/
54			private $domainEventStreamDecorator;
55
56			/**
57			* @param Connection $connection
58			* @param SerializerInterface $payloadSerializer
59			* @param SerializerInterface $metadataSerializer
60			* @param string $tableName
61			*/
62			public function __construct(
63			Connection $connection,
64			SerializerInterface $payloadSerializer,
65			SerializerInterface $metadataSerializer,
66			$tableName
67			) {
68			$this->connection = $connection;
69			$this->payloadSerializer = $payloadSerializer;
70			$this->metadataSerializer = $metadataSerializer;
71			$this->tableName = $tableName;
72			$this->startId = 0;
73			}
74
75			/**
76			* @param int $startId
77			* @return EventStream
78			*/
79			public function withStartId($startId)
80			{
81			if (!is_int($startId)) {
82			throw new \InvalidArgumentException('StartId should have type int.');
83			}
84
85			if ($startId <= 0) {
86			throw new \InvalidArgumentException('StartId should be higher than 0.');
87			}
88
89			$c = clone $this;
90			$c->startId = $startId;
91			return $c;
92			}
93
94			/**
95			* @param string[] $cdbids
96			* @return EventStream
97			*/
98			public function withCdbids($cdbids)
99			{
100			if (!is_array($cdbids)) {
101			throw new \InvalidArgumentException('Cdbids should have type array.');
102			}
103
104			if (empty($cdbids)) {
105			throw new \InvalidArgumentException('Cdbids can\'t be empty.');
106			}
107
108			$c = clone $this;
109			$c->cdbids = $cdbids;
110			return $c;
111			}
112
113			/**
114			* @param EventStreamDecoratorInterface $domainEventStreamDecorator
115			* @return EventStream
116			*/
117			public function withDomainEventStreamDecorator(EventStreamDecoratorInterface $domainEventStreamDecorator)
118			{
119			$c = clone $this;
120			$c->domainEventStreamDecorator = $domainEventStreamDecorator;
121			return $c;
122			}
123
124			public function __invoke()
125			{
126			do {
127			$statement = $this->prepareLoadStatement();
128
129			$events = [];
130			while ($row = $statement->fetch()) {
131			$events[] = $this->deserializeEvent($row);
132			$this->lastProcessedId = $row['id'];
133			// Make sure to increment to prevent endless loop.
134			$this->startId = $row['id'] + 1;
135			}
136
137			/* @var DomainMessage[] $events */
138			if (!empty($events)) {
139			$event = $events[0];
140			$domainEventStream = new DomainEventStream($events);
141
142			if (!is_null($this->domainEventStreamDecorator)) {
143			// Because the load statement always returns one row at a
144			// time, and we always wrap a single domain message in a
145			// stream as a result, we can simply get the aggregate type
146			// and aggregate id from the first domain message in the
147			// stream.
148			$domainEventStream = $this->domainEventStreamDecorator->decorateForWrite(
149			get_class($event->getPayload()),
150			$event->getId(),
151			$domainEventStream
152			);
153			}
154
155			yield $domainEventStream;
156			}
157			} while (!empty($events));
158			}
159
160			/**
161			* @return int
162			*/
163			public function getLastProcessedId()
164			{
165			return $this->lastProcessedId;
166			}
167
168			/**
169			* The load statement can no longer be 'cached' because of using the query
170			* builder. The query builder requires all parameters to be set before
171			* using the execute command. The previous solution used the prepare
172			* statement on the connection, this did not require all parameters to be
173			* set up front.
174			*
175			* @return Statement
176			* @throws DBALException
177			*/
178			protected function prepareLoadStatement()
179			{
180			$queryBuilder = $this->connection->createQueryBuilder();
181
182			$queryBuilder->select('id', 'uuid', 'playhead', 'metadata', 'payload', 'recorded_on')
183			->from($this->tableName)
184			->where('id >= :startid')
185			->setParameter('startid', $this->startId)
186			->orderBy('id', 'ASC')
187			->setMaxResults(1);
188
189			if ($this->cdbids) {
			0 ignored issues – show Bug Best Practice introduced 2017-10-13 12:16 UTC by Report Bug Copy Issue Report The expression `$this->cdbids` of type `string[]` is implicitly converted to a boolean; are you sure this is intended? If so, consider using `! empty($expr)` instead to make it clear that you intend to check for an array without elements. This check marks implicit conversions of arrays to boolean values in a comparison. While in PHP an empty array is considered to be equal (but not identical) to false, this is not always apparent. Consider making the comparison explicit by using `empty(..)` or `! empty(...)` instead. Loading history...
190			$queryBuilder->andWhere('uuid IN (:uuids)')
191			->setParameter('uuids', $this->cdbids, Connection::PARAM_STR_ARRAY);
192			}
193
194			return $queryBuilder->execute();
			0 ignored issues – show Bug Compatibility introduced 2017-04-10 13:02 UTC by Report Bug Copy Issue Report 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\EventSou...m::prepareLoadStatement` of type `Doctrine\DBAL\Driver\Statement`. Loading history...
195			}
196
197			/**
198			* @param $row
199			* @return DomainMessage
200			*/
201			private function deserializeEvent($row)
202			{
203			return new DomainMessage(
204			$row['uuid'],
205			$row['playhead'],
206			$this->metadataSerializer->deserialize(json_decode($row['metadata'], true)),
207			$this->payloadSerializer->deserialize(json_decode($row['payload'], true)),
208			DateTime::fromString($row['recorded_on'])
209			);
210			}
211			}
212

cultuurnet / udb3-php

Pull Request — master (#332)

EventStream::withCdbids() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like