diff --git a/.gitattributes b/.gitattributes deleted file mode 100644 index f328c8b..0000000 --- a/.gitattributes +++ /dev/null @@ -1,7 +0,0 @@ -/test export-ignore -/.github export-ignore -/.gitattributes export-ignore -/.gitignore export-ignore -/.travis.yml export-ignore -/phpunit.travis.xml export-ignore -/phpunit.xml.dist export-ignore diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md deleted file mode 100644 index c23b0dd..0000000 --- a/.github/pull_request_template.md +++ /dev/null @@ -1,6 +0,0 @@ -# NOTE: This is a READ-ONLY repository. - -Please open Pull-Request to main repo: - -- 4.x: https://github.com/windwalker-io/framework -- 3.x: https://github.com/windwalker-io/framework/tree/3.x diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml deleted file mode 100644 index bb9ad46..0000000 --- a/.github/workflows/ci.yml +++ /dev/null @@ -1,45 +0,0 @@ -name: PHP Composer - -on: [push, pull_request] - -jobs: - build: - strategy: - matrix: - php-versions: [ '8.4' ] - runs-on: ubuntu-latest - services: - redis: - image: redis - ports: - - 6379/tcp - options: --health-cmd="redis-cli ping" --health-interval=10s --health-timeout=5s --health-retries=3 - steps: - - uses: actions/checkout@v2 - # MySQL - - name: Setup MySQL - uses: mirromutth/mysql-action@v1.1 - with: - mysql version: 8.1 - mysql database: windwalker_test - mysql root password: ut1234 - # PHP - - name: Setup PHP - uses: shivammathur/setup-php@v2 - with: - php-version: ${{ matrix.php-versions }} - extensions: redis, php-memcached, pdo, pdo_mysql - - name: Get composer cache directory - id: composercache - run: echo "::set-output name=dir::$(composer config cache-files-dir)" - - name: Cache composer dependencies - uses: actions/cache@v4 - with: - path: ${{ steps.composercache.outputs.dir }} - key: ${{ runner.os }}-composer-${{ hashFiles('**/composer.json') }} - restore-keys: ${{ runner.os }}-composer- - - name: Install dependencies - run: composer update --prefer-dist --prefer-stable --no-progress --no-suggest --ignore-platform-reqs - - - name: Run test suite - run: php vendor/bin/phpunit --configuration phpunit.ci.xml diff --git a/.gitignore b/.gitignore deleted file mode 100644 index 1a95239..0000000 --- a/.gitignore +++ /dev/null @@ -1,28 +0,0 @@ -# OS generated files # -.DS_Store -.DS_Store? -ehthumbs.db -Icon? -Thumbs.db - -# IDE Ignores # -.idea/ - -# System # -/tmp/* - -# Composer # -vendor/* -composer.lock - -/coverage - -# JS # -/node_modules - -# Test # -phpunit.xml -.phpunit.result.cache - -# Extra # -!.gitkeep diff --git a/.travis.yml b/.travis.yml deleted file mode 100644 index 9d27506..0000000 --- a/.travis.yml +++ /dev/null @@ -1,20 +0,0 @@ -language: php - -php: - - 8.0 - - nightly - -matrix: - allow_failures: - - php: 8.0 - -services: - - redis - -#before_install: - -before_script: - - composer update --ignore-platform-reqs --prefer-stable - -script: - - php vendor/bin/phpunit --configuration phpunit.travis.xml diff --git a/src/Driver/Bak/BeanstalkdQueueDriver.php b/Driver/BeanstalkdQueueDriver.php similarity index 52% rename from src/Driver/Bak/BeanstalkdQueueDriver.php rename to Driver/BeanstalkdQueueDriver.php index fb98784..692ae0a 100644 --- a/src/Driver/Bak/BeanstalkdQueueDriver.php +++ b/Driver/BeanstalkdQueueDriver.php @@ -1,10 +1,13 @@ -channel = $channel; + $this->queue = $queue; $this->timeout = $timeout; $this->client = $this->getPheanstalk($host); @@ -56,15 +59,15 @@ public function __construct(string $host, string $channel, int $timeout = 60) /** * push * - * @param QueueMessage $message + * @param QueueMessage $message * - * @return string + * @return int|string */ - public function push(QueueMessage $message): string + public function push(QueueMessage $message) { - $channel = $message->getChannel() ?: $this->channel; + $queue = $message->getQueueName() ?: $this->queue; - return (string) $this->client->useTube($channel)->put( + return $this->client->useTube($queue)->put( json_encode($message), PheanstalkInterface::DEFAULT_PRIORITY, $message->getDelay(), @@ -75,15 +78,15 @@ public function push(QueueMessage $message): string /** * pop * - * @param string|null $channel + * @param string $queue * - * @return QueueMessage|null + * @return QueueMessage */ - public function pop(?string $channel = null): ?QueueMessage + public function pop($queue = null) { - $channel = $channel ?: $this->channel; + $queue = $queue ?: $this->queue; - $job = $this->client->watchOnly($channel)->reserve(0); + $job = $this->client->watchOnly($queue)->reserve(0); if (!$job instanceof Job) { return null; @@ -92,10 +95,10 @@ public function pop(?string $channel = null): ?QueueMessage $message = new QueueMessage(); $message->setId($job->getId()); - $message->setAttempts((int) $this->client->statsJob($job)->reserves); + $message->setAttempts($this->client->statsJob($job)->reserves); $message->setBody(json_decode($job->getData(), true)); $message->setRawBody($job->getData()); - $message->setChannel($channel ?: $this->channel); + $message->setQueueName($queue ?: $this->queue); return $message; } @@ -103,15 +106,15 @@ public function pop(?string $channel = null): ?QueueMessage /** * delete * - * @param QueueMessage $message + * @param QueueMessage|string $message * - * @return BeanstalkdQueueDriver + * @return static */ - public function delete(QueueMessage $message): static + public function delete(QueueMessage $message) { - $channel = $message->getChannel() ?: $this->channel; + $queue = $message->getQueueName() ?: $this->queue; - $this->client->useTube($channel)->delete(new Job($message->getId(), '')); + $this->client->useTube($queue)->delete(new Job($message->getId(), '')); return $this; } @@ -119,11 +122,11 @@ public function delete(QueueMessage $message): static /** * release * - * @param QueueMessage|string $message + * @param QueueMessage|string $message * * @return static */ - public function release(QueueMessage $message): static + public function release(QueueMessage $message) { $this->client->release( new Job($message->getId(), ''), @@ -137,15 +140,15 @@ public function release(QueueMessage $message): static /** * getPheanstalk * - * @param string $host + * @param string $host * * @return Pheanstalk - * @throws DomainException + * @throws \DomainException */ - public function getPheanstalk($host = null): Pheanstalk + public function getPheanstalk($host = null) { if (!class_exists(Pheanstalk::class)) { - throw new DomainException('Please install pda/pheanstalk first.'); + throw new \DomainException('Please install pda/pheanstalk first.'); } return new Pheanstalk($host); diff --git a/Driver/DatabaseQueueDriver.php b/Driver/DatabaseQueueDriver.php new file mode 100644 index 0000000..0d49d98 --- /dev/null +++ b/Driver/DatabaseQueueDriver.php @@ -0,0 +1,282 @@ +db = $db; + $this->table = $table; + $this->queue = $queue; + $this->timeout = $timeout; + } + + /** + * push + * + * @param QueueMessage $message + * + * @return int|string + * @throws \Exception + */ + public function push(QueueMessage $message) + { + $time = new \DateTimeImmutable('now'); + + $data = [ + 'queue' => $message->getQueueName() ?: $this->queue, + 'body' => json_encode($message), + 'attempts' => 0, + 'created' => $time->format('Y-m-d H:i:s'), + 'visibility' => $time->modify(sprintf('+%dseconds', $message->getDelay()))->format('Y-m-d H:i:s'), + 'reserved' => null, + ]; + + $this->db->getWriter()->insertOne($this->table, $data, 'id'); + + return $data['id']; + } + + /** + * pop + * + * @param string $queue + * + * @return QueueMessage + * @throws \Exception + */ + public function pop($queue = null) + { + $queue = $queue ?: $this->queue; + + $now = new \DateTimeImmutable('now'); + + $query = $this->db->getQuery(true); + + $query->select('*') + ->from($query->quoteName($this->table)) + ->where('queue = %q', $queue) + ->where('visibility <= %q', $now->format('Y-m-d H:i:s')) + ->orWhere( + function (Query $query) use ($now) { + $query->where('reserved IS NULL') + ->where('reserved < %q', $now->modify('-' . $this->timeout . 'seconds')->format('Y-m-d H:i:s')); + } + ); + + $trans = $this->db->getTransaction()->start(); + + try { + $data = $this->db->setQuery($query . ' FOR UPDATE')->loadOne('assoc'); + + if (!$data) { + return null; + } + + $data['attempts']++; + + $values = ['reserved' => $now->format('Y-m-d H:i:s'), 'attempts' => $data['attempts']]; + + $this->db->getWriter()->updateBatch($this->table, $values, ['id' => $data['id']]); + + $trans->commit(); + } catch (\Throwable $t) { + $trans->rollback(); + } + + $message = new QueueMessage(); + + $message->setId($data['id']); + $message->setAttempts($data['attempts']); + $message->setBody(json_decode($data['body'], true)); + $message->setRawBody($data['body']); + $message->setQueueName($queue); + + return $message; + } + + /** + * delete + * + * @param QueueMessage|string $message + * + * @return static + */ + public function delete(QueueMessage $message) + { + $queue = $message->getQueueName() ?: $this->queue; + + $query = $this->db->getQuery(true); + + $query->delete($query->quoteName($this->table)) + ->where('id = :id') + ->where('queue = :queue') + ->bind('id', $message->getId()) + ->bind('queue', $queue); + + $this->db->setQuery($query)->execute(); + + return $this; + } + + /** + * release + * + * @param QueueMessage|string $message + * + * @return static + * @throws \Exception + */ + public function release(QueueMessage $message) + { + $queue = $message->getQueueName() ?: $this->queue; + + $time = new \DateTimeImmutable('now'); + $time = $time->modify('+' . $message->getDelay() . 'seconds'); + + $values = [ + 'reserved' => null, + 'visibility' => $time->format('Y-m-d H:i:s'), + ]; + + $this->db->getWriter()->updateBatch( + $this->table, + $values, + [ + 'id' => $message->getId(), + 'queue' => $queue, + ] + ); + + return $this; + } + + /** + * Method to get property Table + * + * @return mixed + */ + public function getTable() + { + return $this->table; + } + + /** + * Method to set property table + * + * @param mixed $table + * + * @return static Return self to support chaining. + */ + public function setTable($table) + { + $this->table = $table; + + return $this; + } + + /** + * Method to get property Db + * + * @return AbstractDatabaseDriver + */ + public function getDb() + { + return $this->db; + } + + /** + * Method to set property db + * + * @param AbstractDatabaseDriver $db + * + * @return static Return self to support chaining. + */ + public function setDb($db) + { + $this->db = $db; + + return $this; + } + + /** + * Reconnect database to avoid long connect issues. + * + * @return static + */ + public function reconnect() + { + $this->disconnect(); + + $this->db->connect(); + + return $this; + } + + /** + * Disconnect DB. + * + * @return static + * + * @since 3.5.2 + */ + public function disconnect() + { + $this->db->disconnect(); + + return $this; + } +} diff --git a/src/Driver/Bak/IronmqQueueDriver.php b/Driver/IronmqQueueDriver.php similarity index 50% rename from src/Driver/Bak/IronmqQueueDriver.php rename to Driver/IronmqQueueDriver.php index 5ea8c2e..545bacd 100644 --- a/src/Driver/Bak/IronmqQueueDriver.php +++ b/Driver/IronmqQueueDriver.php @@ -1,10 +1,13 @@ -client = $this->getIronMQ($projectId, $token, $options); - $this->channel = $channel; + $this->queue = $queue; } /** * push * - * @param QueueMessage $message + * @param QueueMessage $message * - * @return string + * @return int|string */ - public function push(QueueMessage $message): string + public function push(QueueMessage $message) { - $channel = $message->getChannel() ?: $this->channel; + $queue = $message->getQueueName() ?: $this->queue; $options = $message->getOptions(); $options['delay'] = $message->getDelay(); - return (string) $this->client->postMessage($channel, json_encode($message), $options)->id; + return $this->client->postMessage($queue, json_encode($message), $options)->id; } /** * pop * - * @param string|null $channel + * @param string $queue * - * @return QueueMessage|null + * @return QueueMessage */ - public function pop(?string $channel = null): ?QueueMessage + public function pop($queue = null) { - $channel = $channel ?: $this->channel; + $queue = $queue ?: $this->queue; - $result = $this->client->reserveMessage($channel); + $result = $this->client->reserveMessage($queue); if (!$result) { return null; @@ -82,10 +85,10 @@ public function pop(?string $channel = null): ?QueueMessage $message = new QueueMessage(); $message->setId($result->id); - $message->setAttempts((int) $result->reserved_count); + $message->setAttempts($result->reserved_count); $message->setBody(json_decode($result->body, true)); $message->setRawBody($result->body); - $message->setChannel($channel ?: $this->channel); + $message->setQueueName($queue ?: $this->queue); $message->set('reservation_id', $result->reservation_id); return $message; @@ -94,15 +97,15 @@ public function pop(?string $channel = null): ?QueueMessage /** * delete * - * @param QueueMessage $message + * @param QueueMessage|string $message * - * @return IronmqQueueDriver + * @return static */ - public function delete(QueueMessage $message): static + public function delete(QueueMessage $message) { - $channel = $message->getChannel() ?: $this->channel; + $queue = $message->getQueueName() ?: $this->queue; - $this->client->deleteMessage($channel, $message->getId(), $message->get('reservation_id')); + $this->client->deleteMessage($queue, $message->getId(), $message->get('reservation_id')); return $this; } @@ -110,16 +113,16 @@ public function delete(QueueMessage $message): static /** * release * - * @param QueueMessage|string $message + * @param QueueMessage|string $message * * @return static */ - public function release(QueueMessage $message): static + public function release(QueueMessage $message) { - $channel = $message->getChannel() ?: $this->channel; + $queue = $message->getQueueName() ?: $this->queue; $this->client->releaseMessage( - $channel, + $queue, $message->getId(), $message->get('reservation_id'), $message->getDelay() @@ -131,16 +134,16 @@ public function release(QueueMessage $message): static /** * getIronMQ * - * @param $projectId - * @param $token - * @param array $options + * @param $projectId + * @param $token + * @param array $options * * @return IronMQ */ - public function getIronMQ($projectId, $token, array $options): IronMQ + public function getIronMQ($projectId, $token, array $options) { if (!class_exists(IronMQ::class)) { - throw new DomainException('Please install iron-io/iron_mq first.'); + throw new \DomainException('Please install iron-io/iron_mq first.'); } $defaultOptions = [ @@ -152,9 +155,4 @@ public function getIronMQ($projectId, $token, array $options): IronMQ return new IronMQ($options); } - - public function defer(QueueMessage $message): static - { - return $this; - } } diff --git a/Driver/NullQueueDriver.php b/Driver/NullQueueDriver.php new file mode 100644 index 0000000..9db3e1f --- /dev/null +++ b/Driver/NullQueueDriver.php @@ -0,0 +1,67 @@ +pdo = $db; + $this->table = $table; + $this->queue = $queue; + $this->timeout = $timeout; + } + + /** + * push + * + * @param QueueMessage $message + * + * @return int|string + * @throws \Exception + */ + public function push(QueueMessage $message) + { + $time = new \DateTimeImmutable('now'); + + $data = [ + ':queue' => $message->getQueueName() ?: $this->queue, + ':body' => json_encode($message), + ':attempts' => 0, + ':created' => $time->format('Y-m-d H:i:s'), + ':visibility' => $time->modify(sprintf('+%dseconds', $message->getDelay()))->format('Y-m-d H:i:s'), + ':reserved' => null, + ]; + + $sql = 'INSERT INTO ' . $this->table . + ' (queue, body, attempts, created, visibility, reserved)' . + ' VALUES (:queue, :body, :attempts, :created, :visibility, :reserved)'; + + $this->pdo->prepare($sql)->execute($data); + + return $this->pdo->lastInsertId(); + } + + /** + * pop + * + * @param string $queue + * + * @return QueueMessage + * @throws \Exception + * @throws \InvalidArgumentException + * @throws \Throwable + */ + public function pop($queue = null) + { + $queue = $queue ?: $this->queue; + + $now = new \DateTimeImmutable('now'); + + $sql = 'SELECT * FROM ' . $this->table . + ' WHERE queue = :queue AND visibility < :visibility' . + ' AND (reserved IS NULL OR reserved < :reserved)' . + ' FOR UPDATE'; + + $this->pdo->beginTransaction(); + + $stat = $this->pdo->prepare($sql); + $stat->bindValue(':queue', $queue); + $stat->bindValue(':visibility', $now->format('Y-m-d H:i:s'), \PDO::PARAM_STR); + $stat->bindValue( + ':reserved', + $now->modify('-' . $this->timeout . 'seconds')->format('Y-m-d H:i:s'), + \PDO::PARAM_STR + ); + + try { + $stat->execute(); + + $data = $stat->fetch(\PDO::FETCH_ASSOC); + + if (!$data) { + $this->pdo->commit(); + + return null; + } + + $data['attempts']++; + + $sql = 'UPDATE ' . $this->table . ' SET reserved = :reserved, attempts = :attempts WHERE id = :id'; + + $stat = $this->pdo->prepare($sql); + $stat->bindValue(':reserved', $now->format('Y-m-d H:i:s')); + $stat->bindValue(':attempts', $data['attempts'] + 1); + $stat->bindValue(':id', $data['id']); + + $stat->execute(); + + $this->pdo->commit(); + } catch (\Throwable $t) { + $this->pdo->rollBack(); + throw $t; + } + + $message = new QueueMessage(); + + $message->setId($data['id']); + $message->setAttempts($data['attempts']); + $message->setBody(json_decode($data['body'], true)); + $message->setRawBody($data['body']); + $message->setQueueName($queue); + + return $message; + } + + /** + * delete + * + * @param QueueMessage|string $message + * + * @return static + */ + public function delete(QueueMessage $message) + { + $queue = $message->getQueueName() ?: $this->queue; + + $sql = 'DELETE FROM ' . $this->table . + ' WHERE id = :id AND queue = :queue'; + + $stat = $this->pdo->prepare($sql); + $stat->bindValue(':id', $message->getId()); + $stat->bindValue(':queue', $queue); + + $stat->execute(); + + return $this; + } + + /** + * release + * + * @param QueueMessage|string $message + * + * @return static + * @throws \Exception + */ + public function release(QueueMessage $message) + { + $queue = $message->getQueueName() ?: $this->queue; + + $time = new \DateTimeImmutable('now'); + $time = $time->modify('+' . $message->getDelay() . 'seconds'); + + $values = [ + 'id' => $message->getId(), + 'queue' => $queue, + 'reserved' => null, + 'visibility' => $time->format('Y-m-d H:i:s'), + ]; + + $sql = 'UPDATE ' . $this->table . + ' SET reserved = :reserved, visibility = :visibility' . + ' WHERE id = :id AND queue = :queue'; + + $stat = $this->pdo->prepare($sql); + + $stat->execute($values); + + return $this; + } + + /** + * Method to get property Table + * + * @return mixed + */ + public function getTable() + { + return $this->table; + } + + /** + * Method to set property table + * + * @param mixed $table + * + * @return static Return self to support chaining. + */ + public function setTable($table) + { + $this->table = $table; + + return $this; + } + + /** + * Method to get property Db + * + * @return \PDO + */ + public function getPdo() + { + return $this->pdo; + } + + /** + * Method to set property db + * + * @param \PDO $pdo + * + * @return static Return self to support chaining. + */ + public function setPdo(\PDO $pdo) + { + $this->pdo = $pdo; + + return $this; + } + + /** + * Reconnect database to avoid long connect issues. + * + * @return static + */ + public function reconnect() + { + // PDO cannot reconnect. + + return $this; + } +} diff --git a/Driver/QueueDriverInterface.php b/Driver/QueueDriverInterface.php new file mode 100644 index 0000000..055bbbf --- /dev/null +++ b/Driver/QueueDriverInterface.php @@ -0,0 +1,55 @@ +client = $this->getAMQPConnection($options); - $this->channel = $channel; + $this->queue = $queue; $this->channel = $this->client->channel(); } /** * push * - * @param QueueMessage $message + * @param QueueMessage $message * - * @return string + * @return int|string */ - public function push(QueueMessage $message): string + public function push(QueueMessage $message) { - $channel = $message->getChannel() ?: $this->channel; + $queue = $message->getQueueName() ?: $this->queue; - $this->channelDeclare($channel); + $this->queueDeclare($queue); $options = [ 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT, @@ -80,26 +83,26 @@ public function push(QueueMessage $message): string $msg = new AMQPMessage(json_encode($message), $options); - $this->channel->basic_publish($msg, '', $channel); + $this->channel->basic_publish($msg, '', $queue); - return '1'; + return 1; } /** * pop * - * @param string|null $channel + * @param string $queue * - * @return QueueMessage|null + * @return QueueMessage */ - public function pop(?string $channel = null): ?QueueMessage + public function pop($queue = null) { - $channel = $channel ?: $this->channel; + $queue = $queue ?: $this->queue; - $this->channelDeclare($channel); + $this->queueDeclare($queue); $this->channel->basic_qos(null, 1, null); - $result = $this->channel->basic_get($channel, false); + $result = $this->channel->basic_get($queue, false); if (!$result) { return null; @@ -110,7 +113,7 @@ public function pop(?string $channel = null): ?QueueMessage $message->setId(0); $message->setBody(json_decode($result->body, true)); $message->setRawBody($result->body); - $message->setChannel($channel ?: $this->channel); + $message->setQueueName($queue ?: $this->queue); // Delivery tag $message->set('delivery_tag', $result->delivery_info['delivery_tag']); @@ -118,7 +121,7 @@ public function pop(?string $channel = null): ?QueueMessage // Attempts $attempts = $message->get('attempts', 0) + 1; - $message->setAttempts((int) $attempts); + $message->setAttempts($attempts); $message->set('attempts', $attempts); return $message; @@ -127,11 +130,11 @@ public function pop(?string $channel = null): ?QueueMessage /** * delete * - * @param QueueMessage $message + * @param QueueMessage|string $message * - * @return RabbitmqQueueDriver + * @return static */ - public function delete(QueueMessage $message): static + public function delete(QueueMessage $message) { $this->channel->basic_ack($message->get('delivery_tag')); @@ -141,11 +144,11 @@ public function delete(QueueMessage $message): static /** * release * - * @param QueueMessage|string $message + * @param QueueMessage|string $message * * @return static */ - public function release(QueueMessage $message): static + public function release(QueueMessage $message) { $this->delete($message); @@ -157,29 +160,29 @@ public function release(QueueMessage $message): static } /** - * channelDeclare + * queueDeclare * - * @param string $channel + * @param string $queue * * @return void */ - protected function channelDeclare(string $channel) + protected function queueDeclare($queue) { - $this->channel->channel_declare($channel, false, true, false, false); + $this->channel->queue_declare($queue, false, true, false, false); } /** * getAMQPConnection * - * @param array $options + * @param array $options * * @return AMQPStreamConnection - * @throws DomainException + * @throws \DomainException */ - public function getAMQPConnection(array $options): AMQPStreamConnection + public function getAMQPConnection(array $options) { if (!class_exists(AMQPStreamConnection::class)) { - throw new DomainException('Please install php-amqplib/php-amqplib first.'); + throw new \DomainException('Please install php-amqplib/php-amqplib first.'); } $defaultOptions = [ @@ -198,8 +201,4 @@ public function getAMQPConnection(array $options): AMQPStreamConnection $options['password'] ); } - - public function defer(QueueMessage $message): static - { - } } diff --git a/Driver/ResqueQueueDriver.php b/Driver/ResqueQueueDriver.php new file mode 100644 index 0000000..6cab153 --- /dev/null +++ b/Driver/ResqueQueueDriver.php @@ -0,0 +1,217 @@ +queue = $queue; + + $this->connect($host, $port); + } + + /** + * push + * + * @param QueueMessage $message + * + * @return int|string + * @throws \DomainException + */ + public function push(QueueMessage $message) + { + $queue = $message->getQueueName() ?: $this->queue; + + if (!$message->getId()) { + $message->set('attempts', 0); + $message->set('queue', $queue); + $message->set('id', Resque::generateJobId()); + $message->set('class', static::JOB_CLASS); + $message->setId($message->getId()); + } + + $delay = $message->getDelay(); + + $data = json_decode(json_encode($message), true); + + if ($delay > 0) { + static::supportDelayed(true); + + \ResqueScheduler::delayedPush(time() + $delay, $data); + } else { + Resque::push($queue, $data); + } + + return $message->getId(); + } + + /** + * pop + * + * @param string $queue + * + * @return QueueMessage + */ + public function pop($queue = null) + { + if (static::supportDelayed()) { + $this->requeueDelayedItems(); + } + + $queue = $queue ?: $this->queue; + + $job = Resque::pop($queue); + + if (!$job) { + return null; + } + + $message = new QueueMessage(); + + $attempts = $job['attempts']; + $attempts++; + + $message->setId($job['id']); + $message->setBody($job); + $message->setRawBody(json_encode($job)); + $message->setQueueName($queue ?: $this->queue); + $message->setAttempts($attempts); + $message->set('attempts', $attempts); + + return $message; + } + + /** + * delete + * + * @param QueueMessage|string $message + * + * @return static + */ + public function delete(QueueMessage $message) + { + $queue = $message->getQueueName() ?: $this->queue; + + Resque::dequeue($queue, [static::JOB_CLASS => $message->getId()]); + + return $this; + } + + /** + * release + * + * @param QueueMessage|string $message + * + * @return static + */ + public function release(QueueMessage $message) + { + $this->push($message); + + return $this; + } + + /** + * Handle delayed items for the next scheduled timestamp. + * + * Searches for any items that are due to be scheduled in Resque + * and adds them to the appropriate job queue in Resque. + * + * @param \DateTime|int $timestamp Search for any items up to this timestamp to schedule. + */ + public function requeueDelayedItems() + { + while (($oldestJobTimestamp = \ResqueScheduler::nextDelayedTimestamp()) !== false) { + $this->enqueueDelayedItemsForTimestamp($oldestJobTimestamp); + } + } + + /** + * Schedule all of the delayed jobs for a given timestamp. + * + * Searches for all items for a given timestamp, pulls them off the list of + * delayed jobs and pushes them across to Resque. + * + * @param \DateTime|int $timestamp Search for any items up to this timestamp to schedule. + */ + public function enqueueDelayedItemsForTimestamp($timestamp) + { + $item = null; + + while ($item = \ResqueScheduler::nextItemForTimestamp($timestamp)) { + Resque::push($item['queue'], $item); + } + } + + /** + * supportDelayed + * + * @param bool $throwError + * + * @return bool + * @throws \DomainException + */ + public static function supportDelayed($throwError = false) + { + if (!class_exists(\ResqueScheduler::class)) { + if ($throwError) { + throw new \DomainException( + 'Please install chrisboulton/php-resque-scheduler to support delayed messages for Resque.' + ); + } + + return false; + } + + return true; + } + + /** + * connect + * + * @param string $host + * @param int $port + * + * @return void + * @throws \DomainException + */ + public function connect($host, $port) + { + if (!class_exists(Resque::class)) { + throw new \DomainException('Please install chrisboulton/php-resque 1.2 to support Resque driver.'); + } + + Resque::setBackend("$host:$port"); + } +} diff --git a/Driver/SqsQueueDriver.php b/Driver/SqsQueueDriver.php new file mode 100644 index 0000000..48267c2 --- /dev/null +++ b/Driver/SqsQueueDriver.php @@ -0,0 +1,207 @@ +client = $this->getSqsClient($key, $secret, $options); + + $this->queue = $queue; + } + + /** + * push + * + * @param QueueMessage $message + * + * @return int|string + */ + public function push(QueueMessage $message) + { + $request = [ + 'QueueUrl' => $this->getQueueUrl($message->getQueueName()), + 'MessageBody' => json_encode($message), + ]; + + $request['DelaySeconds'] = $message->getDelay(); + + $options = $message->getOptions(); + + $request = array_merge($request, $options); + + return $this->client->sendMessage($request)->get('MessageId'); + } + + /** + * pop + * + * @param string $queue + * + * @return QueueMessage|null + */ + public function pop($queue = null) + { + $result = $this->client->receiveMessage( + [ + 'QueueUrl' => $this->getQueueUrl($queue), + 'AttributeNames' => ['ApproximateReceiveCount'], + ] + ); + + if ($result['Messages'] === null) { + return null; + } + + $data = $result['Messages'][0]; + + $res = new QueueMessage(); + + $res->setId($data['MessageId']); + $res->setAttempts($data['Attributes']['ApproximateReceiveCount']); + $res->setBody(json_decode($data['Body'], true)); + $res->setRawBody($data['Body']); + $res->setQueueName($queue ?: $this->queue); + $res->set('ReceiptHandle', $data['ReceiptHandle']); + + return $res; + } + + /** + * delete + * + * @param QueueMessage|string $message + * + * @return static + * @internal param null $queue + * + */ + public function delete(QueueMessage $message) + { + $this->client->deleteMessage( + [ + 'QueueUrl' => $this->getQueueUrl($message->getQueueName()), + 'ReceiptHandle' => $this->getReceiptHandle($message), + ] + ); + + return $this; + } + + /** + * release + * + * @param QueueMessage|string $message + * + * @return static + */ + public function release(QueueMessage $message) + { + $this->client->changeMessageVisibility( + [ + 'QueueUrl' => $this->getQueueUrl($message->getQueueName()), + 'ReceiptHandle' => $this->getReceiptHandle($message), + 'VisibilityTimeout' => $message->getDelay(), + ] + ); + + return $this; + } + + /** + * getQueueUrl + * + * @param string $queue + * + * @return string + */ + public function getQueueUrl($queue = null) + { + $queue = $queue ?: $this->queue; + + if (filter_var($queue, FILTER_VALIDATE_URL) !== false) { + return $queue; + } + + return $this->client->getQueueUrl(['QueueName' => $queue])->get('QueueUrl'); + } + + /** + * getReceiptHandle + * + * @param QueueMessage $message + * + * @return string + */ + public function getReceiptHandle(QueueMessage $message) + { + return $message->get('ReceiptHandle', $message->getId()); + } + + /** + * getSqsClient + * + * @param string $key + * @param string $secret + * @param array $options + * + * @return SqsClient + * @throws \DomainException + */ + public function getSqsClient($key, $secret, array $options = []) + { + if (!class_exists(SqsClient::class)) { + throw new \DomainException('Please install aws/aws-sdk-php first.'); + } + + $defaultOptions = [ + 'region' => 'ap-northeast-1', + 'version' => 'latest', + 'credentials' => [ + 'key' => $key, + 'secret' => $secret, + ], + ]; + + $options = array_merge($defaultOptions, $options); + + return new SqsClient($options); + } +} diff --git a/Driver/SyncQueueDriver.php b/Driver/SyncQueueDriver.php new file mode 100644 index 0000000..160252f --- /dev/null +++ b/Driver/SyncQueueDriver.php @@ -0,0 +1,74 @@ +getJob(); + /** @var JobInterface $job */ + $job = unserialize($job); + + $job->execute(); + + return 0; + } + + /** + * pop + * + * @param string $queue + * + * @return QueueMessage + */ + public function pop($queue = null) + { + return new QueueMessage(); + } + + /** + * delete + * + * @param QueueMessage|string $message + * + * @return static + */ + public function delete(QueueMessage $message) + { + return $this; + } + + /** + * release + * + * @param QueueMessage|string $message + * + * @return static + */ + public function release(QueueMessage $message) + { + return $this; + } +} diff --git a/Exception/MaxAttemptsExceededException.php b/Exception/MaxAttemptsExceededException.php new file mode 100644 index 0000000..f726ecb --- /dev/null +++ b/Exception/MaxAttemptsExceededException.php @@ -0,0 +1,18 @@ +db = $db; + $this->table = $table; + } + + /** + * isSupported + * + * @return bool + */ + public function isSupported() + { + return $this->db->getTable($this->table)->exists(); + } + + /** + * add + * + * @param string $connection + * @param string $queue + * @param string $body + * @param string $exception + * + * @return int|string + */ + public function add($connection, $queue, $body, $exception) + { + $data = get_defined_vars(); + + // For B/C + if (class_exists(Chronos::class)) { + $data['created'] = Chronos::create('now')->toSql(); + } else { + $data['created'] = (new \DateTime('now'))->format('Y-m-d H:i:s'); + } + + $this->db->getWriter()->insertOne($this->table, $data, 'id'); + + return $data['id']; + } + + /** + * all + * + * @return array + * @throws \RuntimeException + * @throws \InvalidArgumentException + */ + public function all() + { + $query = $this->db->getQuery(true); + + $query->select('*') + ->from($query->quoteName($this->table)); + + return $this->db->setQuery($query)->loadAll(null, 'assoc'); + } + + /** + * get + * + * @param mixed $conditions + * + * @return array + */ + public function get($conditions) + { + $query = $this->db->getQuery(true); + + $query->select('*') + ->from($query->quoteName($this->table)) + ->where('id = :id') + ->bind('id', $conditions); + + return $this->db->setQuery($query)->loadOne('assoc'); + } + + /** + * remove + * + * @param mixed $conditions + * + * @return bool + */ + public function remove($conditions) + { + $query = $this->db->getQuery(true); + + $query->delete($query->quoteName($this->table)) + ->where('id = :id') + ->bind('id', $conditions); + + $this->db->setQuery($query)->execute(); + + return true; + } + + /** + * clear + * + * @return bool + */ + public function clear() + { + $this->db->getTable($this->table)->truncate(); + + return true; + } + + /** + * Method to get property Table + * + * @return string + */ + public function getTable() + { + return $this->table; + } + + /** + * Method to set property table + * + * @param string $table + * + * @return static Return self to support chaining. + */ + public function setTable($table) + { + $this->table = $table; + + return $this; + } +} diff --git a/Failer/NullQueueFailer.php b/Failer/NullQueueFailer.php new file mode 100644 index 0000000..3549b64 --- /dev/null +++ b/Failer/NullQueueFailer.php @@ -0,0 +1,76 @@ +pdo = $pdo; $this->table = $table; @@ -47,7 +46,7 @@ public function __construct(PDO $pdo, string $table = 'queue_failed_jobs') * * @return bool */ - public function isSupported(): bool + public function isSupported() { $sql = 'SHOW TABLES LIKE :table'; @@ -55,31 +54,31 @@ public function isSupported(): bool $stat->bindValue(':table', $this->table); $stat->execute(); - return (bool) $stat->fetch(PDO::FETCH_ASSOC); + return (bool) $stat->fetch(\PDO::FETCH_ASSOC); } /** * add * - * @param string $connection - * @param string $channel - * @param string $body - * @param string $exception + * @param string $connection + * @param string $queue + * @param string $body + * @param string $exception * * @return int|string */ - public function add(string $connection, string $channel, string $body, string $exception): int|string + public function add($connection, $queue, $body, $exception) { // For B/C - $created = (new DateTime('now'))->format('Y-m-d H:i:s'); + $created = (new \DateTime('now'))->format('Y-m-d H:i:s'); $sql = 'INSERT INTO ' . $this->table . - ' (connection, channel, body, exception, created)' . - ' VALUES (:connection, :channel, :body, :exception, :created)'; + ' (connection, queue, body, exception, created)' . + ' VALUES (:connection, :queue, :body, :exception, :created)'; $stat = $this->pdo->prepare($sql); $stat->bindValue(':connection', $connection); - $stat->bindValue(':channel', $channel); + $stat->bindValue(':queue', $queue); $stat->bindValue(':body', $body); $stat->bindValue(':exception', $exception); $stat->bindValue(':created', $created); @@ -92,29 +91,27 @@ public function add(string $connection, string $channel, string $body, string $e /** * all * - * @return iterable - * @throws RuntimeException - * @throws InvalidArgumentException + * @return array + * @throws \RuntimeException + * @throws \InvalidArgumentException */ - public function all(): iterable + public function all() { $sql = 'SELECT * FROM ' . $this->table; $stat = $this->pdo->query($sql); - while ($item = $stat->fetch(PDO::FETCH_ASSOC)) { - yield $item; - } + return $stat->fetchAll(\PDO::FETCH_ASSOC); } /** * get * - * @param mixed $conditions + * @param mixed $conditions * - * @return array|null + * @return array */ - public function get(mixed $conditions): ?array + public function get($conditions) { $sql = 'SELECT * FROM ' . $this->table . ' WHERE id = :id'; @@ -123,17 +120,17 @@ public function get(mixed $conditions): ?array $stat->bindValue(':id', $conditions); $stat->execute(); - return $stat->fetch(PDO::FETCH_ASSOC); + return $stat->fetch(\PDO::FETCH_ASSOC); } /** * remove * - * @param mixed $conditions + * @param mixed $conditions * * @return bool */ - public function remove(mixed $conditions): bool + public function remove($conditions) { $sql = 'DELETE FROM ' . $this->table . ' WHERE id = :id'; @@ -149,7 +146,7 @@ public function remove(mixed $conditions): bool * * @return bool */ - public function clear(): bool + public function clear() { $sql = 'TRUNCATE TABLE ' . $this->table; @@ -161,7 +158,7 @@ public function clear(): bool * * @return string */ - public function getTable(): string + public function getTable() { return $this->table; } @@ -169,11 +166,11 @@ public function getTable(): string /** * Method to set property table * - * @param string $table + * @param string $table * * @return static Return self to support chaining. */ - public function setTable(string $table): static + public function setTable($table) { $this->table = $table; diff --git a/Failer/QueueFailerInterface.php b/Failer/QueueFailerInterface.php new file mode 100644 index 0000000..b838c65 --- /dev/null +++ b/Failer/QueueFailerInterface.php @@ -0,0 +1,61 @@ +callback = $callback; + $this->name = $name; + } + + /** + * getName + * + * @return string + */ + public function getName() + { + return $this->name; + } + + /** + * handle + * + * @return void + */ + public function execute() + { + $callback = $this->callback; + + $callback(); + } + + /** + * serialize + * + * @return string + * + * @since 3.5.2 + */ + public function serialize() + { + $serializer = new Serializer(); + + return $serializer->serialize($this->callback); + } + + /** + * unserialize + * + * @param string $serialized + * + * @return void + * + * @since 3.5.2 + */ + public function unserialize($serialized) + { + $serializer = new Serializer(); + + $this->callback = $serializer->unserialize($serialized); + } +} diff --git a/Job/JobInterface.php b/Job/JobInterface.php new file mode 100644 index 0000000..065a08e --- /dev/null +++ b/Job/JobInterface.php @@ -0,0 +1,31 @@ +driver = $driver; + + $this->newInstanceHandler = function ($class) { + return new $class(); + }; + } + + /** + * push + * + * @param mixed $job + * @param int $delay + * @param string $queue + * @param array $options + * + * @return int|string + */ + public function push($job, $delay = 0, $queue = null, array $options = []) + { + $message = $this->getMessageByJob($job); + $message->setDelay($delay); + $message->setQueueName($queue); + $message->setOptions($options); + + return $this->driver->push($message); + } + + /** + * pushRaw + * + * @param string|array $body + * @param int $delay + * @param null $queue + * @param array $options + * + * @return int|string + */ + public function pushRaw($body, $delay = 0, $queue = null, array $options = []) + { + if (is_string($body)) { + json_decode($body, true); + } + + $message = new QueueMessage(); + $message->setBody($body); + $message->setDelay($delay); + $message->setQueueName($queue); + $message->setOptions($options); + + return $this->driver->push($message); + } + + /** + * pop + * + * @param string $queue + * + * @return QueueMessage + */ + public function pop($queue = null) + { + return $this->driver->pop($queue); + } + + /** + * delete + * + * @param QueueMessage|mixed $message + * + * @return void + */ + public function delete($message) + { + if (!$message instanceof QueueMessage) { + $msg = new QueueMessage(); + $msg->setId($message); + } + + $this->driver->delete($message); + + $message->isDeleted(true); + } + + /** + * release + * + * @param QueueMessage|mixed $message + * @param int $delay + * + * @return void + */ + public function release($message, $delay = 0) + { + if (!$message instanceof QueueMessage) { + $msg = new QueueMessage(); + $msg->setId($message); + } + + $message->setDelay($delay); + + $this->driver->release($message); + } + + /** + * getMessage + * + * @param mixed $job + * @param array $data + * + * @return QueueMessage + * @throws \InvalidArgumentException + */ + public function getMessageByJob($job, array $data = []) + { + $message = new QueueMessage(); + + $job = $this->createJobInstance($job); + + $data['class'] = get_class($job); + + $message->setName($job->getName()); + $message->setJob(serialize($job)); + $message->setData($data); + + return $message; + } + + /** + * createJobInstance + * + * @param mixed $job + * + * @return JobInterface + * @throws \InvalidArgumentException + */ + protected function createJobInstance($job) + { + if ($job instanceof JobInterface) { + return $job; + } + + // Create callable + if (is_callable($job)) { + $job = new CallableJob($job, md5(uniqid('', true))); + } + + // Create by class name. + if (is_string($job)) { + if (!class_exists($job) || is_subclass_of($job, JobInterface::class)) { + throw new \InvalidArgumentException( + sprintf('Job should be a class which implements JobInterface, %s given', $job) + ); + } + + $handler = $this->newInstanceHandler; + + $job = $handler($job); + + if (!$job instanceof JobInterface) { + throw new \UnexpectedValueException('Job instance is not a JobInterface.'); + } + } + + if (is_array($job)) { + throw new \InvalidArgumentException('Job should not be array.'); + } + + return $job; + } + + /** + * Method to get property Driver + * + * @return QueueDriverInterface + */ + public function getDriver() + { + return $this->driver; + } + + /** + * Method to set property driver + * + * @param QueueDriverInterface $driver + * + * @return static Return self to support chaining. + */ + public function setDriver(QueueDriverInterface $driver) + { + $this->driver = $driver; + + return $this; + } + + /** + * Method to get property NewInstanceHandler + * + * @return callable + * + * @since 3.3 + */ + public function getNewInstanceHandler() + { + return $this->newInstanceHandler; + } + + /** + * Method to set property newInstanceHandler + * + * @param callable $newInstanceHandler + * + * @return static Return self to support chaining. + * + * @since 3.3 + */ + public function setNewInstanceHandler(callable $newInstanceHandler) + { + $this->newInstanceHandler = $newInstanceHandler; + + return $this; + } +} diff --git a/QueueMessage.php b/QueueMessage.php new file mode 100644 index 0000000..3041816 --- /dev/null +++ b/QueueMessage.php @@ -0,0 +1,365 @@ +setJob($job); + } + + if ($data) { + $this->setData($data); + } + + $this->setDelay($delay); + $this->setOptions($options); + } + + /** + * get + * + * @param string $name + * @param mixed $default + * + * @return mixed + */ + public function get($name, $default = null) + { + if (isset($this->body[$name])) { + return $this->body[$name]; + } + + return $default; + } + + /** + * set + * + * @param string $name + * @param mixed $value + * + * @return static + */ + public function set($name, $value) + { + $this->body[$name] = $value; + + return $this; + } + + /** + * Method to get property Id + * + * @return int|string + */ + public function getId() + { + return $this->id; + } + + /** + * Method to set property id + * + * @param int|string $id + * + * @return static Return self to support chaining. + */ + public function setId($id) + { + $this->id = $id; + + return $this; + } + + /** + * Method to get property Attempts + * + * @return int + */ + public function getAttempts() + { + return $this->attempts; + } + + /** + * Method to set property attempts + * + * @param int $attempts + * + * @return static Return self to support chaining. + */ + public function setAttempts($attempts) + { + $this->attempts = $attempts; + + return $this; + } + + /** + * Method to get property Job + * + * @return string + */ + public function getJob() + { + return Arr::get($this->body, 'job', ''); + } + + /** + * Method to set property job + * + * @param string $job + * + * @return static Return self to support chaining. + */ + public function setJob($job) + { + $this->body['job'] = $job; + + return $this; + } + + /** + * Method to get property Data + * + * @return array + */ + public function getData() + { + return Arr::get($this->body, 'data', []); + } + + /** + * Method to set property data + * + * @param array $data + * + * @return static Return self to support chaining. + */ + public function setData(array $data) + { + $this->body['data'] = $data; + + return $this; + } + + /** + * Method to get property Queue + * + * @return string + */ + public function getQueueName() + { + return Arr::get($this->body, 'queue', ''); + } + + /** + * Method to set property queue + * + * @param string $queue + * + * @return static Return self to support chaining. + */ + public function setQueueName($queue) + { + $this->body['queue'] = $queue; + + return $this; + } + + /** + * Method to get property Body + * + * @return array + */ + public function getBody() + { + return $this->body; + } + + /** + * Method to set property body + * + * @param array $body + * + * @return static Return self to support chaining. + */ + public function setBody($body) + { + $this->body = (array) $body; + + return $this; + } + + /** + * Method to get property RawData + * + * @return string + */ + public function getRawBody() + { + return $this->rawBody; + } + + /** + * Method to set property rawData + * + * @param string $rawBody + * + * @return static Return self to support chaining. + */ + public function setRawBody($rawBody) + { + $this->rawBody = $rawBody; + + return $this; + } + + /** + * Method to get property Name + * + * @return string + */ + public function getName() + { + return Arr::get($this->body, 'name', ''); + } + + /** + * Method to set property name + * + * @param string $name + * + * @return static Return self to support chaining. + */ + public function setName($name) + { + $this->body['name'] = $name; + + return $this; + } + + /** + * Method to get property Delay + * + * @return int + */ + public function getDelay() + { + return $this->delay; + } + + /** + * Method to set property delay + * + * @param int $delay + * + * @return static Return self to support chaining. + */ + public function setDelay($delay) + { + $this->delay = (int) $delay; + + return $this; + } + + /** + * isDeleted + * + * @param bool $bool + * + * @return bool|static + */ + public function isDeleted($bool = null) + { + if ($bool === null) { + return $this->deleted; + } + + $this->deleted = (bool) $bool; + + return $this; + } + + /** + * jsonSerialize + * + * @return array + * + * @throws \InvalidArgumentException + */ + public function jsonSerialize() + { + return $this->body; + } +} diff --git a/README.md b/README.md index 5a7e016..ef5042c 100644 --- a/README.md +++ b/README.md @@ -1,33 +1,444 @@ -

