init complete

Author: banyancheung

.travis.yml

@@ -1,25 +1,39 @@
-# Swoft CronTask
+# Swoft 定时任务组件
 
-Swoft framework CronTask Component 
+swoft定时任务组件，核心大部分基于1.0，在此基础上适配了2.0，实测可用。
 
-## Install
+## 安装
 
 - composer command
 
 ```bash
 composer require banyancheung/cron-task
 ```
 
-## Document
-
-Please see [document site](https://doc.swoft.org)
-
-## Unit testing
+- composer.json
 
 ```bash
-phpunit 
+"require": {
+    ...
+    "banyancheung/cron-task": "dev-master",
+    ...
+}
 ```
 
+然后 `composer update` 即可。
+
+## 使用
+
+1.在`app/beans.php` 文件里的`server`一项，监听 `onPipeMessage` 事件。
+
+![](https://img.yunjes.com/doc/cron-task/1.png)
+
+2.在定义好的task注解里，增加 Scheduled 一项。
+
+![](https://img.yunjes.com/doc/cron-task/2.png)
+
+注意：Scheduled里的task属性是必须的，用于定位是哪个Task，也许会有更好的方法。:)
+
 ## LICENSE
 
 The Component is open-sourced software licensed under the [Apache license](LICENSE).

README.md

@@ -14,7 +14,7 @@
       "email": "i@zhangrong.cc"
     }
   ],
-  "description": "micro-service framework base on swoole",
+  "description": "crontab task component base on swoft",
   "license": "Apache-2.0",
   "require": {
     "swoft/framework": "~2.0.0",
@@ -25,16 +25,6 @@
       "Swoft\\CronTask\\": "src/"
     }
   },
