Merge commit 'ed96e0f8589042fb8e9d6cfee713c5150612bb34'
This commit is contained in:
@@ -0,0 +1,244 @@
|
||||
<?php
|
||||
|
||||
namespace app\common\utils;
|
||||
|
||||
use GatewayWorker\Lib\Gateway;
|
||||
use app\common\utils\WsResponse;
|
||||
use app\common\service\MessageNotificationsService;
|
||||
|
||||
class WsEvents
|
||||
{
|
||||
|
||||
private $registerAddress = '127.0.0.1:1236';
|
||||
protected $clientId;
|
||||
protected $eventData;
|
||||
|
||||
public function __construct()
|
||||
{
|
||||
Gateway::$registerAddress = $this->registerAddress;
|
||||
}
|
||||
|
||||
/**
|
||||
* 监听消息接收
|
||||
*/
|
||||
public function message($client_id, $message)
|
||||
{
|
||||
$this->clientId = $client_id;
|
||||
|
||||
$payload = json_decode($message, true);
|
||||
$event = $payload['event'];
|
||||
$this->eventData = $payload['data'] ?? [];
|
||||
$traceId = $payload['trace_id'] ?? null;
|
||||
|
||||
$method = $this->eventToMethod($event);
|
||||
|
||||
if (method_exists($this, $method)) {
|
||||
try {
|
||||
$this->$method($traceId);
|
||||
} catch (\Throwable $th) {
|
||||
Gateway::sendToClient($client_id, WsResponse::error(
|
||||
$event,
|
||||
500,
|
||||
$th->getMessage(),
|
||||
$traceId
|
||||
));
|
||||
}
|
||||
} else {
|
||||
Gateway::sendToClient($client_id, WsResponse::error(
|
||||
$event,
|
||||
404,
|
||||
"Event handler not found: $method",
|
||||
$traceId
|
||||
));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// ================= 系统事件 =================
|
||||
/**
|
||||
* 认证
|
||||
*/
|
||||
public function auth($client_id, $payload)
|
||||
{
|
||||
try {
|
||||
// 检查来源
|
||||
if (!isset($payload['server']['HTTP_ORIGIN'])) {
|
||||
flg('[x] [TOKEN认证] 未定义HTTP_ORIGIN', false);
|
||||
Gateway::sendToClient($client_id, WsResponse::error(
|
||||
'system.auth',
|
||||
1001,
|
||||
'[x] [TOKEN认证] 未定义HTTP_ORIGIN'
|
||||
));
|
||||
Gateway::closeClient($client_id);
|
||||
return;
|
||||
}
|
||||
|
||||
// 连接是否满足条件
|
||||
// if ($payload['server']['HTTP_ORIGIN'] != 'https://wstool.js.org') {
|
||||
// flg('[x] [TOKEN认证] HTTP_ORIGIN不满足条件', false);
|
||||
// Gateway::sendToClient($client_id, WsResponse::error(
|
||||
// 'system.auth',
|
||||
// 1001,
|
||||
// '[x] [TOKEN认证] HTTP_ORIGIN不满足条件'
|
||||
// ));
|
||||
// Gateway::closeClient($client_id);
|
||||
// return;
|
||||
// }
|
||||
|
||||
// 签名检查
|
||||
if (!isset($payload['get']['token'])) {
|
||||
flg('[x] [TOKEN认证] 未携带TOKEN参数', false);
|
||||
Gateway::sendToClient($client_id, WsResponse::error(
|
||||
'system.auth',
|
||||
1001,
|
||||
'[x] [TOKEN认证] 未携带TOKEN参数'
|
||||
));
|
||||
Gateway::closeClient($client_id);
|
||||
return;
|
||||
}
|
||||
|
||||
// 解析TOKEN
|
||||
$user = (new Jwt())->verifyToken($payload['get']['token']);
|
||||
// 绑定用户ID
|
||||
Gateway::bindUid($client_id, $user->uid);
|
||||
|
||||
// 响应
|
||||
Gateway::sendToClient($client_id, WsResponse::success(
|
||||
'system.auth',
|
||||
['uid' => $user->uid]
|
||||
));
|
||||
} catch (\Throwable $th) {
|
||||
flg('[x] [TOKEN认证] TOKEN认证失败:'.$th->getMessage(), false);
|
||||
Gateway::sendToClient($client_id, WsResponse::error(
|
||||
'system.auth',
|
||||
1001,
|
||||
'[x] [TOKEN认证] TOKEN认证失败:'.$th->getMessage()
|
||||
));
|
||||
Gateway::closeClient($client_id);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 系统心跳
|
||||
*
|
||||
* @param string|null $traceId
|
||||
*/
|
||||
public function systemPing(?string $traceId)
|
||||
{
|
||||
Gateway::sendToClient($this->clientId, WsResponse::success(
|
||||
'system.pong',
|
||||
['time' => time()],
|
||||
$traceId
|
||||
));
|
||||
}
|
||||
|
||||
|
||||
// ================= 消息通知事件 =================
|
||||
/**
|
||||
* 管理端主动推送通知
|
||||
*/
|
||||
public function notifyPush(?string $traceId)
|
||||
{
|
||||
$title = $this->eventData['title'] ?? null;
|
||||
$content = $this->eventData['content'] ?? null;
|
||||
$toUids = $this->eventData['to_uids'] ?? [];
|
||||
|
||||
if (!$title || !$content) {
|
||||
Gateway::sendToClient($this->clientId, WsResponse::error(
|
||||
'notify.push.result',
|
||||
400,
|
||||
'标题和内容不能为空',
|
||||
$traceId
|
||||
));
|
||||
return;
|
||||
}
|
||||
|
||||
// 写消息
|
||||
$msgService = new MessageNotificationsService();
|
||||
$msgId = $msgService->createMessage([
|
||||
'sender_id' => $this->eventData['sender_id'] ?? 0,
|
||||
'sender_type' => $this->eventData['sender_type'] ?? 'system',
|
||||
'title' => $title,
|
||||
'content' => $content,
|
||||
'scene' => $this->eventData['scene'] ?? 0,
|
||||
'recipient' => $this->eventData['to_uids'] ? implode(',', $this->eventData['to_uids']) : '',
|
||||
'push_method' => isset($this->eventData['push_method']) ? $this->eventData['push_method'] : 0
|
||||
]);
|
||||
|
||||
// 写用户状态
|
||||
$uids = !empty($toUids) ? $toUids : $msgService->getAllUserIds();
|
||||
$msgService->addUserStatus($msgId, $uids);
|
||||
|
||||
$payload = WsResponse::success('notify.receive', [
|
||||
'msg_id' => $msgId,
|
||||
'title' => $title,
|
||||
'content' => $content,
|
||||
'created_at' => time()
|
||||
], $traceId);
|
||||
|
||||
if (!empty($toUids)) {
|
||||
// 推送给指定用户
|
||||
foreach ($toUids as $uid) {
|
||||
Gateway::sendToUid($uid, $payload);
|
||||
}
|
||||
} else {
|
||||
// 推送给全体在线用户
|
||||
Gateway::sendToAll($payload);
|
||||
}
|
||||
|
||||
// 回复发送者确认
|
||||
Gateway::sendToClient($this->clientId, WsResponse::success(
|
||||
'notify.push.result',
|
||||
['msg_id' => $msgId],
|
||||
$traceId,
|
||||
'消息发送成功'
|
||||
));
|
||||
}
|
||||
|
||||
/**
|
||||
* 客户端标记通知已读
|
||||
*/
|
||||
public function notifyRead(?string $traceId)
|
||||
{
|
||||
$msgIds = $this->eventData['msg_ids'];
|
||||
if (!is_array($msgIds) && $msgIds !== 'all') {
|
||||
Gateway::sendToClient($this->clientId, WsResponse::error(
|
||||
'notify.read.result',
|
||||
400,
|
||||
'消息ID不能为空',
|
||||
$traceId
|
||||
));
|
||||
return;
|
||||
}
|
||||
|
||||
// 更新消息为已读
|
||||
$msgService = new MessageNotificationsService();
|
||||
$msgService->markAsRead($msgIds, Gateway::getUidByClientId($this->clientId));
|
||||
|
||||
Gateway::sendToClient($this->clientId, WsResponse::success(
|
||||
'notify.read.result',
|
||||
['msg_ids' => $msgIds],
|
||||
$traceId
|
||||
));
|
||||
}
|
||||
|
||||
/**
|
||||
* 推送消息通知
|
||||
*/
|
||||
public function pushWsNotice(array $wsData, int $userId)
|
||||
{
|
||||
Gateway::sendToUid($userId, WsResponse::success('notify.receive', $wsData));
|
||||
}
|
||||
|
||||
// ================= 辅助方法 =================
|
||||
protected function eventToMethod(string $event): string
|
||||
{
|
||||
$parts = explode('.', $event);
|
||||
$method = array_shift($parts);
|
||||
foreach ($parts as $part) {
|
||||
$method .= ucfirst($part);
|
||||
}
|
||||
return $method;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
<?php
|
||||
namespace app\common\utils;
|
||||
|
||||
class WsResponse
|
||||
{
|
||||
/**
|
||||
* 成功返回
|
||||
*
|
||||
* @param string $event 事件名称
|
||||
* @param array $data 返回数据
|
||||
* @param string|null $traceId 请求追踪ID,可选,默认 null
|
||||
* @param int $code 错误码
|
||||
* @param string $message 错误信息
|
||||
* @return string JSON
|
||||
*/
|
||||
public static function success(string $event, array $data = [], ?string $traceId = null, string $message = 'ok', int $code = 0): string
|
||||
{
|
||||
return json_encode([
|
||||
'event' => $event,
|
||||
'code' => $code,
|
||||
'message' => $message,
|
||||
'data' => $data,
|
||||
'trace_id' => $traceId, // 如果没有传就是 null
|
||||
], JSON_UNESCAPED_UNICODE);
|
||||
}
|
||||
|
||||
/**
|
||||
* 错误返回
|
||||
*
|
||||
* @param string $event 事件名称
|
||||
* @param int $code 错误码
|
||||
* @param string $message 错误信息
|
||||
* @param string|null $traceId 请求追踪ID,可选,默认 null
|
||||
* @param array $data 附加数据(可选)
|
||||
* @return string JSON
|
||||
*/
|
||||
public static function error(string $event, int $code, string $message, ?string $traceId = null, array $data = []): string
|
||||
{
|
||||
return json_encode([
|
||||
'event' => $event,
|
||||
'code' => $code,
|
||||
'message' => $message,
|
||||
'data' => $data,
|
||||
'trace_id' => $traceId, // 如果没有传就是 null
|
||||
], JSON_UNESCAPED_UNICODE);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user