@param $batchLimit
@param $batchCount
@param $params
@param $output

@return int
161573
private function doSourceReprocess($batchLimit, $batchCount, $params, $output)
    {
        $timeStart = microtime(true);

        do {
            // 1) Get MAX(date_added) records where reprocess_flag = 1
            $batchStart = microtime(true);
            $params     = $this->getDateParams($params, 'MauticContactLedgerBundle:CampaignSourceStats');

            if ('invalid' == $params['dateFrom']) {
                $output->writeln(
                    '<comment>Exiting without Running. Reprocess Cron has re-written all history for Source (no more reprocessFlag = true). </comment>'
                );
                $batchCount = $batchLimit;
            } else {
                $output->writeln(
                    '<comment>Batch Index: '.(int) $batchCount."  --> Using Parameters:\n \tDate => ".$params['dateFrom'].' and '.$params['dateTo']." UTC,\n \tQuery Cache Directory => ".$params['cacheDir'].'</comment>'
                );

                // 2) Get entities that match date_added
                $entitiesToReprocess = $this->getEntitiesToReprocess($params, 'MauticContactLedgerBundle:CampaignSourceStats');

                // 3) reprocess data using this new date criteria
                $statData = $this->getCampaignSourceStatsData($params);

                foreach ($statData as $stat) {
                    $entity = $this->mapSourceArrayToEntity($stat, $params['dateTo']);
                    $entity->setReprocessFlag(false);
                    $this->em->persist($entity);
                }

                // 4) purge records from step 2
                foreach ($entitiesToReprocess as $entityToDelete) {
                    $this->em->remove($entityToDelete);
                }

                // 5) flush and repeat
                $this->em->flush();
                $timeContext = microtime(true);
                $contextTime = $timeContext - $timeStart;
                $batchRun    = $timeContext - $batchStart;
                $output->writeln(
                    "<comment>\t⌛ Batch Run Time: ".$batchRun.'  --> Total Elapsed time so far: '.$contextTime.'.</comment>'
                );
            }
            ++$batchCount;
        } while ($batchCount < $batchLimit);

        $this->completeRun();

        $output->writeln('<info>Complete With No Errors.</info>');
        $timeEnd     = microtime(true);
        $elapsedTime = $timeEnd - $timeStart;
        $output->writeln("<info>Total Execution Time: $elapsedTime.</info>");
    }
1200452
public function getTotalResult($start, $limit, $sortBy, $sortOrder, $params)
    {
        $results = $this->findAll($start, $limit, $sortBy, $sortOrder, $params);

        return count($results);
    }
145151
public static int getBatchUpdateCount(int[] batchExecuteResult){
		int result = 0;
		for (int c: batchExecuteResult){
			if (c >= 0){
				// real count
				result += c;
			}else if (c == -2){
				// unknown
				return -2;
			}else{
				// error or others
				// do nothing
			}
		}
		return result;
	}
717608
public function batch(AndroidPublisher_InappproductsBatchRequest $postBody, $optParams = array())
  {
    $params = array('postBody' => $postBody);
    $params = array_merge($params, $optParams);
    return $this->call('batch', array($params), "AndroidPublisher_InappproductsBatchResponse");
  }
6089
public function generateLimit($limitCount, $limitOffset, &$limit)
    {
        $limit = "";

        if (empty($limitCount)) {
            return;
        }

        $limit = "$limitCount";

        if (! empty($limitOffset)) {
            $limit .= ",$limitOffset";
        }
    }
1186216
private function rowLimiter(int $limit = 1000): Closure
    {
        return static function () use ($limit): void {
            static $n = 0;

            if (++$n > $limit) {
                $message = I18N::translate('The search returned too many results.');

                throw new InternalServerErrorException($message);
            }
        };
    }
15542
function SelectLimit($sql,$nrows=-1,$offset=-1,$inputarr=false,$secs2cache=0)
	{
		if ($nrows > 0) {
			if ($offset <= 0) $str = " ROWS $nrows ";
			else {
				$a = $offset+1;
				$b = $offset+$nrows;
				$str = " ROWS $a TO $b";
			}
		} else {
			// ok, skip
			$a = $offset + 1;
			$str = " ROWS $a TO 999999999"; // 999 million
		}
		$sql .= $str;

		return ($secs2cache) ?
				$this->CacheExecute($secs2cache,$sql,$inputarr)
			:
				$this->Execute($sql,$inputarr);
	}
