Initial commit: queue workspace

Made-with: Cursor

pro_v3.5.1/vendor/topthink/think-queue/tests/DatabaseConnectorTest.php

@@ -0,0 +1,123 @@
+<?php
+
+namespace think\test\queue;
+
+use Carbon\Carbon;
+use Mockery as m;
+use Mockery\MockInterface;
+use ReflectionClass;
+use stdClass;
+use think\Db;
+use think\queue\Connector;
+use think\queue\connector\Database;
+
+class DatabaseConnectorTest extends TestCase
+{
+    /** @var Database|MockInterface */
+    protected $connector;
+
+    /** @var Db|MockInterface */
+    protected $db;
+
+    protected function setUp()
+    {
+        parent::setUp();
+        $this->db        = m::mock(Db::class);
+        $this->connector = new Database($this->db, 'table', 'default');
+    }
+
+    public function testPushProperlyPushesJobOntoDatabase()
+    {
+        $this->db->shouldReceive('name')->with('table')->andReturn($query = m::mock(stdClass::class));
+
+        $query->shouldReceive('insertGetId')->once()->andReturnUsing(function ($array) {
+            $this->assertEquals('default', $array['queue']);
+            $this->assertEquals(json_encode(['job' => 'foo', 'maxTries' => null, 'timeout' => null, 'data' => ['data']]), $array['payload']);
+            $this->assertEquals(0, $array['attempts']);
+            $this->assertNull($array['reserved_at']);
+            $this->assertInternalType('int', $array['available_at']);
+        });
+        $this->connector->push('foo', ['data']);
+    }
+
+    public function testDelayedPushProperlyPushesJobOntoDatabase()
+    {
+        $this->db->shouldReceive('name')->with('table')->andReturn($query = m::mock(stdClass::class));
+
+        $query->shouldReceive('insertGetId')->once()->andReturnUsing(function ($array) {
+            $this->assertEquals('default', $array['queue']);
+            $this->assertEquals(json_encode(['job' => 'foo', 'maxTries' => null, 'timeout' => null, 'data' => ['data']]), $array['payload']);
+            $this->assertEquals(0, $array['attempts']);
+            $this->assertNull($array['reserved_at']);
+            $this->assertInternalType('int', $array['available_at']);
+        });
+
+        $this->connector->later(10, 'foo', ['data']);
+    }
+
+    public function testFailureToCreatePayloadFromObject()
+    {
+        $this->expectException('InvalidArgumentException');
+
+        $job          = new stdClass;
+        $job->invalid = "\xc3\x28";
+
+        $queue = $this->getMockForAbstractClass(Connector::class);
+        $class = new ReflectionClass(Connector::class);
+
+        $createPayload = $class->getMethod('createPayload');
+        $createPayload->setAccessible(true);
+        $createPayload->invokeArgs($queue, [
+            $job,
+            'queue-name',
+        ]);
+    }
+
+    public function testFailureToCreatePayloadFromArray()
+    {
+        $this->expectException('InvalidArgumentException');
+
+        $queue = $this->getMockForAbstractClass(Connector::class);
+        $class = new ReflectionClass(Connector::class);
+
+        $createPayload = $class->getMethod('createPayload');
+        $createPayload->setAccessible(true);
+        $createPayload->invokeArgs($queue, [
+            ["\xc3\x28"],
+            'queue-name',
+        ]);
+    }
+
+    public function testBulkBatchPushesOntoDatabase()
+    {
+
+        $this->db->shouldReceive('name')->with('table')->andReturn($query = m::mock(stdClass::class));
+
+        Carbon::setTestNow(
+            $now = Carbon::now()->addSeconds()
+        );
+
+        $query->shouldReceive('insertAll')->once()->andReturnUsing(function ($records) use ($now) {
+            $this->assertEquals([
+                [
+                    'queue'        => 'queue',
+                    'payload'      => json_encode(['job' => 'foo', 'maxTries' => null, 'timeout' => null, 'data' => ['data']]),
+                    'attempts'     => 0,
+                    'reserved_at'  => null,
+                    'available_at' => $now->getTimestamp(),
+                    'created_at'   => $now->getTimestamp(),
+                ], [
+                    'queue'        => 'queue',
+                    'payload'      => json_encode(['job' => 'bar', 'maxTries' => null, 'timeout' => null, 'data' => ['data']]),
+                    'attempts'     => 0,
+                    'reserved_at'  => null,
+                    'available_at' => $now->getTimestamp(),
+                    'created_at'   => $now->getTimestamp(),
+                ],
+            ], $records);
+        });
+
+        $this->connector->bulk(['foo', 'bar'], ['data'], 'queue');
+    }
+
+}

