Fix & improve RateLimiter

Author: Oliver Stark

src/BackgroundProcess.php

@@ -35,12 +35,10 @@ public function __construct(QueueCommand $command = null)
     public function start()
     {
         $cmd = $this->command->getPreparedCommand();
-        $cwd = realpath(CRAFT_BASE_PATH);
-
-        $process = Process::fromShellCommandline($cmd, $cwd);
+        $process = Process::fromShellCommandline($cmd);
 
         try {
-            $process->run();
+            $process->start();
         } catch (\Symfony\Component\Process\Exception\RuntimeException $runtimeException) {
             $runtimeException = new RuntimeException($runtimeException->getMessage());
             $runtimeException->setProcess($process);

src/Handlers/BackgroundQueueHandler.php

@@ -54,7 +54,6 @@ public function __invoke(PushEvent $event): void
                     'async-queue'
                 );
             }
-
         }
 
         // Log what's going on

src/QueueCommand.php

@@ -45,7 +45,9 @@ public function getPreparedCommand(callable $wrapper = null): string
             throw new PhpExecutableNotFound('Unable to find php executable.');
         }
 
-        $commandLine = implode(" ", [$php, $this->scriptName, $this->scriptArgs]);
+        $path = realpath(CRAFT_BASE_PATH);
+        $script = $path . DIRECTORY_SEPARATOR . $this->scriptName;
+        $commandLine = implode(" ", [$php, $script, $this->scriptArgs]);
 
         return $this->decorate($commandLine);
     }
@@ -66,7 +68,7 @@ protected function decorate(string $commandLine): string
         }
 
         // default decoration
-        return "nice -n 15 {$commandLine} > /dev/null 2>&1 &";
+        return "nice -n 15 {$commandLine}"; //  > /tmp/null.txt 2>&1 &
     }
 
 }

src/RateLimiter.php

@@ -36,7 +36,12 @@ public function __construct(Queue $queue, Settings $settings)
      */
     public function canIUse(string $context = null): bool
     {
+        if ($this->internalCount >= $this->maxItems) {
+            return false;
+        }
+
         try {
+            $this->queue->channel = 'queue';
             $reserved = $this->queue->getTotalReserved();
         } catch (\Exception) {
             $reserved = 0;
@@ -54,6 +59,11 @@ public function increment(): void
         $this->internalCount++;
     }
 
+    public function getInternalCount(): int
+    {
+        return $this->internalCount;
+    }
+
 
     protected function logAttempt(int $currentUsage, string $context = null): void
     {

tests/BackgroundProcessRunTest.php

@@ -30,22 +30,36 @@ public function test_start_default_dummy_script_success(): void
         $bgProcess = new BackgroundProcess($command);
         $process   = $bgProcess->start();
 
+        $process->wait();
+
         $this->assertEquals(0, $process->getExitCode());
         $this->assertTrue($process->isSuccessful());
         $this->assertEquals(\Symfony\Component\Process\Process::STATUS_TERMINATED, $process->getStatus());
 
-        // Wait 0.25 seconds
-        usleep(250000);
+        $this->assertFileExists(TEST_FILE);
 
         $content = json_decode(file_get_contents(TEST_FILE), true);
-
         $this->assertTrue(is_array($content), 'Unable to read and json_decode test file.');
-        $this->assertContains('craft.php', $content['$argv']);
-        $this->assertContains('queue/run', $content['$argv']);
+        $this->assertStringContainsString('craft.php', $content['$argv'][0]);
+        $this->assertStringContainsString('queue/run', $content['$argv'][1]);
         $this->assertGreaterThanOrEqual($content['timestamp'], time());
+    }
+
+    /**
+     * @covers \ostark\AsyncQueue\BackgroundProcess::start
+     */
+    public function test_process_does_not_block(): void
+    {
+        $command   = new \ostark\AsyncQueue\QueueCommand('craft.php', '--sleep');
+        $bgProcess = new BackgroundProcess($command);
+        $process   = $bgProcess->start();
+        usleep(150000);
 
+        $this->assertFileExists(TEST_FILE);
 
-    }
+        $content = json_decode(file_get_contents(TEST_FILE), true);
+        $this->assertGreaterThanOrEqual($content['timestamp'], time());
 
+    }
 
 }

tests/RateLimiterTest.php

@@ -0,0 +1,73 @@
+<?php
+
+use ostark\AsyncQueue\BackgroundProcess;
+use PHPUnit\Framework\TestCase;
+use yii\queue\PushEvent;
+
+/**
+ * @covers \ostark\AsyncQueue\RateLimiter
+ * @covers \ostark\AsyncQueue\Handlers\BackgroundQueueHandler
+ */
+class RateLimiterTest extends TestCase
+{
+    public $plugin;
+
+    public function setUp(): void
+    {
+        parent::setUp();
+
+        $dummyCommand = new \ostark\AsyncQueue\QueueCommand('craft.php', '--sleep');
+        $this->plugin = new \ostark\AsyncQueue\Plugin('async-queue', null, []);
+        $this->plugin->set('async_process', new BackgroundProcess($dummyCommand));
+    }
+
+    public function tearDown(): void
+    {
+        parent::tearDown();
+        @unlink(TEST_FILE);
+    }
+
+
+    public function test_can_setup_handler_and_invoke_event(): void
+    {
+        $this->plugin = new \ostark\AsyncQueue\Plugin('async-queue', null, []);
+        $handler = new \ostark\AsyncQueue\Handlers\BackgroundQueueHandler($this->plugin);
+
+        $handler->__invoke(new PushEvent());
+
+        $this->assertSame(1, $this->plugin->getRateLimiter()->getInternalCount());
+    }
+
+    public function test_respect_the_limit_when_adding_multiple_jobs_in_one_request(): void
+    {
+        $handler = new \ostark\AsyncQueue\Handlers\BackgroundQueueHandler($this->plugin);
+
+        $handler->__invoke(new PushEvent());
+        $handler->__invoke(new PushEvent());
+        $handler->__invoke(new PushEvent());
+        $handler->__invoke(new PushEvent());
+
+        $this->assertSame(
+            $this->plugin->getRateLimiter()->maxItems,
+            $this->plugin->getRateLimiter()->getInternalCount()
+        );
+    }
+
+    public function test_stop_spawning_processes_when_too_many_jobs_are_reserved(): void
+    {
+        // Fake the reserved count
+        $queue = new class extends \craft\queue\Queue {
+            public function getTotalReserved(): int
+            {
+                return 5;
+            }
+        };
+
+        $this->plugin->set('async_rate_limiter', new \ostark\AsyncQueue\RateLimiter($queue, $this->plugin->getSettings()));
+        $handler = new \ostark\AsyncQueue\Handlers\BackgroundQueueHandler($this->plugin);
+
+        $handler->__invoke(new PushEvent());
+
+        $this->assertSame(0, $this->plugin->getRateLimiter()->getInternalCount());
+    }
+}