-
- Windwalker -
-

+# Windwalker Queue -

Queue

+Windwalker Queue is a queue manager which inspired by Laravel queue package. -

- Windwalker Queue package -

+This package provides an universal interface to wrap different message queue services, and a simple `Job` interface + to easily manage your tasks. Currently we support these drivers: -

- GitHub - GitHub Workflow Status - Packagist Downloads - Packagist Version -

+- [SQS](https://aws.amazon.com/sqs) (Amazon Simple Queue Service) +- [IronMQ](https://www.iron.io/) +- [RabbitMQ](https://www.rabbitmq.com/) (AMQP) +- [Beanstalkd](http://kr.github.io/beanstalkd/) +- [PHP Resque](https://github.com/chrisboulton/php-resque) (Redis) +- Pdo (MySQL) +- Sync (No queue, execute immediately) -### Installation +## Installation via Composer -```bash -composer require windwalker/queue ^4.0 +Add this to the require block in your `composer.json`. + +``` json +{ + "require": { + "windwalker/queue": "~3.0" + } +} +``` + +## Getting Started + +Create a new Queue instance with AWS SQS driver, remember send AWS key and secret into it. + +```php +push(new HelloJob()); + +// Delay 5 seconds +$queue->push(new HelloJob(), 5); + +// Push to anoter queue +$queue->push(new HelloJob(), 0, 'flower'); +``` + +### Add More Information to Job + +Sometimes you need more information to handle things, add them to constructor: + +```php +class HelloJob implements JobInterface +{ + protected $url; + protected $path; + protected $size; + protected $crop; + + public function __construct($url, $path, $size = 600, $crop = true) + { + $this->url = $url; + $this->path = $path; + $this->size = $size; + $this->crop = $crop; + } + + public function getName() + { + return 'hello'; + } + + public function execute() + { + $imgData = (new HttpClient)->get($this->url); + + ImageHelper::load($imgData) + ->resize($this->size, $this->size, $this->crop) + ->save($this->path) + } +} +``` + +Then inject these information when you creating Jobs: + +```php +$queue->push( + new HelloJob( + 'http://example/image.jpg', + __DIR__ . '/../images/image.jpg', + 400, + true + ) +); +``` + +Then Let's run Worker in CLI: + +```php +use Windwalker\Event\Dispatcher; +use Windwalker\Queue\Queue; +use Windwalker\Queue\Worker; + +$worker = new Worker(new Queue($driver), new Dispatcher()); + +// Run once +$worker->runNextJob(['default', 'flower']); + +// Or run as deamon +$worker->loop(['default', 'flower']); +``` + +## Use Queue Object + +### Push + +Simple use push to add a job object as new message: + +```php +$queue->push(new MyJob($data)); +``` + +Push message but wait for 10 seconds later to run: + +```php +$queue->push(new MyJob($data), 10); +``` + +Push to directly queue: + +```php +$queue->push(new MyJob($data), 0, 'flower'); +``` + +Push raw data instead job object: + +```php +$queue->pushRaw(['flower' => 'sakura'], 0, 'flower'); +``` + +### Pop, Delete and Release + +Use `pop()` to get next message: + +```php +$message = $queue->pop(); // QueueMessage object + +$message->getJob(); +$message->getBody(); +$message->getRawBody(); +$message->getId(); +$message->getAttempts(); +$message->get('flower'); // Get data from body +``` + +Delete a message: + +```php +$queue->delete($message); + +// argument should be a QueueMessage object +$message = new \Windwalker\Core\Queue\QueueMessage; +$message->setId($id); + +$queue->delete($message); + +// You can delete by ID +$queue->delete($id); + +// Check this message deleted +$message->isDeleted(); +``` + +Release back to queue list (attempts will auto +1): + +```php +$queue->release($message); + +// You can release by ID +$queue->release($id); + +// Wait a while to run again: +$queue->release($message, 15); +``` + +## Use Worker + +`Worker` will auto fetch new job to run in the background, you can integrate it to your command line programs like +Symfony Console or your own CLI system. + +Create Worker instance: + +```php +use Windwalker\Event\Dispatcher; +use Windwalker\Queue\Queue; +use Windwalker\Queue\Worker; + +$worker = new Worker(new Queue($driver), new Dispatcher()); + +// You can set PSR-3 Logger into it. +$worker = new Worker(new Queue($driver), new Dispatcher(), new MyLogger); +``` + +We recommend that you should prepare a Logger to log all queue messages then you can easily debug and know worker information. + +### Options + +There are many options that you can configure it, you must wrap options array by a `Structure` object: + +```php +use Windwalker\Structure\Structure; + +$options = [ + 'timeout' => 30, // Number of seconds that a job can run. + 'delay' => 0, // Delay time for failed job to wait next run. + 'force' => false, // Force run worker if in pause mode. + 'memory' => 128, // The memory limit in megabytes. + 'sleep' => 1, // Number of seconds to sleep after job run complete. + 'tries' => 5, // Number of times to attempt a job if it failed. + 'restart_signal' => '/path/to/restart_signal_file', // Restart signal +]; + +$worker->loop(['default', 'flower'], new Structure($options)); +``` + +### Pause Mode and `force` option + +You can pause Worker by your CLI process: + +```php +if (fil_get_content('/mode') === 'offline') { + $worker->setState(Worker::STATE_PAUSE); +} else { + $worker->setState(Worker::STATE_ACTIVE); +} +``` + +In pause state, Worker will still run but won't execute any jobs, but if you set `force` option to TRUE, it will +ignore pause state and still run jobs. + +### Restart Signals + +Sometimes you code updated and you wish all background Workers can restart to use new code. You can prepare a file with +timestamp. + +Write restart signal + +```php +format('U')); +``` + +Then if you set `restart_signal` as `/path/to/restart_signal_file`, if Worker found it's start time less than restart time, +it will auto stop process. The the daemon watcher you set in the system will auto raise a new Worker. + +### Listen to Workers + +Works has some events that you can get messages and print to terminal or write logs. + +```php +use Windwalker\Event\Event; + +$worker->getDispatcher() + ->listen('onWorkerBeforeJobRun', function (Event $event) { + $job = $event['job']; + $message = $event['message']; + + // Print to terminal + echo sprintf( + 'Run Job: %s - Message ID: %s', + $job->getName(), + $message->getId() + ); + }) + ->listen('onWorkerJobFailure', function (Event $event) { + $job = $event['job']; + $e = $event['exception']; + $message = $event['message']; + + // Print to terminal + echo sprintf( + 'Job %s failed: %s (%s)', + $job->getName(), + $e->getMessage(), + $message->getId() + ); + + // If be deleted, send to failed table + if ($message->isDeleted()) { + $failer->add( + $this->console->get('queue.connection', 'sync'), + $message->getQueueName(), + json_encode($message), + (string) $e + ); + } + }) + ->listen('onWorkerLoopCycleStart', function (Event $event) { + /** @var Worker $worker */ + $worker = $event['worker']; + + switch ($worker->getState()) { + case $worker::STATE_ACTIVE: + if ($this->console->isOffline()) { + $worker->setState($worker::STATE_PAUSE); + } + break; + + case $worker::STATE_PAUSE: + if ($this->console->isOffline()) { + $worker->setState($worker::STATE_ACTIVE); + } + break; + } + }) + ->listen('onWorkerLoopCycleFailure', function (Event $event) { + /** @var \Exception $e */ + $e = $event['exception']; + + // Print to terminal + echo sprintf( + '%s File: %s (%s)', + $e->getMessage(), + $e->getFile(), + $e->getLine() + ); + }) + ->listen('onWorkerLoopCycleEnd', function (Event $event) { + // + }); +``` + +Available events: + +- onWorkerLoopCycleStart +- onWorkerLoopCycleFailure +- onWorkerLoopCycleEnd +- onWorkerBeforeJobRun +- onWorkerAfterJobRun +- onWorkerStop +- onWorkerJobFailure + +## Drivers + +### Supported Driver + +```php +use Windwalker\Queue\Queue; + +// AWS SQS +$driver = new \Windwalker\Queue\Driver\SqsQueueDriver($key, $secret, 'default'); + +// IronMQ +$driver = new \Windwalker\Queue\Driver\IronmqQueueDriver($projectId, $token, 'default'); + +// RabbitMQ +$driver = new \Windwalker\Queue\Driver\RabbitmqQueueDriver('default'); + +// Beanstalkd +$driver = new \Windwalker\Queue\Driver\BeanstalkdQueueDriver('127.0.0.1', 'default'); + +// PHP Resque (Redis) +$driver = new \Windwalker\Queue\Driver\ResqueQueueDriver('127.0.0.1', 'default'); + +// PDO +$driver = new \Windwalker\Queue\Driver\PdoQueueDriver($pdo, 'default', 'queue_jobs'/* table name*/); +``` + +### PDO Driver + +You must create a table to handle queue, here is an example [SQL file](Resources/sql/queue_jobs.sql). + +## Failed Jobs + +If jobs failed, you may want to log them in a place and retry later or check the error message. Windwalker provides +a database driven failed handler. Please copy [SQL file](Resources/sql/queue_failed_jobs.sql) here. + +Now you must log failed jobs in Worker failure events. + +```php +use Windwalker\Event\Event; + +$failer = new \Windwalker\Queue\Failer\PdoQueueFailer($pdo); + +$worker->getDispatcher() + ->listen('onWorkerJobFailure', function (Event $event) use ($failer) { + $job = $event['job']; + $e = $event['exception']; + $message = $event['message']; + + // Print message to terminal... + + // If be deleted, send to failed table + if ($message->isDeleted()) { + $failer->add( + 'pdo', + $message->getQueueName(), + json_encode($message), + (string) $e + ); + } + }); ``` -### Resources +Use your CLI program to send failed jobs back to queue. + +```php +// In CLI -- [Documentation](https://windwalker.io/documentation/components/queue/) -- [Bug report](https://github.com/windwalker-io/framework) -- [Contributing](https://github.com/windwalker-io/framework) -- [Discussion](https://github.com/windwalker-io/framework/discussions) +$fails = $failer->all(); + +foreach ($fails as $failed) { + $queue->pushRaw(json_decode($failed['body'], true), 5, $failed['queue']); +} +``` +More examples, please see: https://windwalker.io/documentation/3.x/services/queue.html#failed-jobs diff --git a/resources/sql/queue_failed_jobs.sql b/Resources/sql/queue_failed_jobs.sql similarity index 83% rename from resources/sql/queue_failed_jobs.sql rename to Resources/sql/queue_failed_jobs.sql index ce6a8b6..1a665fa 100644 --- a/resources/sql/queue_failed_jobs.sql +++ b/Resources/sql/queue_failed_jobs.sql @@ -1,8 +1,7 @@ -DROP TABLE if EXISTS `queue_failed_jobs`; CREATE TABLE IF NOT EXISTS `queue_failed_jobs` ( `id` int(11) unsigned NOT NULL, `connection` varchar(255) NOT NULL DEFAULT '', - `channel` varchar(255) NOT NULL DEFAULT '', + `queue` varchar(255) NOT NULL DEFAULT '', `body` longtext NOT NULL, `exception` longtext NOT NULL, `created` datetime NOT NULL DEFAULT '1000-01-01 00:00:00' diff --git a/Resources/sql/queue_jobs.sql b/Resources/sql/queue_jobs.sql new file mode 100644 index 0000000..6ac002b --- /dev/null +++ b/Resources/sql/queue_jobs.sql @@ -0,0 +1,16 @@ +CREATE TABLE IF NOT EXISTS `queue_jobs` ( + `id` bigint(20) unsigned NOT NULL, + `queue` varchar(255) NOT NULL DEFAULT '', + `body` longtext NOT NULL, + `attempts` tinyint(4) unsigned NOT NULL DEFAULT '0', + `created` datetime NOT NULL DEFAULT '1000-01-01 00:00:00', + `visibility` datetime NOT NULL DEFAULT '1000-01-01 00:00:00', + `reserved` datetime DEFAULT NULL +) ENGINE=InnoDB DEFAULT CHARSET=utf8; + +ALTER TABLE `queue_jobs` + ADD PRIMARY KEY (`id`), + ADD KEY `idx_queue_jobs_queue` (`queue`); + +ALTER TABLE `queue_jobs` + MODIFY `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT; diff --git a/src/Resque/Resque.php b/Resque/Resque.php similarity index 84% rename from src/Resque/Resque.php rename to Resque/Resque.php index 5d27f75..e9f98c0 100644 --- a/src/Resque/Resque.php +++ b/Resque/Resque.php @@ -1,6 +1,10 @@ - 0) { return self::removeItems($queue, $items); @@ -38,12 +42,12 @@ public static function dequeue(string $queue, array $items = []): int * * @private * - * @param string $queue The name of the queue - * @param array $items + * @param string $queue The name of the queue + * @param array $items * * @return integer number of deleted items */ - protected static function removeItems(string $queue, array $items = []): int + protected static function removeItems($queue, $items = []) { $counter = 0; $originalQueue = 'queue:' . $queue; @@ -97,7 +101,7 @@ protected static function removeItems(string $queue, array $items = []): int * * @return (bool) */ - protected static function matchItem($string, $items): bool + protected static function matchItem($string, $items) { $decoded = json_decode($string, true); @@ -111,10 +115,8 @@ protected static function matchItem($string, $items): bool } elseif (is_array($val)) { $decodedArgs = (array) $decoded['args'][0]; - if ( - $decoded['class'] == $key && count($decodedArgs) > 0 - && count(array_diff($decodedArgs, $val)) == 0 - ) { + if ($decoded['class'] == $key && count($decodedArgs) > 0 + && count(array_diff($decodedArgs, $val)) == 0) { return true; } // class name with ID, example: item[0] = ['class' => 'id'] @@ -137,7 +139,7 @@ protected static function matchItem($string, $items): bool * * @return integer number of deleted items belongs to this list */ - protected static function removeList($queue): int + protected static function removeList($queue) { $counter = self::size($queue); $result = self::redis()->del('queue:' . $queue); @@ -150,7 +152,7 @@ protected static function removeList($queue): int * * @return string */ - public static function generateJobId(): string + public static function generateJobId() { return md5(uniqid('', true)); } diff --git a/Test/test.php b/Test/test.php new file mode 100644 index 0000000..5064912 --- /dev/null +++ b/Test/test.php @@ -0,0 +1,25 @@ + 30, // Number of seconds that a job can run. + 'delay' => 0, // Delay time for failed job to wait next run. + 'force' => false, // Force run worker if in pause mode. + 'memory' => 128, // The memory limit in megabytes. + 'sleep' => 1, // Number of seconds to sleep after job run complete. + 'tries' => 5, // Number of times to attempt a job if it failed. +]; + +$worker->loop(['default', 'flower'], new Structure($options)); diff --git a/Worker.php b/Worker.php new file mode 100644 index 0000000..ac926a0 --- /dev/null +++ b/Worker.php @@ -0,0 +1,486 @@ +manager = $manager; + $this->dispatcher = $dispatcher ?: new Dispatcher(); + $this->logger = $logger ?: new NullLogger(); + } + + /** + * loop + * + * @param string|array $queue + * @param Structure $options + * + * @return void + * @throws \Exception + */ + public function loop($queue, Structure $options) + { + gc_enable(); + + // Last Restart + $this->lastRestart = (new \DateTimeImmutable('now'))->format('U'); + + // Log PID + $this->pid = getmypid(); + + $this->logger->info('A worker start running... PID: ' . $this->pid); + + $this->setState(static::STATE_ACTIVE); + + while (true) { + $this->gc(); + + // @loop start + $this->triggerEvent( + 'onWorkerLoopCycleStart', + [ + 'worker' => $this, + 'manager' => $this->manager, + ] + ); + + // Timeout handler + $this->registerSignals($options); + + if ($this->canLoop() || $options->get('force')) { + try { + $this->runNextJob($queue, $options); + } catch (\Exception $e) { + $msg = sprintf('Worker failure in a loop cycle: %s', $e->getMessage()); + + $this->logger->error($msg); + + $this->triggerEvent( + 'onWorkerLoopCycleFailure', + [ + 'worker' => $this, + 'exception' => $e, + 'message' => $msg, + ] + ); + } + } + + $this->stopIfNecessary($options); + + // @loop end + $this->triggerEvent( + 'onWorkerLoopCycleEnd', + [ + 'worker' => $this, + 'manager' => $this->manager, + ] + ); + + $this->sleep((int) $options->get('sleep', 1)); + } + } + + /** + * runNextJob + * + * @param string|array $queue + * @param Structure $options + * + * @return void + */ + public function runNextJob($queue, Structure $options) + { + $message = $this->getNextMessage($queue); + + if (!$message) { + return; + } + + $this->process($message, $options); + } + + /** + * process + * + * @param QueueMessage $message + * @param Structure $options + * + * @return void + */ + public function process(QueueMessage $message, Structure $options) + { + $maxTries = (int) $options->get('tries', 5); + + $job = $message->getJob(); + /** @var JobInterface $job */ + $job = unserialize($job); + + try { + // @before event + $this->triggerEvent( + 'onWorkerBeforeJobRun', + [ + 'worker' => $this, + 'message' => $message, + 'job' => $job, + 'manager' => $this->manager, + ] + ); + + // Fail if max attempts + if ($maxTries !== 0 && $maxTries < $message->getAttempts()) { + $this->manager->delete($message); + + throw new MaxAttemptsExceededException('Max attempts exceed for Message: ' . $message->getId()); + } + + // run + $job->execute(); + + // @after event + $this->triggerEvent( + 'onWorkerAfterJobRun', + [ + 'worker' => $this, + 'message' => $message, + 'job' => $job, + 'manager' => $this->manager, + ] + ); + + $this->manager->delete($message); + } catch (\Throwable $t) { + $this->handleJobException($job, $message, $options, $t); + } finally { + if (!$message->isDeleted()) { + $this->manager->release($message, (int) $options->get('delay', 0)); + } + } + } + + /** + * canLoop + * + * @return bool + */ + protected function canLoop() + { + return $this->getState() === static::STATE_ACTIVE; + } + + /** + * registerTimeoutHandler + * + * @param Structure $options + * + * @return void + */ + protected function registerSignals(Structure $options) + { + $timeout = (int) $options->get('timeout', 60); + + if (!extension_loaded('pcntl')) { + return; + } + + declare (ticks=1); + + if ($timeout !== 0) { + pcntl_signal( + SIGALRM, + function () use ($timeout) { + $this->stop('A job process over the max timeout: ' . $timeout . ' PID: ' . $this->pid); + } + ); + + pcntl_alarm($timeout + $options->get('sleep')); + } + + // Wait job complete then stop + pcntl_signal(SIGINT, [$this, 'shutdown']); + pcntl_signal(SIGTERM, [$this, 'shutdown']); + } + + /** + * shoutdown + * + * @return void + */ + public function shutdown() + { + $this->setState(static::STATE_EXITING); + } + + /** + * stop + * + * @param string $reason + * + * @return void + */ + public function stop($reason = 'Unkonwn reason') + { + $this->logger->info('Worker stop: ' . $reason); + + $this->triggerEvent( + 'onWorkerStop', + [ + 'worker' => $this, + 'reason' => $reason, + ] + ); + + $this->setState(static::STATE_STOP); + + exit(); + } + + /** + * handleException + * + * @param JobInterface $job + * @param QueueMessage $message + * @param Structure $options + * @param \Exception|\Throwable $e + * + * @return void + */ + protected function handleJobException($job, QueueMessage $message, Structure $options, $e) + { + if (!$job instanceof JobInterface) { + $job = new NullJob(); + } + + $this->logger->error( + sprintf( + 'Job [%s] (%s) failed: %s - Class: %s', + $job->getName(), + $message->getId(), + $e->getMessage(), + get_class($job) + ) + ); + + if (method_exists($job, 'failed')) { + $job->failed($e); + } + + $maxTries = (int) $options->get('tries', 5); + + // Delete and log error if reach max attempts. + if ($maxTries !== 0 && $maxTries <= $message->getAttempts()) { + $this->manager->delete($message); + $this->logger->error( + sprintf( + 'Max attempts exceeded. Job: %s (%s) - Class: %s', + $job->getName(), + $message->getId(), + get_class($job) + ) + ); + } + + $this->dispatcher->triggerEvent( + 'onWorkerJobFailure', + [ + 'worker' => $this, + 'exception' => $e, + 'job' => $job, + 'message' => $message, + ] + ); + } + + /** + * getNextMessage + * + * @param $queue + * + * @return null|QueueMessage + */ + protected function getNextMessage($queue) + { + $queue = (array) $queue; + + foreach ($queue as $queueName) { + if ($message = $this->manager->pop($queueName)) { + return $message; + } + } + + return null; + } + + /** + * sleep + * + * @param int $seconds + * + * @return void + */ + protected function sleep($seconds) + { + usleep($seconds * 1000000); + } + + /** + * Method to perform basic garbage collection and memory management in the sense of clearing the + * stat cache. We will probably call this method pretty regularly in our main loop. + * + * @return void + */ + protected function gc() + { + // Perform generic garbage collection. + gc_collect_cycles(); + + // Clear the stat cache so it doesn't blow up memory. + clearstatcache(); + } + + /** + * Method to get property Manager + * + * @return Queue + */ + public function getManager() + { + return $this->manager; + } + + /** + * Method to get property State + * + * @return string + */ + public function getState() + { + return $this->state; + } + + /** + * setState + * + * @param string $state + * + * @return void + */ + public function setState($state) + { + $this->state = $state; + } + + /** + * stopIfNecessary + * + * @param Structure $options + * + * @return void + */ + public function stopIfNecessary(Structure $options) + { + $restartSignal = $options->get('restart_signal'); + + if (is_file($restartSignal)) { + $signal = file_get_contents($restartSignal); + + if ($this->lastRestart < $signal) { + $this->stop('Receive restart signal. PID: ' . $this->pid); + } + } + + if ((memory_get_usage() / 1024 / 1024) >= (int) $options->get('memory_limit', 128)) { + $this->stop('Memory usage exceeded. PID: ' . $this->pid); + } + + if ($this->getState() === static::STATE_EXITING) { + $this->stop('Shutdown by signal. PID: ' . $this->pid); + } + } +} diff --git a/composer.json b/composer.json index 557aa2b..d06f20e 100644 --- a/composer.json +++ b/composer.json @@ -2,27 +2,19 @@ "name": "windwalker/queue", "type": "windwalker-package", "description": "Windwalker Queue package", - "keywords": [ - "windwalker", - "framework", - "queue" - ], - "homepage": "https://github.com/windwalker-io/queue", - "license": "MIT", + "keywords": ["windwalker", "framework", "queue"], + "homepage": "https://github.com/ventoviro/windwalker-queue", + "license": "LGPL-2.0-or-later", "require": { - "php": ">=8.4.6", - "windwalker/utilities": "^4.0", - "windwalker/event": "^4.0", + "php": ">=7.1.3", + "windwalker/event": "~3.0", + "windwalker/structure": "~3.0", "ext-json": "*" }, "require-dev": { - "phpunit/phpunit": "^10.0||^11.0||^12.0", - "windwalker/test": "^4.0", - "windwalker/data": "^4.0", - "windwalker/database": "^4.0", - "asika/sql-splitter": "^1.0", - "jdorn/sql-formatter": "^1.2", - "laravel/serializable-closure": "^1.0" + "windwalker/test": "~3.0", + "windwalker/database": "~3.0", + "jdorn/sql-formatter": "1.*" }, "suggest": { "aws/aws-sdk-php": "If you want to use AWS SQS as queue service.", @@ -30,35 +22,17 @@ "php-amqplib/php-amqplib": "If you want to use RabbitMQ as queue service.", "pda/pheanstalk": "If you want to use Beanstalkd queue as service.", "chrisboulton/php-resque": "Install <= 1.2 if you want to use PHP Resque (Redis) as queue service.", - "laravel/serializable-closure": "Install ^1.0 to support Closure jobs." + "jeremeamia/SuperClosure": "Install ~2.0 to support anonymous function jobs." }, "minimum-stability": "beta", "autoload": { "psr-4": { - "Windwalker\\Queue\\": "src/" - }, - "files": [ - "src/bootstrap.php" - ] - }, - "autoload-dev": { - "psr-4": { - "Windwalker\\Queue\\Test\\": "test/" + "Windwalker\\Queue\\": "" } }, "extra": { - "windwalker": { - "packages": [ - "Windwalker\\Queue\\QueuePackage" - ] - }, "branch-alias": { - "dev-master": "4.x-dev" - } - }, - "config": { - "platform": { - "php": "8.4.6" + "dev-master": "3.x-dev" } } } diff --git a/etc/queue.config.php b/etc/queue.config.php deleted file mode 100644 index e7bcba5..0000000 --- a/etc/queue.config.php +++ /dev/null @@ -1,88 +0,0 @@ - [ - - 'default' => env('QUEUE_DEFAULT') ?: 'sync', - - 'failer_default' => env('QUEUE_FAILER_DEFAULT') ?: 'database', - - 'providers' => [ - QueuePackage::class, - ], - - 'listeners' => [ - // - ], - - 'bindings' => [ - // - ], - - 'init_scripts' => [ - // - ], - - 'loop_end_scripts' => [ - // - ], - - 'job_end_scripts' => [ - // - ], - - 'enqueuer' => [ - 'handlers' => [ - __DIR__ . '/../../resources/registry/enqueuers.php', - ], - 'init_scripts' => [ - // - ], - 'loop_end_scripts' => [ - // - ], - ], - - 'factories' => [ - 'instances' => [ - 'sync' => static fn() => QueueFactory::syncAdapter( - handler: QueueFactory::createSyncHandler() - ), - 'sqs' => static fn() => QueueFactory::sqsAdapter( - key: env('QUEUE_SQS_KEY'), - secret: env('QUEUE_SQS_SECRET'), - channel: 'default', - options: [ - 'region' => env('QUEUE_SQS_REGION') ?: 'us-west-2', - 'version' => env('QUEUE_SQS_VERSION') ?: 'latest', - ] - ), - 'database' => static fn(DatabaseAdapter $db) => QueueFactory::databaseAdapter( - db: $db, - channel: 'default', - table: 'queue_jobs', - timeout: 60, - idType: DatabaseIdType::INT - ), - ], - 'failers' => [ - 'database' => static fn(DatabaseServiceFactory $factory) => new DatabaseQueueFailer( - db: $factory->get(), - table: 'queue_failed_jobs', - idType: DatabaseIdType::INT - ), - ], - ], -]; diff --git a/phpunit.ci.xml b/phpunit.ci.xml deleted file mode 100644 index 2f528c3..0000000 --- a/phpunit.ci.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - - - - - - - - test - - - diff --git a/phpunit.travis.xml b/phpunit.travis.xml deleted file mode 100644 index 2f528c3..0000000 --- a/phpunit.travis.xml +++ /dev/null @@ -1,19 +0,0 @@ - - - - - - - - - - - test - - - diff --git a/phpunit.xml.dist b/phpunit.xml.dist deleted file mode 100644 index 3bd0ea0..0000000 --- a/phpunit.xml.dist +++ /dev/null @@ -1,19 +0,0 @@ - - - - - - - - - - - test - - - diff --git a/resources/sql/queue_jobs.sql b/resources/sql/queue_jobs.sql deleted file mode 100644 index 1eeeb93..0000000 --- a/resources/sql/queue_jobs.sql +++ /dev/null @@ -1,17 +0,0 @@ -DROP TABLE IF EXISTS `queue_jobs`; -CREATE TABLE IF NOT EXISTS `queue_jobs` ( - `id` bigint(20) unsigned NOT NULL, - `channel` varchar(255) NOT NULL DEFAULT '', - `body` longtext NOT NULL, - `attempts` tinyint(4) NOT NULL DEFAULT '0', - `created` datetime DEFAULT NULL, - `visibility` datetime DEFAULT NULL, - `reserved` datetime DEFAULT NULL -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; - -ALTER TABLE `queue_jobs` - ADD PRIMARY KEY (`id`), - ADD KEY `idx_queue_jobs_channel` (`channel`); - -ALTER TABLE `queue_jobs` - MODIFY `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT; diff --git a/src/AbstractRunner.php b/src/AbstractRunner.php deleted file mode 100644 index 34f98de..0000000 --- a/src/AbstractRunner.php +++ /dev/null @@ -1,362 +0,0 @@ -lastRestart = (int) new DateTimeImmutable('now')->format('U'); - - // Run Times - $this->runTimes = 0; - - // Log PID - $this->pid = getmypid(); - - $this->logger->info("A {$this->getRunnerName()} start running... PID: {$this->pid}"); - - $this->setState(static::STATE_ACTIVE); - - while (!$this->isStop()) { - $this->gc(); - - $worker = $this; - $queue = $this->queue; - - // @loop start - $this->emit(new LoopStartEvent(runner: $worker, queue: $queue)); - - // Timeout handler - $this->registerSignals(); - - if (($this->options->force ?? null) || $this->canLoop()) { - try { - $this->next($channel); - } catch (UnrecoverableException $e) { - $this->stop('[STOP] Unrecoverable error: ' . $e->getMessage(), 1, true); - } catch (Exception $exception) { - $message = sprintf( - '%s failure in a loop cycle: %s', - $this->getRunnerName(), - $exception->getMessage() - ); - - $this->logger->error($message); - - $this->emit(new LoopFailureEvent(runner: $worker, message: $message, exception: $exception)); - } - } - - $this->stopIfNecessary(); - - // @loop end - $this->emit(new LoopEndEvent(runner: $worker, queue: $queue)); - - $this->sleep((float) $this->options->sleep); - } - } - - /** - * canLoop - * - * @return bool - */ - protected function canLoop(): bool - { - return $this->getState() === static::STATE_ACTIVE; - } - - /** - * @return void - */ - protected function registerSignals(): void - { - $timeout = $this->options->timeout; - - if (!extension_loaded('pcntl')) { - return; - } - - declare(ticks=1); - - if ($timeout !== 0) { - pcntl_signal( - SIGALRM, - function () use ($timeout) { - $this->stop('A loop process over the max timeout: ' . $timeout . ' PID: ' . $this->pid); - }, - ); - - pcntl_alarm((int) ($timeout + $this->options->sleep)); - } - - // Wait job complete then stop - pcntl_signal(SIGINT, $this->shutdown(...)); - pcntl_signal(SIGTERM, $this->shutdown(...)); - } - - /** - * shoutdown - * - * @return void - */ - public function shutdown(): void - { - $this->setState(static::STATE_EXITING); - } - - /** - * stop - * - * @param string $reason - * @param int $code - * @param bool $instant - * - * @return void - */ - public function stop(string $reason = 'Unknown reason', int $code = 0, bool $instant = false): void - { - $this->logger->info($this->getRunnerName() . ' stop: ' . $reason); - - $this->emit( - new StopEvent( - reason: $reason, - runner: $this, - queue: $this->queue, - ), - ); - - $this->setState(static::STATE_STOP); - - if ($instant) { - exit($code); - } - } - - public function isStop(): bool - { - return in_array( - $this->getState(), - [ - static::STATE_EXITING, - static::STATE_STOP, - ], - true, - ); - } - - /** - * sleep - * - * @param float $seconds - * - * @return void - */ - protected function sleep(float $seconds): void - { - usleep((int) ($seconds * 1000_000)); - } - - /** - * Method to perform basic garbage collection and memory management in the sense of clearing the - * stat cache. We will probably call this method pretty regularly in our main loop. - * - * @return void - */ - protected function gc(): void - { - // Perform generic garbage collection. - gc_collect_cycles(); - - // Clear the stat cache so it doesn't blow up memory. - clearstatcache(); - } - - /** - * Method to get property Manager - * - * @return Queue - */ - public function getQueue(): Queue - { - return $this->queue; - } - - /** - * Method to get property State - * - * @return string - */ - public function getState(): string - { - return $this->state; - } - - /** - * setState - * - * @param string $state - * - * @return void - */ - public function setState(string $state): void - { - $this->state = $state; - } - - /** - * @return void - */ - public function stopIfNecessary(): void - { - $restartSignal = $this->options->restartSignal; - - if ($restartSignal && is_file($restartSignal)) { - $signal = file_get_contents($restartSignal); - - if ($this->lastRestart < $signal) { - $this->stop('Receive restart signal. PID: ' . $this->pid); - } - } - - if ($this->options->maxRuns > 0 && ++$this->runTimes >= $this->options->maxRuns) { - $this->stop('Max run times reached. PID: ' . $this->pid); - } - - if ( - $this->options->lifetime > 0 - && $this->lastRestart - && (time() - $this->lastRestart) >= $this->options->lifetime - ) { - $this->stop('Max lifetime reached. PID: ' . $this->pid); - } - - if ((memory_get_usage() / 1024 / 1024) >= $this->options->memoryLimit) { - $this->stop('Memory usage exceeded. PID: ' . $this->pid); - } - - if ($this->getState() === static::STATE_EXITING) { - $this->stop('Shutdown by signal. PID: ' . $this->pid); - } - } - - public function getInvoker(): callable - { - return $this->invoker ??= $this->getDefaultInvoker(); - } - - /** - * @param ?callable $invoker - * - * @return static Return self to support chaining. - */ - public function setInvoker(?callable $invoker): static - { - $this->invoker = $invoker; - - return $this; - } - - abstract protected function getDefaultInvoker(): \Closure; - - protected function createControllerLogger(): LoggerInterface - { - return new CallbackLogger( - function ($level, string|\Stringable $message, array $context = []) { - $message = (string) $message; - $this->logger->log($level, strip_tags($message), $context); - - return $this->emit( - new DebugOutputEvent( - level: $level, - message: $message, - context: $context, - ) - ); - } - ); - } -} diff --git a/src/Attributes/JobBackoff.php b/src/Attributes/JobBackoff.php deleted file mode 100644 index 5c85243..0000000 --- a/src/Attributes/JobBackoff.php +++ /dev/null @@ -1,74 +0,0 @@ -= 3 | (1, 2, 9, 27, 81, 243, ...) - * - adaptive = pow: >= 2, adaptive: > 1 | (1, 4, 9, 16, 25, 36, ...) - * - * @param int $n - * @param int $base - * @param int $pow - * @param int|null $maxTimes - * @param int $maxSeconds - * @param int $adaptive - * - * @return int|false - */ - public static function backoff( - int $n, - int $base = 1, - int $pow = 2, - ?int $maxTimes = null, - int $maxSeconds = 86400, - int $adaptive = 1, - ): int|false { - if ($maxTimes !== null && $n >= $maxTimes) { - return false; - } - - return min($maxSeconds, $base * ($pow ** $n) * $adaptive); - } - - public static function resolve(int $attempts, array|int|null|false $backoff): int|null|false - { - if (is_array($backoff)) { - return $backoff[$attempts - 1] ?? false; - } - - return $backoff; - } - - public static function fromController(JobController $controller): false|int|null - { - // Find Attribute if set on the job class. - if ( - $attr = AttributesAccessor::getFirstAttributeInstance( - $controller->job, - static::class, - \ReflectionAttribute::IS_INSTANCEOF - ) - ) { - $backoff = $attr->backoff; - } else { - // Find Attribute if set on the method, only get once. - $backoff = $controller->invokeMethodsWithAttribute(static::class)->current(); - } - - return static::resolve($controller->attempts, $backoff); - } -} diff --git a/src/Attributes/JobEntry.php b/src/Attributes/JobEntry.php deleted file mode 100644 index 36ff09f..0000000 --- a/src/Attributes/JobEntry.php +++ /dev/null @@ -1,10 +0,0 @@ -channel = $channel; - - $this->connect($host, $port); - } - - /** - * push - * - * @param QueueMessage $message - * - * @return string - * @throws DomainException - */ - public function push(QueueMessage $message): string - { - $channel = $message->getChannel() ?: $this->channel; - - if (!$message->getId()) { - $message->set('attempts', 0); - $message->set('channel', $channel); - $message->set('id', Resque::generateJobId()); - $message->set('class', static::JOB_CLASS); - $message->setId($message->getId()); - } - - $delay = $message->getDelay(); - - $data = json_decode(json_encode($message), true); - - if ($delay > 0) { - static::supportDelayed(true); - - ResqueScheduler::delayedPush(time() + $delay, $data); - } else { - Resque::push($channel, $data); - } - - return (string) $message->getId(); - } - - /** - * pop - * - * @param string|null $channel - * - * @return QueueMessage|null - */ - public function pop(?string $channel = null): ?QueueMessage - { - if (static::supportDelayed()) { - $this->rechannelDelayedItems(); - } - - $channel = $channel ?: $this->channel; - - $job = Resque::pop($channel); - - if (!$job) { - return null; - } - - $message = new QueueMessage(); - - $attempts = $job['attempts']; - $attempts++; - - $message->setId($job['id']); - $message->setBody($job); - $message->setRawBody(json_encode($job)); - $message->setChannel($channel ?: $this->channel); - $message->setAttempts($attempts); - $message->set('attempts', $attempts); - - return $message; - } - - /** - * delete - * - * @param QueueMessage $message - * - * @return ResqueQueueDriver - */ - public function delete(QueueMessage $message): static - { - $channel = $message->getChannel() ?: $this->channel; - - Resque::dechannel($channel, [static::JOB_CLASS => $message->getId()]); - - return $this; - } - - /** - * release - * - * @param QueueMessage $message - * - * @return static - */ - public function release(QueueMessage $message): static - { - $this->push($message); - - return $this; - } - - /** - * Handle delayed items for the next scheduled timestamp. - * - * Searches for any items that are due to be scheduled in Resque - * and adds them to the appropriate job channel in Resque. - */ - public function rechannelDelayedItems(): void - { - while (($oldestJobTimestamp = ResqueScheduler::nextDelayedTimestamp()) !== false) { - $this->enchannelDelayedItemsForTimestamp($oldestJobTimestamp); - } - } - - /** - * Schedule all of the delayed jobs for a given timestamp. - * - * Searches for all items for a given timestamp, pulls them off the list of - * delayed jobs and pushes them across to Resque. - * - * @param DateTime|int $timestamp Search for any items up to this timestamp to schedule. - */ - public function enchannelDelayedItemsForTimestamp(DateTime|int $timestamp) - { - $item = null; - - while ($item = ResqueScheduler::nextItemForTimestamp($timestamp)) { - Resque::push($item['channel'], $item); - } - } - - /** - * supportDelayed - * - * @param bool $throwError - * - * @return bool - * @throws DomainException - */ - public static function supportDelayed($throwError = false): bool - { - if (!class_exists(ResqueScheduler::class)) { - if ($throwError) { - throw new DomainException( - 'Please install chrisboulton/php-resque-scheduler to support delayed messages for Resque.' - ); - } - - return false; - } - - return true; - } - - /** - * connect - * - * @param string $host - * @param int $port - * - * @return void - * @throws DomainException - */ - public function connect(string $host, int $port): void - { - if (!class_exists(Resque::class)) { - throw new DomainException('Please install chrisboulton/php-resque 1.2 to support Resque driver.'); - } - - Resque::setBackend("$host:$port"); - } -} diff --git a/src/Driver/DatabaseQueueDriver.php b/src/Driver/DatabaseQueueDriver.php deleted file mode 100644 index 4d3d16f..0000000 --- a/src/Driver/DatabaseQueueDriver.php +++ /dev/null @@ -1,306 +0,0 @@ -idType = $idType; - } - - /** - * push - * - * @param QueueMessage $message - * - * @return string - * @throws Exception - */ - public function push(QueueMessage $message): string - { - $now = new DateTimeImmutable('now'); - $time = $message->createdAt ?? $now; - - $data = [ - 'channel' => $message->getChannel() ?: $this->channel, - 'body' => json_encode($message, JSON_THROW_ON_ERROR), - 'attempts' => 0, - 'created' => $time->format('Y-m-d H:i:s'), - 'visibility' => $now->modify(sprintf('+%dseconds', $message->getDelay())), - 'reserved' => null, - ]; - - if ($this->idType->isUuid()) { - $data['id'] = $this->generateUuidString(); - } - - $data = $this->db->getWriter()->insertOne($this->table, $data, 'id'); - - return (string) $this->idType->toWritable($data['id']); - } - - /** - * pop - * - * @param string|null $channel - * - * @return QueueMessage|null - * @throws Throwable - */ - public function pop(?string $channel = null): ?QueueMessage - { - $channel = $channel ?: $this->channel; - - $now = new DateTimeImmutable('now'); - - $query = $this->db->getQuery(true); - - $do = $this->checkCanSkipLocked() ? 'SKIP LOCKED' : ''; - - $query->select('*') - ->from($this->table) - ->where('channel', $channel) - ->where('visibility', '<=', $now) - ->orWhere( - function (Query $query) use ($now) { - $query->where('reserved', null) - // After a timeout if a job was not released or hanging. - ->where('reserved', '<', $now->modify('-' . $this->timeout . 'seconds')); - } - ) - ->order('visibility', 'ASC') - ->order('id', 'ASC') - ->forUpdate($do); - - $data = $this->db->transaction( - function () use ($now, $query) { - try { - $data = $this->db->prepare($query)->get(); - } catch (\Exception $e) { - if ($this->db->getDriver()::shouldAutoReconnect($e)) { - // DB should already auto reconnect, if reconnect failed, we force queue restart. - UnrecoverableException::throwFrom($e); - } - - throw $e; - } - - if (!$data) { - return null; - } - - $rawId = $data['id']; - $data['id'] = $this->idType->toReadable($data['id']); - - $data['attempts']++; - - $values = ['reserved' => $now, 'attempts' => $data['attempts']]; - - $this->db->getWriter()->updateWhere($this->table, $values, ['id' => $rawId]); - - return $data; - } - ); - - if ($data === null) { - return null; - } - - $message = new QueueMessage(); - - $message->setId($data['id']); - $message->setAttempts((int) $data['attempts']); - $message->setBody(json_decode($data['body'], true, 512, JSON_THROW_ON_ERROR)); - $message->setRawBody($data['body']); - $message->setChannel($channel); - $message->setCreatedAt($data['created']); - - return $message; - } - - /** - * delete - * - * @param QueueMessage $message - * - * @return static - */ - public function delete(QueueMessage $message): static - { - $channel = $message->getChannel() ?: $this->channel; - - $this->db->delete($this->table) - ->where('id', $this->idType->toWritable($message->getId())) - ->where('channel', $channel) - ->execute(); - - return $this; - } - - /** - * @param QueueMessage $message - * - * @return static - * @throws \DateMalformedStringException - */ - public function release(QueueMessage $message): static - { - $channel = $message->getChannel() ?: $this->channel; - - $time = new DateTimeImmutable('now'); - $time = $time->modify('+' . $message->getDelay() . 'seconds'); - - $values = [ - 'reserved' => null, - 'visibility' => $time, - 'attempts' => $message->getAttempts(), - ]; - - $this->db->getWriter()->updateWhere( - $this->table, - $values, - [ - 'id' => $this->idType->toWritable($message->getId()), - 'channel' => $channel, - ] - ); - - return $this; - } - - public function defer(QueueMessage $message): static - { - $message->setAttempts($message->getAttempts() - 1); - - return $this->release($message); - } - - /** - * Method to get property Table - * - * @return string - */ - public function getTable(): string - { - return $this->table; - } - - /** - * Method to set property table - * - * @param mixed $table - * - * @return static Return self to support chaining. - */ - public function setTable(string $table): static - { - $this->table = $table; - - return $this; - } - - /** - * Method to get property Db - * - * @return DatabaseAdapter - */ - public function getDb(): DatabaseAdapter - { - return $this->db; - } - - /** - * Method to set property db - * - * @param DatabaseAdapter $db - * - * @return static Return self to support chaining. - */ - public function setDb(DatabaseAdapter $db): static - { - $this->db = $db; - - return $this; - } - - /** - * Reconnect database to avoid long connect issues. - * - * @return static - */ - public function reconnect(): static - { - $this->disconnect(); - - $this->db->connect(); - - return $this; - } - - /** - * Disconnect DB. - * - * @return static - * - * @since 3.5.2 - */ - public function disconnect(): static - { - $this->db->disconnect(); - - return $this; - } - - protected function checkCanSkipLocked(): bool - { - if ($this->canSkipLocked !== null) { - return $this->canSkipLocked; - } - - try { - $this->db->createQuery()->selectRaw('1')->from($this->table)->forUpdate('SKIP LOCKED')->get(); - $this->canSkipLocked = true; - } catch (Throwable) { - $this->canSkipLocked = false; - } - - return $this->canSkipLocked; - } -} diff --git a/src/Driver/InfinityQueueDriver.php b/src/Driver/InfinityQueueDriver.php deleted file mode 100644 index 791aaf1..0000000 --- a/src/Driver/InfinityQueueDriver.php +++ /dev/null @@ -1,50 +0,0 @@ -job = new ClosureJob($job); - } - - public function push(QueueMessage $message): string - { - return $message->getId() ?: ''; - } - - public function pop(?string $channel = null): ?QueueMessage - { - return new QueueMessage($this->job) - ->setId(uid('mq__')) - ->setCreatedAt('now'); - } - - public function delete(QueueMessage $message): static - { - // Do nothing, as this is an infinite queue. - return $this; - } - - public function release(QueueMessage $message): static - { - // Do nothing, as this is an infinite queue. - return $this; - } - - public function defer(QueueMessage $message): static - { - // Do nothing, as this is an infinite queue. - return $this; - } -} diff --git a/src/Driver/NullQueueDriver.php b/src/Driver/NullQueueDriver.php deleted file mode 100644 index a439d29..0000000 --- a/src/Driver/NullQueueDriver.php +++ /dev/null @@ -1,68 +0,0 @@ -idType = $idType; - } - - /** - * push - * - * @param QueueMessage $message - * - * @return string - * @throws Exception - */ - public function push(QueueMessage $message): string - { - $time = $message->createdAt ?? new DateTimeImmutable('now'); - - $data = [ - ':channel' => $message->getChannel() ?: $this->channel, - ':body' => json_encode($message), - ':attempts' => 0, - ':created' => $time->format('Y-m-d H:i:s'), - ':visibility' => $time->modify(sprintf('+%dseconds', $message->getDelay()))->format('Y-m-d H:i:s'), - ':reserved' => null, - ]; - - if ($this->idType->isUuid()) { - $id = $this->generateUuidString(); - $idColumn = 'id, '; - $idValue = ':id, '; - $data[':id'] = $id; - } - - $sql = 'INSERT INTO ' . $this->table . - " ({$idColumn}channel, body, attempts, created, visibility, reserved)" . - " VALUES ({$idValue}:channel, :body, :attempts, :created, :visibility, :reserved)"; - - $this->pdo->prepare($sql)->execute($data); - - return (string) $this->pdo->lastInsertId(); - } - - /** - * pop - * - * @param string|null $channel - * - * @return QueueMessage|null - * @throws Exception - * @throws InvalidArgumentException - * @throws Throwable - */ - public function pop(?string $channel = null): ?QueueMessage - { - $channel = $channel ?: $this->channel; - - $now = new DateTimeImmutable('now'); - - $sql = 'SELECT * FROM ' . $this->table . - ' WHERE channel = :channel AND visibility <= :visibility' . - ' AND (reserved IS NULL OR reserved < :reserved)' . - ' FOR UPDATE'; - - $this->pdo->beginTransaction(); - - $stat = $this->pdo->prepare($sql); - $stat->bindValue(':channel', $channel, PDO::PARAM_STR); - $stat->bindValue(':visibility', $now->format('Y-m-d H:i:s'), PDO::PARAM_STR); - $stat->bindValue( - ':reserved', - $now->modify('-' . $this->timeout . 'seconds')->format('Y-m-d H:i:s'), - PDO::PARAM_STR - ); - - try { - $stat->execute(); - - $data = $stat->fetch(PDO::FETCH_ASSOC); - - if (!$data) { - $this->pdo->commit(); - - return null; - } - - $rawId = $data['id']; - $data['id'] = $this->idType->toReadable($data['id']); - $data['attempts']++; - - $sql = 'UPDATE ' . $this->table . ' SET reserved = :reserved, attempts = :attempts WHERE id = :id'; - - $stat = $this->pdo->prepare($sql); - $stat->bindValue(':reserved', $now->format('Y-m-d H:i:s')); - $stat->bindValue(':attempts', $data['attempts'] + 1); - $stat->bindValue(':id', $rawId); - - $stat->execute(); - - $this->pdo->commit(); - } catch (Throwable $t) { - $this->pdo->rollBack(); - throw $t; - } - - $message = new QueueMessage(); - - $message->setId($data['id']); - $message->setAttempts((int) $data['attempts']); - $message->setBody(json_decode($data['body'], true)); - $message->setRawBody($data['body']); - $message->setChannel($channel); - - return $message; - } - - /** - * delete - * - * @param QueueMessage $message - * - * @return PdoQueueDriver - */ - public function delete(QueueMessage $message): static - { - $channel = $message->getChannel() ?: $this->channel; - - $sql = 'DELETE FROM ' . $this->table . - ' WHERE id = :id AND channel = :channel'; - - $stat = $this->pdo->prepare($sql); - $stat->bindValue(':id', $this->idType->toWritable($message->getId())); - $stat->bindValue(':channel', $channel); - - $stat->execute(); - - return $this; - } - - /** - * @param QueueMessage $message - * - * @return static - * @throws \DateMalformedStringException - */ - public function release(QueueMessage $message): static - { - $channel = $message->getChannel() ?: $this->channel; - - $time = new DateTimeImmutable('now'); - $time = $time->modify('+' . $message->getDelay() . 'seconds'); - - $values = [ - 'id' => $this->idType->toWritable($message->getId()), - 'channel' => $channel, - 'reserved' => null, - 'visibility' => $time->format('Y-m-d H:i:s'), - 'attempts' => $message->getAttempts(), - ]; - - $sql = 'UPDATE ' . $this->table . - ' SET reserved = :reserved, visibility = :visibility, attempts = :attempts' . - ' WHERE id = :id AND channel = :channel'; - - $stat = $this->pdo->prepare($sql); - - $stat->execute($values); - - return $this; - } - - public function defer(QueueMessage $message): static - { - $this->delete($message); - - $message->setDeleted(false); - $message->setId(''); - $message->setAttempts($message->getAttempts() - 1); - - $this->push($message); - - return $this; - } - - /** - * Method to get property Table - * - * @return string - */ - public function getTable(): string - { - return $this->table; - } - - /** - * Method to set property table - * - * @param string $table - * - * @return static Return self to support chaining. - */ - public function setTable(string $table): static - { - $this->table = $table; - - return $this; - } - - /** - * Method to get property Db - * - * @return PDO - */ - public function getPdo(): PDO - { - return $this->pdo; - } - - /** - * Method to set property db - * - * @param PDO $pdo - * - * @return static Return self to support chaining. - */ - public function setPdo(PDO $pdo): static - { - $this->pdo = $pdo; - - return $this; - } - - /** - * Reconnect database to avoid long connect issues. - * - * @return static - */ - public function reconnect(): static - { - // PDO cannot reconnect. - - return $this; - } -} diff --git a/src/Driver/QueueDriverInterface.php b/src/Driver/QueueDriverInterface.php deleted file mode 100644 index 365928e..0000000 --- a/src/Driver/QueueDriverInterface.php +++ /dev/null @@ -1,54 +0,0 @@ -client = $this->getSqsClient($key, $secret, $options); - - $this->channel = $channel; - } - - /** - * push - * - * @param QueueMessage $message - * - * @return string - * @throws JsonException - */ - public function push(QueueMessage $message): string - { - $channel = $message->getChannel() ?: $this->channel; - - $request = [ - 'QueueUrl' => $this->getQueueUrl($channel), - 'MessageBody' => json_encode($message, JSON_THROW_ON_ERROR), - ]; - - $request['DelaySeconds'] = $message->getDelay(); - - $options = $message->getOptions(); - - if (str_ends_with($channel, '.fifo')) { - $options['MessageGroupId'] ??= '1'; - } - - $request = array_merge($request, $options); - - return (string) $this->client->sendMessage($request)->get('MessageId'); - } - - /** - * pop - * - * @param string|null $channel - * - * @return QueueMessage|null - */ - public function pop(?string $channel = null): ?QueueMessage - { - $result = $this->client->receiveMessage( - [ - 'QueueUrl' => $this->getQueueUrl($channel), - 'AttributeNames' => ['ApproximateReceiveCount'], - ] - ); - - if ($result['Messages'] === null) { - return null; - } - - $data = $result['Messages'][0]; - - $message = new QueueMessage(); - - $message->setId($data['MessageId']); - $message->setAttempts((int) $data['Attributes']['ApproximateReceiveCount']); - $message->setBody(json_decode($data['Body'], true)); - $message->setRawBody($data['Body']); - $message->setChannel($channel ?: $this->channel); - $message->set('ReceiptHandle', $data['ReceiptHandle']); - - return $message; - } - - /** - * delete - * - * @param QueueMessage $message - * - * @return static - */ - public function delete(QueueMessage $message): static - { - $this->client->deleteMessage( - [ - 'QueueUrl' => $this->getQueueUrl($message->getChannel()), - 'ReceiptHandle' => $this->getReceiptHandle($message), - ] - ); - - return $this; - } - - /** - * @param QueueMessage $message - * - * @return static - */ - public function release(QueueMessage $message): static - { - $this->client->changeMessageVisibility( - [ - 'QueueUrl' => $this->getQueueUrl($message->getChannel()), - 'ReceiptHandle' => $this->getReceiptHandle($message), - 'VisibilityTimeout' => $message->getDelay(), - ] - ); - - return $this; - } - - public function defer(QueueMessage $message): static - { - // Do nothing, as SQS does not support deferring messages. - // Let's make the job timeout, so it will be retried later without increasing attempts. - return $this; - } - - /** - * getQueueUrl - * - * @param string|null $channel - * - * @return string - */ - public function getQueueUrl(?string $channel = null): string - { - $channel = $channel ?: $this->channel; - - if (filter_var($channel, FILTER_VALIDATE_URL) !== false) { - return $channel; - } - - return $this->client->getQueueUrl(['QueueName' => $channel])->get('QueueUrl'); - } - - /** - * getReceiptHandle - * - * @param QueueMessage $message - * - * @return string - */ - public function getReceiptHandle(QueueMessage $message): string - { - return $message->get('ReceiptHandle', $message->getId()); - } - - /** - * getSqsClient - * - * @param string $key - * @param string $secret - * @param array $options - * - * @return SqsClient - * @throws DomainException - */ - public function getSqsClient(string $key, string $secret, array $options = []): SqsClient - { - if (!class_exists(SqsClient::class)) { - throw new DomainException('Please install aws/aws-sdk-php first.'); - } - - $defaultOptions = [ - 'region' => 'ap-northeast-1', - 'version' => 'latest', - 'credentials' => [ - 'key' => $key, - 'secret' => $secret, - ], - ]; - - $options = array_merge($defaultOptions, $options); - - return new SqsClient($options); - } -} diff --git a/src/Driver/SyncQueueDriver.php b/src/Driver/SyncQueueDriver.php deleted file mode 100644 index dce2247..0000000 --- a/src/Driver/SyncQueueDriver.php +++ /dev/null @@ -1,139 +0,0 @@ -handler = $handler ?? static::getDefaultHandler(); - $this->prepareOptions( - [ - 'debug' => false - ], - $options - ); - } - - /** - * push - * - * @param QueueMessage $message - * - * @return string - */ - public function push(QueueMessage $message): string - { - $output = ($this->handler)($message); - - $debug = $this->getOption('debug'); - - if ($debug) { - if (is_callable($debug)) { - $debug($output, $this); - } else { - show($output); - } - } - - return (string) TypeCast::tryString($output); - } - - /** - * @return callable - */ - public function getHandler(): callable - { - return $this->handler; - } - - /** - * @param callable $handler - * - * @return static Return self to support chaining. - */ - public function setHandler(callable $handler): static - { - $this->handler = $handler; - - return $this; - } - - protected function runJob(callable $job) - { - return $job(); - } - - /** - * pop - * - * @param string|null $channel - * - * @return QueueMessage|null - */ - public function pop(?string $channel = null): ?QueueMessage - { - return null; - } - - /** - * delete - * - * @param QueueMessage $message - * - * @return SyncQueueDriver - */ - public function delete(QueueMessage $message): static - { - return $this; - } - - /** - * release - * - * @param QueueMessage $message - * - * @return static - */ - public function release(QueueMessage $message): static - { - return $this; - } - - public function defer(QueueMessage $message): static - { - return $this; - } - - public static function getDefaultHandler(): \Closure - { - return static function (QueueMessage $message) { - return $message->run(); - }; - } -} diff --git a/src/Driver/UuidDriverTrait.php b/src/Driver/UuidDriverTrait.php deleted file mode 100644 index 30e59cf..0000000 --- a/src/Driver/UuidDriverTrait.php +++ /dev/null @@ -1,35 +0,0 @@ -checkDependencyAndGenerateUuid(); - - return $this->idType === DatabaseIdType::UUID_BIN ? $uuid->getBytes() : $uuid->toString(); - } - - public function checkDependencyAndGenerateUuid(): UuidInterface - { - if (!class_exists(Uuid::class)) { - throw new \DomainException('Please install `ramsey/uuid` to use UUID queue driver.'); - } - - return $this->generateUuid(); - } - - public function generateUuid(): UuidInterface - { - return Uuid::uuid7(); - } -} diff --git a/src/Enqueuer.php b/src/Enqueuer.php deleted file mode 100644 index ad49e6f..0000000 --- a/src/Enqueuer.php +++ /dev/null @@ -1,134 +0,0 @@ -enqueue($chan) === null && $empty; - } - - if ($this->options->stopWhenEmpty && $empty) { - $this->stop('Nothing to enqueue or return from handlers, exit.'); - } - } - - public function enqueue(string $channel): mixed - { - $handler = $this->channelHandlers[$channel] ?? $this->defaultHandler ?? null; - - if (!$handler) { - return null; - } - - if ($this->options->maxRuns > 0) { - $this->runTimes++; - } - - $controller = $this->createEnqueuerController($channel); - $controller->queue = $this->queue; - - $event = $this->emit( - new BeforeEnqueueEvent( - controller: $controller, - enqueuer: $this, - queue: $this->queue, - ) - ); - - $controller = $event->controller; - - try { - $result = $controller->run($handler); - - $event = $this->emit( - new AfterEnqueueEvent( - controller: $controller, - enqueuer: $this, - queue: $this->queue, - result: $result, - ) - ); - - return $event->result; - } catch (\Throwable $e) { - $this->emit( - new EnqueueFailureEvent( - exception: $e, - controller: $controller, - enqueuer: $this, - queue: $this->queue, - ), - ); - - return null; - } - } - - public function default(callable $defaultHandler): static - { - $this->defaultHandler = $defaultHandler(...); - - return $this; - } - - public function channel(string $channel, callable $handler): void - { - $this->channelHandlers[$channel] = $handler; - } - - public function setChannelHandlers(array $channelHandlers): Enqueuer - { - $this->channelHandlers = $channelHandlers; - - return $this; - } - - protected function getDefaultInvoker(): \Closure - { - return fn(EnqueuerController $controller, callable $invokable) => $invokable($controller); - } - - public function createEnqueuerController(string $channel): EnqueuerController - { - if ($this->options->controllerFactory) { - return ($this->options->controllerFactory)( - $channel, - $this->getInvoker(), - $this->createControllerLogger(), - ); - } - - return new EnqueuerController($channel, $this->getInvoker(), $this->createControllerLogger()); - } -} diff --git a/src/Enqueuer/EnqueuerController.php b/src/Enqueuer/EnqueuerController.php deleted file mode 100644 index ff06c3a..0000000 --- a/src/Enqueuer/EnqueuerController.php +++ /dev/null @@ -1,47 +0,0 @@ -invoker = $invoker ?? fn(self $controller, callable $invokable, array $args = []) => $invokable( - $controller, - ...$args - ); - } - - public function run(callable $invokable, array $args = []): mixed - { - return ($this->invoker)($this, $invokable, $args); - } - - public function log(string|array $message, string $level = LogLevel::INFO, array $context = []): static - { - foreach ((array) $message as $msg) { - $this->logger->log($level, $msg, $context); - } - - return $this; - } - - public function enqueue(mixed $job, int $delay = 0, ?string $channel = null, array $options = []): int|string - { - return $this->queue->push($job, $delay, $channel ?? $this->channel, $options); - } -} diff --git a/src/Enum/DatabaseIdType.php b/src/Enum/DatabaseIdType.php deleted file mode 100644 index 36eedad..0000000 --- a/src/Enum/DatabaseIdType.php +++ /dev/null @@ -1,89 +0,0 @@ -getBytes(); - } - - if (strlen($id) === 16) { - return $id; - } - - if (strlen((string) $id) !== 36) { - throw new \InvalidArgumentException('Invalid UUID string: ' . (string) $id); - } - - return Uuid::fromString($id)->getBytes(); - } - - if ($this === self::UUID) { - if ($id instanceof UuidInterface) { - return (string) $id; - } - - if (strlen($id) === 16) { - return Uuid::fromBytes($id)->toString(); - } - - if (strlen((string) $id) !== 36) { - throw new \InvalidArgumentException('Invalid UUID string: ' . (string) $id); - } - - return (string) $id; - } - - if (!is_numeric($id)) { - throw new \InvalidArgumentException('ID must be integer: ' . (string) $id); - } - - return $id; - } - - public function toReadable(mixed $id): int|string - { - if ($this === self::INT) { - return (string) $id; - } - - return (string) to_uuid($id); - } -} diff --git a/src/Event/AfterEnqueueEvent.php b/src/Event/AfterEnqueueEvent.php deleted file mode 100644 index cda0945..0000000 --- a/src/Event/AfterEnqueueEvent.php +++ /dev/null @@ -1,23 +0,0 @@ -runner = $runner; - $this->queue = $queue; - $this->controller = $controller; - } -} diff --git a/src/Event/BeforeEnqueueEvent.php b/src/Event/BeforeEnqueueEvent.php deleted file mode 100644 index 9faba60..0000000 --- a/src/Event/BeforeEnqueueEvent.php +++ /dev/null @@ -1,21 +0,0 @@ -controller = $controller; - $this->runner = $runner; - $this->queue = $queue; - } -} diff --git a/src/Event/DebugOutputEvent.php b/src/Event/DebugOutputEvent.php deleted file mode 100644 index bb9df5d..0000000 --- a/src/Event/DebugOutputEvent.php +++ /dev/null @@ -1,18 +0,0 @@ -exception = $exception; - } -} diff --git a/src/Event/JobEventTrait.php b/src/Event/JobEventTrait.php deleted file mode 100644 index 2591544..0000000 --- a/src/Event/JobEventTrait.php +++ /dev/null @@ -1,32 +0,0 @@ - $this->runner; - } - - public QueueMessage $message { - get => $this->controller->message; - } - - // phpcs:disable - public object $job { - get => $this->message->getJob(); - } -} diff --git a/src/Event/JobFailureEvent.php b/src/Event/JobFailureEvent.php deleted file mode 100644 index fed3cca..0000000 --- a/src/Event/JobFailureEvent.php +++ /dev/null @@ -1,34 +0,0 @@ -exception = $exception; - $this->controller = $controller; - $this->runner = $runner; - $this->queue = $queue; - } -} diff --git a/src/Event/LoopEndEvent.php b/src/Event/LoopEndEvent.php deleted file mode 100644 index 8fc8edf..0000000 --- a/src/Event/LoopEndEvent.php +++ /dev/null @@ -1,27 +0,0 @@ -runner = $runner; - $this->queue = $queue; - } -} diff --git a/src/Event/LoopFailureEvent.php b/src/Event/LoopFailureEvent.php deleted file mode 100644 index ac110f8..0000000 --- a/src/Event/LoopFailureEvent.php +++ /dev/null @@ -1,27 +0,0 @@ -runner = $runner; - $this->queue = $queue; - } -} diff --git a/src/Event/QueueEventTrait.php b/src/Event/QueueEventTrait.php deleted file mode 100644 index 5755e3a..0000000 --- a/src/Event/QueueEventTrait.php +++ /dev/null @@ -1,21 +0,0 @@ -runner = $runner; - $this->queue = $queue; - } -} diff --git a/src/Exception/AbandonedException.php b/src/Exception/AbandonedException.php deleted file mode 100644 index 82bdb79..0000000 --- a/src/Exception/AbandonedException.php +++ /dev/null @@ -1,27 +0,0 @@ -getMessage(), $e->getCode(), $e); - } - - public static function throwFrom(\Throwable $e): never - { - throw static::from($e); - } - - public function toReasonText(): string - { - if ($this->getMessage()) { - return 'Job abandoned, reason: ' . $this->getMessage(); - } - - return 'Job abandoned.'; - } -} diff --git a/src/Exception/DeferredException.php b/src/Exception/DeferredException.php deleted file mode 100644 index 856c41e..0000000 --- a/src/Exception/DeferredException.php +++ /dev/null @@ -1,32 +0,0 @@ -getMessage(), $e->getCode(), $e); - } - - public static function throwFrom(int $delay, \Throwable $e): never - { - throw static::from($delay, $e); - } - - public function getReasonText(): string - { - if ($this->getMessage()) { - return 'released after ' . $this->delay . ' seconds, reason: ' . $this->getMessage(); - } - - return 'released after ' . $this->delay . ' seconds.'; - } -} diff --git a/src/Exception/MaxAttemptsExceededException.php b/src/Exception/MaxAttemptsExceededException.php deleted file mode 100644 index 35902f2..0000000 --- a/src/Exception/MaxAttemptsExceededException.php +++ /dev/null @@ -1,16 +0,0 @@ -getMessage(), (int) $e->getCode(), $e); - } - - public static function throwFrom(\Throwable $e): never - { - throw static::from($e); - } -} diff --git a/src/Failer/DatabaseQueueFailer.php b/src/Failer/DatabaseQueueFailer.php deleted file mode 100644 index d1411ef..0000000 --- a/src/Failer/DatabaseQueueFailer.php +++ /dev/null @@ -1,195 +0,0 @@ -idType = $idType; - } - - /** - * isSupported - * - * @return bool - */ - public function isSupported(): bool - { - return $this->db->getTableManager($this->table)->exists(); - } - - /** - * add - * - * @param string $connection - * @param string $channel - * @param string $body - * @param string $exception - * - * @return int|string - * @throws JsonException - */ - public function add(string $connection, string $channel, string $body, string $exception): int|string - { - $data = compact( - 'connection', - 'channel', - 'body', - 'exception' - ); - - if ($this->idType->isUuid()) { - $data['id'] = $this->generateUuidString(); - } - - $data['created'] = new DateTime('now'); - - $data = $this->db->getWriter()->insertOne($this->table, $data, 'id'); - - return $data['id']; - } - - /** - * all - * - * @return iterable - * @throws RuntimeException - * @throws InvalidArgumentException - */ - public function all(): iterable - { - $query = $this->db->select('*') - ->from($this->table); - - /** @var Collection $item */ - foreach ($query as $item) { - $item->id = $this->idType->toReadable($item->id); - - yield $item; - } - } - - /** - * get - * - * @param mixed $conditions - * - * @return array|null - */ - public function get(mixed $conditions): ?array - { - $conditions = $this->makeConditionsWritable($conditions); - - $item = $this->db->select('*') - ->from($this->table) - ->where('id', $conditions) - ->get(); - - if (!$item) { - return null; - } - - $item->id = $this->idType->toReadable($item->id); - - return $item->dump(); - } - - /** - * remove - * - * @param mixed $conditions - * - * @return bool - */ - public function remove(mixed $conditions): bool - { - $conditions = $this->makeConditionsWritable($conditions); - - $this->db->delete($this->table) - ->where('id', $conditions) - ->execute() - ->countAffected(); - - return true; - } - - /** - * clear - * - * @return bool - */ - public function clear(): bool - { - $this->db->getTableManager($this->table)->truncate(); - - return true; - } - - /** - * Method to get property Table - * - * @return string - */ - public function getTable(): string - { - return $this->table; - } - - /** - * Method to set property table - * - * @param string $table - * - * @return static Return self to support chaining. - */ - public function setTable(string $table): static - { - $this->table = $table; - - return $this; - } - - /** - * @param mixed $conditions - * - * @return int|mixed - */ - public function makeConditionsWritable(mixed $conditions): mixed - { - if (!is_array($conditions)) { - $conditions = $this->idType->toWritable($conditions); - } elseif (isset($conditions['id'])) { - $conditions['id'] = $this->idType->toWritable($conditions['id']); - } - - return $conditions; - } -} diff --git a/src/Failer/NullQueueFailer.php b/src/Failer/NullQueueFailer.php deleted file mode 100644 index 1a8f63d..0000000 --- a/src/Failer/NullQueueFailer.php +++ /dev/null @@ -1,72 +0,0 @@ -job = $job; - } - - /** - * Find Backoff attribute from job closure. - * - * @return array|int|null - * - * @throws \ReflectionException - */ - #[JobBackoff] - public function backoff(): array|int|null - { - $attr = AttributesAccessor::getFirstAttributeInstance( - $this->getJobClosure(), - JobBackoff::class, - \ReflectionAttribute::IS_INSTANCEOF - ); - - if (!$attr) { - return null; - } - - return $attr->backoff; - } - - /** - * @return SerializableClosure - */ - public function getJob(): SerializableClosure - { - return $this->job; - } - - /** - * @return Closure - */ - public function getJobClosure(): Closure - { - return $this->job->getClosure(); - } - - public function __invoke(JobController $controller): void - { - $callback = $this->getJobClosure(); - - $callback($controller); - } -} diff --git a/src/Job/JobController.php b/src/Job/JobController.php deleted file mode 100644 index 0c3add1..0000000 --- a/src/Job/JobController.php +++ /dev/null @@ -1,357 +0,0 @@ - $this->message->getAttempts(); - } - - public string|int $id { - get => $this->message->getId(); - } - - public int $delay { - get => $this->message->getDelay(); - } - - public bool $deleted { - get => $this->message->isDeleted(); - } - - public array $body { - get => $this->message->getBody(); - } - - public object $job { - get => $this->message->getJob(); - } - - public bool $failed { - get => $this->exception !== null; - } - - public array $extra = []; - - public protected(set) ?DeferredException $defer = null; - - public ?\Throwable $exception = null; - - public protected(set) ?AbandonedException $abandoned = null; - - public bool $maxAttemptsExceeds = false; - - public bool $shouldDelete { - get => $this->maxAttemptsExceeds || $this->abandoned; - } - - public bool $willRetry { - get => !$this->shouldDelete; - } - - public \Closure $invoker; - - protected \Generator $middlewares; - - /** - * @psalm-var 'pending'|'middleware'|'running' - */ - protected string $context = 'pending'; - - public function __construct( - readonly public QueueMessage $message, - ?\Closure $invoker = null, - public LoggerInterface $logger = new NullLogger(), - ) { - $this->invoker = $invoker ?? fn(JobController $controller, callable $invokable, array $args = []) => $invokable( - $controller, - ...$args - ); - } - - public function log(string|array $message, string $level = LogLevel::INFO, array $context = []): static - { - foreach ((array) $message as $msg) { - $this->logger->log($level, $msg, $context); - } - - return $this; - } - - public function defer(int $delay = 0, string $reason = ''): DeferredException - { - return $this->defer = new DeferredException($delay, $reason); - } - - public function resume(): static - { - $this->defer = null; - - return $this; - } - - public function failed(\Throwable|string $e, ?int $code = null): \Throwable - { - if (is_string($e)) { - $e = new \RuntimeException($e, $code ?? 0); - } - - return $this->exception = $e; - } - - public function success(): static - { - $this->exception = null; - - return $this; - } - - public function abandoned(string|\Throwable $reason = ''): AbandonedException - { - if (is_string($reason)) { - $e = new AbandonedException($reason); - } else { - $e = AbandonedException::from($reason); - } - - return $this->abandoned = $e; - } - - public function keep(): static - { - $this->abandoned = null; - - return $this; - } - - /** - * @template T - * - * @param class-string $attrName - * - * @return \Generator }> - */ - protected function findMethodsAttributes(string $attrName): \Generator - { - if (!is_object($this->job)) { - return; - } - - foreach (new \ReflectionObject($this->job)->getMethods() as $method) { - if ($attrs = $method->getAttributes($attrName, \ReflectionAttribute::IS_INSTANCEOF)) { - yield $method->getName() => [$method, $attrs[0]]; - } - } - } - - public function invokeMethodsWithAttribute(string $attrName, ...$args): \Generator - { - foreach ($this->findMethodsAttributes($attrName) as $name => [$method, $attr]) { - yield $name => $this->invoke($method->getClosure($this->job), $args); - } - } - - public function invoke(callable $invokable, array $args = []) - { - return ($this->invoker)($this, $invokable, $args); - } - - public function run(): static - { - if ($this->context !== 'pending') { - throw new \RuntimeException( - sprintf('JobController is already in %s context, cannot run again.', $this->context) - ); - } - - if (is_callable($this->job)) { - // Job is invokable, we can call it directly. - $last = function () { - $this->invoke($this->job); - - return $this; - }; - } else { - // Find JobEntry Attribute - $entry = $this->findMethodsAttributes(JobEntry::class)->current(); - - if (!$entry) { - throw new \RuntimeException( - sprintf( - 'Job %s must have a method with %s attribute or be invokable.', - get_debug_type($this->job), - JobEntry::class - ) - ); - } - - /** @var \ReflectionMethod $method */ - $method = $entry[0]; - $last = function () use ($method) { - $this->context = 'running'; - - $this->invoke($method->getClosure($this->job)); - - return $this; - }; - } - - $this->compileMiddlewares( - $this->job, - $last - ); - - $this->context = 'middleware'; - - $controller = $this->next(); - - $this->context = 'pending'; - - return $controller; - } - - public function next(): static - { - if ($this->context !== 'middleware') { - throw new \RuntimeException( - sprintf('JobController is already in %s context, cannot run next middleware.', $this->context) - ); - } - - $current = $this->middlewares->current(); - - $this->middlewares->next(); - - return $this->invokeMiddleware($current); - } - - protected function compileMiddlewares(mixed $job, \Closure $last): static - { - if (!is_object($job)) { - return $this; - } - - $middlewares = function () use ($last, $job) { - $queue = new PriorityQueue(); - - // Get middlewares from JobMiddlewaresProvider attributes. - foreach ($this->invokeMethodsWithAttribute(JobMiddlewaresProvider::class) as $methodName => $items) { - if (!is_iterable($items)) { - throw new \RuntimeException( - sprintf( - '%s::%s() must return an iterable of middleware list, %s given.', - get_debug_type($this->job), - $methodName, - get_debug_type($items) - ) - ); - } - - foreach ($items as $i => $item) { - $attr = AttributesAccessor::getFirstAttributeInstance( - $item, - JobMiddleware::class, - \ReflectionAttribute::IS_INSTANCEOF - ); - - $queue->insert($item, $attr?->order ?? $i); - } - } - - $o = $queue->count() - 1; - - // Get middlewares from JobMiddleware attributes. - foreach ($this->findMethodsAttributes(JobMiddleware::class) as [$method, $attr]) { - /** @var JobMiddleware $attrInstance */ - $attrInstance = $attr->newInstance(); - $o++; - - $queue->insert($method->getClosure($job), $attrInstance->order ?? $o); - } - - // PriorityQueue is ordered by ascending order, we need to reverse it to process from small to large. - foreach (array_reverse($queue->toArray()) as $middleware) { - yield $middleware; - } - - // Finally, yield the last callable. - yield $last; - }; - - $this->middlewares = $middlewares(); - - return $this; - } - - protected function invokeMiddleware(mixed $middleware): mixed - { - try { - if ($middleware instanceof \Closure) { - $result = $this->invoke($middleware); - } elseif ($middleware instanceof QueueMiddlewareInterface) { - $result = $middleware->process($this); - } elseif ($middleware instanceof self) { - $result = $middleware->next(); - } else { - throw new \InvalidArgumentException( - sprintf( - 'Invalid middleware queue entry: %s. Middleware must implement %s or be an instance of %s.', - $middleware, - QueueMiddlewareInterface::class, - self::class - ) - ); - } - - if ($result === null) { - throw new \UnexpectedValueException( - 'Middleware returns null, you may forgot to return JobController::next() in middleware.' - ); - } - - if ($result instanceof \Throwable) { - throw $result; - } - - if ($result->failed) { - throw $result->exception; - } - - return $result; - } catch (DeferredException $e) { - $this->defer = $e; - - if ($e->getPrevious()) { - $this->failed($e->getPrevious()); - } - } catch (AbandonedException $e) { - $this->abandoned = $e; - - if ($e->getPrevious()) { - $this->failed($e->getPrevious()); - } - } - - return $this; - } -} diff --git a/src/Job/JobInterface.php b/src/Job/JobInterface.php deleted file mode 100644 index 7da61c1..0000000 --- a/src/Job/JobInterface.php +++ /dev/null @@ -1,17 +0,0 @@ -driver = $driver; - } - - /** - * push - * - * @param mixed $job - * @param int $delay - * @param string|null $channel - * @param array $options - * - * @return int|string - */ - public function push(mixed $job, int $delay = 0, ?string $channel = null, array $options = []): int|string - { - if (!$job instanceof QueueMessage) { - $message = $this->getMessageByJob($job); - } else { - $message = $job; - } - - $message->setDelay($delay); - $message->setChannel($channel); - $message->setOptions($options); - $message->setCreatedAt('now'); - - return $this->driver->push($message); - } - - /** - * pushRaw - * - * @param string|array $body - * @param int $delay - * @param string|null $channel - * @param array $options - * - * @return int|string - * @throws JsonException - */ - public function pushRaw( - string|array $body, - int $delay = 0, - ?string $channel = null, - array $options = [] - ): int|string { - if (is_string($body)) { - json_decode($body, true, 512, JSON_THROW_ON_ERROR); - } - - $message = new QueueMessage(); - $message->setBody($body); - $message->setDelay($delay); - $message->setChannel($channel); - $message->setOptions($options); - $message->setCreatedAt('now'); - - return $this->driver->push($message); - } - - /** - * pop - * - * @param string|null $channel - * - * @return QueueMessage|null - */ - public function pop(?string $channel = null): ?QueueMessage - { - return $this->driver->pop($channel); - } - - /** - * delete - * - * @param QueueMessage|mixed $message - * - * @return void - */ - public function delete(mixed $message): void - { - if (!$message instanceof QueueMessage) { - $msg = new QueueMessage(); - $msg->setId($message); - - $message = $msg; - } - - $this->driver->delete($message); - - $message->setDeleted(true); - } - - /** - * @param QueueMessage|mixed $message - * @param int $delay - * - * @return void - */ - public function release(mixed $message, int $delay = 0): void - { - if (!$message instanceof QueueMessage) { - $msg = new QueueMessage(); - $msg->setId($message); - } - - $message->setDelay($delay); - - $this->driver->release($message); - } - - public function defer(mixed $message, int $delay = 0): void - { - if (!$message instanceof QueueMessage) { - $msg = new QueueMessage(); - $msg->setId($message); - } - - $message->setDelay($delay); - - $this->driver->defer($message); - } - - /** - * getMessage - * - * @param mixed $job - * @param array $data - * - * @return QueueMessage - * @throws InvalidArgumentException - */ - public function getMessageByJob(mixed $job, array $data = []): QueueMessage - { - $message = new QueueMessage(); - - $job = $this->createJobInstance($job); - - $data['class'] = $job::class; - - $message->setName(get_debug_type($job)); - $message->setJob($job)->serializeJob(); - $message->setData($data); - - return $message; - } - - /** - * @param object|callable|string $job - * - * @return object - */ - protected function createJobInstance(mixed $job): object - { - // Create callable - if ($job instanceof \Closure) { - $instance = new ClosureJob($job); - } elseif (is_string($job)) { - // Create by class name. - if (!class_exists($job) || method_exists($job, '__invoke')) { - throw new InvalidArgumentException( - 'Job should be a class which has __invoke() method.' - ); - } - - $instance = $this->createJobByClassName($job); - } elseif ($job instanceof DefinitionInterface) { - $instance = $this->createJobByClassName($job); - } else { - $instance = $job; - } - - return $instance; - } - - /** - * Method to get property Driver - * - * @return QueueDriverInterface - */ - public function getDriver(): QueueDriverInterface - { - return $this->driver; - } - - /** - * Method to set property driver - * - * @param QueueDriverInterface $driver - * - * @return static Return self to support chaining. - */ - public function setDriver(QueueDriverInterface $driver): static - { - $this->driver = $driver; - - return $this; - } - - /** - * createJobByClassName - * - * @param mixed $job Can be class name or DI definition. - * @param mixed ...$args Arguments. - * - * @return object - */ - protected function createJobByClassName(mixed $job, ...$args): object - { - return $this->getObjectBuilder()->getBuilder()($job, ...$args); - } -} diff --git a/src/QueueMessage.php b/src/QueueMessage.php deleted file mode 100644 index 9af14f8..0000000 --- a/src/QueueMessage.php +++ /dev/null @@ -1,439 +0,0 @@ - $this->body['job'] ?? null; - } - - // phpcs:enable - - /** - * QueueMessage constructor. - * - * @param callable|null $job - * @param array $data - * @param int $delay - * @param array $options - */ - public function __construct(?callable $job = null, array $data = [], int $delay = 0, array $options = []) - { - if ($job !== null) { - $this->setJob($job); - } - - if ($data) { - $this->setData($data); - } - - $this->setDelay($delay); - $this->setOptions($options); - } - - /** - * get - * - * @param string $name - * @param mixed $default - * - * @return mixed - */ - public function get(string $name, mixed $default = null): mixed - { - return $this->body[$name] ?? $default; - } - - /** - * set - * - * @param string $name - * @param mixed $value - * - * @return static - */ - public function set(string $name, mixed $value): static - { - $this->body[$name] = $value; - - return $this; - } - - /** - * Method to get property Id - * - * @return int|string - */ - public function getId(): int|string - { - return $this->id; - } - - /** - * Method to set property id - * - * @param int|string $id - * - * @return static Return self to support chaining. - */ - public function setId(int|string $id): static - { - $this->id = $id; - - return $this; - } - - /** - * Method to get property Attempts - * - * @return int - */ - public function getAttempts(): int - { - return $this->attempts; - } - - /** - * Method to set property attempts - * - * @param int $attempts - * - * @return static Return self to support chaining. - */ - public function setAttempts(int $attempts): static - { - $this->attempts = $attempts; - - return $this; - } - - /** - * Method to get property Job - * - * @return object - */ - public function getJob(): object - { - $this->unserializeJob(); - - return $this->unserializedJob; - } - - /** - * Method to set property job - * - * @param object $job - * - * @return static Return self to support chaining. - */ - public function setJob(object $job): static - { - $this->unserializedJob = $job; - - unset($this->body['job']); - - return $this; - } - - /** - * Method to get property Data - * - * @return array - */ - public function getData(): array - { - return $this->body['data'] ?? []; - } - - /** - * Method to set property data - * - * @param array $data - * - * @return static Return self to support chaining. - */ - public function setData(array $data): static - { - $this->body['data'] = $data; - - return $this; - } - - /** - * Method to get property Queue - * - * @return string|null - */ - public function getChannel(): ?string - { - return $this->body['channel'] ?? null; - } - - /** - * Method to set property queue - * - * @param ?string $channel - * - * @return static Return self to support chaining. - */ - public function setChannel(?string $channel): static - { - $this->body['channel'] = $channel; - - return $this; - } - - /** - * Method to get property Body - * - * @return array - */ - public function getBody(): array - { - return $this->body; - } - - /** - * Method to set property body - * - * @param array $body - * - * @return static Return self to support chaining. - */ - public function setBody(array $body): static - { - $this->body = $body; - - return $this; - } - - /** - * Method to get property RawData - * - * @return string - */ - public function getRawBody(): string - { - return $this->rawBody; - } - - /** - * Method to set property rawData - * - * @param string $rawBody - * - * @return static Return self to support chaining. - */ - public function setRawBody(string $rawBody): static - { - $this->rawBody = $rawBody; - - return $this; - } - - /** - * Method to get property Name - * - * @return string - */ - public function getName(): string - { - return $this->body['name'] ?? ''; - } - - /** - * Method to set property name - * - * @param string $name - * - * @return static Return self to support chaining. - */ - public function setName(string $name): static - { - $this->body['name'] = $name; - - return $this; - } - - /** - * Method to get property Delay - * - * @return int - */ - public function getDelay(): int - { - return $this->delay; - } - - /** - * Method to set property delay - * - * @param int $delay - * - * @return static Return self to support chaining. - */ - public function setDelay(int $delay): static - { - $this->delay = $delay; - - return $this; - } - - /** - * @return array - * - * @throws InvalidArgumentException - */ - public function jsonSerialize(): array - { - $new = clone $this; - $new->serializeJob(); - - return $new->body; - } - - /** - * @return bool - */ - public function isDeleted(): bool - { - return $this->deleted; - } - - /** - * @param bool $deleted - * - * @return static Return self to support chaining. - */ - public function setDeleted(bool $deleted): static - { - $this->deleted = $deleted; - - return $this; - } - - public function serializeJob(): void - { - if ($this->serializedJob) { - return; - } - - $this->body['job'] = serialize($this->unserializedJob); - } - - public function unserializeJob(): void - { - $this->unserializedJob ??= unserialize($this->body['job'] ?? '', ['allowed_classes' => true]); - } - - public function makeJobController( - ?\Closure $invoker = null, - LoggerInterface $logger = new NullLogger(), - ): JobController { - return new JobController($this, $invoker, $logger); - } - - public function run( - ?\Closure $invoker = null, - LoggerInterface $logger = new NullLogger(), - ): JobController { - return $this->makeJobController($invoker, $logger)->run(); - } - - public function __serialize(): array - { - $this->serializeJob(); - - return get_object_vars($this); - } - - public function __unserialize(array $data): void - { - foreach ($data as $key => $value) { - if ($key === 'serializedJob') { - continue; - } - - if (property_exists($this, $key)) { - $this->{$key} = $value; - } - } - - $this->unserializeJob(); - } - - public function setCreatedAt(\DateTimeImmutable|string|null $createdAt): static - { - if (is_string($createdAt)) { - $createdAt = new \DateTimeImmutable($createdAt); - } - - $this->createdAt = $createdAt; - - return $this; - } -} diff --git a/src/QueuePackage.php b/src/QueuePackage.php deleted file mode 100644 index df9d448..0000000 --- a/src/QueuePackage.php +++ /dev/null @@ -1,43 +0,0 @@ -installConfig(__DIR__ . '/../etc/*.php', 'config'); - } - - public function register(Container $container): void - { - // $container->prepareSharedObject(Worker::class); - $container->prepareSharedObject(QueueFactory::class); - $container->prepareSharedObject(QueueFailerFactory::class); - $container->bindShared( - Queue::class, - function (Container $container, ?string $tag = null) { - return $container->get(QueueFactory::class)->get($tag); - } - ); - $container->bindShared( - QueueFailerInterface::class, - function (Container $container, ?string $tag = null) { - return $container->get(QueueFailerFactory::class)->get($tag); - } - ); - } -} diff --git a/src/RunnerOptions.php b/src/RunnerOptions.php deleted file mode 100644 index 3c88fc1..0000000 --- a/src/RunnerOptions.php +++ /dev/null @@ -1,24 +0,0 @@ -enqueueIfAvailable($channel); - - $message ??= $this->getNextMessage($channel); - - if (!$message) { - if ($this->options->stopWhenEmpty) { - $this->stop('No more messages in queue.'); - } - - return; - } - - if ($this->options->maxRuns > 0) { - $this->runTimes++; - } - - $this->process($message); - } - - protected function enqueueIfAvailable(array|string $defaultChannel): ?QueueMessage - { - if (!$this->enqueuer) { - return null; - } - - $channel = $this->enqueuerChannels ?? $defaultChannel; - - $channel = (array) $channel; - - foreach ($channel as $channelName) { - $result = $this->enqueuer->enqueue($channelName); - - if ($result && !$result instanceof QueueMessage) { - $result = $this->queue->getMessageByJob($result); - } - - if ($result instanceof QueueMessage) { - return $result; - } - } - - return null; - } - - /** - * @param string|array $channel - * - * @return void - * - * @deprecated Use next() instead. - */ - public function runNextJob(string|array $channel): void - { - $this->next($channel); - } - - /** - * @param QueueMessage $message - * - * @return void - */ - public function process(QueueMessage $message): void - { - $maxTries = $this->options->tries; - - $controller = $this->createJonController($message); - $backoff = JobBackoff::fromController($controller); - - // @before event - $event = $this->emit( - new BeforeJobRunEvent( - controller: $controller, - runner: $this, - queue: $this->queue, - ), - ); - - $message = $event->message; - $controller = $event->controller; - - try { - // Fail if max attempts - if ($backoff === false || ($maxTries !== 0 && $maxTries < $message->getAttempts())) { - $this->queue->delete($message); - - throw new MaxAttemptsExceededException('Max attempts exceed for Message: ' . $message->getId()); - } - - $controller = $controller->run(); - - if ($controller->failed) { - throw $controller->exception; - } - - // @after event - $event = $this->emit( - new AfterJobRunEvent( - controller: $controller, - runner: $this, - queue: $this->queue, - ), - ); - - $controller = $event->controller; - $this->settleJob($controller); - } catch (Throwable $t) { - $this->handleJobException($controller, $t, $backoff); - } - } - - protected function handleJobException( - JobController $controller, - Throwable $e, - int|false|null $backoff = null, - ): void { - $controller->failed($e); - - $job = $controller->job; - $message = $controller->message; - - $maxTries = $this->options->tries; - - // Delete and log error if reach max attempts. - $maxAttemptsExceeds = ($backoff === false || ($maxTries !== 0 && $maxTries <= $message->getAttempts())); - $controller->maxAttemptsExceeds = $maxAttemptsExceeds; - - // Run through JobFailed methods. - iterator_count($controller->invokeMethodsWithAttribute(JobFailed::class)); - - if ($controller->shouldDelete) { - $this->queue->delete($message); - $this->logger->error( - sprintf( - 'Job: [%s] (%s) failed. %s - Class: %s', - get_debug_type($job), - $message->getId(), - $maxAttemptsExceeds - ? 'Max attempts exceeded.' - : $controller->abandoned->toReasonText(), - get_debug_type($job), - ), - ); - - $backoff = false; - } else { - if ($message->getId()) { - $this->queue->release( - $message, - $backoff ??= $this->options->backoff - ); - } else { - $this->queue->push( - $message, - $backoff ??= $this->options->backoff - ); - } - $this->logger->error( - sprintf( - 'Job: [%s] (%s) failed, will retry after %d seconds - Class: %s', - get_debug_type($job), - $message->getId(), - $backoff, - get_debug_type($job) - ), - ); - } - - $this->emit( - new JobFailureEvent( - exception: $e, - controller: $controller, - runner: $this, - queue: $this->queue, - backoff: $backoff, - ), - ); - } - - protected function settleJob(JobController $controller): void - { - if ($controller->failed) { - throw $controller->exception; - } - - $message = $controller->message; - - // Release job if it has a release delay. - if ($controller->defer) { - $message->setDeleted(false); - $message->setDelay($controller->defer->delay); - - $this->queue->defer( - $message, - $controller->defer->delay - ); - - return; - } - - $this->queue->delete($message); - } - - /** - * getNextMessage - * - * @param string|array $channel - * - * @return null|QueueMessage - */ - protected function getNextMessage(string|array $channel): ?QueueMessage - { - $channel = (array) $channel; - - foreach ($channel as $channelName) { - if ($message = $this->queue->pop($channelName)) { - return $message; - } - } - - return null; - } - - /** - * @param QueueMessage $message - * - * @return JobController - */ - public function createJonController(QueueMessage $message): JobController - { - if ($this->options->controllerFactory) { - return ($this->options->controllerFactory)( - $message, - $this->getInvoker(), - $this->createControllerLogger(), - ); - } - - return $message->makeJobController( - $this->getInvoker(), - $this->createControllerLogger() - ); - } - - protected function getDefaultInvoker(): \Closure - { - return fn(JobController $controller, callable $invokable) => $invokable($controller); - } -} diff --git a/src/bootstrap.php b/src/bootstrap.php deleted file mode 100644 index 13fbe8c..0000000 --- a/src/bootstrap.php +++ /dev/null @@ -1,5 +0,0 @@ -expectException(\Error::class); - - $event = new class () { - use JobEventTrait; - }; - - $event->job = 123; - } -} diff --git a/test/Failer/DatabaseQueueFailerTest.php b/test/Failer/DatabaseQueueFailerTest.php deleted file mode 100644 index 67bc4da..0000000 --- a/test/Failer/DatabaseQueueFailerTest.php +++ /dev/null @@ -1,188 +0,0 @@ -instance->add( - 'test', - 'default', - '{}', - RuntimeException::class - ); - $r = $this->instance->add( - 'hello', - 'world', - '{}', - RuntimeException::class - ); - - self::assertEquals( - 2, - $r - ); - - $item = self::$db->select('*') - ->from('queue_failed_jobs') - ->where('id', 1) - ->get(); - - self::assertEquals( - 'test', - $item->connection - ); - self::assertEquals( - '{}', - $item->body - ); - self::assertEquals( - RuntimeException::class, - $item->exception - ); - } - - /** - * @see DatabaseQueueFailer::__construct - */ - public function testConstruct(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see DatabaseQueueFailer::setTable - */ - public function testSetTable(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see DatabaseQueueFailer::get - */ - public function testGet(): void - { - $item = $this->instance->get(2); - - self::assertEquals( - 2, - $item['id'] - ); - - self::assertEquals( - 'hello', - $item['connection'] - ); - - self::assertEquals( - 'world', - $item['channel'] - ); - } - - /** - * @see DatabaseQueueFailer::all - */ - public function testAll(): void - { - $items = $this->instance->all(); - - self::assertEquals( - ['test', 'hello'], - array_column(iterator_to_array($items), 'connection') - ); - } - - /** - * @see DatabaseQueueFailer::remove - */ - public function testRemove(): void - { - $this->instance->remove(1); - - self::assertNull( - self::$db->select('*') - ->from('queue_failed_jobs') - ->where('id', 1) - ->get() - ); - } - - /** - * @see DatabaseQueueFailer::clear - */ - public function testClear(): void - { - $this->instance->clear(); - - self::assertEquals( - [], - self::$db->select('*') - ->from('queue_failed_jobs') - ->all() - ->dump() - ); - } - - /** - * @see DatabaseQueueFailer::getTable - */ - public function testGetTable(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see DatabaseQueueFailer::isSupported - */ - public function testIsSupported(): void - { - self::assertTrue($this->instance->isSupported()); - } - - protected function setUp(): void - { - $this->instance = new DatabaseQueueFailer( - self::$db - ); - } - - protected function tearDown(): void - { - } - - /** - * setupDatabase - * - * @return void - */ - protected static function setupDatabase(): void - { - self::createDatabase('pdo_mysql'); - - self::importFromFile(__DIR__ . '/../../resources/sql/queue_failed_jobs.sql'); - } -} diff --git a/test/Failer/PdoQueueFailerTest.php b/test/Failer/PdoQueueFailerTest.php deleted file mode 100644 index 2970346..0000000 --- a/test/Failer/PdoQueueFailerTest.php +++ /dev/null @@ -1,28 +0,0 @@ -getDriver()->getConnection(); - - $this->instance = new PdoQueueFailer( - $conn->get() - ); - - $conn->release(); - } - - protected function tearDown(): void - { - } -} diff --git a/test/QueueDatabaseTest.php b/test/QueueDatabaseTest.php deleted file mode 100644 index 69b944e..0000000 --- a/test/QueueDatabaseTest.php +++ /dev/null @@ -1,179 +0,0 @@ -instance->push($job, 0); - - self::assertEquals( - '1', - $result - ); - - $job = self::$db->select('*') - ->from('queue_jobs') - ->where('id', 1) - ->get(); - - self::assertEquals( - 'default', - $job->channel - ); - - $body = json_decode($job->body, true); - - self::assertEquals( - ['Hello'], - unserialize($body['job'])->logs - ); - } - - /** - * @see Queue::pop - */ - public function testPop(): void - { - $message = $this->instance->pop(); - - self::assertEquals( - 1, - $message->getAttempts() - ); - - /** @var TestJob $job */ - $job = $message->getJob(); - $job->__invoke(); - - self::assertEquals( - ['Hello'], - $job->executed - ); - } - - /** - * @see Queue::setDriver - */ - public function testSetDriver(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Queue::release - */ - public function testRelease(): void - { - $job = new TestJob(['Welcome']); - - $result = $this->instance->push($job, 0); - - $message = $this->instance->pop(); - - $r = self::$db->select('*') - ->from('queue_jobs') - ->where('id', 2) - ->get(); - - self::assertNotNull($r); - - $this->instance->release($message); - - $item = self::$db->select('*') - ->from('queue_jobs') - ->where('id', 2) - ->get(); - - self::assertNull($item->reserved); - } - - /** - * @see Queue::delete - */ - public function testDelete(): void - { - $this->instance->delete(2); - - $items = self::$db->select('*') - ->from('queue_jobs') - ->all(); - - self::assertCount(1, $items); - } - - /** - * @see Queue::getDriver - */ - public function testGetDriver(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Queue::__construct - */ - public function testConstruct(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Queue::getMessageByJob - */ - public function testGetMessageByJob(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Queue::pushRaw - */ - public function testPushRaw(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - protected function setUp(): void - { - $this->instance = new Queue( - new DatabaseQueueDriver(static::$db) - ); - } - - protected function tearDown(): void - { - } - - /** - * setupDatabase - * - * @return void - */ - protected static function setupDatabase(): void - { - self::createDatabase('pdo_mysql'); - self::importFromFile(__DIR__ . '/../resources/sql/queue_jobs.sql'); - } -} diff --git a/test/QueuePdoTest.php b/test/QueuePdoTest.php deleted file mode 100644 index 74437c5..0000000 --- a/test/QueuePdoTest.php +++ /dev/null @@ -1,38 +0,0 @@ -getDriver()->getConnection(); - - $this->instance = new Queue(new PdoQueueDriver($conn->get())); - - $conn->release(); - } - - protected function tearDown(): void - { - } - - /** - * setupDatabase - * - * @return void - */ - protected static function setupDatabase(): void - { - self::createDatabase('pdo_mysql'); - self::importFromFile(__DIR__ . '/../resources/sql/queue_jobs.sql'); - } -} diff --git a/test/Stub/TestJob.php b/test/Stub/TestJob.php deleted file mode 100644 index 8d4e031..0000000 --- a/test/Stub/TestJob.php +++ /dev/null @@ -1,47 +0,0 @@ -logs = $logs; - } - - /** - * getName - * - * @return string - */ - public function getName(): string - { - return 'test'; - } - - /** - * execute - * - * @return void - */ - public function __invoke(): void - { - $this->executed = $this->logs; - } -} diff --git a/test/WorkerTest.php b/test/WorkerTest.php deleted file mode 100644 index 92902c8..0000000 --- a/test/WorkerTest.php +++ /dev/null @@ -1,155 +0,0 @@ -instance->getQueue()->push( - static function () { - file_put_contents( - __DIR__ . '/../tmp/test.log', - uid() - ); - static::$logs[] = 'Job executed.'; - }, - 0, - 'hello' - ); - - $this->instance->on(LoopEndEvent::class, fn() => $this->instance->stop()); - - $this->instance->loop(['default', 'hello'], ['sleep' => 0.1]); - - self::assertEquals( - 'Job executed.', - static::$logs[0] - ); - - self::$logs = []; - } - - /** - * @see Worker::stopIfNecessary - */ - public function testStopIfNecessary(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Worker::getState - */ - public function testGetState(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Worker::stop - */ - public function testStop(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Worker::runNextJob - */ - public function testRunNextJob(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Worker::setState - */ - public function testSetState(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Worker::process - */ - public function testProcess(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Worker::__construct - */ - public function testConstruct(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Worker::shutdown - */ - public function testShutdown(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - /** - * @see Worker::getQueue - */ - public function testGetAdapter(): void - { - self::markTestIncomplete(); // TODO: Complete this test - } - - protected function setUp(): void - { - $this->instance = new Worker( - new Queue( - new DatabaseQueueDriver(self::$db) - ) - ); - } - - protected function tearDown(): void - { - } - - /** - * setupDatabase - * - * @return void - */ - protected static function setupDatabase(): void - { - self::createDatabase('pdo_mysql'); - self::importFromFile(__DIR__ . '/../resources/sql/queue_jobs.sql'); - } -} diff --git a/tmp/.gitignore b/tmp/.gitignore deleted file mode 100644 index d6b7ef3..0000000 --- a/tmp/.gitignore +++ /dev/null @@ -1,2 +0,0 @@ -* -!.gitignore