MongoSnapshot::__construct() - Code Metrics - Inspection of "Add snapshot" - borobudur-php/borobudur-es - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 8e5e55...6fad72 )

by Iqbal

created 2016-08-22 19:30 UTC

MongoSnapshot::__construct() A

↳ Parent: MongoSnapshot

Complexity

Conditions	1
Paths	1

Size

Total Lines	6
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	1

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

<?php

namespace Borobudur\EventSourcing\Snapshot;

use Borobudur\Cqrs\IdentifierInterface;
use Borobudur\EventSourcing\Entity\AggregateRootInterface;
use Borobudur\EventSourcing\Exception\LogicException;
use Borobudur\EventSourcing\Storage\Mongo\MongoConfig;
use Borobudur\EventSourcing\Storage\TransactionalTrait;
use DateTime;
use MongoDB\Driver\BulkWrite;
use MongoDB\Driver\Command;
use MongoDB\Driver\Manager;
use MongoDB\Driver\Query;

/**
 * @author      Iqbal Maulana <[email protected]>
 * @created     8/22/16
 */
class MongoSnapshot implements SnapshotInterface
{
    use TransactionalTrait;

    /**
     * @var Manager
     */
    private $conn;

    /**
     * @var string
     */
    private $db;

    /**
     * @var int
     */
    private $snapshotPeriodical;

    /**
     * Constructor.
     *
     * @param MongoConfig $config
     * @param int         $snapshotPeriodical
     */
    public function __construct(MongoConfig $config, $snapshotPeriodical = SnapshotInterface::SNAPSHOT_PERIODICAL)
    {
        $this->conn = new Manager($config->dsn);
        $this->db = $config->database;
        $this->snapshotPeriodical = $snapshotPeriodical;
    }

    /**
     * {@inheritdoc}
     */
    public function append(AggregateRootInterface $aggregate, $table)
    {
        if ($this->isNeedToSnapshot($table, $aggregate->getVersion())) {
            if ($this->has($aggregate->getId(), $aggregate->getVersion(), $table)) {
                throw new LogicException(
                    sprintf(
                        'Snapshot for "%s" with id "%s" and version "%d" already exist.',
                        $table,
                        (string) $aggregate->getId(),
                        $aggregate->getVersion()
                    )
                );
            }

            $record = $this->serialize($aggregate);
            $this->watch($table, $record);

            $bulk = new BulkWrite;
            $record['_id'] = $bulk->insert($record);
            $this->conn->executeBulkWrite($this->getNamespace($table), $bulk);
        }
    }

    /**
     * {@inheritdoc}
     */
    public function has(IdentifierInterface $id, $version, $table)
    {
        return 0 !== $this->count($table, array('uuid' => $id, 'version' => $version));
    }

    /**
     * {@inheritdoc}
     */
    public function find(IdentifierInterface $id, $class, $table)
    {
        $query = new Query(
            array('uuid' => (string) $id),
            array('sort' => array('version' => -1), 'limit' => 1)
        );
        $cursor = $this->conn->executeQuery($this->getNamespace($table), $query);
        $cursor->setTypeMap(array('document' => 'array'));
        $results = $cursor->toArray();
        
        if (count($results)) {
            $results = current($results);

            return $this->deserialize($class, (array) $results);
        }

        return null;
    }

    /**
     * {@inheritdoc}
     */
    public function drop($table)
    {
        $bulk = new BulkWrite;
        $bulk->delete(array());
        $this->conn->executeBulkWrite($this->getNamespace($table), $bulk);
    }
    
    /**
     * {@inheritdoc}
     */
    public function getSnapshotPeriodical()
    {
        return $this->snapshotPeriodical;
    }

    /**
     * {@inheritdoc}
     */
    public function setSnapshotPeriodical($snapshotPeriodical)
    {
        $this->snapshotPeriodical = $snapshotPeriodical;
    }

    /**
     * {@inheritdoc}
     */
    protected function performCommit($table, array $records)

    {
        // Do nothing (handle by the trait)
    }

    /**
     * {@inheritdoc}
     */
    protected function performRollback($table, array $records)
    {
        $bulk = new BulkWrite;
        foreach ($records as $record) {
            $id = $record['_id'];
            unset($record['_id']);
            $bulk->delete(array('_id' => $id));
        }
        
        $this->conn->executeBulkWrite($this->getNamespace($table), $bulk);
    }

