EventStream::getPreviousId() - Code Metrics - Inspection of "III-1490 Merge the separate event stores." - cultuurnet/udb3-php - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#255)

by Luc

created 2016-12-16 14:32 UTC

EventStream::getPreviousId() A

↳ Parent: EventStream

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
c	0
b	0
f	0
dl	0
loc	4
rs	10
cc	1
eloc	2
nc	1
nop	0

<?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\DBALException;
use Doctrine\DBAL\Query\QueryBuilder;

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

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

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

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

    /**
     * @var QueryBuilder
     */
    protected $queryBuilder;

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

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

    /**
     * @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->previousId = 0;
        $this->primaryKey = 'id';
class MyClass { }

$x = new MyClass();
$x->foo = true;
        $this->aggregateType = '';
        $this->domainEventStreamDecorator = null;
    }

    /**
     * @param int $startId
     * @return EventStream
     */
    public function withStartId($startId)
    {
        $c = clone $this;
        $c->previousId = $startId - 1;
        return $c;
    }

    /**
     * @param string $aggregateType
     * @return EventStream
     */
    public function withAggregateType($aggregateType)
    {
        $c = clone $this;
        $c->aggregateType = $aggregateType;
        return $c;
    }

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

    public function __invoke()
    {
        $queryBuilder = $this->prepareLoadQuery();

        do {
            $queryBuilder->setParameter(':previousId', $this->previousId);
            $statement = $queryBuilder->execute();

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

            /* @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 getPreviousId()
    {
        return $this->previousId;
    }

    /**
     * @return QueryBuilder
     * @throws DBALException
     */
    protected function prepareLoadQuery()
    {
        if (null === $this->queryBuilder) {

            $this->queryBuilder = $this->connection->createQueryBuilder();

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

            if (!empty($this->aggregateType)) {
                $this->queryBuilder->andWhere('aggregate_type = :aggregate_type');
                $this->queryBuilder->setParameter('aggregate_type', $this->aggregateType);
            }
        }

        return $this->queryBuilder;
    }

    /**
     * @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\DBALException;
12		use Doctrine\DBAL\Query\QueryBuilder;
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 QueryBuilder
38		*/
39		protected $queryBuilder;
40
41		/**
42		* @var int
43		*/
44		protected $previousId;
45
46		/**
47		* @var string
48		*/
49		protected $aggregateType;
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->previousId = 0;
73		$this->primaryKey = 'id';
		0 ignored issues – show Bug introduced 2016-12-16 14:36 UTC by Report Bug Copy Issue Report The property `primaryKey` does not exist. Did you maybe forget to declare it? In PHP it is possible to write to properties without declaring them. For example, the following is perfectly valid PHP code: class MyClass { } $x = new MyClass(); $x->foo = true; Generally, it is a good practice to explictly declare properties to avoid accidental typos and provide IDE auto-completion: class MyClass { public $foo; } $x = new MyClass(); $x->foo = true; Loading history...
74		$this->aggregateType = '';
75		$this->domainEventStreamDecorator = null;
76		}
77
78		/**
79		* @param int $startId
80		* @return EventStream
81		*/
82		public function withStartId($startId)
83		{
84		$c = clone $this;
85		$c->previousId = $startId - 1;
86		return $c;
87		}
88
89		/**
90		* @param string $aggregateType
91		* @return EventStream
92		*/
93		public function withAggregateType($aggregateType)
94		{
95		$c = clone $this;
96		$c->aggregateType = $aggregateType;
97		return $c;
98		}
99
100		/**
101		* @param EventStreamDecoratorInterface $domainEventStreamDecorator
102		* @return EventStream
103		*/
104		public function withDomainEventStreamDecorator(EventStreamDecoratorInterface $domainEventStreamDecorator)
105		{
106		$c = clone $this;
107		$c->domainEventStreamDecorator = $domainEventStreamDecorator;
108		return $c;
109		}
110
111		public function __invoke()
112		{
113		$queryBuilder = $this->prepareLoadQuery();
114
115		do {
116		$queryBuilder->setParameter(':previousId', $this->previousId);
117		$statement = $queryBuilder->execute();
118
119		$events = [];
120		while ($row = $statement->fetch()) {
121		$events[] = $this->deserializeEvent($row);
122		$this->previousId = $row[$this->primaryKey];
123		}
124
125		/* @var DomainMessage[] $events */
126		if (!empty($events)) {
127		$event = $events[0];
128		$domainEventStream = new DomainEventStream($events);
129
130		if (!is_null($this->domainEventStreamDecorator)) {
131		// Because the load statement always returns one row at a
132		// time, and we always wrap a single domain message in a
133		// stream as a result, we can simply get the aggregate type
134		// and aggregate id from the first domain message in the
135		// stream.
136		$domainEventStream = $this->domainEventStreamDecorator->decorateForWrite(
137		get_class($event->getPayload()),
138		$event->getId(),
139		$domainEventStream
140		);
141		}
142
143		yield $domainEventStream;
144		}
145		} while (!empty($events));
146		}
147
148		/**
149		* @return int
150		*/
151		public function getPreviousId()
152		{
153		return $this->previousId;
154		}
155
156		/**
157		* @return QueryBuilder
158		* @throws DBALException
159		*/
160		protected function prepareLoadQuery()
161		{
162		if (null === $this->queryBuilder) {
163
164		$this->queryBuilder = $this->connection->createQueryBuilder();
165
166		$this->queryBuilder->select(
167		[
168		$this->primaryKey,
169		'uuid',
170		'playhead',
171		'payload',
172		'metadata',
173		'recorded_on'
174		]
175		)
176		->from($this->tableName)
177		->where($this->primaryKey . ' > :previousId')
178		->orderBy($this->primaryKey, 'ASC')
179		->setMaxResults(1);
180
181		if (!empty($this->aggregateType)) {
182		$this->queryBuilder->andWhere('aggregate_type = :aggregate_type');
183		$this->queryBuilder->setParameter('aggregate_type', $this->aggregateType);
184		}
185		}
186
187		return $this->queryBuilder;
188		}
189
190		/**
191		* @param $row
192		* @return DomainMessage
193		*/
194	View Code Duplication	private function deserializeEvent($row)
		0 ignored issues – show Duplication introduced 2016-12-13 15:29 UTC by Report Bug Copy Issue Report This method seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
195		{
196		return new DomainMessage(
197		$row['uuid'],
198		$row['playhead'],
199		$this->metadataSerializer->deserialize(json_decode($row['metadata'], true)),
200		$this->payloadSerializer->deserialize(json_decode($row['payload'], true)),
201		DateTime::fromString($row['recorded_on'])
202		);
203		}
204		}
205

cultuurnet / udb3-php

Pull Request — master (#255)

EventStream::getPreviousId() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like