DoctrineDbalEventStore::initialize() - Code Metrics - Inspection of "Find aggregate by DateTime" - DDDominio/event-sourcing - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 8f1ad4...65e262 )

by Daniel

created 2017-02-27 18:51 UTC

DoctrineDbalEventStore::initialize() B

↳ Parent: DoctrineDbalEventStore

Complexity

Conditions	2
Paths	1

Size

Total Lines	26
Code Lines	19

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	21
CRAP Score	2

Importance

Changes

Metric	Value
dl	0
loc	26
c	0
b	0
f	0
ccs	21
cts	21
cp	1
rs	8.8571
cc	2
eloc	19
nc	1
nop	0
crap	2

<?php

namespace DDDominio\EventSourcing\EventStore\Vendor;

use DDDominio\EventSourcing\Common\EventStream;
use DDDominio\EventSourcing\Common\EventStreamInterface;
use DDDominio\EventSourcing\EventStore\AbstractEventStore;
use DDDominio\EventSourcing\EventStore\ConcurrencyException;
use DDDominio\EventSourcing\EventStore\StoredEvent;
use Doctrine\DBAL\Connection;
use DDDominio\EventSourcing\Serialization\SerializerInterface;
use DDDominio\EventSourcing\Versioning\EventUpgrader;
use DDDominio\EventSourcing\Versioning\Version;
use Doctrine\DBAL\Schema\Schema;
use DDDominio\EventSourcing\EventStore\InitializableInterface;

class DoctrineDbalEventStore extends AbstractEventStore implements InitializableInterface
{
    const MAX_UNSIGNED_BIG_INT = 9223372036854775807;
    const STREAMS_TABLE = 'streams';
    const EVENTS_TABLE = 'events';

    /**
     * @var Connection
     */
    private $connection;

    /**
     * @param Connection $connection
     * @param SerializerInterface $serializer
     * @param EventUpgrader $eventUpgrader
     */
    public function __construct($connection, $serializer, $eventUpgrader)
    {
        parent::__construct($serializer, $eventUpgrader);
        $this->connection = $connection;
    }

    /**
     * @param string $streamId
     * @param int $start
     * @param int $count
     * @return EventStreamInterface
     */
    public function readStreamEvents($streamId, $start = 1, $count = null)

    {
        if (!isset($count)) {
            $count = self::MAX_UNSIGNED_BIG_INT;
        }
        $stmt = $this->connection->prepare(
            'SELECT *
             FROM events
             WHERE stream_id = :streamId
             LIMIT :limit
             OFFSET :offset'
        );
        $stmt->bindValue(':streamId', $streamId);
        $stmt->bindValue(':offset', (int) $start - 1, \PDO::PARAM_INT);
        $stmt->bindValue(':limit', $count, \PDO::PARAM_INT);
        $stmt->execute();
        $results = $stmt->fetchAll();

        $storedEvents = array_map(function($result) {
            return new StoredEvent(
                $result['id'],
                $result['stream_id'],
                $result['type'],
                $result['event'],
                $result['metadata'],
                new \DateTimeImmutable($result['occurred_on']),
                Version::fromString($result['version'])
            );
        }, $results);

        return $this->domainEventStreamFromStoredEvents($storedEvents);
    }

    /**
     * @param string $streamId
     * @param \DateTimeImmutable $datetime
     * @param int $start
     * @return EventStreamInterface
     */
    public function readStreamEventsUntil($streamId, $datetime, $start = 1)
    {
        // TODO: Implement readStreamEventsUntil() method.
    }

    /**
     * @param string $streamId
     * @return EventStreamInterface
     */
    public function readFullStream($streamId)

    {
        $stmt = $this->connection->prepare(
            'SELECT *
             FROM events
             WHERE stream_id = :streamId'
        );
        $stmt->bindValue(':streamId', $streamId);
        $stmt->execute();
        $results = $stmt->fetchAll();

        $storedEvents = array_map(function($result) {
            return new StoredEvent(
                $result['id'],
                $result['stream_id'],
                $result['type'],
                $result['event'],
                $result['metadata'],
                new \DateTimeImmutable($result['occurred_on']),
                Version::fromString($result['version'])
            );
        }, $results);

        return $this->domainEventStreamFromStoredEvents($storedEvents);
    }

