Roll-out all events to reconstitute the state of the aggregate.

@param EventStream $eventStream

@return static
193999
final public static function loadFromHistory(EventStream $eventStream)
    {
        $aggregate = new static();

        while ($eventStream->hasNext()) {
            $aggregate->replayChange($eventStream->next());
        }

        return $aggregate;
    }
913601
public function reconstituteFrom(AggregateRootEventList $history)
    {
        if (!$this->history->isEmpty()) {
            throw new ReconstitutionError('Trying to reconstitute history on an already initialized aggregate-root.');
        }

        $first = true;
        foreach ($history as $past_event) {
            if ($first) {
                $first = false;
                if (!$past_event instanceof AggregateRootCreatedEvent) {
                    throw new ReconstitutionError(
                        sprintf(
                            'The first event given within a history to reconstitute from must be by the type of "%s".' .
                            ' Instead "%s" was given for AR %s.',
                            AggregateRootCreatedEvent::CLASS,
                            get_class($past_event),
                            $past_event->getAggregateRootIdentifier()
                        )
                    );
                }
            }

            $this->history->push($this->applyEvent($past_event, false));
        }

        return $this->isValid();
    }
608616
public function serialize(DomainEventStream $domainEventStream)
    {
        $serializedDomainMessages = [];
        foreach ($domainEventStream as $domainMessage) {
            $serializedDomainMessages[] = $this->serializer->serialize($domainMessage);
        }

        return json_encode($serializedDomainMessages);
    }
294344
private function appendToEventsForStream(&$eventsForStream, $serializedEvents)
    {
        // TODO: remove support of non array values by upcasters in next major release
        if (!is_array($serializedEvents)) {
            @trigger_error(
                'Upcasters need to return an array collection of upcasted events, ' .
                'non array return values are deprecated and support will be removed in the next major release',
                E_USER_DEPRECATED
            );

            $serializedEvents = [$serializedEvents];
        }

        foreach ($serializedEvents as $serializedEvent) {
            $eventsForStream[] = $this->eventFactory->createFromSerializedEvent($serializedEvent);
        }
    }
1213094
protected function limitPartStream(Stream $stream)
    {
        // Limit what is read from the stream to the part size.
        return new Psr7\LimitStream(
            $stream,
            $this->state->getPartSize(),
            $this->source->tell()
        );
    }
977279
public function createSuccessFromEventCollection(
        EventCollection $events
    ): EventCollection {
        Assert::assertThat(true, Assert::isTrue());

        return $events;
    }
174122
public function rollback()
    {
        parent::rollback();
        foreach ($this->getOriginalAll() as $collection) {
            if ($collection instanceof Collection) {
                $collection->rollback();
            }
        }
        return $this;
    }
798289
public function writeDouble($stream)
    {
        $stream = pack('d', $stream);
        if (!$this->_bigEndian) {
            $stream = strrev($stream);
        }
        $this->_stream.= $stream;
        return $this;
    }
334083
public static function buildEventFromEventStreamObject(EventStreamObject $eventStreamObject): EventInterface
    {
        /** @var EventInterface $eventClass */
        $eventClass = $eventStreamObject->getEvent();
        /** @var \RevisionTen\CQRS\Interfaces\CommandInterface $commandClass */
        $commandClass = $eventClass::getCommandClass();

        $user = $eventStreamObject->getUser();
        $commandUuid = $eventStreamObject->getCommandUuid();
        $uuid = $eventStreamObject->getUuid();
        // Its safe to assume the command was created on the previous version, it would have failed otherwise.
        $onVersion = $eventStreamObject->getVersion() - 1;
        $payload = $eventStreamObject->getPayload();

        /** @var \RevisionTen\CQRS\Interfaces\CommandInterface $command */
        $command = new $commandClass($user, $commandUuid, $uuid, $onVersion, $payload);

        return new $eventClass($command);
    }
1071465
public function useGlobally()
    {
        if (!isset($this->handle)) {
            throw new \RuntimeException("The stream is closed");
        }
        
        if ($this->isGlobal()) {
            return $this;
        }
        
        $this->assertOutputBuffering();
        
        $output = $this->createOutputStream();
        if ($output === false) {
            throw new \RuntimeException("Failed to create temp stream");
        }
        
        $this->obClean();
        $this->rewind();
        stream_copy_to_stream($this->handle, $output);
        
        parent::close();
        $this->handle = $output;
        
        return $this;
    }
