Merge branch 'master' of ssh://118.195.187.246:10022/ysun/EdgeManager

.dockerignore

@@ -0,0 +1 @@
+*

EdgeManager.php

@@ -1,5 +1,6 @@
 <?php
 use Workerman\Worker;
+use \Workerman\Connection\AsyncTcpConnection;
 use Workerman\Connection\TcpConnection;
 use Workerman\Protocols\Http\Request;
 use Workerman\Protocols\Http\Response;
@@ -7,19 +8,36 @@ require_once __DIR__ . '/vendor/autoload.php';
 
 use EdgeManager\EDataCapture\{ EDataCapture, ENodeConfigure };
 
-$options = getopt('h::', ['server_name:', 'user:', 'password:', 'help::']);
+$options = getopt('h::', ['no_dup_code', 'server_name:', 'port::', 'user:', 'password:', 'help::']);
 
-init_db($options['server_name'], $options['user'], $options['password']);
+if (array_key_exists('h', $options) or array_key_exists('help', $options)) {
+    print_r(
+        "EdgeManager使用说明：
+
+        --no_dup_code   禁止code在不同的working subclass间复用
+        --server_name   pg实例的FQDN
+        --user          pg实例的用户名
+        --port          pg实例的端口号
+        --password      pg实例的密码
+        -h, --help      显示此帮助信息
+"
+    );
+    exit;
+}
+
+init_db($options['server_name'], $options['port'] ?? 5432, $options['user'], $options['password']);
 
 $worker = new Worker('http://0.0.0.0:8888');
-$worker -> name = 'CaptureWorker';
+$worker -> name = 'EntryPoint';
 $worker -> count = 20;
 
 $worker -> onWorkerStart = function(Worker $worker) {
     global $options, $dbconn;
 
-    $dbconn = pg_connect(sprintf(  "host=%s dbname=scada user=%s password=%s",
-                                    $options['server_name'], $options['user'], $options['password']));
+    $dbconn = pg_connect(sprintf(
+        "host=%s port=%s dbname=scada user=%s password=%s",
+        $options['server_name'], $options['port'] ?? 5432, $options['user'], $options['password']
+    ));
 };
 
 $worker -> onMessage = function(TcpConnection $connection, Request $request) {
@@ -44,9 +62,9 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
                 $response = new Response(200, [
                     'Content-Type' => 'application/json;charset=utf-8',
                 ], json_encode(array(
-                    'action' => '错得太离谱以至于我也不知道这是什么动作',
-                    'errcode' => 4001,
-                    'errmsg' => 'POST的内容不是JSON，也并非可按照JSON解析的字符串！或者检查一下你的JSON是不是有注释？'
+                    'action'    => '错得太离谱以至于我也不知道这是什么动作',
+                    'errcode'   => 4001,
+                    'errmsg'    => 'POST的内容不是JSON，也并非可按照JSON解析的字符串！或者检查一下你的JSON是不是有注释？'
                 )));
                 $connection -> send($response);
             }
@@ -61,26 +79,38 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
                 // 因为pg_insert()不需要action字段
                 $action = $post -> action;
                 unset($post -> action);
