public function getNextRecords($count, $documentType = null, $shopId = null)
    {
        $count = (int)$count;
        if ($count === 0) {
            return [];
        }

        $connection = $this->getConnection();
        $connection->beginTransaction();

        $tableName = $connection->quoteIdentifier($this->getTableName($shopId));

        $baseParams = [
            ['limit', $count, \PDO::PARAM_INT],
        ];

        $documentTypeCondition = '';
        if (!empty($documentType) && is_string($documentType)) {
            $documentTypeCondition = ' AND `document_type` = :documentType';
            $baseParams[] = ['documentType', $documentType, \PDO::PARAM_STR];
        }

        // Select records for update.
        $sqlSelectForUpdate = sprintf(
            "SELECT *, :shopId AS `shop_id` FROM {$tableName}
            WHERE
                `status` = :status %s
            ORDER BY `timestamp` ASC, `id` ASC
            LIMIT :limit
            FOR UPDATE",
            $documentTypeCondition
        );

        $params = [
            ['shopId', $shopId, \PDO::PARAM_INT],
            ['status', self::STATUS_NEW, \PDO::PARAM_INT],
        ];

        $statement = $connection->prepare($sqlSelectForUpdate);
        $this->bindParams($statement, array_merge_recursive($params, $baseParams));
        $statement->execute();
        $nextRecords = $statement->fetchAll();

        // Update status.
        $sqlUpdate = sprintf(
            "UPDATE {$tableName}
            SET `status` = :toStatus
            WHERE
                `status` = :fromStatus %s
            ORDER BY `timestamp` ASC, `id` ASC
            LIMIT :limit",
            $documentTypeCondition
        );

        $params = [
            ['fromStatus', self::STATUS_NEW, \PDO::PARAM_INT],
            ['toStatus', self::STATUS_IN_PROGRESS, \PDO::PARAM_INT],
        ];

        $statement = $connection->prepare($sqlUpdate);
        $this->bindParams($statement, array_merge_recursive($params, $baseParams));
        $statement->execute();
        $connection->commit();

        return $nextRecords;
    }
public function getNextRecords($count, $documentType = null, $shopId = null)
    {
        $count = (int)$count;
        if ($count === 0) {
            return [];
        }

        $connection = $this->getConnection();
        $connection->beginTransaction();

        $tableName = $connection->quoteIdentifier($this->getTableName($shopId));

        $baseParams = [
            ['limit', $count, \PDO::PARAM_INT],
        ];

        $documentTypeCondition = '';
        if (!empty($documentType) && is_string($documentType)) {
            $documentTypeCondition = ' AND `document_type` = :documentType';
            $baseParams[] = ['documentType', $documentType, \PDO::PARAM_STR];
        }

        // Select records for update.
        $sqlSelectForUpdate = sprintf(
            "SELECT *, :shopId AS `shop_id` FROM {$tableName}
            WHERE
                `status` = :status %s
            ORDER BY `timestamp` ASC, `id` ASC
            LIMIT :limit
            FOR UPDATE",
            $documentTypeCondition
        );

        $params = [
            ['shopId', $shopId, \PDO::PARAM_INT],
            ['status', self::STATUS_NEW, \PDO::PARAM_INT],
        ];

        $statement = $connection->prepare($sqlSelectForUpdate);
        $this->bindParams($statement, array_merge_recursive($params, $baseParams));
        $statement->execute();
        $nextRecords = $statement->fetchAll();

        // Update status.
        $sqlUpdate = sprintf(
            "UPDATE {$tableName}
            SET `status` = :toStatus
            WHERE
                `status` = :fromStatus %s
            ORDER BY `timestamp` ASC, `id` ASC
            LIMIT :limit",
            $documentTypeCondition
        );

        $params = [
            ['fromStatus', self::STATUS_NEW, \PDO::PARAM_INT],
            ['toStatus', self::STATUS_IN_PROGRESS, \PDO::PARAM_INT],
        ];

        $statement = $connection->prepare($sqlUpdate);
        $this->bindParams($statement, array_merge_recursive($params, $baseParams));
        $statement->execute();
        $connection->commit();

        return $nextRecords;
    }
{@inheritdoc}