Borobudur\EventSourcing\Snapshot\MongoSnapshot

Complexity

Total Complexity

21

Size/Duplication

Total Lines	224
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	4

Importance

Changes	1
Bugs	0	Features	1

15 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	append()	0	22	3
A	has()	0	4	1
A	find()	0	18	2
A	drop()	0	6	1
A	getSnapshotPeriodical()	0	4	1
A	setSnapshotPeriodical()	0	4	1
A	performCommit()	0	4	1
A	performRollback()	0	11	2
A	isNeedToSnapshot()	0	6	3
A	isEmpty()	0	4	1
A	count()	0	8	1
A	serialize()	0	9	1
A	deserialize()	0	11	1
A	getNamespace()	0	4	1

<?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 <iq.bluejack@gmail.com>
 * @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)

The parameter $table is not used and could be removed.

The parameter $records is not used and could be removed.

    {
        // 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);
    }
}