1. <?php
  3. /**
  4. * Task Queue Batch Processor CLI
  5. */
  7. require_once 'vendor/autoload.php'; // Autoload Composer dependencies
  9. use Symfony\Component\Console\Application;
  10. use Symfony\Component\Console\Command\Command;
  11. use Symfony\Component\Console\Input\InputInterface;
  12. use Symfony\Component\Console\Output\OutputInterface;
  13. use League\Queue\Laravel\LaravelQueue; //Example Queue implementation
  15. class BatchProcessorCommand extends Command
  16. {
  17. protected $name = 'batch-process';
  18. protected $description = 'Processes tasks from a queue in batches.';
  20. private $queue;
  21. private $batchSize;
  23. public function __construct(LaravelQueue $queue, $batchSize = 10)
  24. {
  25. parent::__construct();
  26. $this->queue = $queue;
  27. $this->batchSize = $batchSize;
  28. }
  30. /**
  31. * Set the batch size.
  32. *
  33. * @param string $batchSize
  34. * @return void
  35. */
  36. public function setBatchSize(string $batchSize)
  37. {
  38. $this->batchSize = (int)$batchSize;
  39. }
  42. /**
  43. * Define the command arguments.
  44. *
  45. * @param InputInterface $input
  46. * @param OutputInterface $output
  47. * @return void
  48. */
  49. protected function defineArguments(InputInterface $input, OutputInterface $output)
  50. {
  51. $output->registerArgument('batch-size', 'The number of tasks to process in each batch (default: 10)', 1);
  52. }
  54. /**
  55. * Execute the command.
  56. *
  57. * @param InputInterface $input
  58. * @param OutputInterface $output
  59. * @return void
  60. */
  61. public function run(InputInterface $input, OutputInterface $output)
  62. {
  63. $output->writeln('<info>Starting batch processing...</info>');
  65. try {
  66. $this->processQueueInBatches();
  67. } catch (\Exception $e) {
  68. $output->writeln('<error>Error during batch processing: ' . $e->getMessage() . '</error>');
  69. }
  71. $output->writeln('<info>Batch processing completed.</info>');
  72. }
  74. private function processQueueInBatches()
  75. {
  76. $queue = $this->queue;
  78. while (true) {
  79. $tasks = $queue->getTasks($this->batchSize);
  81. if (empty($tasks)) {
  82. break; // No more tasks in the queue
  83. }
  85. $output->writeln('<info>Processing batch of ' . count($tasks) . ' tasks...</info>');
  87. foreach ($tasks as $task) {
  88. try {
  89. $task->process(); // Execute the task
  90. } catch (\Exception $e) {
  91. $output->writeln('<error>Error processing task ' . $task->id . ': ' . $e->getMessage() . '</error>');
  92. // Optionally, handle task failures (e.g., retry, log to a dead-letter queue)
  93. }
  94. }
  95. }
  96. }
  97. }
  99. // Create the console application
  100. $application = new Application();
  101. $application->addCommand(new BatchProcessorCommand(new LaravelQueue()));
  103. $application->run();
  105. ?>

Add your comment