-  "autoload-dev": {
-    "psr-4": {
-      "SwoftTest\\CronTask\\": "test/case"
-    }
-  },
-  "require-dev": {
-  },
-  "scripts": {
-    "test": "./vendor/bin/phpunit -c phpunit.xml"
-  },
   "repositories": {
     "packagist": {
       "type": "composer",

composer.json

@@ -4,6 +4,7 @@
 
 use Swoft\App;
 use Swoft\Bean\Annotation\Mapping\Bean;
+use Swoft\CronTask\PipeMessage;
 use Swoft\CronTask\TaskRegister;
 use Swoole\Table as SwooleTable;
 
@@ -180,7 +181,7 @@ public function getRunTimeTable(): Table
      * 获取key值
      *
      * @param string $rule 规则
-     * @param string $taskClass 任务类
+     * @param string $task 任务名
      * @param string $taskMethod 任务方法
      * @param string $min 分
      * @param string $sec 时间戳
@@ -221,7 +222,6 @@ public function loadTableTask()
     private function initRunTimeTableData(array $task, array $parseResult): bool
     {
         $runTimeTable = $this->getRunTimeTable();
-
         $min = date('YmdHi');
         $sec = strtotime(date('Y-m-d H:i'));
 
@@ -309,6 +309,20 @@ public function startTask($key)
         return $this->getRunTimeTable()->set($key, ['runStatus' => self::START]);
     }
 
+    /**
+     * @param string $task
+     * @param string $method
+     */
+    public function deliverTask(string $task, string $method): void
+    {
+        /* @var PipeMessageInterface $pipeMessage */
+        $message = PipeMessage::pack(PipeMessage::MESSAGE_TYPE_TASK, [
+            'name' => $task,
+            'method' => $method
+        ]);
+        \server()->getSwooleServer()->sendMessage($message, 0);
+    }
+
     /**
      * 完成任务
      *

phpunit.xml

@@ -63,7 +63,6 @@ class TableCrontab
      */
     public static function init(int $taskCount = null, int $taskQueue = null)
     {
-        echo 'init tableCrontab' . "\n";
         self::$taskCount = $taskCount == null ? self::$taskCount : $taskCount;
         self::$taskQueue = $taskQueue == null ? self::$taskQueue : $taskQueue;
         self::getInstance();

src/Crontab/Crontab.php

@@ -30,29 +30,25 @@ class AttachCronProcessHandler implements EventHandlerInterface
      */
     public function handle(EventInterface $event): void
     {
-        echo "init event..."."\n";
         $setting = \config('cron');
         // Init crontab share memory table
         if (isset($setting['cronable']) && (int)$setting['cronable'] === 1) {
             $this->initCrontabMemoryTable();
             Crontab::init();
             $this->initProcessByEvent($event);
         }
-        echo "init event succeed"."\n";
     }
 
     /**
      * init table of crontab
      */
     private function initCrontabMemoryTable(): void
     {
-        echo "init memory table..."."\n";
         /** @var array[] $settings */
         $setting = \config('cron');
         $taskCount = isset($setting['task_count']) && $setting['task_count'] > 0 ? $setting['task_count'] : null;
         $taskQueue = isset($setting['task_queue']) && $setting['task_queue'] > 0 ? $setting['task_queue'] : null;
         TableCrontab::init($taskCount, $taskQueue);
-        echo "init memory table succeed"."\n";
     }
 
     /**
@@ -61,13 +57,11 @@ private function initCrontabMemoryTable(): void
      */
     private function initProcessByEvent(EventInterface $event): void
     {
-        echo "init process..."."\n";
         $swooleServer = $event->getTarget()->getSwooleServer();
         $execProcess = \bean(CronExecProcess::class)->create();
         $timerProcess = \bean(CronTimerProcess::class)->create();
         $swooleServer->addProcess($execProcess);
         $swooleServer->addProcess($timerProcess);
-        echo "init process done."."\n";
     }
 
 }

src/Crontab/TableCrontab.php

@@ -0,0 +1,35 @@
+<?php declare(strict_types=1);
+
+namespace Swoft\CronTask\Listener;
+
+use Swoft\CronTask\PipeMessage;
+use Swoft\Event\Annotation\Mapping\Listener;
+use Swoft\Event\EventHandlerInterface;
+use Swoft\Server\Swoole\SwooleEvent;
+use Swoft\Event\EventInterface;
+use Swoft\Task\Task;
+
+/**
+ * The pipe message listener
+ *
+ * @Listener(event=SwooleEvent::PIPE_MESSAGE)
+ */
+class PipeMessageHandler implements EventHandlerInterface
+{
+
+    public function handle(EventInterface $event): void
+    {
+        $params = $event->getParams();
+        if (count($params) < 3) {
+            return;
+        }
+        list($server, $srcWorkerId, $data) = $params;
+        $message = PipeMessage::unpack($data);
+        $type = $message['type'];
+        $taskName = $message['message']['name'];
+        $methodName = $message['message']['method'];
+        // delever task
+        Task::async($taskName, $methodName);
+
+    }
+}

src/Listener/AttachCronProcessHandler.php

@@ -0,0 +1,45 @@
+<?php declare(strict_types=1);
+
+namespace Swoft\CronTask;
+
+
+use Swoft\Stdlib\Helper\JsonHelper;
+
+/**
+ * Pipe message
+ */
+class PipeMessage
+{
+    /**
+     * Task message type
+     */
+    const MESSAGE_TYPE_TASK = 'task';
+
+
+    /**
+     * @param string $type
+     * @param array $data
+     *
+     * @return string
+     */
+    public static function pack(string $type, array $data): string
+    {
+        $data = [
+            'type' => $type,
+            'message' => $data,
+        ];
+        return JsonHelper::encode($data, JSON_UNESCAPED_UNICODE);
+    }
+
+    /**
+     * @param string $message
+     *
+     * @return array
+     */
+    public static function unpack(string $message): array
+    {
+        $messageAry = json_decode($message, true);
+        return $messageAry;
+    }
+
+}

src/Listener/PipeMessageHandler.php

@@ -34,7 +34,7 @@ public function handle(Process $process)
             if (!empty($tasks)) {
                 foreach ($tasks as $task) {
                     // Diliver task
-                    Task::async($task['task'], $task['taskMethod']);
+                    $cron->deliverTask($task['task'], $task['taskMethod']);
                     $cron->finishTask($task['key']);
                 }
             }

src/PipeMessage.php

@@ -0,0 +1,34 @@
+<?php declare(strict_types=1);
+
+
+namespace Swoft\CronTask\Swoole;
+
+
+use function go;
+use Swoft;
+use Swoft\Server\Swoole\PipeMessageInterface;
+use Swoft\Bean\Annotation\Mapping\Bean;
+use Swoft\Server\Swoole\SwooleEvent;
+use Swoole\Server;
+
+/**
+ * Class PipeMessageListener
+ *
+ * @since 2.0
+ *
+ * @Bean()
+ */
+class PipeMessageListener implements PipeMessageInterface
+{
+    /**
+     * @param Server $server
+     * @param int $taskId
+     * @param string $data
+     */
+    public function onPipeMessage(Server $server, int $srcWorkerId, $message): void
+    {
+        go(function () use ($server, $srcWorkerId, $message) {
+            Swoft::trigger(SwooleEvent::PIPE_MESSAGE, null, $server, $srcWorkerId, $message);
+        });
+    }
+}