From 837c1e54d6c5a11000437133ad7947b4a333575c Mon Sep 17 00:00:00 2001 From: Yu Sun Date: Thu, 4 Aug 2022 15:27:43 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=B6=E4=BD=9C=E4=B8=80=E4=B8=AAtext?= =?UTF-8?q?=E5=8D=8F=E8=AE=AE=E7=9A=84worker=E6=9D=A5=E6=B6=88=E8=B4=B9?= =?UTF-8?q?=E6=96=B0=E5=A2=9E=E8=8A=82=E7=82=B9=E7=9A=84=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- EdgeManager.php | 103 +++++++++++++++++++++++++++++++++++++----------- 1 file changed, 79 insertions(+), 24 deletions(-) diff --git a/EdgeManager.php b/EdgeManager.php index febaa18..8e90634 100644 --- a/EdgeManager.php +++ b/EdgeManager.php @@ -1,5 +1,6 @@ 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 dbname=scada user=%s password=%s", + $options['server_name'], $options['user'], $options['password'] + )); }; $worker -> onMessage = function(TcpConnection $connection, Request $request) { @@ -44,9 +47,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,24 +64,36 @@ $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, 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); if ($data_capture -> check_res === 'WRONG_WORKING_SUBCLASS') { @@ -206,4 +221,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 dbname=scada user=%s password=%s", + $options['server_name'], $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, 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' => '同一工序单元内的节点编码不可重复！' + ))); + else if ($res === false) { + $connection -> send(json_encode(array( + 'code' => 1, + 'msg' => '服务器内部逻辑错误，请联系开发者！' + ))); + } + } +}; + Worker::runAll();