-                $enode_configure = new ENodeConfigure($dbconn, post: $post);
-                $res = $enode_configure -> $action();
-                if ($res === true)
-                    $connection -> send(json_encode(array(
-                        'code'  => 0,
-                        'msg'   => 'Success'
-                    )));
-                else if ($res === "REPLICATED")
-                    $connection -> send(json_encode(array(
-                        'code'  => 1,
-                        'msg'   => '同一工序单元内的节点编码不可重复！'
-                    )));
-                else if ($res === false) {
-                    $connection -> send(json_encode(array(
-                        'code'  => 1,
-                        'msg'   => '服务器内部逻辑错误，请联系开发者！'
-                    )));
+
+                if ($action === 'add_node') {
+                    $task_connection = new AsyncTcpConnection('Text://127.0.0.1:1888');
+                    $task_data = array(
+                        'action' => 'add_node',
+                        'data'   => $post,
+                    );
+                    $task_connection -> send(json_encode($task_data));
+                    $task_connection -> onMessage = function(AsyncTcpConnection $task_connection, $task_result) use ($connection) {
+                        $task_connection -> close();
+                        $connection -> send($task_result);
+                    };
+                    // 执行异步连接
+                    $task_connection->connect();
+                } else {
+                    $enode_configure = new ENodeConfigure($dbconn, no_dup_code: $options['no_dup_code'] ?? true, post: $post);
+                    $res = $enode_configure -> $action();
+                    
+                    if ($res === true)
+                        $connection -> send(json_encode(array(
+                            'code'  => 0,
+                            'msg'   => 'Success'
+                        )));
+                    else if ($res === false) {
+                        $connection -> send(json_encode(array(
+                            'code'  => 1,
+                            'msg'   => '服务器内部逻辑错误，请联系开发者！'
+                        )));
+                    }
                 }                
             } else if ($post -> action === 'set_node_data') {
-                $data_capture = new EDataCapture($dbconn, post: $post);
+                $data_capture = new EDataCapture($dbconn, no_dup_code: $options['no_dup_code'] ?? true, post: $post);
                 if ($data_capture -> check_res === 'WRONG_WORKING_SUBCLASS') {
                     $response = new Response(200, [
                         'Content-Type' => 'application/json;charset=utf-8',
@@ -141,7 +171,7 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
     $get = $request -> get();
     if (isset($get['query'])) {
         if ($get['query'] == 'nodes') {
-            $enode_configure = new ENodeConfigure($dbconn, get: $get);
+            $enode_configure = new ENodeConfigure($dbconn, no_dup_code: $options['no_dup_code'] ?? true, get: $get);
             $nodes = $enode_configure -> get_nodes();
             if (is_null($nodes))
                 $connection -> send(json_encode(array(
@@ -154,7 +184,7 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
                     'data'  => $nodes
                 )));
         } else if ($get['query'] == 'working_subclasses') {
-            $enode_configure = new ENodeConfigure($dbconn, get: $get);
+            $enode_configure = new ENodeConfigure($dbconn, no_dup_code: $options['no_dup_code'] ?? true, get: $get);
             $working_subclasses = $enode_configure -> get_working_subclasses($dbconn);
             if (is_null($working_subclasses))
                 $connection -> send(json_encode(array(
@@ -167,7 +197,7 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
                     'data'  => $working_subclasses
                 )));
         } else if ($get['query'] == 'codes') {
-            $enode_configure = new ENodeConfigure($dbconn, get: $get);
+            $enode_configure = new ENodeConfigure($dbconn, no_dup_code: $options['no_dup_code'] ?? true, get: $get);
             $codes = $enode_configure -> get_codes_by_working_subclasses();
             if (is_null($codes))
                 $connection -> send(json_encode(array(
@@ -180,7 +210,7 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
                     'data'  => $codes
                 )));
         } else if ($get['query'] == 'node_data') {
-            $data_capture = new EDataCapture($dbconn, get: $get);
+            $data_capture = new EDataCapture($dbconn, no_dup_code: $options['no_dup_code'] ?? true, get: $get);
             $data = $data_capture -> get_node_data();
             if (is_null($data))
                 $connection -> send(json_encode(array(
@@ -206,4 +236,44 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
     }
 };
 
+$consumer = new Worker('Text://0.0.0.0:1888');
+$consumer -> name = 'Consumer';
+
+$consumer -> onWorkerStart = function(Worker $consumer) {
+    global $options, $consumer_dbconn;
+
+    $consumer_dbconn = pg_connect(sprintf(
+        "host=%s port=%s dbname=scada user=%s password=%s",
+        $options['server_name'], $options['port'] ?? 5432, $options['user'], $options['password']
+    ));
+};
+
+$consumer -> onMessage = function(TcpConnection $connection, $task_data) {
+    global $consumer_dbconn;
+
+    $task_data = json_decode($task_data);
+
+    if ($task_data -> action === 'add_node') {
+        $enode_configure = new ENodeConfigure($consumer_dbconn, no_dup_code: $options['no_dup_code'] ?? true, post: $task_data -> data);
+        $res = $enode_configure -> add_node();
+    
+        if ($res === true)
+            $connection -> send(json_encode(array(
+                'code'  => 0,
+                'msg'   => 'Success'
+            )));
+        else if ($res === "REPLICATED")
+            $connection -> send(json_encode(array(
+                'code'  => 1,
+                'msg'   => isset($options['no_dup_code']) ? '节点编码不可重复！' : '同一工序单元内的节点编码不可重复！'
+            )));
+        else if ($res === false) {
+            $connection -> send(json_encode(array(
+                'code'  => 1,
+                'msg'   => '服务器内部逻辑错误，请联系开发者！'
+            )));
+        }
+    }
+};
+
 Worker::runAll();

EdgeManager/EDataCapture/EDataCapture.php

@@ -4,30 +4,44 @@ namespace EdgeManager\EDataCapture;
 class EDataCapture {
     function __construct(
         protected $dbconn,
+        protected $no_dup_code,
         protected $post = NULL,
         protected $get = NULL,
         public $check_res = NULL,
         protected $working_subclass = NULL,
         protected $code_type = [],
-        protected $data = []
+        protected $data = [],
     ) {
         if (!is_null($this -> post)) {
-            if (!in_array(
-                $this -> post -> param -> working_subclass,
-                ENodeConfigure::get_working_subclasses($this -> dbconn)
-            )) {
-                $this -> check_res = 'WRONG_WORKING_SUBCLASS';
-                return;
-            } else {
-                $this -> working_subclass =  $this -> post -> param -> working_subclass;
+            if(isset($this -> post -> param -> working_subclass)){
+                if (!in_array(
+                    $this -> post -> param -> working_subclass,
+                    ENodeConfigure::get_working_subclasses($this -> dbconn)
+                )) {
+                    $this -> check_res = 'WRONG_WORKING_SUBCLASS';
+                    return;
+                } else {
+                    $this -> working_subclass =  $this -> post -> param -> working_subclass;
+                }
+            }else{
+                //根据code查询working_subclasses
+                $working_subclass = ENodeConfigure::get_working_subclasses_by_codes($this -> dbconn, $this -> post -> param -> data[0] -> code);
+                $this -> working_subclass = $working_subclass[0];
             }
     
-            $res = pg_fetch_all(pg_query($this -> dbconn, sprintf(
-                "SELECT code, type
-                FROM hf_mes_scada_data_capture_node_configure
-                WHERE working_subclass = '%s'",
-                $this -> post -> param -> working_subclass
-            )));
+            if ($this -> no_dup_code) {
+                $res = pg_fetch_all(pg_query($this -> dbconn, 
+                    "SELECT code, type
+                    FROM hf_mes_scada_data_capture_node_configure"
+                ));
+            } else {
+                $res = pg_fetch_all(pg_query($this -> dbconn, sprintf(
+                    "SELECT code, type
+                    FROM hf_mes_scada_data_capture_node_configure
+                    WHERE working_subclass = '%s'",
+                    $this -> working_subclass
+                )));
+            }
 
             $code_type = &$this -> code_type;
             array_walk($res, function(&$v, $k) use (&$code_type) {

EdgeManager/EDataCapture/ENodeConfigure.php

@@ -4,24 +4,37 @@ namespace EdgeManager\EDataCapture;
 class ENodeConfigure {
     function __construct(
         protected $dbconn,
+        protected $no_dup_code,
         protected $post = NULL,
-        protected $get = NULL
+        protected $get = NULL,
     ) {}
 
     function add_node() {
-        $exists = pg_query($this -> dbconn, sprintf(
-            "SELECT EXISTS(
-                SELECT 1 FROM hf_mes_scada_data_capture_node_configure
-                WHERE code = '%s'
-                AND working_subclass = '%s'
-            )", $this -> post -> code, $this -> post -> working_subclass
-        ));
-        if (pg_fetch_assoc($exists)['exists'] == 't') {
+        pg_query($this -> dbconn, "BEGIN");
+        if ($this -> no_dup_code) {
+            $exists = pg_query($this -> dbconn, sprintf(
+                "SELECT EXISTS(
+                    SELECT 1 FROM hf_mes_scada_data_capture_node_configure
+                    WHERE code = '%s'
+                )", $this -> post -> code
+            ));
+        } else {
+            $exists = pg_query($this -> dbconn, sprintf(
+                "SELECT EXISTS(
+                    SELECT 1 FROM hf_mes_scada_data_capture_node_configure
+                    WHERE code = '%s'
+                    AND working_subclass = '%s'
+                )", $this -> post -> code, $this -> post -> working_subclass
+            ));
+        }
+        
+        if (pg_fetch_assoc($exists)['exists'] === 't') {
+            pg_query($this -> dbconn, "ROLLBACK");
             return "REPLICATED";
         }
 
         $table_name = 'hf_mes_scada_data_capture_node_data_' . $this -> post -> working_subclass;
-        pg_query($this -> dbconn, "BEGIN");
+        // https://www.postgresql.org/docs/current/when-can-parallel-query-be-used.html
         $res[] = pg_query($this -> dbconn, sprintf(
             'CREATE TABLE IF NOT EXISTS "%s" (
                 id serial8,
@@ -98,7 +111,7 @@ class ENodeConfigure {
             $this -> dbconn,
             'hf_mes_scada_data_capture_node_configure',
             (array) $this -> post,
-            ['code' => $this -> post -> code]
+            ['id' => $this -> post -> id]
         );
     }
 
@@ -112,6 +125,16 @@ class ENodeConfigure {
         return pg_fetch_all_columns($res, 0);
     }
 
+    static function get_working_subclasses_by_codes($dbconn, $code) {
+        $res = pg_query($dbconn, sprintf(
+            "SELECT working_subclass
+            FROM hf_mes_scada_data_capture_node_configure
+            WHERE code = '%s'",
+            $code
+        ));
+        return pg_fetch_all_columns($res, 0);
+    }
+
     function get_codes_by_working_subclasses() {
         $res = pg_query($this -> dbconn, sprintf(
             "SELECT code

EdgeManager/Init.php

@@ -1,13 +1,13 @@
 <?php
-function init_db($server_name, $user, $password) {
-    $dbconn = pg_connect(sprintf("host=%s user=%s password=%s", $server_name, $user, $password));
+function init_db($server_name, $port, $user, $password) {
+    $dbconn = pg_connect(sprintf("host=%s port=%s user=%s password=%s", $server_name, $port, $user, $password));
     $res = pg_query($dbconn, "select datname from pg_database");
     
     if (!in_array('scada', pg_fetch_all_columns($res, 0)))
         pg_query($dbconn, "CREATE DATABASE scada");
     pg_close($dbconn);
 
-    $dbconn = pg_connect(sprintf("host=%s dbname=scada user=%s password=%s", $server_name, $user, $password));
+    $dbconn = pg_connect(sprintf("host=%s port=%s dbname=scada user=%s password=%s", $server_name, $port, $user, $password));
     // 全局未启用TSDB的时候在这里对scada数据库启用就好
     pg_query($dbconn, "CREATE EXTENSION IF NOT EXISTS timescaledb");
     pg_query($dbconn, "CREATE TABLE IF NOT EXISTS hf_mes_scada_data_capture_node_configure (

README.md

@@ -8,6 +8,7 @@
     - [后端](#后端)
     - [前端](#前端)
   - [开发环境](#开发环境)
+  - [调试环境](#调试环境)
   - [技术细节](#技术细节)
     - [0. `EdgeManager\EDataCapture\EDataCapture -> set_data()`为什么是以`6710885`为大小chunked的？](#0-edgemanageredatacaptureedatacapture---set_data为什么是以6710885为大小chunked的)
 
@@ -257,6 +258,7 @@ cd EdgeManager
 一键部署PHP workerman和TimescaleDB环境：
 
 ```bash
+# docker build --network host -t edge_manager .
 docker compose up -d
 ```
 
@@ -270,7 +272,7 @@ docker exec -it edge_manager bash
 
 ```bash
 # In container
-php EdgeManager.php --server_name=GPU-server-01 --user=postgres --password=big_dick start
+php EdgeManager.php --no_dup_code --server_name=GPU-server-01 --user=postgres --password=big_dick start
 ```
 
 前端调试：
@@ -292,6 +294,29 @@ psql -h localhost -U postgres
 \l
 ```
 
+## 调试环境
+
+不使用`docker compose`创建两个container分别运行EdgeManager和pg：
+
+```bash
+# cd EdgeManager  # 先定位到项目目录，方便创建image和挂载
+# 创建EdgeManager的image
+docker build --network host -t edge_manager .
+# 创建container运行EdgeManager
+docker run -d --name edge_manager_test -v $PWD:/EdgeManager --network host -it edge_manager
+# 创建container运行pg，将端口映射到host的55432
+docker run -d --name pg_test -v $PWD/config/postgresql.conf:/etc/postgresql/postgresql.conf -p 55432:5432 -e POSTGRES_PASSWORD=big_dick -it timescale/timescaledb-ha:pg14-latest postgres -c 'config_file=/etc/postgresql/postgresql.conf'
+# 进入交互式Prompt
+docker exec -it edge_manager_test bash
+# 启动EdgeManager（workerman）命令省略...
+
+# 常用命令
+# 查看全部container
+docker ps -a
+# 启动已停止的container
+docker start [container]
+```
+
 ## 技术细节
 
 ### 0. `EdgeManager\EDataCapture\EDataCapture -> set_data()`为什么是以`6710885`为大小chunked的？

docker-compose.yml

@@ -1,5 +1,5 @@
 services:
-  scada:
+  edge_manager:
     build:
       context: .
     container_name: edge_manager