secutiry improvement and add new callback API with onDownloaderConnectToParser for downloader worker

Author: blogdaren

Docs/README.md

@@ -108,6 +108,10 @@ return array(
         //set the max number of the task queue, 0 indicates no limit (optional, default `0`)
         'max_number'      => 1000,
 
+        //specifies the max number of connections each downloader process can connect to the parser
+        //(optional, default `1`, minimum value 1, maximum value 1000)
+        //'max_connections' => 1,
+
         //set the max number of the request for each socket connection,  
         //if the cumulative number of socket requests exceeds the max number of requests,
         //the parser will close the connection and try to reconect automatically.

Examples/start.php

@@ -177,9 +177,12 @@
     //任务队列最大task数量, 0代表无限制 (可选项，默认0)
     //'max_number'      => 1000,
 
+    //特指每个下载器进程可以建立到解析器的最大连接数 (可选项，默认1，最小值为1，最大值为1000)
+    //'max_connections' => 1,
+    
     //当前Socket连接累计最大请求数，0代表无限制 (可选项，默认0)
     //如果当前Socket连接的累计请求数超过最大请求数时，
-    //parser端会主动关闭连接，同时客户端会自动尝试重连
+    //parser端会主动关闭连接，同时客户端会自动尝试重连.
     //'max_request'     => 1000,
 
     //限定爬取站点域，留空表示不受限
@@ -355,6 +358,10 @@ function startAppDownloader()
     $downloader->onDownloaderStart = function($downloader){
     };
 
+    $downloader->onDownloaderConnectToParser = function($connection){
+        //$connection->bufferFull = true;
+    };
+
     //回调【onBeforeDownload】的新增别名是【onDownloadBefore】
     $downloader->onDownloadBefore = function($downloader, $task){
         //disable http ssl verify in any of the following two ways 
@@ -377,7 +384,9 @@ function startAppDownloader()
         //pprint($error, $task);
     };
 
-    $downloader->onTaskEmpty = function($downloader){
+    //回调【onTaskEmpty】的新增别名是【onDownloadTaskEmpty】
+    $downloader->onDownloadTaskEmpty= function($downloader){
+        //$downloader->removeTimer();
     };
 
     //使用无头浏览器回调或者直接使用无头浏览器相关API

README.md

@@ -153,6 +153,7 @@ $config['redis'] = [
 $config['task'] = array( 
     //'crawl_interval'  => 1,
     //'max_number'      => 1000,
+    //'max_connections' => 1,
     //'max_request'     => 1000,
     'context' => [
         'cache_enabled'    => true,
@@ -285,10 +286,11 @@ function startAppDownloader()
     //$downloader->onDownloaderStart = function($downloader){};
     //$downloader->onDownloaderStop  = function($downloader){};
     //$downloader->onDownloaderMessage = function($downloader, $parser_reply){};
+    //$downloader->onDownloaderConnectToParser = function($connection){};
     //$downloader->onDownloadStart = function($downloader, $task){};
     //$downloader->onDownloadAfter = function($downloader, $download_data, $task){};
     //$downloader->onDownloadFail  = function($downloader, $error, $task){};
-    //$downloader->onTaskEmpty = function($downloader){};
+    //$downloader->onDownloadTaskEmpty = function($downloader){};
 }
 
 function startAppParser()

src/Downloader.php

@@ -123,7 +123,7 @@ public function onWorkerStart($worker)
         }
 
         //try to connect to parser asynchronously then execute consumeOneTask by interval
-        $this->connectToParser();
+        $this->_connectToParser();
     }
 
     /**
@@ -158,7 +158,7 @@ public function onWorkerReload($worker)
      *
      * @return   void
      */
-    public function connectToParser()
+    private function _connectToParser()
     {
         //get all task connections
         $task_connections = $this->getAsyncTaskConnection();
@@ -230,6 +230,10 @@ public function onConnectToParser($connection)
             'downloader_client_address' => 'tcp://' . $connection->getLocalAddress(),
         ]));
 
+        //trigger callback
+        $returning = $this->triggerUserCallback('onDownloaderConnectToParser', $connection);
+        if(false === $returning) return false;
+
         //install task timer
         $connection->taskTimerId = Timer::add($this->getTaskCrawlInterval(), [$this, 'consumeOneTask'], [$connection->channel], 1);
 

src/Kernel/PHPCreeper.php

@@ -40,7 +40,7 @@ class PHPCreeper extends Worker
      *
      * @var string
      */
-    public const CURRENT_VERSION = '1.9.3';
+    public const CURRENT_VERSION = '1.9.4';
 
     /**
      * engine name
@@ -324,6 +324,8 @@ class PHPCreeper extends Worker
         'onDownloaderStop'     => null,
         'onDownloaderReload'   => null,
         'onDownloaderMessage'  => null,
+        'onDownloaderConnectToParser' => null,
+        'onTaskEmpty'          => null,
         'onBeforeDownload'     => null,
         'onStartDownload'      => null,
         'onAfterDownload'      => null,
@@ -344,7 +346,6 @@ class PHPCreeper extends Worker
         'onServerBufferFull'   => null,
         'onServerBufferDrain'  => null,
         'onServerError'        => null,
-        'onTaskEmpty'          => null,
         'onHeadlessBrowserOpenPage' => null,
     );
 
@@ -364,6 +365,7 @@ class PHPCreeper extends Worker
         'onStartDownload'   => ['onDownloadStart'],
         'onAfterDownload'   => ['onDownloadAfter'],
         'onFailDownload'    => ['onDownloadFail'],
+        'onTaskEmpty'       => ['onDownloadTaskEmpty'],
     );
 
     /**