000-RepositoryInterface.php

<?php

/*
 * This file is part of the broadway/broadway package.
 *
 * (c) Qandidate.com <opensource@qandidate.com>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Broadway\Repository;

use Broadway\Domain\AggregateRoot;

/**
 * Repository for aggregate roots.
 */
interface RepositoryInterface
{
    public function add(AggregateRoot $aggregate);

    /**
     * Loads an aggregate from the given id.
     *
     * @param mixed $id
     *
     * @throws AggregateNotFoundException
     */
    public function load($id);
    
    public function getIdForAggregateRoot(AggregateRoot $aggregateRoot);
}

001-EventSourcingRepository.php

<?php

//
// Default implementation can use BC behaviour.
//

/*
 * This file is part of the broadway/broadway package.
 *
 * (c) Qandidate.com <opensource@qandidate.com>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Broadway\EventSourcing;

use Assert\Assertion as Assert;
use Assert\InvalidArgumentException;
use Broadway\Domain\AggregateRoot;
use Broadway\Domain\DomainEventStream;
use Broadway\EventHandling\EventBusInterface;
use Broadway\EventStore\EventStoreInterface;
use Broadway\EventStore\EventStreamNotFoundException;
use Broadway\Repository\AggregateNotFoundException;
use Broadway\Repository\RepositoryInterface;

/**
 * Naive initial implementation of an event sourced aggregate repository.
 */
class EventSourcingRepository implements RepositoryInterface
{
    private $eventStore;
    private $eventBus;
    private $aggregateClass;

    /**
     * @param string $aggregateClass
     */
    public function __construct(
        EventStoreInterface $eventStore,
        EventBusInterface $eventBus,
        $aggregateClass,
        array $eventStreamDecorators = array()
    ) {
        $this->assertExtendsEventSourcedAggregateRoot($aggregateClass);

        $this->eventStore            = $eventStore;
        $this->eventBus              = $eventBus;
        $this->aggregateClass        = $aggregateClass; // todo: aggregate factory
        $this->eventStreamDecorators = $eventStreamDecorators;
    }

    /**
     * {@inheritDoc}
     */
    public function load($id)
    {
        try {
            $domainEventStream = $this->eventStore->load($id);

            $aggregate = new $this->aggregateClass();
            $aggregate->initializeState($domainEventStream);

            return $aggregate;
        } catch (EventStreamNotFoundException $e) {
            throw AggregateNotFoundException::create($id, $e);
        }
    }

    public function add(AggregateRoot $aggregate)
    {
        // maybe we can get generics one day.... ;)
        Assert::isInstanceOf($aggregate, $this->aggregateClass);

        $domainEventStream = $aggregate->getUncommittedEvents();
        $eventStream       = $this->decorateForWrite($aggregate, $domainEventStream);
        $this->eventStore->append($aggregate->getId(), $eventStream);
        $this->eventBus->publish($domainEventStream);
    }

    private function decorateForWrite(AggregateRoot $aggregate, DomainEventStream $eventStream)
    {
        $aggregateType       = $this->getType();
        $aggregateIdentifier = $aggregate->getId();

        foreach ($this->eventStreamDecorators as $eventStreamDecorator) {
            $eventStream = $eventStreamDecorator->decorateForWrite($aggregateType, $aggregateIdentifier, $eventStream);
        }

        return $eventStream;
    }

    // todo: move to assert lib?
    private function assertExtendsEventSourcedAggregateRoot($class)
    {
        $parents = class_parents($class);

        if (! in_array('Broadway\EventSourcing\EventSourcedAggregateRoot', $parents)) {
            throw new InvalidArgumentException(sprintf("Class '%s' is not an EventSourcedAggregateRoot.", $class), -1);
        }
    }

    private function getType()
    {
        return $this->aggregateClass;
    }

    
    public function getIdForAggregateRoot(AggregateRoot $aggregateRoot)
    {
        if (method_exists(array($aggregateRoot, 'getId'))) {
            // Not sure if we need this guard or not?
            return $aggregateRoot->getId();
        }
        
        throw new Runtime Exception("something descriptive here...");
    }

}

002-UserRepository.php

<?php

//
// Implement one new method here if I want to use a non-standard
// ID location.
//

namespace Boogio\IdentityAccess\Port\Adapter\Identity\Persistence\Repository\EventSourcing;