1210690
protected function execute(InputInterface $input, OutputInterface $output) : int
    {
        $logger = LoggerHelper::createDefaultLogger($output);

        /** @var string $cacheOption */
        $cacheOption = $input->getOption('cache');
        $fileCache = new FilesystemCache($cacheOption);
        $cache = new SimpleCacheAdapter($fileCache);

        /** @var string $file */
        $file = $input->getArgument('file');
        if (! $file) {
            $file = $this->defaultIniFile;
        }

        $output->writeln(sprintf('write fetched file to %s', $file));

        $logger->info('started fetching remote file');

        $browscap = new BrowscapUpdater($cache, $logger);

        /** @var string $remoteFileOption */
        $remoteFileOption = $input->getOption('remote-file');

        try {
            $browscap->fetch($file, $remoteFileOption);
        } catch (ErrorCachedVersionException $e) {
            $logger->debug($e);

            return 3;
        } catch (FetcherException $e) {
            $logger->debug($e);

            return 9;
        } catch (Exception $e) {
            $logger->debug($e);

            return 10;
        }

        $logger->info('finished fetching remote file');

        return 0;
    }
15961
public function exec($statement, $params = array())
    {
		
        if (count($params) > 0){
        	$sth = $this->dbConnector->prepare($statement);
        	$this->preparedExec = true;
        	$this->lastPreparedExecStatement = $sth;
        	$sth->execute($params);
        	$returnValue = $sth->rowCount();
        }
        else{
        	$this->preparedExec = false;
        	try {
        	    $returnValue = $this->dbConnector->exec($statement);
        	} catch (PDOException $e) {
        	    common_Logger::w('Error in statement: '.$statement);
        	    throw $e;
        	}
        }

        return (int) $returnValue;
    }
1003935
@Override
    public int[] executeBatch() throws SQLException
    {
        checkClosed();
        closeCurrentResult();
        if (batch == null || batch.size() == 0) {
            return new int[0];
        }

        int[] updateCounts = new int[batch.size()];
        // keep a running total of update counts
        int runningUpdateCount = 0;

        int i = 0;
        try {
            for (; i < batch.size(); i++) {

                setCurrentResult(
                        null,
                        (int) batch.get(i).execute(
                                sourceConnection.NativeConnection,
                                this.m_timeout,
                                sourceConnection.queryTimeOutUnit)[0].fetchRow(
                                0).getLong(0));
                updateCounts[i] = this.lastUpdateCount;
                runningUpdateCount += this.lastUpdateCount;
            }
        } catch (SQLException x) {
            updateCounts[i] = EXECUTE_FAILED;
            throw new BatchUpdateException(Arrays.copyOf(updateCounts, i + 1),
                    x);
        } finally {
            clearBatch();
        }
        // replace the update count from the last statement with the update count
        // from the last batch.
        this.lastUpdateCount = runningUpdateCount;

        return updateCounts;
    }
844007
int calculateMaxProgress(MappingTarget mappingTarget) {
    int batches = mappingTarget.getEntityMappings().stream().mapToInt(this::countBatches).sum();
    if (mappingTarget.hasSelfReferences()) {
      batches *= 2;
    }
    return batches;
  }
440717
private static <T> void batchCount(T[] array, int start, int end, CountProcedure<T> castProcedure)
    {
        for (int i = start; i < end; i++)
        {
            castProcedure.value(array[i]);
        }
    }
1162981
public function batch(Closure $processor, int $batchSize = 1000, int $offsetStart = 0, ?int $offsetEnd = null): void
	{
		$this->limit($batchSize);

		while(true)
		{
			$query = clone $this;

			if($offsetEnd !== null && $offsetStart >= $offsetEnd)
			{
				break;
			}

			if($offsetStart !== 0)
			{
				$query->offset($offsetStart);
			}

			$results = $query->all();

			if(count($results) > 0)
			{
				$processor($results);

				$offsetStart += $batchSize;
			}
			else
			{
				break;
			}
		}
	}
