<commit_before><?php
/**
 * This file is part of the prooph/pdo-event-store.
 * (c) 2016-2016 prooph software GmbH <<EMAIL>>
 * (c) 2016-2016 <NAME> <<EMAIL>>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

declare(strict_types=1);

namespace Prooph\EventStore\PDO;

use Iterator;
use PDO;
use PDOStatement;
use Prooph\Common\Messaging\Message;
use Prooph\Common\Messaging\MessageFactory;

final class PDOStreamIterator implements Iterator
{
    /**
     * @var PDO
     */
    private $connection;

    /**
     * @var PDOStatement
     */
    private $statement;

    /**
     * @var array
     */
    private $sql;

    /**
     * @var MessageFactory
     */
    private $messageFactory;

    /**
     * @var array|false
     */
    private $currentItem = null;

    /**
     * @var int
     */
    private $currentKey = -1;

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

    private $batchSize;

    /**
     * @var int
     */
    private $fromNumber;

    /**
     * @var int|null
     */
    private $count;

    /**
     * @var bool
     */
    private $forward;

    public function __construct(
        PDO $connection,
        PDOStatement $statement,
        MessageFactory $messageFactory,
        array $sql,
        ?int $batchSize,
        int $fromNumber,
        int $count,
        bool $forward
    ) {
        $this->connection = $connection;
        $this->statement = $statement;
        $this->messageFactory = $messageFactory;
        $this->sql = $sql;
        $this->batchSize = $batchSize;
        $this->fromNumber = $fromNumber;
        $this->count = $count;
        $this->forward = $forward;

        $this->next();
    }

    /**
     * @return null|Message
     */
    public function current()
    {
        if (false === $this->currentItem) {
            return;
        }

        $createdAt = \DateTimeImmutable::createFromFormat(
            'Y-m-d\TH:i:s.u',
            $this->currentItem->created_at,
            new \DateTimeZone('UTC')
        );

        return $this->messageFactory->createMessageFromArray($this->currentItem->event_name, [
            'uuid' => $this->currentItem->event_id,
            'created_at' => $createdAt,
            'payload' => json_decode($this->currentItem->payload, true),
            'metadata' => json_decode($this->currentItem->metadata, true),
        ]);
    }

    /**
     * Next
     */
    public function next()
    {
        if (($this->count - 1) === $this->currentKey) {
            $this->currentKey = -1;
            $this->currentItem = false;

            return;
        }

        $this->currentItem = $this->statement->fetch();

        if (false !== $this->currentItem) {
            $this->currentKey++;
        } else {
            $this->batchPosition++;
            if ($this->forward) {
                $limit = $this->batchSize * $this->batchPosition + $this->fromNumber;
            } else {
                $limit = $this->fromNumber - $this->batchSize * $this->batchPosition;
            }
            $this->statement = $this->buildStatement($this->sql, $limit);
            $this->statement->execute();
            $this->statement->setFetchMode(PDO::FETCH_OBJ);

            $this->currentItem = $this->statement->fetch();

            if (false === $this->currentItem) {
                $this->currentKey = -1;
            } else {
                $this->currentKey++;
            }
        }
    }

    /**
     * @return bool|int
     */
    public function key()
    {
        if (-1 === $this->currentKey) {
            return false;
        }

        return $this->currentKey;
    }

    /**
     * @return bool
     */
    public function valid()
    {
        return false !== $this->currentItem;
    }

    /**
     * Rewind
     */
    public function rewind()
    {
        //Only perform rewind if current item is not the first element
        if ($this->currentKey !== 0) {
            $this->batchPosition = 0;
            $this->statement = $this->buildStatement($this->sql, $this->fromNumber);

            $this->statement->execute();
            $this->statement->setFetchMode(PDO::FETCH_OBJ);

            $this->currentItem = null;
            $this->currentKey = -1;

            $this->next();
        }
    }

    private function buildStatement(array $sql, int $fromNumber): PDOStatement
    {
        $limit = $this->count < ($this->batchSize * ($this->batchPosition + 1))
            ? $this->count - ($this->batchSize * $this->batchPosition)
            : $this->batchSize;

        if ($this->forward) {
            $query = $sql['from'] . " WHERE no >= $fromNumber";<commit_msg>add return type to PDO Stream Iterator<commit_after><?php
/**
 * This file is part of the prooph/pdo-event-store.
 * (c) 2016-2016 prooph software GmbH <<EMAIL>>
 * (c) 2016-2016 <NAME> <<EMAIL>>
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

declare(strict_types=1);

namespace Prooph\EventStore\PDO;

use Iterator;
use PDO;
use PDOStatement;
use Prooph\Common\Messaging\Message;
use Prooph\Common\Messaging\MessageFactory;

final class PDOStreamIterator implements Iterator
{
    /**
     * @var PDO
     */
    private $connection;

    /**
     * @var PDOStatement
     */
    private $statement;

    /**
     * @var array
     */
    private $sql;

    /**
     * @var MessageFactory
     */
    private $messageFactory;

    /**
     * @var array|false
     */
    private $currentItem = null;

    /**
     * @var int
     */
    private $currentKey = -1;

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

    private $batchSize;

    /**
     * @var int
     */
    private $fromNumber;

    /**
     * @var int|null
     */
    private $count;

    /**
     * @var bool
     */
    private $forward;

    public function __construct(
        PDO $connection,
        PDOStatement $statement,
        MessageFactory $messageFactory,
        array $sql,
        ?int $batchSize,
        int $fromNumber,
        int $count,
        bool $forward
    ) {
        $this->connection = $connection;
        $this->statement = $statement;
        $this->messageFactory = $messageFactory;
        $this->sql = $sql;
        $this->batchSize = $batchSize;
        $this->fromNumber = $fromNumber;
        $this->count = $count;
        $this->forward = $forward;

        $this->next();
    }

    /**
     * @return null|Message
     */
    public function current(): ?Message
    {
        if (false === $this->currentItem) {
            return null;
        }

        $createdAt = \DateTimeImmutable::createFromFormat(
            'Y-m-d\TH:i:s.u',
            $this->currentItem->created_at,
            new \DateTimeZone('UTC')
        );

        return $this->messageFactory->createMessageFromArray($this->currentItem->event_name, [
            'uuid' => $this->currentItem->event_id,
            'created_at' => $createdAt,
            'payload' => json_decode($this->currentItem->payload, true),
            'metadata' => json_decode($this->currentItem->metadata, true),
        ]);
    }

    public function next(): void
    {
        if (($this->count - 1) === $this->currentKey) {
            $this->currentKey = -1;
            $this->currentItem = false;

            return;
        }

        $this->currentItem = $this->statement->fetch();

        if (false !== $this->currentItem) {
            $this->currentKey++;
        } else {
            $this->batchPosition++;
            if ($this->forward) {
                $limit = $this->batchSize * $this->batchPosition + $this->fromNumber;
            } else {
                $limit = $this->fromNumber - $this->batchSize * $this->batchPosition;
            }
            $this->statement = $this->buildStatement($this->sql, $limit);
            $this->statement->execute();
            $this->statement->setFetchMode(PDO::FETCH_OBJ);

            $this->currentItem = $this->statement->fetch();

            if (false === $this->currentItem) {
                $this->currentKey = -1;
            } else {
                $this->currentKey++;
            }
        }
    }

    /**
     * @return bool|int
     */
    public function key()
    {
        if (-1 === $this->currentKey) {
            return false;
        }

        return $this->currentKey;
    }

    /**
     * @return bool
     */
    public function valid(): bool
    {
        return false !== $this->currentItem;
    }

    public function rewind(): void
    {
        //Only perform rewind if current item is not the first element
        if ($this->currentKey !== 0) {
            $this->batchPosition = 0;
            $this->statement = $this->buildStatement($this->sql, $this->fromNumber);

            $this->statement->execute();
            $this->statement->setFetchMode(PDO::FETCH_OBJ);

            $this->currentItem = null;
            $this->currentKey = -1;

            $this->next();
        }
    }

    private function buildStatement(array $sql, int $fromNumber): PDOStatement
    {
        $limit = $this->count < ($this->batchSize * ($this->batchPosition + 1))
            ? $this->count - ($this->batchSize * $this->batchPosition)
            : $this->batchSize;

        if ($this->forward) {
            $query = $sql['from'] . " WHERE no >= $fromNumber";
{'fraction_non_alphanumeric': 0.14882589380156547, 'fraction_numerical': 0.005711867992384176, 'mean_word_length': 1.777614571092832, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 175, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6686934', 'n_tokens_mistral': 2827, 'n_tokens_neox': 2705, 'n_words': 572}