285247
public function persist(array $events)
    {
        $factory = new EventFactory();
        $events = array_map(
            function ($event) use ($factory) {
                return is_array($event) ? $factory->create($event) : $event;
            },
            $events
        );
        $this->persister()->logEvents($events);
    }
806625
public static function fromStream(StreamInterface $stream)
    {
        if (! $stream->isReadable() || ! $stream->isSeekable()) {
            throw new InvalidArgumentException('Message stream must be both readable and seekable');
        }

        $stream->rewind();

        list($version, $status, $reasonPhrase) = self::getStatusLine($stream);
        list($headers, $body)                  = self::splitStream($stream);

        return (new Response($body, $status, $headers))
            ->withProtocolVersion($version)
            ->withStatus($status, $reasonPhrase);
    }
736649
public function loadFromHistory(AggregateInterface $aggregate, array $eventStreamObjects): AggregateInterface
    {
        // Arrays start at zero ;)
        $count = \count($eventStreamObjects) - 1;

        /**
         * Get events and replay them.
         *
         * @var EventStreamObject $eventStreamObject
         */
        foreach ($eventStreamObjects as $key => $eventStreamObject) {
            $uuid = $eventStreamObject->getUuid();

            // Check if the uuid of the provided Event Stream Object matches.
            if ($uuid !== $aggregate->getUuid()) {
                continue;
            }

            if (0 === $key && null === $aggregate->getCreated()) {
                // Set Created from first EventStreamObject.
                $aggregate->setCreated($eventStreamObject->getCreated());
            }

            if ($key === $count) {
                // Set Modified from last EventStreamObject.
                $aggregate->setModified($eventStreamObject->getCreated());
            }

            $event = EventStore::buildEventFromEventStreamObject($eventStreamObject);

            $aggregate = $this->apply($aggregate, $event);

            // Update the Aggregate history.
            $aggregate->addToHistory([
                'version' => $eventStreamObject->getVersion(),
                'message' => $eventStreamObject->getMessage(),
                'created' => $eventStreamObject->getCreated(),
                'payload' => $eventStreamObject->getPayload(),
            ]);
        }

        return $this->applyChanges($aggregate);
    }
923948
public function writeAsync(array $events = []): Promise
    {
        if ($this->isRolledBack) {
            throw new \RuntimeException('Cannot commit a rolledback transaction');
        }

        if ($this->isCommitted) {
            throw new \RuntimeException('Transaction is already committed');
        }

        return $this->connection->transactionalWriteAsync($this, $events, $this->userCredentials);
    }
248184
public function withDefaultThrowableEvent(string $eventClass): self
    {
        if (false === \is_a($eventClass, ExecutionFailedEvent::class, true))
        {
            throw new InvalidEventType(
                \sprintf(
                    'Event class "%s" must implement "%s" interface',
                    $eventClass,
                    ExecutionFailedEvent::class
                )
            );
        }

        $defaultValidationFailedEvent = $this->defaultValidationFailedEvent;

        /**
         * @psalm-var class-string $eventClass
         * @psalm-var class-string|null $defaultValidationFailedEvent
         */

        return new self(
            $this->isEventListener,
            $this->isCommandHandler,
            $this->validationEnabled,
            $this->validationGroups,
            $defaultValidationFailedEvent,
            $eventClass
        );
    }
961172
public function createEncryptingStream(EventStreamInterface $eventStream)
    {
        return new CryptoEventStream(
            $eventStream,
            $this->cryptos,
            $this->encryptionKey,
            CryptoEventStream::ENCRYPT
        );
    }
113724
public function withLocalScope()
    {
        if ($this->isClosed()) {
            throw new \RuntimeException("The stream is closed");
        }
        
        if (!$this->isGlobal()) {
            return $this;
        }
        
        $stream = clone $this;
        
        fwrite($stream->handle, (string)$this);
        
        return $stream;
    }