pro_v3.5.1/vendor/topthink/think-queue/tests/ListenerTest.php

@@ -0,0 +1,80 @@
+<?php
+
+namespace think\test\queue;
+
+use Mockery as m;
+use Mockery\MockInterface;
+use Symfony\Component\Process\Process;
+use think\queue\Listener;
+
+class ListenerTest extends TestCase
+{
+    /** @var Process|MockInterface */
+    protected $process;
+    /** @var Listener|MockInterface */
+    protected $listener;
+
+    public function testRunProcessCallsProcess()
+    {
+        /** @var Process|MockInterface $process */
+        $process = m::mock(Process::class)->makePartial();
+        $process->shouldReceive('run')->once();
+        /** @var Listener|MockInterface $listener */
+        $listener = m::mock(Listener::class)->makePartial();
+        $listener->shouldReceive('memoryExceeded')->once()->with(1)->andReturn(false);
+
+        $listener->runProcess($process, 1);
+    }
+
+    public function testListenerStopsWhenMemoryIsExceeded()
+    {
+        /** @var Process|MockInterface $process */
+        $process = m::mock(Process::class)->makePartial();
+        $process->shouldReceive('run')->once();
+        /** @var Listener|MockInterface $listener */
+        $listener = m::mock(Listener::class)->makePartial();
+        $listener->shouldReceive('memoryExceeded')->once()->with(1)->andReturn(true);
+        $listener->shouldReceive('stop')->once();
+
+        $listener->runProcess($process, 1);
+    }
+
+    public function testMakeProcessCorrectlyFormatsCommandLine()
+    {
+        $listener = new Listener(__DIR__);
+
+        $process = $listener->makeProcess('connection', 'queue', 1, 3, 0, 2, 3);
+        $escape  = '\\' === DIRECTORY_SEPARATOR ? '"' : '\'';
+
+        $this->assertInstanceOf(Process::class, $process);
+        $this->assertEquals(__DIR__, $process->getWorkingDirectory());
+        $this->assertEquals(3, $process->getTimeout());
+        $this->assertEquals($escape . PHP_BINARY . $escape . " {$escape}think{$escape} {$escape}queue:work{$escape} {$escape}connection{$escape} {$escape}--once{$escape} {$escape}--queue=queue{$escape} {$escape}--delay=1{$escape} {$escape}--memory=2{$escape} {$escape}--sleep=3{$escape} {$escape}--tries=0{$escape}", $process->getCommandLine());
+    }
+
+    public function testMakeProcessCorrectlyFormatsCommandLineWithAnEnvironmentSpecified()
+    {
+        $listener = new Listener(__DIR__);
+
+        $process = $listener->makeProcess('connection', 'queue', 1, 3, 0, 2, 3);
+        $escape  = '\\' === DIRECTORY_SEPARATOR ? '"' : '\'';
+
+        $this->assertInstanceOf(Process::class, $process);
+        $this->assertEquals(__DIR__, $process->getWorkingDirectory());
+        $this->assertEquals(3, $process->getTimeout());
+        $this->assertEquals($escape . PHP_BINARY . $escape . " {$escape}think{$escape} {$escape}queue:work{$escape} {$escape}connection{$escape} {$escape}--once{$escape} {$escape}--queue=queue{$escape} {$escape}--delay=1{$escape} {$escape}--memory=2{$escape} {$escape}--sleep=3{$escape} {$escape}--tries=0{$escape}", $process->getCommandLine());
+    }
+
+    public function testMakeProcessCorrectlyFormatsCommandLineWhenTheConnectionIsNotSpecified()
+    {
+        $listener = new Listener(__DIR__);
+
+        $process = $listener->makeProcess(null, 'queue', 1, 3, 0, 2, 3);
+        $escape  = '\\' === DIRECTORY_SEPARATOR ? '"' : '\'';
+
+        $this->assertInstanceOf(Process::class, $process);
+        $this->assertEquals(__DIR__, $process->getWorkingDirectory());
+        $this->assertEquals(3, $process->getTimeout());
+        $this->assertEquals($escape . PHP_BINARY . $escape . " {$escape}think{$escape} {$escape}queue:work{$escape} {$escape}--once{$escape} {$escape}--queue=queue{$escape} {$escape}--delay=1{$escape} {$escape}--memory=2{$escape} {$escape}--sleep=3{$escape} {$escape}--tries=0{$escape}", $process->getCommandLine());
+    }
+}

