wyw/EdgeManager
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

制作一个text协议的worker来消费新增节点的任务

Browse Source
This commit is contained in:
Yu Sun
2022-08-04 15:27:43 +08:00
parent 6e86509e27
commit 837c1e54d6
1 changed files with 79 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
EdgeManager.php
View File

@@ -1,5 +1,6 @@
<?php <?php
use Workerman\Worker; use Workerman\Worker;
use \Workerman\Connection\AsyncTcpConnection;
use Workerman\Connection\TcpConnection; use Workerman\Connection\TcpConnection;
use Workerman\Protocols\Http\Request; use Workerman\Protocols\Http\Request;
use Workerman\Protocols\Http\Response; use Workerman\Protocols\Http\Response;
@@ -12,14 +13,16 @@ $options = getopt('h::', ['server_name:', 'user:', 'password:', 'help::']);
init_db($options['server_name'], $options['user'], $options['password']); init_db($options['server_name'], $options['user'], $options['password']);
$worker = new Worker('http://0.0.0.0:8888'); $worker = new Worker('http://0.0.0.0:8888');
$worker -> name = 'CaptureWorker'; $worker -> name = 'EntryPoint';
$worker -> count = 20; $worker -> count = 20;
$worker -> onWorkerStart = function(Worker $worker) { $worker -> onWorkerStart = function(Worker $worker) {
global $options, $dbconn; global $options, $dbconn;
$dbconn = pg_connect(sprintf( "host=%s dbname=scada user=%s password=%s", $dbconn = pg_connect(sprintf(
$options['server_name'], $options['user'], $options['password'])); "host=%s dbname=scada user=%s password=%s",
$options['server_name'], $options['user'], $options['password']
));
}; };
$worker -> onMessage = function(TcpConnection $connection, Request $request) { $worker -> onMessage = function(TcpConnection $connection, Request $request) {
@@ -44,9 +47,9 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
$response = new Response(200, [ $response = new Response(200, [
'Content-Type' => 'application/json;charset=utf-8', 'Content-Type' => 'application/json;charset=utf-8',
], json_encode(array( ], json_encode(array(
'action' => '错得太离谱以至于我也不知道这是什么动作', 'action' => '错得太离谱以至于我也不知道这是什么动作',
'errcode' => 4001, 'errcode' => 4001,
'errmsg' => 'POST的内容不是JSON，也并非可按照JSON解析的字符串！或者检查一下你的JSON是不是有注释？' 'errmsg' => 'POST的内容不是JSON，也并非可按照JSON解析的字符串！或者检查一下你的JSON是不是有注释？'
))); )));
$connection -> send($response); $connection -> send($response);
} }
@@ -61,23 +64,35 @@ $worker -> onMessage = function(TcpConnection $connection, Request $request) {
// 因为pg_insert()不需要action字段 // 因为pg_insert()不需要action字段
$action = $post -> action; $action = $post -> action;
unset($post -> action); unset($post -> action);
$enode_configure = new ENodeConfigure($dbconn, post: $post);
$res = $enode_configure -> $action(); if ($action === 'add_node') {
if ($res === true) $task_connection = new AsyncTcpConnection('Text://127.0.0.1:1888');
$connection -> send(json_encode(array( $task_data = array(
'code' => 0, 'action' => 'add_node',
'msg' => 'Success' 'data' => $post,
))); );
else if ($res === "REPLICATED") $task_connection -> send(json_encode($task_data));
$connection -> send(json_encode(array( $task_connection -> onMessage = function(AsyncTcpConnection $task_connection, $task_result) use ($connection) {
'code' => 1, $task_connection -> close();
'msg' => '同一工序单元内的节点编码不可重复！' $connection -> send($task_result);
))); };
else if ($res === false) { // 执行异步连接
$connection -> send(json_encode(array( $task_connection->connect();
'code' => 1, } else {
'msg' => '服务器内部逻辑错误，请联系开发者！' $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') { } else if ($post -> action === 'set_node_data') {
$data_capture = new EDataCapture($dbconn, post: $post); $data_capture = new EDataCapture($dbconn, post: $post);
@@ -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(); Worker::runAll();