Borobudur\EventSourcing\Storage\Mongo\MongoStorage

Complexity

Total Complexity

25

Size/Duplication

Total Lines	248
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	7

Importance

Changes	6
Bugs	1	Features	3

13 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	find()	0	21	3
A	append()	0	23	3
A	drop()	0	6	1
A	isTransactional()	0	4	1
A	getTransactionalLevel()	0	4	1
A	beginTransaction()	0	8	2
B	rollback()	0	22	4
A	commit()	0	12	3
A	serialize()	0	12	1
A	deserialize()	0	9	1
A	build()	0	20	2
A	assertKeyExists()	0	6	2

<?php
/*
 * This file is part of the Borobudur-Event-Sourcing package.
 *
 * (c) Hexacodelabs <http://hexacodelabs.com>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Borobudur\EventSourcing\Storage\Mongo;

use Borobudur\Cqrs\IdentifierInterface;
use Borobudur\Cqrs\Message\DomainEventInterface;
use Borobudur\EventSourcing\Domain\DomainEventStream;
use Borobudur\EventSourcing\Domain\DomainMessage;
use Borobudur\EventSourcing\Domain\DomainMessageInterface;
use Borobudur\EventSourcing\Exception\LogicException;
use Borobudur\EventSourcing\Storage\StorageInterface;
use MongoDB\Driver\BulkWrite;
use MongoDB\Driver\Manager;
use MongoDB\Driver\Query;

/**
 * @author      Iqbal Maulana <iq.bluejack@gmail.com>
 * @created     8/20/15
 */
class MongoStorage implements StorageInterface
{
    /**
     * @var Manager
     */
    private $conn;

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

    /**
     * @var array
     */
    private $watched = array();

    /**
     * @var bool
     */
    private $transactional = false;

    /**
     * @var int
     */
    private $transactionalLevel = 0;

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

    /**
     * {@inheritdoc}
     */
    public function find($table, IdentifierInterface $id)
    {
        $query = new Query(
            array('uuid' => $id->toString(), 'status' => 'OK'),
            array('sort' => array('version' => 1))
        );
        $cursor = $this->conn->executeQuery(sprintf('%s.%s', $this->db, $table), $query);
        $cursor->setTypeMap(array('document' => 'array'));
        $results = $cursor->toArray();
        $events = array();

        if (count($results)) {
            foreach ($results as $record) {
                $events[] = $this->deserialize((array) $record);
            }

            return new DomainEventStream($events);
        }

        throw new LogicException(sprintf('Aggregate root with id "%s" is not found.', (string) $id));
    }

    /**
     * {@inheritdoc}
     */
    public function append($table, DomainMessageInterface $message, $status, array $exception = null)
    {
        $serialized = $this->serialize($message);
        $record = array_merge(
            $serialized,
            array(
                'status'    => $status,
                'exception' => $exception,
            )
        );

        if (true === $this->transactional) {
            if (!isset($this->watched[$table])) {
                $this->watched[$table] = array();
            }

            $this->watched[$table][] = &$record;
        }

        $bulk = new BulkWrite;
        $record['_id'] = $bulk->insert($record);
        $this->conn->executeBulkWrite(sprintf('%s.%s', $this->db, $table), $bulk);
    }

    /**
     * {@inheritdoc}
     */
    public function drop($table)
    {
        $bulk = new BulkWrite;
        $bulk->delete(array());
        $this->conn->executeBulkWrite(sprintf('%s.%s', $this->db, $table), $bulk);
    }

    /**
     * @return boolean
     */
    public function isTransactional()
    {
        return $this->transactional;
    }

    /**
     * @return int
     */
    public function getTransactionalLevel()
    {
        return $this->transactionalLevel;
    }

    /**
     * Begin transaction
     */
    public function beginTransaction()
    {
        if (0 === $this->transactionalLevel) {
            $this->transactional = true;
        }

        $this->transactionalLevel += 1;
    }

    /**
     * Rollback transaction
     */
    public function rollback()
    {
        if (false === $this->transactional) {
            throw new LogicException('Cannot rollback without begin transaction.');
        }

        foreach ($this->watched as $table => $records) {
            $bulk = new BulkWrite;
            foreach ($records as $record) {
                $id = $record['_id'];
                unset($record['_id']);
                $record['status'] = 'ROLLBACK';
                $bulk->update(array('_id' => $id), array('$set' => $record));
            }

            $this->conn->executeBulkWrite(sprintf('%s.%s', $this->db, $table), $bulk);
        }

        $this->transactionalLevel = 0;
        $this->transactional = false;
        $this->watched = array();
    }

    /**
     * Commit transaction.
     */
    public function commit()
    {
        if (false === $this->transactional) {
            throw new LogicException('Cannot commit without begin transaction.');
        }

        $this->transactionalLevel -= 1;
        if (0 === $this->transactionalLevel) {
            $this->transactional = false;
            $this->watched = array();
        }
    }

    /**
     * Serialize domain message.
     *
     * @param DomainMessageInterface $domain
     *
     * @return array
     */
    private function serialize(DomainMessageInterface $domain)
    {
        $payload = $domain->getPayload();

        return array(
            'uuid'    => (string) $domain->getId(),
            'version' => $domain->getVersion(),
            'type'    => $domain->getType(),
            'created' => $domain->getCreated()->format('Y-m-d\TH:i:s.uP'),
            'payload' => array('class' => get_class($payload), 'body' => $payload->serialize()),
        );
    }

    /**
     * Deserialize data to domain message.
     *
     * @param array $record
     *
     * @return DomainMessage
     */
    private function deserialize(array $record)
    {
        return new DomainMessage(
            $record['uuid'],
            $record['version'],
            new \DateTime($record['created']),
            $this->build((array) $record['payload'])
        );
    }

    /**
     * Build an object from serialized data
     *
     * @param array $serializedObject
     *
     * @return DomainEventInterface
     */
    private function build(array $serializedObject)
    {
        $this->assertKeyExists($serializedObject, 'class');
        $this->assertKeyExists($serializedObject, 'body');

        if (!in_array(
            'Borobudur\Serialization\DeserializableInterface',
            class_implements($serializedObject['class'])
        )
        ) {
            throw new \RuntimeException(
                sprintf(
                    'Class "%s" does not implement "\Borobudur\Serialization\DeserializableInterface"',
                    $serializedObject['class']
                )
            );
        }

        return $serializedObject['class']::{'deserialize'}((array) $serializedObject['body']);
    }

    /**
     * Assert array key
     *
     * @param array  $serializeObject
     * @param string $key
     */
    private function assertKeyExists(array $serializeObject, $key)
    {
        if (!array_key_exists($key, $serializeObject)) {
            throw new \RuntimeException(sprintf('Key "%s" should be set', $key));
        }
    }
}