pro_v3.5.1/vendor/topthink/think-queue/tests/QueueTest.php

@@ -0,0 +1,49 @@
+<?php
+
+namespace think\test\queue;
+
+use InvalidArgumentException;
+use Mockery as m;
+use think\Config;
+use think\Queue;
+use think\queue\connector\Sync;
+
+class QueueTest extends TestCase
+{
+    /** @var Queue */
+    protected $queue;
+
+    protected function setUp()
+    {
+        parent::setUp();
+        $this->queue = new Queue($this->app);
+    }
+
+    public function testDefaultConnectionCanBeResolved()
+    {
+        $sync = new Sync();
+
+        $this->app->shouldReceive('invokeClass')->once()->with('\think\queue\connector\Sync', [['driver' => 'sync']])->andReturn($sync);
+
+        $config = m::mock(Config::class);
+
+        $config->shouldReceive('get')->twice()->with('queue.connectors.sync', ['driver' => 'sync'])->andReturn(['driver' => 'sync']);
+        $config->shouldReceive('get')->once()->with('queue.default', 'sync')->andReturn('sync');
+
+        $this->app->shouldReceive('get')->times(3)->with('config')->andReturn($config);
+
+        $this->assertSame($sync, $this->queue->driver('sync'));
+        $this->assertSame($sync, $this->queue->driver());
+    }
+
+    public function testNotSupportDriver()
+    {
+        $config = m::mock(Config::class);
+
+        $config->shouldReceive('get')->once()->with('queue.connectors.hello', ['driver' => 'sync'])->andReturn(['driver' => 'hello']);
+        $this->app->shouldReceive('get')->once()->with('config')->andReturn($config);
+
+        $this->expectException(InvalidArgumentException::class);
+        $this->queue->driver('hello');
+    }
+}

pro_v3.5.1/vendor/topthink/think-queue/tests/TestCase.php

@@ -0,0 +1,23 @@
+<?php
+
+namespace think\test\queue;
+
+use Mockery as m;
+use Mockery\MockInterface;
+use think\App;
+
+abstract class TestCase extends \PHPUnit\Framework\TestCase
+{
+    /** @var App|MockInterface */
+    protected $app;
+
+    public function tearDown()
+    {
+        m::close();
+    }
+
+    protected function setUp()
+    {
+        $this->app = m::mock(App::class)->makePartial();
+    }
+}

pro_v3.5.1/vendor/topthink/think-queue/tests/WorkerTest.php