    /**
     * @return EventStreamInterface[]
     */
    public function readAllStreams()
    {
        // TODO: Implement readAllStreams() method.
    }

    /**
     * @return EventStreamInterface
     */
    public function readAllEvents()
    {
        // TODO: Implement readAllEvents() method.
    }

    /**
     * @param string $type
     * @param Version $version
     * @return EventStreamInterface
     */
    protected function readStoredEventsOfTypeAndVersion($type, $version)

    {
        $stmt = $this->connection->prepare(
            'SELECT *
             FROM events
             WHERE type = :type
             AND version = :version'
        );
        $stmt->bindValue(':type', $type);
        $stmt->bindValue(':version', $version);
        $stmt->execute();
        $results = $stmt->fetchAll();

        $storedEvents = array_map(function($result) {
            return new StoredEvent(
                $result['id'],
                $result['stream_id'],
                $result['type'],
                $result['event'],
                $result['metadata'],
                new \DateTimeImmutable($result['occurred_on']),
                Version::fromString($result['version'])
            );
        }, $results);

        return new EventStream($storedEvents);
    }

    /**
     * @param string $streamId
     * @param StoredEvent[] $storedEvents
     * @param int $expectedVersion
     */
    protected function appendStoredEvents($streamId, $storedEvents, $expectedVersion)
    {
        $this->connection->transactional(function() use ($streamId, $storedEvents, $expectedVersion) {
            if (!$this->streamExists($streamId)) {

                $stmt = $this->connection
                    ->prepare('INSERT INTO streams (id) VALUES (:streamId)');
                $stmt->bindValue(':streamId', $streamId);
                $stmt->execute();
            }
            foreach ($storedEvents as $storedEvent) {

                $stmt = $this->connection->prepare(
                    'INSERT INTO events (stream_id, type, event, metadata, occurred_on, version)
                 VALUES (:streamId, :type, :event, :metadata, :occurredOn, :version)'
                );
                $stmt->bindValue(':streamId', $streamId);
                $stmt->bindValue(':type', $storedEvent->type());
                $stmt->bindValue(':event', $storedEvent->data());
                $stmt->bindValue(':metadata', $storedEvent->metadata());
                $stmt->bindValue(':occurredOn', $storedEvent->occurredOn()->format('Y-m-d H:i:s'));
                $stmt->bindValue(':version', $storedEvent->version());
                $stmt->execute();
            }
            $streamFinalVersion = $this->streamVersion($streamId);
            if (count($storedEvents) !== $streamFinalVersion - $expectedVersion) {
                throw ConcurrencyException::fromVersions(
                    $this->streamVersion($streamId),
                    $expectedVersion
                );
            }
        });
    }
    /**
     * @param string $streamId
     * @return int
     */
    protected function streamVersion($streamId)

    {
        $stmt = $this->connection
            ->prepare('SELECT COUNT(*) FROM events WHERE stream_id = :streamId');
        $stmt->bindValue(':streamId', $streamId);
        $stmt->execute();
        return intval($stmt->fetchColumn());
    }

    /**
     * @param string $streamId
     * @return bool
     */
    protected function streamExists($streamId)

    {
        $stmt = $this->connection
            ->prepare('SELECT COUNT(*) FROM streams WHERE id = :streamId');
        $stmt->bindValue(':streamId', $streamId);
        $stmt->execute();
        return boolval($stmt->fetchColumn());
    }