    /**
     * @param string $table
     * @param int    $version
     *
     * @return bool
     */
    private function isNeedToSnapshot($table, $version)
    {
        return $version >= $this->snapshotPeriodical
        && (0 === $version % $this->snapshotPeriodical
            || $this->isEmpty($table));
    }

    /**
     * @param string $table
     *
     * @return bool
     */
    private function isEmpty($table)
    {
        return 0 === $this->count($table);
    }

    /**
     * Count collection.
     *
     * @param string $table
     * @param array  $query
     *
     * @return int
     */
    private function count($table, array $query = array())
    {
        $command = new Command(array('count' => $table, 'query' => $query));
        $cursor = $this->conn->executeCommand($this->db, $command);
        $results = (array) current($cursor->toArray());

        return (int) $results['n'];
    }
    
    /**
     * Serialize aggregate root with metadata.
     *
     * @param AggregateRootInterface $aggregate
     *
     * @return array
     */
    private function serialize(AggregateRootInterface $aggregate)
    {
        return [
            'uuid'    => (string) $aggregate->getId(),
            'version' => $aggregate->getVersion(),
            'created' => (new DateTime())->format('Y-m-d\TH:i:s.uP'),
            'payload' => array('class' => get_class($aggregate), 'body' => $aggregate->serialize()),
        ];
    }

    /**
     * Deserialize from array to aggregate root.
     *
     * @param string $class
     * @param array  $serialized
     *
     * @return AggregateRootInterface
     */
    private function deserialize($class, array $serialized)
    {
        return $class::{'deserialize'}(
            array_merge(
                $serialized['payload']['body'],
                array(
                    'version' => $serialized['version'],
                )
            )
        );
    }

    /**
     * @param string $table
     *
     * @return string
     */
    private function getNamespace($table)
    {
        return sprintf('%s.%s', $this->db, $table);
    }
}


