addArgument('table', InputArgument::REQUIRED);
$this->addOption(
'database',
null,
InputOption::VALUE_OPTIONAL,
'The address of the database to read from',
'mysql://goteo:goteo@mariadb:3306/benzina'
);
$this->addOption(
'batch-size',
null,
InputOption::VALUE_OPTIONAL,
'The number of rows to process at once',
99
);
$this->addOption(
'skip-pumped',
null,
InputOption::VALUE_OPTIONAL,
'Skips feeding already pumped records in a batch',
false
);
$this->addUsage('app:benzina:pump --no-debug user');
$this->setHelp(
<<<'EOF'
The %command.name% processes the data in the database table and supplies it to the supporting pumps:
%command.full_name%
You can avoid possible memory leaks caused by the Symfony profiler with the no-debug flag:
%command.full_name% --no-debug
EOF
);
}
/**
* @param ConsoleOutputInterface $output
*/
protected function execute(InputInterface $input, OutputInterface $output): int
{
$batchSize = $input->getOption('batch-size');
$stream = new PdoStream(
$input->getOption('database'),
$input->getArgument('table'),
$batchSize
);
$streamSize = $stream->size();
$streamBatches = $streamSize / $batchSize;
$streamSection = new SymfonyStyle($input, $output->section());
if ($streamSize < 1) {
$streamSection->writeln('No data found at the given source. Skipping execution.');
return Command::SUCCESS;
}
$streamSection->writeln(sprintf('Streaming %d records (%d batches).', $streamSize, $streamBatches));
$pumps = $this->benzina->getPumpsFor($stream);
$pumpsCount = \count($pumps);
$pumpsSection = new SymfonyStyle($input, $output->section());
if ($pumpsCount < 1) {
$pumpsSection->writeln('No pumps support the streamed sample. Skipping execution.');
return Command::SUCCESS;
}
$pumpsSection->writeln(sprintf('Streaming to %d pumps.', $pumpsCount));
$pumpsSection->listing(\array_map(function (PumpInterface $pump) {
return $pump::class;
}, $pumps));
$streamedBatchesSection = $output->section();
$streamedBatchesSection->writeln("\tStreamed batches:");
$streamedBatches = new ProgressBar($streamedBatchesSection);
$streamedBatches->start($streamBatches);
$streamedRecordsSection = $output->section();
$streamedRecordsSection->writeln("\tStreamed records:");
$streamedRecords = new ProgressBar($streamedRecordsSection);
$streamedRecords->start($streamSize);
$memUsageSection = $output->section();
$memUsageSection->writeln("\tMemory usage:");
$memUsage = new ProgressBar($memUsageSection);
$memUsage->start($this->getMemoryLimit());
while (!$stream->eof()) {
\memory_reset_peak_usage();
$batch = $stream->read();
$streamBatches = $streamBatches - 1;
$streamedBatches->advance();
foreach ($pumps as $key => $pump) {
$pump->setConfig([
'skip-pumped' => $input->getOption('skip-pumped'),
]);
$pump->pump($batch);
$memUsage->setProgress(\memory_get_peak_usage(false));
}
$streamed = $stream->tell();
$streamedRecords->setProgress($streamed);
}
$endSection = new SymfonyStyle($input, $output->section());
$endSection->success('Data processed successfully!');
return Command::SUCCESS;
}
private function getMemoryLimit(): int
{
$limit = \ini_get('memory_limit');
if (preg_match('/^(\d+)(.)$/', $limit, $matches)) {
if ($matches[2] == 'M') {
$limit = $matches[1] * 1024 * 1024; // nnnM -> nnn MB
} elseif ($matches[2] == 'K') {
$limit = $matches[1] * 1024; // nnnK -> nnn KB
}
}
return $limit;
}
}