AggregateAwareDBALEventStore::insertMessage() - 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-15 14:07 UTC

AggregateAwareDBALEventStore::insertMessage() A

↳ Parent: AggregateAwareDBALEventStore

Complexity

Conditions	1
Paths	1

Size

Total Lines	14
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
dl	0
loc	14
rs	9.4285
c	0
b	0
f	0
cc	1
eloc	10
nc	1
nop	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\EventStore\Exception\InvalidIdentifierException;
use Broadway\Serializer\SerializerInterface;
use Doctrine\DBAL\Connection;
use Doctrine\DBAL\DBALException;
use Doctrine\DBAL\Schema\Schema;
use Doctrine\DBAL\Schema\Table;
use Doctrine\DBAL\Version;
use Rhumsaa\Uuid\Uuid;

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;

    /**
     * @var bool
     */
    private $useBinary;

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

        if ($this->useBinary && Version::compare('2.5.0') >= 0) {
            throw new \InvalidArgumentException(
                'The Binary storage is only available with Doctrine DBAL >= 2.5.0'
            );
        }
    }

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

        $events = array();
        while ($row = $statement->fetch()) {
            if ($this->useBinary) {
                $row['uuid'] = $this->convertStorageValueToIdentifier($row['uuid']);
            }
            $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)
    {
        // noop to ensure that an error will be thrown early if the ID
        // is not something that can be converted to a string. If we
        // let this move on without doing this DBAL will eventually
        // give us a hard time but the true reason for the problem
        // will be obfuscated.
        $id = (string) $id;
$myVar = 'Value';
$higher = false;

if (rand(1, 6) > 3) {
    $higher = true;
} else {
    $higher = false;
}

        $this->connection->beginTransaction();

        try {
            foreach ($eventStream as $domainMessage) {
                $this->insertMessage($this->connection, $domainMessage);
            }

            $this->connection->commit();
        } catch (DBALException $exception) {
            $this->connection->rollback();

            throw DBALEventStoreException::create($exception);
        }
    }

    /**
     * @param Connection $connection
     * @param DomainMessage $domainMessage
     */
    private function insertMessage(Connection $connection, DomainMessage $domainMessage)
    {
        $data = array(
            'uuid'           => $this->convertIdentifierToStorageValue((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();

        $uuidColumnDefinition = array(
            'type'   => 'guid',
            'params' => array(
                'length' => 36,
            ),
        );

        if ($this->useBinary) {
            $uuidColumnDefinition['type']   = 'binary';
            $uuidColumnDefinition['params'] = array(
                'length' => 16,
                'fixed'  => true,
            );
        }

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

        $table->addColumn('id', 'integer', array('autoincrement' => true));
        $table->addColumn('uuid', $uuidColumnDefinition['type'], $uuidColumnDefinition['params']);
        $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(

                $queryBuilder->getSQL()
            );
        }

        return $this->loadStatement;
    }

    /**
     * @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)),
            BroadwayDateTime::fromString($row['recorded_on'])
        );
    }

    /**
     * @param $id
     * @return mixed
     */
    private function convertIdentifierToStorageValue($id)
    {
        if ($this->useBinary) {
            try {
                return Uuid::fromString($id)->getBytes();
            } catch (\Exception $e) {
                throw new InvalidIdentifierException(
                    'Only valid UUIDs are allowed to by used with the binary storage mode.'
                );
            }
        }

        return $id;
    }

    /**
     * @param $id
     * @return mixed
     */
    private function convertStorageValueToIdentifier($id)
    {
        if ($this->useBinary) {
            try {
                return Uuid::fromBytes($id)->toString();
            } catch (\Exception $e) {
                throw new InvalidIdentifierException(
                    'Could not convert binary storage value to UUID.'
                );
            }
        }

        return $id;
    }
}


1		<?php
2
3		namespace CultuurNet\UDB3\EventSourcing\DBAL;
4
5		use Broadway\Domain\DateTime as BroadwayDateTime;
6		use Broadway\Domain\DomainEventStream;
7		use Broadway\Domain\DomainEventStreamInterface;
8		use Broadway\Domain\DomainMessage;
9		use Broadway\EventStore\DBALEventStoreException;
10		use Broadway\EventStore\EventStoreInterface;
11		use Broadway\EventStore\EventStreamNotFoundException;
12		use Broadway\EventStore\Exception\InvalidIdentifierException;
13		use Broadway\Serializer\SerializerInterface;
14		use Doctrine\DBAL\Connection;
15		use Doctrine\DBAL\DBALException;
16		use Doctrine\DBAL\Schema\Schema;
17		use Doctrine\DBAL\Schema\Table;
18		use Doctrine\DBAL\Version;
19		use Rhumsaa\Uuid\Uuid;
20
21		class AggregateAwareDBALEventStore implements EventStoreInterface
22		{
23		/**
24		* @var Connection
25		*/
26		private $connection;
27
28		/**
29		* @var SerializerInterface
30		*/
31		private $payloadSerializer;
32
33		/**
34		* @var SerializerInterface
35		*/
36		private $metadataSerializer;
37
38		/**
39		* @var null
40		*/
41		private $loadStatement = null;
42
43		/**
44		* @var string
45		*/
46		private $tableName;
47
48		/**
49		* @var string
50		*/
51		private $aggregateType;
52
53		/**
54		* @var bool
55		*/
56		private $useBinary;
57
58		/**
59		* @param Connection $connection
60		* @param SerializerInterface $payloadSerializer
61		* @param SerializerInterface $metadataSerializer
62		* @param string $tableName
63		* @param string $aggregateType
64		* @param bool $useBinary
65		*/
66		public function __construct(
67		Connection $connection,
68		SerializerInterface $payloadSerializer,
69		SerializerInterface $metadataSerializer,
70		$tableName,
71		$aggregateType,
72		$useBinary = false
73		) {
74		$this->connection = $connection;
75		$this->payloadSerializer = $payloadSerializer;
76		$this->metadataSerializer = $metadataSerializer;
77		$this->tableName = $tableName;
78		$this->aggregateType = $aggregateType;
79		$this->useBinary = (bool) $useBinary;
80
81		if ($this->useBinary && Version::compare('2.5.0') >= 0) {
82		throw new \InvalidArgumentException(
83		'The Binary storage is only available with Doctrine DBAL >= 2.5.0'
84		);
85		}
86		}
87
88		/**
89		* {@inheritDoc}
90		*/
91		public function load($id)
92		{
93		$statement = $this->prepareLoadStatement();
94		$statement->bindValue('uuid', $this->convertIdentifierToStorageValue($id));
95		$statement->execute();
96
97		$events = array();
98		while ($row = $statement->fetch()) {
99		if ($this->useBinary) {
100		$row['uuid'] = $this->convertStorageValueToIdentifier($row['uuid']);
101		}
102		$events[] = $this->deserializeEvent($row);
103		}
104
105		if (empty($events)) {
106		throw new EventStreamNotFoundException(sprintf('EventStream not found for aggregate with id %s', $id));
107		}
108
109		return new DomainEventStream($events);
110		}
111
112		/**
113		* {@inheritDoc}
114		*/
115		public function append($id, DomainEventStreamInterface $eventStream)
116		{
117		// noop to ensure that an error will be thrown early if the ID
118		// is not something that can be converted to a string. If we
119		// let this move on without doing this DBAL will eventually
120		// give us a hard time but the true reason for the problem
121		// will be obfuscated.
122		$id = (string) $id;
		0 ignored issues – show Unused Code introduced 2016-12-13 15:29 UTC by Report Bug Copy Issue Report `$id` is not used, you could remove the assignment. This check looks for variable assignements that are either overwritten by other assignments or where the variable is not used subsequently. $myVar = 'Value'; $higher = false; if (rand(1, 6) > 3) { $higher = true; } else { $higher = false; } Both the `$myVar` assignment in line 1 and the `$higher` assignment in line 2 are dead. The first because `$myVar` is never used and the second because `$higher` is always overwritten for every possible time line. Loading history...
123
124		$this->connection->beginTransaction();
125
126		try {
127		foreach ($eventStream as $domainMessage) {
128		$this->insertMessage($this->connection, $domainMessage);
129		}
130
131		$this->connection->commit();
132		} catch (DBALException $exception) {
133		$this->connection->rollback();
134
135		throw DBALEventStoreException::create($exception);
136		}
137		}
138
139		/**
140		* @param Connection $connection
141		* @param DomainMessage $domainMessage
142		*/
143		private function insertMessage(Connection $connection, DomainMessage $domainMessage)
144		{
145		$data = array(
146		'uuid' => $this->convertIdentifierToStorageValue((string) $domainMessage->getId()),
147		'playhead' => $domainMessage->getPlayhead(),
148		'metadata' => json_encode($this->metadataSerializer->serialize($domainMessage->getMetadata())),
149		'payload' => json_encode($this->payloadSerializer->serialize($domainMessage->getPayload())),
150		'recorded_on' => $domainMessage->getRecordedOn()->toString(),
151		'type' => $domainMessage->getType(),
152		'aggregate_type' => $this->aggregateType
153		);
154
155		$connection->insert($this->tableName, $data);
156		}
157
158		/**
159		* @param Schema $schema
160		* @return Table\|null
161		*/
162		public function configureSchema(Schema $schema)
163		{
164		if ($schema->hasTable($this->tableName)) {
165		return null;
166		}
167
168		return $this->configureTable();
169		}
170
171		/**
172		* @return mixed
173		*/
174		public function configureTable()
175		{
176		$schema = new Schema();
177
178		$uuidColumnDefinition = array(
179		'type' => 'guid',
180		'params' => array(
181		'length' => 36,
182		),
183		);
184
185		if ($this->useBinary) {
186		$uuidColumnDefinition['type'] = 'binary';
187		$uuidColumnDefinition['params'] = array(
188		'length' => 16,
189		'fixed' => true,
190		);
191		}
192
193		$table = $schema->createTable($this->tableName);
194
195		$table->addColumn('id', 'integer', array('autoincrement' => true));
196		$table->addColumn('uuid', $uuidColumnDefinition['type'], $uuidColumnDefinition['params']);
197		$table->addColumn('playhead', 'integer', array('unsigned' => true));
198		$table->addColumn('payload', 'text');
199		$table->addColumn('metadata', 'text');
200		$table->addColumn('recorded_on', 'string', array('length' => 32));
201		$table->addColumn('type', 'string', array('length' => 128));
202		$table->addColumn('aggregate_type', 'string', array('length' => 128));
203
204		$table->setPrimaryKey(array('id'));
205		$table->addUniqueIndex(array('uuid', 'playhead'));
206
207		return $table;
208		}
209
210		/**
211		* @return \Doctrine\DBAL\Driver\Statement\|null
212		*/
213		private function prepareLoadStatement()
214		{
215		if (null === $this->loadStatement) {
216		$queryBuilder = $this->connection->createQueryBuilder();
217
218		$queryBuilder->select(
219		['uuid', 'playhead', 'metadata', 'payload', 'recorded_on']
220		)
221		->from($this->tableName)
222		->where('uuid = :uuid')
223		->orderBy('playhead', 'ASC');
224
225		$this->loadStatement = $this->connection->prepare(
		0 ignored issues – show Documentation Bug introduced 2016-12-15 09:21 UTC by Report Bug Copy Issue Report It seems like `$this->connection->prepa...queryBuilder->getSQL())` of type `object<Doctrine\DBAL\Driver\Statement>` is incompatible with the declared type `null` of property `$loadStatement`. Our type inference engine has found an assignment to a property that is incompatible with the declared type of that property. Either this assignment is in error or the assigned type should be added to the documentation/type hint for that property.. Loading history...
226		$queryBuilder->getSQL()
227		);
228		}
229
230		return $this->loadStatement;
231		}
232
233		/**
234		* @param $row
235		* @return DomainMessage
236		*/
237	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...
238		{
239		return new DomainMessage(
240		$row['uuid'],
241		$row['playhead'],
242		$this->metadataSerializer->deserialize(json_decode($row['metadata'], true)),
243		$this->payloadSerializer->deserialize(json_decode($row['payload'], true)),
244		BroadwayDateTime::fromString($row['recorded_on'])
245		);
246		}
247
248		/**
249		* @param $id
250		* @return mixed
251		*/
252		private function convertIdentifierToStorageValue($id)
253		{
254		if ($this->useBinary) {
255		try {
256		return Uuid::fromString($id)->getBytes();
257		} catch (\Exception $e) {
258		throw new InvalidIdentifierException(
259		'Only valid UUIDs are allowed to by used with the binary storage mode.'
260		);
261		}
262		}
263
264		return $id;
265		}
266
267		/**
268		* @param $id
269		* @return mixed
270		*/
271		private function convertStorageValueToIdentifier($id)
272		{
273		if ($this->useBinary) {
274		try {
275		return Uuid::fromBytes($id)->toString();
276		} catch (\Exception $e) {
277		throw new InvalidIdentifierException(
278		'Could not convert binary storage value to UUID.'
279		);
280		}
281		}
282
283		return $id;
284		}
285		}
286

cultuurnet / udb3-php

Pull Request — master (#255)

AggregateAwareDBALEventStore::insertMessage() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like