@@ -0,0 +1,420 @@
+<?php
+
+namespace think\test\queue;
+
+use Carbon\Carbon;
+use Mockery as m;
+use Mockery\MockInterface;
+use RuntimeException;
+use think\Cache;
+use think\Event;
+use think\exception\Handle;
+use think\Queue;
+use think\queue\connector\Sync;
+use think\queue\event\JobExceptionOccurred;
+use think\queue\event\JobFailed;
+use think\queue\event\JobProcessed;
+use think\queue\event\JobProcessing;
+use think\queue\exception\MaxAttemptsExceededException;
+
+class WorkerTest extends TestCase
+{
+    /** @var Handle|MockInterface */
+    protected $handle;
+
+    /** @var Event|MockInterface */
+    protected $event;
+
+    /** @var Cache|MockInterface */
+    protected $cache;
+
+    /** @var Queue|MockInterface */
+    protected $queue;
+
+    protected function setUp()
+    {
+        parent::setUp();
+        $this->queue  = m::mock(Queue::class);
+        $this->handle = m::spy(Handle::class);
+        $this->event  = m::spy(Event::class);
+        $this->cache  = m::spy(Cache::class);
+    }
+
+    public function testJobCanBeFired()
+    {
+
+        $worker = $this->getWorker(['default' => [$job = new WorkerFakeJob]]);
+
+        $this->event->shouldReceive('trigger')->with(m::type(JobProcessing::class))->once();
+        $this->event->shouldReceive('trigger')->with(m::type(JobProcessed::class))->once();
+
+        $worker->runNextJob('sync', 'default');
+    }
+
+    public function testWorkerCanWorkUntilQueueIsEmpty()
+    {
+        $worker = $this->getWorker(['default' => [
+            $firstJob = new WorkerFakeJob,
+            $secondJob = new WorkerFakeJob,
+        ]]);
+
+        $this->expectException(LoopBreakerException::class);
+
+        $worker->daemon('sync', 'default');
+
+        $this->assertTrue($firstJob->fired);
+
+        $this->assertTrue($secondJob->fired);
+
+        $this->assertSame(0, $worker->stoppedWithStatus);
+
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobProcessing::class))->twice();
+
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobProcessed::class))->twice();
+    }
+
+    public function testJobCanBeFiredBasedOnPriority()
+    {
+        $worker = $this->getWorker([
+            'high' => [
+                $highJob = new WorkerFakeJob,
+                $secondHighJob = new WorkerFakeJob,
+            ],
+            'low'  => [$lowJob = new WorkerFakeJob],
+        ]);
+
+        $worker->runNextJob('sync', 'high,low');
+
+        $this->assertTrue($highJob->fired);
+        $this->assertFalse($secondHighJob->fired);
+        $this->assertFalse($lowJob->fired);
+
+        $worker->runNextJob('sync', 'high,low');
+        $this->assertTrue($secondHighJob->fired);
+        $this->assertFalse($lowJob->fired);
+
+        $worker->runNextJob('sync', 'high,low');
+        $this->assertTrue($lowJob->fired);
+    }
+
+    public function testExceptionIsReportedIfConnectionThrowsExceptionOnJobPop()
+    {
+        $e = new RuntimeException();
+
+        $sync = m::mock(Sync::class);
+
+        $sync->shouldReceive('pop')->andReturnUsing(function () use ($e) {
+            throw $e;
+        });
+
+        $this->queue->shouldReceive('driver')->with('sync')->andReturn($sync);
+
+        $worker = new Worker($this->queue, $this->event, $this->handle);
+
+        $worker->runNextJob('sync', 'default');
+
+        $this->handle->shouldHaveReceived('report')->with($e);
+    }
+
+    public function testWorkerSleepsWhenQueueIsEmpty()
+    {
+        $worker = $this->getWorker(['default' => []]);
+        $worker->runNextJob('sync', 'default', 0, 5);
+        $this->assertEquals(5, $worker->sleptFor);
+    }
+
+    public function testJobIsReleasedOnException()
+    {
+        $e = new RuntimeException;
+
+        $job = new WorkerFakeJob(function () use ($e) {
+            throw $e;
+        });
+
+        $worker = $this->getWorker(['default' => [$job]]);
+        $worker->runNextJob('sync', 'default', 10);
+
+        $this->assertEquals(10, $job->releaseAfter);
+        $this->assertFalse($job->deleted);
+        $this->handle->shouldHaveReceived('report')->with($e);
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobExceptionOccurred::class))->once();
+        $this->event->shouldNotHaveReceived('trigger', [m::type(JobProcessed::class)]);
+    }
+
+    public function testJobIsNotReleasedIfItHasExceededMaxAttempts()
+    {
+        $e = new RuntimeException;
+
+        $job           = new WorkerFakeJob(function ($job) use ($e) {
+            // In normal use this would be incremented by being popped off the queue
+            $job->attempts++;
+
+            throw $e;
+        });
+        $job->attempts = 1;
+
+        $worker = $this->getWorker(['default' => [$job]]);
+        $worker->runNextJob('sync', 'default', 0, 3, 1);
+
+        $this->assertNull($job->releaseAfter);
+        $this->assertTrue($job->deleted);
+        $this->assertEquals($e, $job->failedWith);
+        $this->handle->shouldHaveReceived('report')->with($e);
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobExceptionOccurred::class))->once();
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobFailed::class))->once();
+        $this->event->shouldNotHaveReceived('trigger', [m::type(JobProcessed::class)]);
+    }
+
+    public function testJobIsNotReleasedIfItHasExpired()
+    {
+        $e = new RuntimeException;
+
+        $job = new WorkerFakeJob(function ($job) use ($e) {
+            // In normal use this would be incremented by being popped off the queue
+            $job->attempts++;
+
+            throw $e;
+        });
+
+        $job->timeoutAt = Carbon::now()->addSeconds(1)->getTimestamp();
+
+        $job->attempts = 0;
+
+        Carbon::setTestNow(
+            Carbon::now()->addSeconds(1)
+        );
+
+        $worker = $this->getWorker(['default' => [$job]]);
+        $worker->runNextJob('sync', 'default');
+
+        $this->assertNull($job->releaseAfter);
+        $this->assertTrue($job->deleted);
+        $this->assertEquals($e, $job->failedWith);
+        $this->handle->shouldHaveReceived('report')->with($e);
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobExceptionOccurred::class))->once();
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobFailed::class))->once();
+        $this->event->shouldNotHaveReceived('trigger', [m::type(JobProcessed::class)]);
+    }
+
+    public function testJobIsFailedIfItHasAlreadyExceededMaxAttempts()
+    {
+        $job = new WorkerFakeJob(function ($job) {
+            $job->attempts++;
+        });
+
+        $job->attempts = 2;
+
+        $worker = $this->getWorker(['default' => [$job]]);
+        $worker->runNextJob('sync', 'default', 0, 3, 1);
+
+        $this->assertNull($job->releaseAfter);
+        $this->assertTrue($job->deleted);
+        $this->assertInstanceOf(MaxAttemptsExceededException::class, $job->failedWith);
+        $this->handle->shouldHaveReceived('report')->with(m::type(MaxAttemptsExceededException::class));
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobExceptionOccurred::class))->once();
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobFailed::class))->once();
+        $this->event->shouldNotHaveReceived('trigger', [m::type(JobProcessed::class)]);
+    }
+
+    public function testJobIsFailedIfItHasAlreadyExpired()
+    {
+        $job = new WorkerFakeJob(function ($job) {
+            $job->attempts++;
+        });
+
+        $job->timeoutAt = Carbon::now()->addSeconds(2)->getTimestamp();
+
+        $job->attempts = 1;
+
+        Carbon::setTestNow(
+            Carbon::now()->addSeconds(3)
+        );
+
+        $worker = $this->getWorker(['default' => [$job]]);
+        $worker->runNextJob('sync', 'default');
+
+        $this->assertNull($job->releaseAfter);
+        $this->assertTrue($job->deleted);
+        $this->assertInstanceOf(MaxAttemptsExceededException::class, $job->failedWith);
+        $this->handle->shouldHaveReceived('report')->with(m::type(MaxAttemptsExceededException::class));
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobExceptionOccurred::class))->once();
+        $this->event->shouldHaveReceived('trigger')->with(m::type(JobFailed::class))->once();
+        $this->event->shouldNotHaveReceived('trigger', [m::type(JobProcessed::class)]);
+    }
+
+    public function testJobBasedMaxRetries()
+    {
+        $job = new WorkerFakeJob(function ($job) {
+            $job->attempts++;
+        });
+
+        $job->attempts = 2;
+
+        $job->maxTries = 10;
+
+        $worker = $this->getWorker(['default' => [$job]]);
+        $worker->runNextJob('sync', 'default', 0, 3, 1);
+
+        $this->assertFalse($job->deleted);
+        $this->assertNull($job->failedWith);
+    }
+
+    protected function getWorker($jobs)
+    {
+        $sync = m::mock(Sync::class);
+
+        $sync->shouldReceive('pop')->andReturnUsing(function ($queue) use (&$jobs) {
+            return array_shift($jobs[$queue]);
+        });
+
+        $this->queue->shouldReceive('driver')->with('sync')->andReturn($sync);
+
+        return new Worker($this->queue, $this->event, $this->handle, $this->cache);
+    }
+}
+
+class WorkerFakeConnector
+{
+    public $jobs = [];
+
+    public function __construct($jobs)
+    {
+        $this->jobs = $jobs;
+    }
+
+    public function pop($queue)
+    {
+        return array_shift($this->jobs[$queue]);
+    }
+}
+
+class Worker extends \think\queue\Worker
+{
+    public $sleptFor;
+
+    public $stoppedWithStatus;
+
+    public function sleep($seconds)
+    {
+        $this->sleptFor = $seconds;
+    }
+
+    public function stop($status = 0)
+    {
+        $this->stoppedWithStatus = $status;
+
+        throw new LoopBreakerException;
+    }
+
+    protected function stopIfNecessary($job, $lastRestart, $memory)
+    {
+        if (is_null($job)) {
+            $this->stop();
+        } else {
+            parent::stopIfNecessary($job, $lastRestart, $memory);
+        }
+    }
+}
+
+class WorkerFakeJob
+{
+
+    public $fired    = false;
+    public $callback;
+    public $deleted  = false;
+    public $releaseAfter;
+    public $released = false;
+    public $maxTries;
+    public $timeoutAt;
+    public $attempts = 0;
+    public $failedWith;
+    public $failed   = false;
+    public $connectionName;
+
+    public function __construct($callback = null)
+    {
+        $this->callback = $callback ?: function () {
+            //
+        };
+    }
+
+    public function fire()
+    {
+        $this->fired = true;
+        $this->callback->__invoke($this);
+    }
+
+    public function payload()
+    {
+        return [];
+    }
+
+    public function maxTries()
+    {
+        return $this->maxTries;
+    }
+
+    public function timeoutAt()
+    {
+        return $this->timeoutAt;
+    }
+
+    public function delete()
+    {
+        $this->deleted = true;
+    }
+
+    public function isDeleted()
+    {
+        return $this->deleted;
+    }
+
+    public function release($delay)
+    {
+        $this->released = true;
+
+        $this->releaseAfter = $delay;
+    }
+
+    public function isReleased()
+    {
+        return $this->released;
+    }
+
+    public function attempts()
+    {
+        return $this->attempts;
+    }
+
+    public function markAsFailed()
+    {
+        $this->failed = true;
+    }
+
+    public function failed($e)
+    {
+        $this->markAsFailed();
+
+        $this->failedWith = $e;
+    }
+
+    public function hasFailed()
+    {
+        return $this->failed;
+    }
+
+    public function timeout()
+    {
+        return time() + 60;
+    }
+
+    public function getName()
+    {
+        return 'WorkerFakeJob';
+    }
+}
+
+class LoopBreakerException extends RuntimeException
+{
+    //
+}

pro_v3.5.1/vendor/topthink/think-queue/tests/bootstrap.php

@@ -0,0 +1,3 @@
+<?php
+
+include __DIR__.'/../vendor/autoload.php';