440400
protected function migrateFiles(
        $totalFileCount = null,
        $bulkCount,
        $dryRun,
        OutputInterface $output
    ) {
        $progress = new ProgressBar($output, $totalFileCount);
        if ($totalFileCount) {
            $progress->setFormat("%message%\n %current%/%max% [%bar%] %percent:3s%% %elapsed:6s%/%estimated:-6s% %memory:6s%");
        } else {
            $progress->setFormat("%message%\n %current% [%bar%] %elapsed:6s% %memory:6s%");
        }

        $output->writeln('');
        $progress->setMessage('Starting migration...');
        $progress->start();

        $elapsedFileCount = 0;
        $timestamp = microtime(true);
        $updateFrequency = 1;

        foreach ($this->fileListers as $fileLister) {
            $pass = 0;
            do {
                $metadataList = $fileLister->loadMetadataList($bulkCount, $pass * $bulkCount);

                foreach ($metadataList as $metadata) {
                    if (!$dryRun) {
                        $this->fileMigrator->migrateFile($metadata);
                    }

                    $progress->setMessage('Updated file ' . $metadata->id);
                    $progress->advance();
                    ++$elapsedFileCount;

                    // Magic that ensures the progressbar is updated ca. once per second
                    if (($elapsedFileCount % $updateFrequency) === 0) {
                        $newTimestamp = microtime(true);
                        if ($newTimestamp - $timestamp > 0.5 && $updateFrequency > 1) {
                            $updateFrequency = (int)($updateFrequency / 2);
                            $progress->setRedrawFrequency($updateFrequency);
                        } elseif ($newTimestamp - $timestamp < 0.1 && $updateFrequency < 10000) {
                            $updateFrequency = $updateFrequency * 2;
                            $progress->setRedrawFrequency($updateFrequency);
                        }
                        $timestamp = $newTimestamp;
                    }
                }

                ++$pass;
            } while (count($metadataList) > 0);
        }

        $progress->setMessage('');
        $progress->finish();

        $output->writeln("\n\nFinished processing $elapsedFileCount files.");
        if ($totalFileCount && $totalFileCount > $elapsedFileCount) {
            $output->writeln([
                'Files that could not be migrated: ' . ($totalFileCount - $elapsedFileCount),
                '',
            ]);
        }
    }
1062663
public function numRows($result_object,$which_connection='')
    {
        if (!is_object($result_object)) {
            return false;
        }

        if ($which_connection == '') {
            $which_connection = $this->default_db;
        }

        return $result_object->RecordCount();
    }
260941
protected function execute(InputInterface $input, OutputInterface $output)
    {
        $errOutput = $output instanceof ConsoleOutputInterface ? $output->getErrorOutput() : $output;
        $output->setVerbosity(OutputInterface::VERBOSITY_QUIET);
        $errOutput->setVerbosity(OutputInterface::VERBOSITY_NORMAL);

        $batchSize = NumberUtils::bound($input->getOption('batch-size'), 1, 1000);
        $batchDelay = NumberUtils::bound($input->getOption('batch-delay'), 100, 600000);
        $since = $input->getOption('since');
        $until = $input->getOption('until');
        $context = json_decode($input->getOption('context') ?: '{}', true);
        $context['tenant_id'] = (string)$input->getOption('tenant-id');
        $streamId = $input->getArgument('stream-id') ? StreamId::fromString($input->getArgument('stream-id')) : null;

        if (!empty($since)) {
            $since = Microtime::fromString(str_pad($since, 16, '0'));
        }

        if (!empty($until)) {
            $until = Microtime::fromString(str_pad($until, 16, '0'));
        }

        $i = 0;
        $receiver = function (Event $event, StreamId $streamId) use ($errOutput, $batchSize, $batchDelay, &$i) {
            ++$i;

            try {
                echo json_encode($event) . PHP_EOL;
            } catch (\Throwable $e) {
                $errOutput->writeln($e->getMessage());
            }

            if (0 === $i % $batchSize) {
                if ($batchDelay > 0) {
                    usleep($batchDelay * 1000);
                }
            }
        };

        if ($streamId) {
            $this->getPbjx()->getEventStore()->pipeEvents($streamId, $receiver, $since, $until, $context);
        } else {
            $this->getPbjx()->getEventStore()->pipeAllEvents($receiver, $since, $until, $context);
        }
    }
1249584
protected function execute(InputInterface $input, OutputInterface $output) : int
    {
        /* @var ConsoleOutput $output */
        $output->setPrefix(ConsoleOutput::PREFIX_TIMESTAMP | ConsoleOutput::PREFIX_CLASSNAME);

        $this->queue = self::queue($input->getOption('queue'));

        if ($this->input->getOption('stop-when-empty') && !$this->queue->getStorage() instanceof CountableInterface) {
            throw new UserException("Can't use stop-when-empty on '" . $this->queue->getType() . "' storage");
        }

        return parent::execute($input, $output);
    }