    /**
     * Initialize the Event Store
     */
    public function initialize()
    {
        $schema = new Schema();

        $streamTable = $schema->createTable(self::STREAMS_TABLE);
        $streamTable->addColumn('id', 'string');
        $streamTable->setPrimaryKey(array("id"));

        $eventsTable = $schema->createTable(self::EVENTS_TABLE);
        $eventsTable->addColumn('id', 'integer', ['autoincrement' => true]);
        $eventsTable->addColumn('stream_id', 'string');
        $eventsTable->addColumn('type', 'string');
        $eventsTable->addColumn('event', 'text');
        $eventsTable->addColumn('metadata', 'text');
        $eventsTable->addColumn('occurred_on', 'datetime');
        $eventsTable->addColumn('version', 'string');
        $eventsTable->setPrimaryKey(['id']);
        $eventsTable->addForeignKeyConstraint($streamTable, ['stream_id'], ['id']);

        $queries = $schema->toSql($this->connection->getDatabasePlatform());
        $this->connection->transactional(function(Connection $connection) use ($queries) {
            foreach ($queries as $query) {
                $connection->exec($query);
            }
        });
    }

    /**
     * Check if the Event Store has been initialized
     *
     * @return bool
     */
    public function initialized()
    {
        return $this->connection->getSchemaManager()->tablesExist([self::STREAMS_TABLE]);
    }

    /**
     * @param string $streamId
     * @param \DateTimeImmutable $datetime
     * @return int
     */
    public function getStreamVersionAt($streamId, \DateTimeImmutable $datetime)
    {
        // TODO: Implement findStreamVersionAt() method.
    }
}


