webmozarts · theofidry · Oct 20, 2022 · Oct 20, 2022
diff --git a/src/ParallelExecutor.php b/src/ParallelExecutor.php
@@ -25,20 +25,16 @@
 use Webmozarts\Console\Parallelization\Logger\Logger;
 use Webmozarts\Console\Parallelization\Process\ProcessLauncher;
 use Webmozarts\Console\Parallelization\Process\ProcessLauncherFactory;
-use Webmozarts\Console\Parallelization\Process\StandardSymfonyProcessFactory;
 use function array_filter;
 use function array_map;
 use function array_merge;
 use function array_slice;
 use function implode;
 use function mb_strlen;
 use function sprintf;
-use function usleep;
 
 final class ParallelExecutor
 {
-    private const CHILD_POLLING_IN_MICRO_SECONDS = 1000;    // 1ms
-
     /**
      * @var callable(InputInterface):iterable<string>
      */
@@ -110,6 +106,11 @@ final class ParallelExecutor
 
     private ProcessLauncherFactory $processLauncherFactory;
 
+    /**
+     * @var callable(): void
+     */
+    private $processTick;
+
     /**
      * @param callable(InputInterface):iterable<string>                    $fetchItems
      * @param callable(string, InputInterface, OutputInterface):void       $runSingleCommand
@@ -122,6 +123,7 @@ final class ParallelExecutor
      * @param callable(InputInterface, OutputInterface, list<string>):void $runBeforeBatch
      * @param callable(InputInterface, OutputInterface, list<string>):void $runAfterBatch
      * @param array<string, string>                                        $extraEnvironmentVariables
+     * @param callable(): void                                             $processTick
      */
     public function __construct(
         callable $fetchItems,
@@ -142,7 +144,8 @@ public function __construct(
         string $scriptPath,
         string $workingDirectory,
         ?array $extraEnvironmentVariables,
-        ProcessLauncherFactory $processLauncherFactory
+        ProcessLauncherFactory $processLauncherFactory,
+        callable $processTick
     ) {
         self::validateBatchSize($batchSize);
         self::validateSegmentSize($segmentSize);
@@ -168,6 +171,7 @@ public function __construct(
         $this->workingDirectory = $workingDirectory;
         $this->extraEnvironmentVariables = $extraEnvironmentVariables;
         $this->processLauncherFactory = $processLauncherFactory;
+        $this->processTick = $processTick;
     }
 
     public function execute(
@@ -363,10 +367,7 @@ private function createProcessLauncher(
             $segmentSize,
             $logger,
             fn (string $type, string $buffer) => $this->processChildOutput($buffer, $logger),
-            // TODO: make this configurable?
-            static fn () => usleep(self::CHILD_POLLING_IN_MICRO_SECONDS),
-            // TODO: make this configurable?
-            new StandardSymfonyProcessFactory(),
+            $this->processTick,
         );
     }
 

diff --git a/src/ParallelExecutorFactory.php b/src/ParallelExecutorFactory.php
@@ -19,6 +19,7 @@
 use Webmozarts\Console\Parallelization\ErrorHandler\ErrorHandler;
 use Webmozarts\Console\Parallelization\Process\PhpExecutableFinder;
 use Webmozarts\Console\Parallelization\Process\ProcessLauncherFactory;
+use Webmozarts\Console\Parallelization\Process\StandardSymfonyProcessFactory;
 use Webmozarts\Console\Parallelization\Process\SymfonyProcessLauncherFactory;
 use function chr;
 use function Safe\getcwd;
@@ -27,6 +28,8 @@
 
 final class ParallelExecutorFactory
 {
+    private const CHILD_POLLING_IN_MICRO_SECONDS = 1000;    // 1ms
+
     /**
      * @var callable(InputInterface):iterable<string>
      */
@@ -98,6 +101,11 @@ final class ParallelExecutorFactory
 
     private ProcessLauncherFactory $processLauncherFactory;
 
+    /**
+     * @var callable(): void
+     */
+    private $processTick;
+
     /**
      * @param callable(InputInterface):iterable<string>                    $fetchItems
      * @param callable(string, InputInterface, OutputInterface):void       $runSingleCommand
@@ -110,6 +118,7 @@ final class ParallelExecutorFactory
      * @param callable(InputInterface, OutputInterface, list<string>):void $runBeforeBatch
      * @param callable(InputInterface, OutputInterface, list<string>):void $runAfterBatch
      * @param array<string, string>                                        $extraEnvironmentVariables
+     * @param callable(): void                                             $processTick
      */
     private function __construct(
         callable $fetchItems,
@@ -130,7 +139,8 @@ private function __construct(
         string $scriptPath,
         string $workingDirectory,
         ?array $extraEnvironmentVariables,
-        ProcessLauncherFactory $processLauncherFactory
+        ProcessLauncherFactory $processLauncherFactory,
+        callable $processTick
     ) {
         $this->fetchItems = $fetchItems;
         $this->runSingleCommand = $runSingleCommand;
@@ -151,6 +161,7 @@ private function __construct(
         $this->workingDirectory = $workingDirectory;
         $this->extraEnvironmentVariables = $extraEnvironmentVariables;
         $this->processLauncherFactory = $processLauncherFactory;
+        $this->processTick = $processTick;
     }
 
     /**
@@ -185,7 +196,10 @@ public static function create(
             self::getScriptPath(),
             getcwd(),
             null,
-            new SymfonyProcessLauncherFactory(),
+            new SymfonyProcessLauncherFactory(
+                new StandardSymfonyProcessFactory(),
+            ),
+            static fn () => usleep(self::CHILD_POLLING_IN_MICRO_SECONDS),
         );
     }
 
@@ -362,6 +376,17 @@ public function withProcessLauncherFactory(ProcessLauncherFactory $processLaunch
         return $clone;
     }
 
+    /**
+     * @param callable(): void $processTick
+     */
+    public function withProcessTick(callable $processTick): self
+    {
+        $clone = clone $this;
+        $clone->processTick = $processTick;
+
+        return $clone;
+    }
+
     public function build(): ParallelExecutor
     {
         return new ParallelExecutor(
@@ -384,6 +409,7 @@ public function build(): ParallelExecutor
             $this->workingDirectory,
             $this->extraEnvironmentVariables,
             $this->processLauncherFactory,
+            $this->processTick,
         );
     }
 

diff --git a/src/Process/ProcessLauncherFactory.php b/src/Process/ProcessLauncherFactory.php
@@ -33,7 +33,6 @@ public function create(
         int $segmentSize,
         Logger $logger,
         callable $callback,
-        callable $tick,
-        SymfonyProcessFactory $processFactory
+        callable $tick
     ): ProcessLauncher;
 }
diff --git a/src/Process/SymfonyProcessLauncherFactory.php b/src/Process/SymfonyProcessLauncherFactory.php
@@ -17,6 +17,13 @@
 
 final class SymfonyProcessLauncherFactory implements ProcessLauncherFactory
 {
+    private SymfonyProcessFactory $processFactory;
+
+    public function __construct(SymfonyProcessFactory $processFactory)
+    {
+        $this->processFactory = $processFactory;
+    }
+
     /**
      * @param list<string>                   $command
      * @param array<string, string>|null     $extraEnvironmentVariables
@@ -33,8 +40,7 @@ public function create(
         int $segmentSize,
         Logger $logger,
         callable $callback,
-        callable $tick,
-        SymfonyProcessFactory $processFactory
+        callable $tick
     ): ProcessLauncher {
         return new SymfonyProcessLauncher(
             $command,
@@ -45,7 +51,7 @@ public function create(
             $logger,
             $callback,
             $tick,
-            $processFactory,
+            $this->processFactory,
         );
     }
 }
diff --git a/tests/ParallelExecutorFactoryTest.php b/tests/ParallelExecutorFactoryTest.php
@@ -49,6 +49,7 @@ public function test_it_can_create_a_configured_executor(): void
         $callable4 = self::createCallable(4);
         $callable5 = self::createCallable(5);
         $callable6 = self::createCallable(6);
+        $callable7 = self::createCallable(7);
 
         $childSourceStream = StringStream::fromString('');
         $batchSize = 10;
@@ -78,6 +79,7 @@ public function test_it_can_create_a_configured_executor(): void
             ->withWorkingDirectory(self::FILE_3)
             ->withExtraEnvironmentVariables($extraEnvironmentVariables)
             ->withProcessLauncherFactory($processLauncherFactory)
+            ->withProcessTick($callable7)
             ->build();
 
         $expected = new ParallelExecutor(
@@ -100,6 +102,7 @@ public function test_it_can_create_a_configured_executor(): void
             self::FILE_3,
             $extraEnvironmentVariables,
             $processLauncherFactory,
+            $callable7,
         );
 
         self::assertEquals($expected, $executor);

diff --git a/tests/ParallelExecutorTest.php b/tests/ParallelExecutorTest.php
@@ -36,7 +36,6 @@
 use Webmozarts\Console\Parallelization\Process\FakeProcessLauncherFactory;
 use Webmozarts\Console\Parallelization\Process\ProcessLauncher;
 use Webmozarts\Console\Parallelization\Process\ProcessLauncherFactory;
-use Webmozarts\Console\Parallelization\Process\StandardSymfonyProcessFactory;
 use function array_fill;
 use function func_get_args;
 use function getcwd;
@@ -427,7 +426,6 @@ public function test_it_can_launch_configured_child_processes(): void
                 $logger,
                 Argument::type('callable'),
                 Argument::type('callable'),
-                Argument::type(StandardSymfonyProcessFactory::class),
             )
             ->willReturn($processLauncherProphecy->reveal());
 
@@ -451,6 +449,7 @@ public function test_it_can_launch_configured_child_processes(): void
             $workingDirectory,
             $extraEnvironmentVariables,
             $processLauncherFactoryProphecy->reveal(),
+            static function (): void {},
         );
 
         $executor->execute(
@@ -1018,6 +1017,7 @@ private static function createChildProcessExecutor(
             __DIR__,
             null,
             new FakeProcessLauncherFactory(),
+            static function (): void {},
         );
     }
 
@@ -1069,6 +1069,7 @@ private static function createMainProcessExecutor(
             __DIR__,
             null,
             $processLauncherFactory,
+            static function (): void {},
         );
     }
 

diff --git a/tests/Process/FakeProcessLauncherFactory.php b/tests/Process/FakeProcessLauncherFactory.php
@@ -26,8 +26,7 @@ public function create(
         int $segmentSize,
         Logger $logger,
         callable $callback,
-        callable $tick,
-        SymfonyProcessFactory $processFactory
+        callable $tick
     ): ProcessLauncher {
         throw new DomainException('Unexpected call.');
     }