問題描述

workerman實現(xiàn)客戶端訂閱第三方，訂閱數(shù)據(jù)取自外部怎么實現(xiàn)，外部數(shù)據(jù)變動自動訂閱而不是重啟服務(wù)

代碼實現(xiàn)已經(jīng)可以正常訂閱并轉(zhuǎn)發(fā)給系統(tǒng)內(nèi)部啟動的gateway進(jìn)程

namespace app\pushserver\controller;

use Workerman\Worker;
use Workerman\Connection\AsyncTcpConnection;
use GatewayWorker\Lib\Gateway;
use app\common\model\Currency;

class Huobisub{
private $worker;
//訂閱K線數(shù)據(jù)
private $subscribed;
//訂閱k線時序
private $kline_periods;
public function __construct(){
$this->worker = new Worker();
$this->worker->count = 10;
$this->worker->name = 'Huobi Websocket';
$this->worker->onWorkerStart = function($worker){
$klinesub = Currency::getKlineSub();
$this->subscribed = $klinesub['subscribed'];
$this->kline_periods = $klinesub['kline_periods'];
// ssl需要訪問443端口
$con = new AsyncTcpConnection('ws://api.exmple.com:443/linear-swap-ws');
// 設(shè)置以ssl加密方式訪問，使之成為wss
$con->transport = 'ssl';
$con->onConnect = function(AsyncTcpConnection $con) use ($worker) {
echo "進(jìn)程啓動:".$worker->id. PHP_EOL;
foreach($this->subscribed as $topic=>$subinfo){
$data = ['sub' => $topic, 'id' => $topic];
$con->send(json_encode($data));
}
};
$con->onMessage = function(AsyncTcpConnection $con, $data) use ($worker) {
//echo "進(jìn)程答復(fù):".$worker->id. PHP_EOL;
$data = gzdecode($data);
Gateway::sendToGroup($data['ch'],$data);
};
$con->onClose = function ($con) {
// 如果連接斷開，則在1秒后重連
echo 'work exited,reconnecting......' . PHP_EOL;
$con->reConnect(1);
};
$con->connect();
};
//運行所有Worker;
Worker::runAll();
}
}