1			<?php
2
3			namespace DDDominio\EventSourcing\EventStore\Vendor;
4
5			use DDDominio\EventSourcing\Common\EventStream;
6			use DDDominio\EventSourcing\Common\EventStreamInterface;
7			use DDDominio\EventSourcing\EventStore\AbstractEventStore;
8			use DDDominio\EventSourcing\EventStore\ConcurrencyException;
9			use DDDominio\EventSourcing\EventStore\StoredEvent;
10			use Doctrine\DBAL\Connection;
11			use DDDominio\EventSourcing\Serialization\SerializerInterface;
12			use DDDominio\EventSourcing\Versioning\EventUpgrader;
13			use DDDominio\EventSourcing\Versioning\Version;
14			use Doctrine\DBAL\Schema\Schema;
15			use DDDominio\EventSourcing\EventStore\InitializableInterface;
16
17			class DoctrineDbalEventStore extends AbstractEventStore implements InitializableInterface
18			{
19			const MAX_UNSIGNED_BIG_INT = 9223372036854775807;
20			const STREAMS_TABLE = 'streams';
21			const EVENTS_TABLE = 'events';
22
23			/**
24			* @var Connection
25			*/
26			private $connection;
27
28			/**
29			* @param Connection $connection
30			* @param SerializerInterface $serializer
31			* @param EventUpgrader $eventUpgrader
32			*/
33	12		public function __construct($connection, $serializer, $eventUpgrader)
34			{
35	12		parent::__construct($serializer, $eventUpgrader);
36	12		$this->connection = $connection;
37	12		}
38
39			/**
40			* @param string $streamId
41			* @param int $start
42			* @param int $count
43			* @return EventStreamInterface
44			*/
45	4	View Code Duplication	public function readStreamEvents($streamId, $start = 1, $count = null)
			0 ignored issues – show Duplication introduced 2017-02-11 16:14 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...
46			{
47	4		if (!isset($count)) {
48	3		$count = self::MAX_UNSIGNED_BIG_INT;
49			}
50	4		$stmt = $this->connection->prepare(
51			'SELECT *
52			FROM events
53			WHERE stream_id = :streamId
54			LIMIT :limit
55	4		OFFSET :offset'
56			);
57	4		$stmt->bindValue(':streamId', $streamId);
58	4		$stmt->bindValue(':offset', (int) $start - 1, \PDO::PARAM_INT);
59	4		$stmt->bindValue(':limit', $count, \PDO::PARAM_INT);
60	4		$stmt->execute();
61	4		$results = $stmt->fetchAll();
62
63			$storedEvents = array_map(function($result) {
64	3		return new StoredEvent(
65	3		$result['id'],
66	3		$result['stream_id'],
67	3		$result['type'],
68	3		$result['event'],
69	3		$result['metadata'],
70	3		new \DateTimeImmutable($result['occurred_on']),
71	3		Version::fromString($result['version'])
72			);
73	4		}, $results);
74
75	4		return $this->domainEventStreamFromStoredEvents($storedEvents);
76			}
77
78			/**
79			* @param string $streamId
80			* @param \DateTimeImmutable $datetime
81			* @param int $start
82			* @return EventStreamInterface
83			*/
84			public function readStreamEventsUntil($streamId, $datetime, $start = 1)
85			{
86			// TODO: Implement readStreamEventsUntil() method.
87			}
88
89			/**
90			* @param string $streamId
91			* @return EventStreamInterface
92			*/
93	6	View Code Duplication	public function readFullStream($streamId)
			0 ignored issues – show Duplication introduced 2017-02-11 16:14 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...
94			{
95	6		$stmt = $this->connection->prepare(
96			'SELECT *
97			FROM events
98	6		WHERE stream_id = :streamId'
99			);
100	6		$stmt->bindValue(':streamId', $streamId);
101	6		$stmt->execute();
102	6		$results = $stmt->fetchAll();
103
104			$storedEvents = array_map(function($result) {
105	5		return new StoredEvent(
106	5		$result['id'],
107	5		$result['stream_id'],
108	5		$result['type'],
109	5		$result['event'],
110	5		$result['metadata'],
111	5		new \DateTimeImmutable($result['occurred_on']),
112	5		Version::fromString($result['version'])
113			);
114	6		}, $results);
115
116	6		return $this->domainEventStreamFromStoredEvents($storedEvents);
117			}
118
119			/**
120			* @return EventStreamInterface[]
121			*/
122			public function readAllStreams()
123			{
124			// TODO: Implement readAllStreams() method.
125			}
126
127			/**
128			* @return EventStreamInterface
129			*/
130			public function readAllEvents()
131			{
132			// TODO: Implement readAllEvents() method.
133			}
134
135			/**
136			* @param string $type
137			* @param Version $version
138			* @return EventStreamInterface
139			*/
140	1	View Code Duplication	protected function readStoredEventsOfTypeAndVersion($type, $version)
			0 ignored issues – show Duplication introduced 2017-02-11 16:14 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...
141			{
142	1		$stmt = $this->connection->prepare(
143			'SELECT *
144			FROM events
145			WHERE type = :type
146	1		AND version = :version'
147			);
148	1		$stmt->bindValue(':type', $type);
149	1		$stmt->bindValue(':version', $version);
150	1		$stmt->execute();
151	1		$results = $stmt->fetchAll();
152
153			$storedEvents = array_map(function($result) {
154	1		return new StoredEvent(
155	1		$result['id'],
156	1		$result['stream_id'],
157	1		$result['type'],
158	1		$result['event'],
159	1		$result['metadata'],
160	1		new \DateTimeImmutable($result['occurred_on']),
161	1		Version::fromString($result['version'])
162			);
163	1		}, $results);
164
165	1		return new EventStream($storedEvents);
166			}
167
168			/**
169			* @param string $streamId
170			* @param StoredEvent[] $storedEvents
171			* @param int $expectedVersion
172			*/
173	7		protected function appendStoredEvents($streamId, $storedEvents, $expectedVersion)
174			{
175			$this->connection->transactional(function() use ($streamId, $storedEvents, $expectedVersion) {
176	7	View Code Duplication	if (!$this->streamExists($streamId)) {
			0 ignored issues – show Duplication introduced 2017-02-11 16:14 UTC by Report Bug Copy Issue Report This code seems to be duplicated across 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...
177	7		$stmt = $this->connection
178	7		->prepare('INSERT INTO streams (id) VALUES (:streamId)');
179	7		$stmt->bindValue(':streamId', $streamId);
180	7		$stmt->execute();
181			}
182	7	View Code Duplication	foreach ($storedEvents as $storedEvent) {
			0 ignored issues – show Duplication introduced 2017-02-11 16:14 UTC by Report Bug Copy Issue Report This code seems to be duplicated across 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...
183	7		$stmt = $this->connection->prepare(
184			'INSERT INTO events (stream_id, type, event, metadata, occurred_on, version)
185	7		VALUES (:streamId, :type, :event, :metadata, :occurredOn, :version)'
186			);
187	7		$stmt->bindValue(':streamId', $streamId);
188	7		$stmt->bindValue(':type', $storedEvent->type());
189	7		$stmt->bindValue(':event', $storedEvent->data());
190	7		$stmt->bindValue(':metadata', $storedEvent->metadata());
191	7		$stmt->bindValue(':occurredOn', $storedEvent->occurredOn()->format('Y-m-d H:i:s'));
192	7		$stmt->bindValue(':version', $storedEvent->version());
193	7		$stmt->execute();
194			}
195	7		$streamFinalVersion = $this->streamVersion($streamId);
196	7		if (count($storedEvents) !== $streamFinalVersion - $expectedVersion) {
197			throw ConcurrencyException::fromVersions(
198			$this->streamVersion($streamId),
199			$expectedVersion
200			);
201			}
202	7		});
203	7		}
204			/**
205			* @param string $streamId
206			* @return int
207			*/
208	7	View Code Duplication	protected function streamVersion($streamId)
			0 ignored issues – show Duplication introduced 2017-02-11 16:14 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...
209			{
210	7		$stmt = $this->connection
211	7		->prepare('SELECT COUNT(*) FROM events WHERE stream_id = :streamId');
212	7		$stmt->bindValue(':streamId', $streamId);
213	7		$stmt->execute();
214	7		return intval($stmt->fetchColumn());
215			}
216
217			/**
218			* @param string $streamId
219			* @return bool
220			*/
221	8	View Code Duplication	protected function streamExists($streamId)
			0 ignored issues – show Duplication introduced 2017-02-11 16:14 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...
222			{
223	8		$stmt = $this->connection
224	8		->prepare('SELECT COUNT(*) FROM streams WHERE id = :streamId');
225	8		$stmt->bindValue(':streamId', $streamId);
226	8		$stmt->execute();
227	8		return boolval($stmt->fetchColumn());
228			}
229
230			/**
231			* Initialize the Event Store
232			*/
233	12		public function initialize()
234			{
235	12		$schema = new Schema();
236
237	12		$streamTable = $schema->createTable(self::STREAMS_TABLE);
238	12		$streamTable->addColumn('id', 'string');
239	12		$streamTable->setPrimaryKey(array("id"));
240
241	12		$eventsTable = $schema->createTable(self::EVENTS_TABLE);
242	12		$eventsTable->addColumn('id', 'integer', ['autoincrement' => true]);
243	12		$eventsTable->addColumn('stream_id', 'string');
244	12		$eventsTable->addColumn('type', 'string');
245	12		$eventsTable->addColumn('event', 'text');
246	12		$eventsTable->addColumn('metadata', 'text');
247	12		$eventsTable->addColumn('occurred_on', 'datetime');
248	12		$eventsTable->addColumn('version', 'string');
249	12		$eventsTable->setPrimaryKey(['id']);
250	12		$eventsTable->addForeignKeyConstraint($streamTable, ['stream_id'], ['id']);
251
252	12		$queries = $schema->toSql($this->connection->getDatabasePlatform());
253	12		$this->connection->transactional(function(Connection $connection) use ($queries) {
254	12		foreach ($queries as $query) {
255	12		$connection->exec($query);
256			}
257	12		});
258	12		}
259
260			/**
261			* Check if the Event Store has been initialized
262			*
263			* @return bool
264			*/
265			public function initialized()
266			{
267			return $this->connection->getSchemaManager()->tablesExist([self::STREAMS_TABLE]);
268			}
269
270			/**
271			* @param string $streamId
272			* @param \DateTimeImmutable $datetime
273			* @return int
274			*/
275			public function getStreamVersionAt($streamId, \DateTimeImmutable $datetime)
276			{
277			// TODO: Implement findStreamVersionAt() method.
278			}
279			}
280

DDDominio / event-sourcing

Push — master ( 8f1ad4...65e262 )

DoctrineDbalEventStore::initialize() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like