MySqlJsonEventStore::initialize() - Code Metrics - Inspection of "Add MySqlJsonEventStore migration test" - DDDominio/event-sourcing - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 075b21...a5631d )

by Daniel

created 2017-02-28 19:19 UTC

MySqlJsonEventStore::initialize() B

↳ Parent: MySqlJsonEventStore

Complexity

Conditions	2
Paths	5

Size

Total Lines	33
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	10
CRAP Score	2.0491

Importance

Changes

Metric	Value
dl	0
loc	33
c	0
b	0
f	0
ccs	10
cts	13
cp	0.7692
rs	8.8571
cc	2
eloc	11
nc	5
nop	0
crap	2.0491

<?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\EventStreamDoesNotExistException;
use DDDominio\EventSourcing\EventStore\InitializableInterface;
use DDDominio\EventSourcing\EventStore\StoredEvent;
use DDDominio\EventSourcing\Serialization\SerializerInterface;
use DDDominio\EventSourcing\Versioning\EventUpgrader;
use DDDominio\EventSourcing\Versioning\Version;

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

    /**
     * @var \PDO
     */
    private $connection;

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

    /**
     * @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($event) {
            return new StoredEvent(
                $event['id'],
                $event['stream_id'],
                $event['type'],
                $event['event'],
                $event['metadata'],
                new \DateTimeImmutable($event['occurred_on']),
                Version::fromString($event['version'])
            );
        }, $results);

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

    /**
     * @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($event) {
            return new StoredEvent(
                $event['id'],
                $event['stream_id'],
                $event['type'],
                $event['event'],
                $event['metadata'],
                new \DateTimeImmutable($event['occurred_on']),
                Version::fromString($event['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 $streamId
     * @param StoredEvent[] $storedEvents
     * @param int $expectedVersion
     * @throws \Exception
     */
    protected function appendStoredEvents($streamId, $storedEvents, $expectedVersion)
    {
        $this->connection->beginTransaction();
        try {
            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
                );
            }
            $this->connection->commit();
        } catch (\Exception $e) {
            $this->connection->rollBack();
            throw $e;
        }
    }

    /**
     * @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());
    }
    /**
     * @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 $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);
    }

    public function initialize()
    {
        try {
            $this->connection->beginTransaction();

            $this->connection->exec(
                'CREATE TABLE `'.self::STREAMS_TABLE.'` (
                    `id` varchar(255) NOT NULL,
                    PRIMARY KEY (`id`)
                )'
            );

            $this->connection->exec(
                'CREATE TABLE `'.self::EVENTS_TABLE.'` (
                    `id` int(11) NOT NULL AUTO_INCREMENT,
                    `stream_id` varchar(255) NOT NULL,
                    `type` varchar(255) NOT NULL,
                    `event` json NOT NULL,
                    `metadata` json NOT NULL,
                    `occurred_on` datetime NOT NULL,
                    `version` varchar(255) NOT NULL,
                    PRIMARY KEY (`id`),
                    KEY `stream_id` (`stream_id`),
                    CONSTRAINT `events_ibfk_1` FOREIGN KEY (`stream_id`) REFERENCES `streams` (`id`)
                )'
            );

            $this->connection->commit();
        } catch (\Exception $e) {
            $this->connection->rollBack();
            throw $e;
        }
    }

    /**
     * @return bool
     */
    public function initialized()

    {
        try {
            $result = $this->connection->query('SELECT 1 FROM `'.self::STREAMS_TABLE.'` LIMIT 1');
        } catch (\Exception $e) {
            return false;
        }
        return $result !== false;
    }

    /**
     * @param string $streamId
     * @param \DateTimeImmutable $datetime
     * @return int
     * @throws EventStreamDoesNotExistException
     */
    public function getStreamVersionAt($streamId, \DateTimeImmutable $datetime)

    {
        if (!$this->streamExists($streamId)) {
            throw EventStreamDoesNotExistException::fromStreamId($streamId);
        }
        $stmt = $this->connection->prepare(
            'SELECT COUNT(*)
             FROM events
             WHERE stream_id = :streamId
             AND occurred_on <= :occurred_on'
        );
        $stmt->bindValue(':streamId', $streamId);
        $stmt->bindValue(':occurred_on', $datetime->format('Y-m-d H:i:s'));
        $stmt->execute();
        return intval($stmt->fetchColumn());
    }
}


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

DDDominio / event-sourcing

Push — master ( 075b21...a5631d )

MySqlJsonEventStore::initialize() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like