1			<?php
2
3			namespace Borobudur\EventSourcing\Snapshot;
4
5			use Borobudur\Cqrs\IdentifierInterface;
6			use Borobudur\EventSourcing\Entity\AggregateRootInterface;
7			use Borobudur\EventSourcing\Exception\LogicException;
8			use Borobudur\EventSourcing\Storage\Mongo\MongoConfig;
9			use Borobudur\EventSourcing\Storage\TransactionalTrait;
10			use DateTime;
11			use MongoDB\Driver\BulkWrite;
12			use MongoDB\Driver\Command;
13			use MongoDB\Driver\Manager;
14			use MongoDB\Driver\Query;
15
16			/**
17			* @author Iqbal Maulana <[email protected]>
18			* @created 8/22/16
19			*/
20			class MongoSnapshot implements SnapshotInterface
21			{
22			use TransactionalTrait;
23
24			/**
25			* @var Manager
26			*/
27			private $conn;
28
29			/**
30			* @var string
31			*/
32			private $db;
33
34			/**
35			* @var int
36			*/
37			private $snapshotPeriodical;
38
39			/**
40			* Constructor.
41			*
42			* @param MongoConfig $config
43			* @param int $snapshotPeriodical
44			*/
45			public function __construct(MongoConfig $config, $snapshotPeriodical = SnapshotInterface::SNAPSHOT_PERIODICAL)
46			{
47			$this->conn = new Manager($config->dsn);
48			$this->db = $config->database;
49			$this->snapshotPeriodical = $snapshotPeriodical;
50			}
51
52			/**
53			* {@inheritdoc}
54			*/
55			public function append(AggregateRootInterface $aggregate, $table)
56			{
57			if ($this->isNeedToSnapshot($table, $aggregate->getVersion())) {
58			if ($this->has($aggregate->getId(), $aggregate->getVersion(), $table)) {
59			throw new LogicException(
60			sprintf(
61			'Snapshot for "%s" with id "%s" and version "%d" already exist.',
62			$table,
63			(string) $aggregate->getId(),
64			$aggregate->getVersion()
65			)
66			);
67			}
68
69			$record = $this->serialize($aggregate);
70			$this->watch($table, $record);
71
72			$bulk = new BulkWrite;
73			$record['_id'] = $bulk->insert($record);
74			$this->conn->executeBulkWrite($this->getNamespace($table), $bulk);
75			}
76			}
77
78			/**
79			* {@inheritdoc}
80			*/
81			public function has(IdentifierInterface $id, $version, $table)
82			{
83			return 0 !== $this->count($table, array('uuid' => $id, 'version' => $version));
84			}
85
86			/**
87			* {@inheritdoc}
88			*/
89			public function find(IdentifierInterface $id, $class, $table)
90			{
91			$query = new Query(
92			array('uuid' => (string) $id),
93			array('sort' => array('version' => -1), 'limit' => 1)
94			);
95			$cursor = $this->conn->executeQuery($this->getNamespace($table), $query);
96			$cursor->setTypeMap(array('document' => 'array'));
97			$results = $cursor->toArray();
98
99			if (count($results)) {
100			$results = current($results);
101
102			return $this->deserialize($class, (array) $results);
103			}
104
105			return null;
106			}
107
108			/**
109			* {@inheritdoc}
110			*/
111			public function drop($table)
112			{
113			$bulk = new BulkWrite;
114			$bulk->delete(array());
115			$this->conn->executeBulkWrite($this->getNamespace($table), $bulk);
116			}
117
118			/**
119			* {@inheritdoc}
120			*/
121			public function getSnapshotPeriodical()
122			{
123			return $this->snapshotPeriodical;
124			}
125
126			/**
127			* {@inheritdoc}
128			*/
129			public function setSnapshotPeriodical($snapshotPeriodical)
130			{
131			$this->snapshotPeriodical = $snapshotPeriodical;
132			}
133
134			/**
135			* {@inheritdoc}
136			*/
137			protected function performCommit($table, array $records)
			0 ignored issues – show Unused Code introduced 2016-08-22 19:41 UTC by Report Bug Copy Issue Report The parameter `$table` is not used and could be removed. This check looks from parameters that have been defined for a function or method, but which are not used in the method body. Loading history... Unused Code introduced 2016-08-22 19:41 UTC by Report Bug Copy Issue Report The parameter `$records` is not used and could be removed. This check looks from parameters that have been defined for a function or method, but which are not used in the method body. Loading history...
138			{
139			// Do nothing (handle by the trait)
140			}
141
142			/**
143			* {@inheritdoc}
144			*/
145			protected function performRollback($table, array $records)
146			{
147			$bulk = new BulkWrite;
148			foreach ($records as $record) {
149			$id = $record['_id'];
150			unset($record['_id']);
151			$bulk->delete(array('_id' => $id));
152			}
153
154			$this->conn->executeBulkWrite($this->getNamespace($table), $bulk);
155			}
156
157			/**
158			* @param string $table
159			* @param int $version
160			*
161			* @return bool
162			*/
163			private function isNeedToSnapshot($table, $version)
164			{
165			return $version >= $this->snapshotPeriodical
166			&& (0 === $version % $this->snapshotPeriodical
167			\|\| $this->isEmpty($table));
168			}
169
170			/**
171			* @param string $table
172			*
173			* @return bool
174			*/
175			private function isEmpty($table)
176			{
177			return 0 === $this->count($table);
178			}
179
180			/**
181			* Count collection.
182			*
183			* @param string $table
184			* @param array $query
185			*
186			* @return int
187			*/
188			private function count($table, array $query = array())
189			{
190			$command = new Command(array('count' => $table, 'query' => $query));
191			$cursor = $this->conn->executeCommand($this->db, $command);
192			$results = (array) current($cursor->toArray());
193
194			return (int) $results['n'];
195			}
196
197			/**
198			* Serialize aggregate root with metadata.
199			*
200			* @param AggregateRootInterface $aggregate
201			*
202			* @return array
203			*/
204			private function serialize(AggregateRootInterface $aggregate)
205			{
206			return [
207			'uuid' => (string) $aggregate->getId(),
208			'version' => $aggregate->getVersion(),
209			'created' => (new DateTime())->format('Y-m-d\TH:i:s.uP'),
210			'payload' => array('class' => get_class($aggregate), 'body' => $aggregate->serialize()),
211			];
212			}
213
214			/**
215			* Deserialize from array to aggregate root.
216			*
217			* @param string $class
218			* @param array $serialized
219			*
220			* @return AggregateRootInterface
221			*/
222			private function deserialize($class, array $serialized)
223			{
224			return $class::{'deserialize'}(
225			array_merge(
226			$serialized['payload']['body'],
227			array(
228			'version' => $serialized['version'],
229			)
230			)
231			);
232			}
233
234			/**
235			* @param string $table
236			*
237			* @return string
238			*/
239			private function getNamespace($table)
240			{
241			return sprintf('%s.%s', $this->db, $table);
242			}
243			}
244

borobudur-php / borobudur-es

Push — master ( 8e5e55...6fad72 )

MongoSnapshot::__construct() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like