Borobudur\EventSourcing\Storage\Mongo\MongoStorage

Complexity

Total Complexity

27

Size/Duplication

Total Lines	287
Duplicated Lines	0 %

Coupling/Cohesion

Components	2
Dependencies	9

Importance

Changes	9
Bugs	1	Features	4

18 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	6	1
A	attachSnapshot()	0	4	1
A	getSnapshot()	0	4	1
A	findEventFromBeginning()	0	12	2
A	findEventFromVersion()	0	9	1
A	append()	0	16	1
A	drop()	0	6	1
A	beginTransaction()	0	8	2
A	rollback()	0	8	2
A	commit()	0	8	2
A	performCommit()	0	4	1
A	performRollback()	0	12	2
A	serialize()	0	12	1
A	deserialize()	0	9	1
A	build()	0	20	2
A	assertKeyExists()	0	6	2
A	getNamespace()	0	4	1
A	query()	0	17	3

<?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\Snapshot\SnapshotInterface;
use Borobudur\EventSourcing\Storage\StorageInterface;
use Borobudur\EventSourcing\Storage\TransactionalTrait;
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
{
    use TransactionalTrait {
        TransactionalTrait::beginTransaction as protected parentBeginTransaction;
        TransactionalTrait::commit as protected parentCommit;
        TransactionalTrait::rollback as protected parentRollback;
    }

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

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

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

    /**
     * @param SnapshotInterface $snapshot
     */
    public function attachSnapshot(SnapshotInterface $snapshot)
    {
        $this->snapshot = $snapshot;
    }

    /**
     * @return SnapshotInterface
     */
    public function getSnapshot()
    {
        return $this->snapshot;
    }

    /**
     * {@inheritdoc}
     */
    public function findEventFromBeginning($table, IdentifierInterface $id)
    {
        $query = new Query(
            array('uuid' => $id->toString(), 'status' => 'OK'),
            array('sort' => array('version' => 1))
        );
        if (null !== $events = $this->query($query, $this->getNamespace($table))) {
            return $events;
        }

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

    /**
     * {@inheritdoc}
     */
    public function findEventFromVersion($table, IdentifierInterface $id, $version)
    {
        $query = new Query(
            array('uuid' => $id->toString(), 'status' => 'OK', 'version' => array('$gt' => $version)),
            array('sort' => array('version' => 1))
        );

        return $this->query($query, $this->getNamespace($table));
    }

    /**
     * {@inheritdoc}
     */
    public function append($table, DomainMessageInterface $message, $status, array $exception = null)
    {
        $serialized = $this->serialize($message);
        $record = array_merge(
            $serialized,
            array(
                'status'    => $status,
                'exception' => $exception,
            )
        );
        $this->watch($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($this->getNamespace($table), $bulk);
    }

    /**
     * {@inheritdoc}
     */
    public function beginTransaction()
    {
        if (null !== $this->snapshot) {
            $this->snapshot->beginTransaction();
        }

        $this->parentBeginTransaction();
    }

    /**
     * {@inheritdoc}
     */
    public function rollback()
    {
        if (null !== $this->snapshot) {
            $this->snapshot->rollback();
        }

        $this->parentRollback();
    }

    /**
     * {@inheritdoc}
     */
    public function commit()
    {
        if (null !== $this->snapshot) {
            $this->snapshot->commit();
        }

        $this->parentCommit();
    }

    /**
     * {@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']);
            $record['status'] = 'ROLLBACK';
            $bulk->update(array('_id' => $id), array('$set' => $record));
        }

        $this->conn->executeBulkWrite($this->getNamespace($table), $bulk);
    }

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

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

    /**
     * Execute mongodb query.
     *
     * @param Query  $query
     * @param string $namespace
     *
     * @return DomainEventStream|null
     */
    private function query(Query $query, $namespace)
    {
        $cursor = $this->conn->executeQuery($namespace, $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);
        }

        return null;
    }
}