use Boogio\IdentityAccess\Domain\Model\Identity\User;
use Boogio\IdentityAccess\Domain\Model\Identity\UserId;
use Boogio\IdentityAccess\Domain\Model\Identity\UserRepository;
use Broadway\EventHandling\EventBusInterface;
use Broadway\EventSourcing\EventSourcingRepository;
use Broadway\EventStore\EventStoreInterface;

class EventSourcingUserRepository extends EventSourcingRepository
{
    public function __construct()
    {
        parent::__construct($eventStore, $eventBus, User::class);
    }

    public function getIdForAggregateRoot(AggregateRoot $aggregateRoot)
    {
        return (string) $aggregateRoot->getUserId();
    }
}

003-User.php

<?php

//
// No longer need getId method, I can use getUserId and return
// a UserId instance as I want.
//

namespace Boogio\IdentityAccess\Domain\Model\Identity;

use Boogio\IdentityAccess\Domain\Model\Identity\Event\UserCredentialsWereUpdated;
use Boogio\IdentityAccess\Domain\Model\Identity\Event\UserWasDeactivated;
use Boogio\IdentityAccess\Domain\Model\Identity\Event\UserWasReactivated;
use Boogio\IdentityAccess\Domain\Model\Identity\Event\UserWasRegistered;
use Boogio\IdentityAccegss\Domain\Model\Identity\Event\UserWasSuspended;
use Boogio\IdentityAccess\Domain\Model\Identity\Event\UserWasUnsuspended;
use Boogio\IdentityAccess\Domain\Model\Identity\Event\UserPersonalContactInformationWasChanged;
use Boogio\IdentityAccess\Domain\Model\Identity\Event\UserPersonalFullNameWasChanged;
use Broadway\EventSourcing\EventSourcedAggregateRoot;

class User extends EventSourcedAggregateRoot
{
    /**
     * @var UserId
     */
    private $userId;

    /**
     * @var Person
     */
    private $person;

    /**
     * @var Credentials
     */
    private $credentials;

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

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

    public function __construct() { }

    /**
     * @return UserId
     */
    public function getUserId()
    {
        return $this->userId;
    }

    public static function register(Person $person, Credentials $credentials)
    {
        return static::registerWithId(
            UserId::generate(),
            $person,
            $credentials
        );
    }

    public function suspend()
    {
        if ($this->isSuspended) {
            return;
        }

        $this->apply(UserWasSuspended::with($this->userId));
    }

    public function unsuspend()
    {
        if (! $this->isSuspended) {
            return;
        }

        $this->apply(UserWasUnsuspended::with($this->userId));
    }

    public function reactivate()
    {
        if (! $this->isDeactivated) {
            return;
        }

        $this->apply(UserWasReactivated::with($this->userId));
    }

    public function deactivate()
    {
        if ($this->isDeactivated) {
            return;
        }

        $this->apply(UserWasDeactivated::with($this->userId));
    }

    public function changePersonalContactInformation(ContactInformation $contactInformation)
    {
        $this->apply(UserPersonalContactInformationWasChanged::with($this->userId, $contactInformation));
    }

    public function changePersonalFullName(FullName $fullName)
    {
        $this->apply(UserPersonalFullNameWasChanged::with($this->userId, $fullName));
    }

    public static function registerWithId(UserId $userId, Person $person, Credentials $credentials)
    {
        $instance = new static();

        $instance->apply(UserWasRegistered::with($userId, $person));
        $instance->apply(UserCredentialsWereUpdated::with($userId, $credentials));

        return $instance;
    }

    protected function applyUserWasRegistered(UserWasRegistered $event)
    {
        $this->userId = $event->getUserId();
        $this->person = $event->getPerson();
    }

    protected function applyUserCredentialsWereUpdated(UserCredentialsWereUpdated $event)
    {
        $this->credentials = $event->getCredentials();
    }

    protected function applyUserWasSuspended(UserWasSuspended $event)
    {
        $this->isSuspended = true;
    }

    protected function applyUserWasUnsuspended(UserWasUnsuspended $event)
    {
        $this->isSuspended = false;
    }

    protected function applyUserWasDeactivated(UserWasDeactivated $event)
    {
        $this->isDeactivated = true;
    }

    protected function applyUserWasReactivated(UserWasReactivated $event)
    {
        $this->isDeactivated = false;
    }
}