molong/laravel_swoole
Template
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

格式化代码，websocket功能完善

Browse Source
This commit is contained in:
molong
2026-02-18 21:50:05 +08:00
parent 6543e2ccdd
commit b6c133952b
101 changed files with 15829 additions and 10739 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
app/Services/Auth/UserService.php
View File

@@ -3,6 +3,8 @@
namespace App\Services\Auth;
use App\Models\Auth\User;
use App\Models\System\Notification;
use App\Services\System\NotificationService;
use Illuminate\Support\Facades\Auth;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Hash;
@@ -16,10 +18,12 @@ use App\Jobs\Auth\UserExportJob;
class UserService
{
protected $departmentService;
protected $notificationService;
public function __construct(DepartmentService $departmentService)
public function __construct(DepartmentService $departmentService, NotificationService $notificationService)
{
$this->departmentService = $departmentService;
$this->notificationService = $notificationService;
}
/**
@@ -29,6 +33,7 @@ class UserService
{
return Auth::guard('admin')->id();
}
/**
* 获取用户列表
*/
@@ -215,6 +220,12 @@ class UserService
}
DB::commit();
// 发送更新通知（如果更新的是其他用户）
// if ($id !== $this->getCurrentUserId()) {
$this->sendUserUpdateNotification($user, $data);
// }
return $user;
} catch (\Exception $e) {
DB::rollBack();
@@ -352,4 +363,52 @@ class UserService
'updated_at' => $user->updated_at->toDateTimeString(),
];
}
/**
* 发送用户更新通知
*/
private function sendUserUpdateNotification(User $user, array $data): void
{
// 收集被更新的字段
$changes = [];
$fieldLabels = [
'username' => '用户名',
'real_name' => '姓名',
'email' => '邮箱',
'phone' => '手机号',
'department_id' => '所属部门',
'avatar' => '头像',
'status' => '状态',
'password' => '密码',
'role_ids' => '角色',
];
foreach ($data as $key => $value) {
if (isset($fieldLabels[$key])) {
$changes[] = $fieldLabels[$key];
}
}
if (empty($changes)) {
return;
}
// 生成通知内容
$content = '您的账户信息已被管理员更新，更新的内容：' . implode('、', $changes);
// 发送通知
$this->notificationService->sendToUser(
$user->id,
'个人信息已更新',
$content,
Notification::TYPE_INFO,
Notification::CATEGORY_SYSTEM,
[
'user_id' => $user->id,
'updated_fields' => $changes,
'action_type' => Notification::ACTION_NONE,
]
);
}
}

867
app/Services/WebSocket/WebSocketHandler.php
View File

@@ -3,513 +3,530 @@
namespace App\Services\WebSocket;
use Hhxsv5\LaravelS\Swoole\WebSocketHandlerInterface;
use Illuminate\Support\Facades\Log;
use Swoole\Http\Request;
use Swoole\Http\Response;
use Swoole\WebSocket\Frame;
use Swoole\WebSocket\Server;
use Illuminate\Support\Facades\Log;
use App\Services\Auth\UserOnlineService;
use Tymon\JWTAuth\Facades\JWTAuth;
/**
* WebSocket Handler
* WebSocket 处理器
*
* Handles WebSocket connections, messages, and disconnections
* 处理 WebSocket 连接事件：onOpen, onMessage, onClose
*/
class WebSocketHandler implements WebSocketHandlerInterface
{
/**
* @var UserOnlineService
*/
protected $userOnlineService;
/**
* Get wsTable instance
*
* @return \Swoole\Table
*/
protected function getWsTable(): \Swoole\Table
{
return app('swoole')->wsTable;
}
/**
* WebSocketHandler constructor
* WebSocketHandlerInterface 需要的构造函数
* wsTable 直接从 handler 方法的 $server 参数中访问
*/
public function __construct()
{
$this->userOnlineService = app(UserOnlineService::class);
// 空构造函数 - wsTable 从 server 参数中访问
}
/**
* Handle WebSocket connection open event
* 处理 WebSocket 握手（可选）
*
* @param Server $server
* @param Request $request
* @param Request $request 请求对象
* @param Response $response 响应对象
* @return void
*/
// public function onHandShake(Request $request, Response $response)
// {
// // 自定义握手逻辑（如果需要）
// // 握手成功后，onOpen 事件会自动触发
// }
/**
* 处理连接打开事件
*
* @param Server $server WebSocket 服务器对象
* @param Request $request 请求对象
* @return void
*/
public function onOpen(Server $server, Request $request): void
{
try {
$fd = $request->fd;
$path = $request->server['path_info'] ?? $request->server['request_uri'] ?? '/';
// 从服务器获取 wsTable
$wsTable = $server->wsTable;
Log::info('WebSocket connection opened', [
'fd' => $fd,
'path' => $path,
'ip' => $request->server['remote_addr'] ?? 'unknown'
]);
// 从查询字符串获取 user_id 和 token
$userId = (int)($request->get['user_id'] ?? 0);
$token = $request->get['token'] ?? '';
// Extract user ID from query parameters if provided
$userId = $request->get['user_id'] ?? null;
$token = $request->get['token'] ?? null;
if ($userId && $token) {
// Store user connection mapping
$this->getWsTable()->set('uid:' . $userId, [
'value' => $fd,
'expiry' => time() + 3600, // 1 hour expiry
]);
$this->getWsTable()->set('fd:' . $fd, [
'value' => $userId,
'expiry' => time() + 3600
]);
// Update user online status
$this->userOnlineService->updateUserOnlineStatus($userId, $fd, true);
Log::info('User connected to WebSocket', [
'user_id' => $userId,
'fd' => $fd
]);
// Send welcome message to client
$server->push($fd, json_encode([
'type' => 'welcome',
'data' => [
'message' => 'WebSocket connection established',
'user_id' => $userId,
'timestamp' => time()
]
]));
} else {
Log::warning('WebSocket connection without authentication', [
'fd' => $fd
]);
// Send error message
$server->push($fd, json_encode([
// 用户认证
if (!$userId || !$token) {
$server->push($request->fd, json_encode([
'type' => 'error',
'data' => [
'message' => 'Authentication required. Please provide user_id and token.',
'message' => '认证失败：缺少 user_id 或 token',
'code' => 401
]
]));
$server->disconnect($request->fd);
return;
}
} catch (\Exception $e) {
Log::error('WebSocket onOpen error', [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
// 验证 JWT token
try {
$payload = JWTAuth::setToken($token)->getPayload();
// 验证 token 中的用户 ID 是否匹配
$tokenUserId = $payload['sub'] ?? null;
if ($tokenUserId != $userId) {
Log::warning('WebSocket 认证失败：用户 ID 不匹配', [
'fd' => $request->fd,
'token_user_id' => $tokenUserId,
'query_user_id' => $userId
]);
$server->push($request->fd, json_encode([
'type' => 'error',
'data' => [
'message' => '认证失败：用户 ID 不匹配',
'code' => 401
]
]));
$server->disconnect($request->fd);
return;
}
// 验证 token 是否过期
if (isset($payload['exp']) && $payload['exp'] < time()) {
Log::warning('WebSocket 认证失败：token 已过期', [
'fd' => $request->fd,
'user_id' => $userId,
'exp' => $payload['exp'],
'current_time' => time()
]);
$server->push($request->fd, json_encode([
'type' => 'error',
'data' => [
'message' => '认证失败：token 已过期',
'code' => 401
]
]));
$server->disconnect($request->fd);
return;
}
Log::info('WebSocket 认证成功', [
'fd' => $request->fd,
'user_id' => $userId
]);
} catch (\Exception $e) {
Log::warning('WebSocket 认证失败：无效的 token', [
'fd' => $request->fd,
'user_id' => $userId,
'error' => $e->getMessage()
]);
$server->push($request->fd, json_encode([
'type' => 'error',
'data' => [
'message' => '认证失败：无效的 token',
'code' => 401
]
]));
$server->disconnect($request->fd);
return;
}
// 存储连接映射：uid:{userId} -> fd
$wsTable->set('uid:' . $userId, [
'value' => $request->fd,
'expiry' => time() + 3600 // 1 小时过期
]);
// 存储反向映射：fd:{fd} -> userId
$wsTable->set('fd:' . $request->fd, [
'value' => $userId,
'expiry' => time() + 3600
]);
// 发送欢迎消息
$server->push($request->fd, json_encode([
'type' => 'connected',
'data' => [
'message' => '欢迎连接到 LaravelS WebSocket',
'user_id' => $userId,
'fd' => $request->fd,
'timestamp' => time()
]
]));
Log::info('WebSocket 连接已打开', [
'fd' => $request->fd,
'user_id' => $userId,
'ip' => $request->server['remote_addr']
]);
} catch (\Exception $e) {
Log::error('WebSocket onOpen 错误', [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString(),
'fd' => $request->fd
]);
$server->push($request->fd, json_encode([
'type' => 'error',
'data' => [
'message' => '连接错误：' . $e->getMessage(),
'code' => 500
]
]));
$server->disconnect($request->fd);
}
}
/**
* Handle WebSocket message event
* 处理接收消息事件
*
* @param Server $server
* @param Frame $frame
* @param Server $server WebSocket 服务器对象
* @param Frame $frame WebSocket 帧对象
* @return void
*/
public function onMessage(Server $server, Frame $frame): void
{
try {
$fd = $frame->fd;
$data = $frame->data;
// 从服务器获取 wsTable
$wsTable = $server->wsTable;
Log::info('WebSocket message received', [
'fd' => $fd,
'data' => $data,
'opcode' => $frame->opcode
]);
// 从 fd 映射获取 user_id
$fdInfo = $wsTable->get('fd:' . $frame->fd);
if ($fdInfo === false) {
$server->disconnect($frame->fd);
return;
}
// Parse incoming message
$message = json_decode($data, true);
$userId = (int)$fdInfo['value'];
if (!$message) {
$server->push($fd, json_encode([
// 解析消息
$message = json_decode($frame->data, true);
if (!$message || !isset($message['type'])) {
$server->push($frame->fd, json_encode([
'type' => 'error',
'data' => [
'message' => 'Invalid JSON format',
'message' => '无效的消息格式',
'code' => 400
]
]));
return;
}
// Handle different message types
$this->handleMessage($server, $fd, $message);
} catch (\Exception $e) {
Log::error('WebSocket onMessage error', [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
$type = $message['type'];
$data = $message['data'] ?? [];
Log::info('收到 WebSocket 消息', [
'fd' => $frame->fd,
'user_id' => $userId,
'type' => $type
]);
}
}
/**
* Handle WebSocket message based on type
*
* @param Server $server
* @param int $fd
* @param array $message
* @return void
*/
protected function handleMessage(Server $server, int $fd, array $message): void
{
$type = $message['type'] ?? 'unknown';
$data = $message['data'] ?? [];
// 处理不同类型的消息
switch ($type) {
case 'ping':
// 响应 ping
$server->push($frame->fd, json_encode([
'type' => 'pong',
'data' => $data
]));
break;
switch ($type) {
case 'auth':
// Handle authentication confirmation
$this->handleAuth($server, $fd, $data);
break;
case 'heartbeat':
// 心跳确认
$server->push($frame->fd, json_encode([
'type' => 'heartbeat_ack',
'data' => array_merge($data, [
'timestamp' => time()
])
]));
break;
case 'ping':
// Respond to ping with pong
$server->push($fd, json_encode([
'type' => 'pong',
'data' => [
'timestamp' => time()
]
]));
break;
case 'chat':
// 私聊消息
$this->handleChatMessage($server, $wsTable, $frame, $userId, $data);
break;
case 'heartbeat':
// Handle heartbeat
$server->push($fd, json_encode([
'type' => 'heartbeat_ack',
'data' => [
'timestamp' => time()
]
]));
break;
case 'broadcast':
// 广播消息给所有用户
$this->handleBroadcast($server, $wsTable, $userId, $data);
break;
case 'chat':
// Handle chat message
$this->handleChatMessage($server, $fd, $data);
break;
case 'subscribe':
// 订阅频道
$this->handleSubscribe($server, $wsTable, $frame, $userId, $data);
break;
case 'broadcast':
// Handle broadcast message (admin only)
$this->handleBroadcast($server, $fd, $data);
break;
case 'unsubscribe':
// 取消订阅频道
$this->handleUnsubscribe($server, $wsTable, $frame, $userId, $data);
break;
case 'subscribe':
// Handle channel subscription
$this->handleSubscribe($server, $fd, $data);
break;
case 'unsubscribe':
// Handle channel unsubscription
$this->handleUnsubscribe($server, $fd, $data);
break;
default:
$server->push($fd, json_encode([
'type' => 'error',
'data' => [
'message' => 'Unknown message type: ' . $type,
'code' => 400
]
]));
break;
}
}
/**
* Handle authentication confirmation
*
* @param Server $server
* @param int $fd
* @param array $data
* @return void
*/
protected function handleAuth(Server $server, int $fd, array $data): void
{
$userId = $data['user_id'] ?? null;
$token = $data['token'] ?? null;
// Get the user ID from wsTable (set during connection)
$storedUserId = $this->getWsTable()->get('fd:' . $fd)['value'] ?? null;
if ($storedUserId && $storedUserId == $userId) {
// Authentication confirmed, send success response
$server->push($fd, json_encode([
'type' => 'connected',
'data' => [
'user_id' => $storedUserId,
'message' => 'Authentication confirmed',
'timestamp' => time()
]
]));
Log::info('WebSocket authentication confirmed', [
'fd' => $fd,
'user_id' => $userId
]);
} else {
// Authentication failed
$server->push($fd, json_encode([
'type' => 'error',
'data' => [
'message' => 'Authentication failed. User ID mismatch.',
'code' => 401
]
]));
Log::warning('WebSocket authentication failed', [
'fd' => $fd,
'stored_user_id' => $storedUserId,
'provided_user_id' => $userId
]);
}
}
/**
* Handle chat message
*
* @param Server $server
* @param int $fd
* @param array $data
* @return void
*/
protected function handleChatMessage(Server $server, int $fd, array $data): void
{
$toUserId = $data['to_user_id'] ?? null;
$content = $data['content'] ?? '';
if (!$toUserId || !$content) {
$server->push($fd, json_encode([
'type' => 'error',
'data' => [
'message' => 'Missing required fields: to_user_id and content',
'code' => 400
]
]));
return;
}
// Get target user's connection
$targetFd = $this->getWsTable()->get('uid:' . $toUserId);
if ($targetFd && $targetFd['value']) {
$server->push((int)$targetFd['value'], json_encode([
'type' => 'chat',
'data' => [
'from_user_id' => $this->getWsTable()->get('fd:' . $fd)['value'] ?? null,
'content' => $content,
'timestamp' => time()
]
]));
// Send delivery receipt to sender
$server->push($fd, json_encode([
'type' => 'message_delivered',
'data' => [
'to_user_id' => $toUserId,
'content' => $content,
'timestamp' => time()
]
]));
} else {
$server->push($fd, json_encode([
'type' => 'error',
'data' => [
'message' => 'Target user is not online',
'code' => 404
]
]));
}
}
/**
* Handle broadcast message
*
* @param Server $server
* @param int $fd
* @param array $data
* @return void
*/
protected function handleBroadcast(Server $server, int $fd, array $data): void
{
$message = $data['message'] ?? '';
$userId = $this->getWsTable()->get('fd:' . $fd)['value'] ?? null;
// TODO: Check if user has admin permission to broadcast
// For now, allow any authenticated user
if (!$message) {
$server->push($fd, json_encode([
'type' => 'error',
'data' => [
'message' => 'Message content is required',
'code' => 400
]
]));
return;
}
// Broadcast to all connected clients except sender
$broadcastData = json_encode([
'type' => 'broadcast',
'data' => [
'from_user_id' => $userId,
'message' => $message,
'timestamp' => time()
]
]);
foreach ($server->connections as $connectionFd) {
if ($server->isEstablished($connectionFd) && $connectionFd !== $fd) {
$server->push($connectionFd, $broadcastData);
default:
// 未知消息类型
$server->push($frame->fd, json_encode([
'type' => 'error',
'data' => [
'message' => '未知的消息类型：' . $type,
'code' => 400
]
]));
break;
}
} catch (\Exception $e) {
Log::error('WebSocket onMessage 错误', [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString(),
'fd' => $frame->fd
]);
}
// Send confirmation to sender
$server->push($fd, json_encode([
'type' => 'broadcast_sent',
'data' => [
'message' => $message,
'timestamp' => time()
]
]));
}
/**
* Handle channel subscription
* 处理连接关闭事件
*
* @param Server $server
* @param int $fd
* @param array $data
* @return void
*/
protected function handleSubscribe(Server $server, int $fd, array $data): void
{
$channel = $data['channel'] ?? '';
if (!$channel) {
$server->push($fd, json_encode([
'type' => 'error',
'data' => [
'message' => 'Channel name is required',
'code' => 400
]
]));
return;
}
// Store subscription in wsTable
$this->getWsTable()->set('channel:' . $channel . ':fd:' . $fd, [
'value' => 1,
'expiry' => time() + 7200 // 2 hours
]);
$server->push($fd, json_encode([
'type' => 'subscribed',
'data' => [
'channel' => $channel,
'timestamp' => time()
]
]));
Log::info('User subscribed to channel', [
'fd' => $fd,
'channel' => $channel
]);
}
/**
* Handle channel unsubscription
*
* @param Server $server
* @param int $fd
* @param array $data
* @return void
*/
protected function handleUnsubscribe(Server $server, int $fd, array $data): void
{
$channel = $data['channel'] ?? '';
if (!$channel) {
$server->push($fd, json_encode([
'type' => 'error',
'data' => [
'message' => 'Channel name is required',
'code' => 400
]
]));
return;
}
// Remove subscription from wsTable
$this->getWsTable()->del('channel:' . $channel . ':fd:' . $fd);
$server->push($fd, json_encode([
'type' => 'unsubscribed',
'data' => [
'channel' => $channel,
'timestamp' => time()
]
]));
Log::info('User unsubscribed from channel', [
'fd' => $fd,
'channel' => $channel
]);
}
/**
* Handle WebSocket connection close event
*
* @param Server $server
* @param $fd
* @param $reactorId
* @param Server $server WebSocket 服务器对象
* @param int $fd 文件描述符
* @param int $reactorId 反应器 ID
* @return void
*/
public function onClose(Server $server, $fd, $reactorId): void
{
try {
Log::info('WebSocket connection closed', [
'fd' => $fd,
'reactor_id' => $reactorId
]);
// 从服务器获取 wsTable
$wsTable = $server->wsTable;
// Get user ID from wsTable
$userId = $this->getWsTable()->get('fd:' . $fd)['value'] ?? null;
// 从 fd 映射获取 user_id
$fdInfo = $wsTable->get('fd:' . $fd);
if ($userId) {
// Remove user connection mapping
$this->getWsTable()->del('uid:' . $userId);
$this->getWsTable()->del('fd:' . $fd);
if ($fdInfo !== false) {
$userId = (int)$fdInfo['value'];
// Update user online status
$this->userOnlineService->updateUserOnlineStatus($userId, $fd, false);
// 删除 uid 映射
$wsTable->del('uid:' . $userId);
Log::info('User disconnected from WebSocket', [
// 删除该用户的所有频道订阅
$this->removeUserFromAllChannels($wsTable, $userId, $fd);
Log::info('WebSocket 连接已关闭', [
'fd' => $fd,
'user_id' => $userId,
'fd' => $fd
'reactor_id' => $reactorId
]);
}
// Clean up channel subscriptions
// Note: In production, you might want to iterate through all channel keys
// and remove the ones associated with this fd
// 删除 fd 映射
$wsTable->del('fd:' . $fd);
} catch (\Exception $e) {
Log::error('WebSocket onClose error', [
Log::error('WebSocket onClose 错误', [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
'trace' => $e->getTraceAsString(),
'fd' => $fd
]);
}
}
/**
* 处理私聊消息
*
* @param Server $server WebSocket 服务器对象
* @param \Swoole\Table $wsTable WebSocket 表
* @param Frame $frame WebSocket 帧对象
* @param int $fromUserId 发送者用户 ID
* @param array $data 消息数据
* @return void
*/
protected function handleChatMessage(Server $server, \Swoole\Table $wsTable, Frame $frame, int $fromUserId, array $data): void
{
$toUserId = $data['to_user_id'] ?? 0;
if (!$toUserId) {
$server->push($frame->fd, json_encode([
'type' => 'error',
'data' => [
'message' => '缺少 to_user_id',
'code' => 400
]
]));
return;
}
// 获取接收者的 fd
$recipientInfo = $wsTable->get('uid:' . $toUserId);
if ($recipientInfo === false) {
$server->push($frame->fd, json_encode([
'type' => 'error',
'data' => [
'message' => '用户不在线',
'to_user_id' => $toUserId,
'code' => 404
]
]));
return;
}
$toFd = (int)$recipientInfo['value'];
// 发送消息给接收者
$server->push($toFd, json_encode([
'type' => 'chat',
'data' => array_merge($data, [
'from_user_id' => $fromUserId,
'timestamp' => time()
])
]));
}
/**
* 处理广播消息
*
* @param Server $server WebSocket 服务器对象
* @param \Swoole\Table $wsTable WebSocket 表
* @param int $userId 用户 ID
* @param array $data 消息数据
* @return void
*/
protected function handleBroadcast(Server $server, \Swoole\Table $wsTable, int $userId, array $data): void
{
$excludeUserId = $data['exclude_user_id'] ?? null;
$message = json_encode([
'type' => 'broadcast',
'data' => array_merge($data, [
'from_user_id' => $userId,
'timestamp' => time()
])
]);
// 发送消息给所有连接的用户
foreach ($wsTable as $key => $row) {
if (strpos($key, 'uid:') === 0) {
$targetUserId = (int)substr($key, 4); // 移除 'uid:' 前缀
$fd = (int)$row['value'];
// 跳过排除的用户
if ($excludeUserId && $targetUserId == $excludeUserId) {
continue;
}
if ($server->isEstablished($fd)) {
$server->push($fd, $message);
}
}
}
}
/**
* 处理频道订阅
*
* @param Server $server WebSocket 服务器对象
* @param \Swoole\Table $wsTable WebSocket 表
* @param Frame $frame WebSocket 帧对象
* @param int $userId 用户 ID
* @param array $data 消息数据
* @return void
*/
protected function handleSubscribe(Server $server, \Swoole\Table $wsTable, Frame $frame, int $userId, array $data): void
{
$channel = $data['channel'] ?? '';
if (!$channel) {
$server->push($frame->fd, json_encode([
'type' => 'error',
'data' => [
'message' => '缺少频道名称',
'code' => 400
]
]));
return;
}
// 存储频道订阅
$channelKey = 'channel:' . $channel . ':fd:' . $frame->fd;
$wsTable->set($channelKey, [
'value' => $userId,
'expiry' => time() + 3600
]);
$server->push($frame->fd, json_encode([
'type' => 'subscribed',
'data' => [
'channel' => $channel,
'message' => '成功订阅频道：' . $channel,
'timestamp' => time()
]
]));
Log::info('用户订阅频道', [
'user_id' => $userId,
'channel' => $channel,
'fd' => $frame->fd
]);
}
/**
* 处理频道取消订阅
*
* @param Server $server WebSocket 服务器对象
* @param \Swoole\Table $wsTable WebSocket 表
* @param Frame $frame WebSocket 帧对象
* @param int $userId 用户 ID
* @param array $data 消息数据
* @return void
*/
protected function handleUnsubscribe(Server $server, \Swoole\Table $wsTable, Frame $frame, int $userId, array $data): void
{
$channel = $data['channel'] ?? '';
if (!$channel) {
$server->push($frame->fd, json_encode([
'type' => 'error',
'data' => [
'message' => '缺少频道名称',
'code' => 400
]
]));
return;
}
// 删除频道订阅
$channelKey = 'channel:' . $channel . ':fd:' . $frame->fd;
$wsTable->del($channelKey);
$server->push($frame->fd, json_encode([
'type' => 'unsubscribed',
'data' => [
'channel' => $channel,
'message' => '成功取消订阅频道：' . $channel,
'timestamp' => time()
]
]));
Log::info('用户取消订阅频道', [
'user_id' => $userId,
'channel' => $channel,
'fd' => $frame->fd
]);
}
/**
* 从所有频道中移除用户
*
* @param \Swoole\Table $wsTable WebSocket 表
* @param int $userId 用户 ID
* @param int $fd 文件描述符
* @return void
*/
protected function removeUserFromAllChannels(\Swoole\Table $wsTable, int $userId, int $fd): void
{
foreach ($wsTable as $key => $row) {
if (strpos($key, 'channel:') === 0 && strpos($key, ':fd:' . $fd) !== false) {
$wsTable->del($key);
}
}
}
}

559
app/Services/WebSocket/WebSocketService.php
View File

@@ -6,91 +6,90 @@ use Illuminate\Support\Facades\Log;
use Swoole\WebSocket\Server;
/**
* WebSocket Service
* WebSocket 服务
*
* Provides helper functions for WebSocket operations
* 提供 WebSocket 操作的便捷方法
*/
class WebSocketService
{
/**
* Get Swoole WebSocket Server instance
* 获取 Swoole Server 实例
*
* @return Server|null
* @return Server
*/
public function getServer(): ?Server
protected function getServer(): Server
{
// Check if Laravel-S is running
if (!class_exists('Hhxsv5\LaravelS\Illuminate\Laravel') || !defined('IN_LARAVELS')) {
return null;
}
try {
// Try to get the Swoole server from the Laravel-S container
$laravelS = \Hhxsv5\LaravelS\Illuminate\Laravel::getInstance();
if ($laravelS && $laravelS->getSwooleServer()) {
return $laravelS->getSwooleServer();
}
} catch (\Exception $e) {
Log::warning('Failed to get Swoole server instance', [
'error' => $e->getMessage()
]);
}
return null;
/** @var Server $server */
$server = app('swoole');
return $server;
}
/**
* Send message to a specific user
* 获取 WebSocket 表
*
* @param int $userId
* @param array $data
* @return \Swoole\Table
*/
protected function getWsTable(): \Swoole\Table
{
return app('swoole')->wsTable;
}
/**
* 发送消息给指定用户
*
* @param int $userId 用户 ID
* @param array $data 消息数据
* @return bool
*/
public function sendToUser(int $userId, array $data): bool
{
$server = $this->getServer();
try {
$wsTable = $this->getWsTable();
$server = $this->getServer();
if (!$server) {
Log::warning('WebSocket server not available', ['user_id' => $userId]);
// 获取用户的 fd
$fdInfo = $wsTable->get('uid:' . $userId);
if ($fdInfo === false) {
return false;
}
$fd = (int)$fdInfo['value'];
// 检查连接是否仍然建立
if (!$server->isEstablished($fd)) {
// 删除过期连接
$wsTable->del('uid:' . $userId);
$wsTable->del('fd:' . $fd);
return false;
}
// 发送消息
$result = $server->push($fd, json_encode($data));
Log::info('消息已发送给用户', [
'user_id' => $userId,
'fd' => $fd,
'success' => $result
]);
return $result;
} catch (\Exception $e) {
Log::error('发送消息给用户失败', [
'user_id' => $userId,
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
]);
return false;
}
$wsTable = app('swoole')->wsTable;
$fdInfo = $wsTable->get('uid:' . $userId);
if (!$fdInfo || !$fdInfo['value']) {
Log::info('User not connected to WebSocket', ['user_id' => $userId]);
return false;
}
$fd = (int)$fdInfo['value'];
if (!$server->isEstablished($fd)) {
Log::info('WebSocket connection not established', ['user_id' => $userId, 'fd' => $fd]);
// Clean up stale connection
$wsTable->del('uid:' . $userId);
$wsTable->del('fd:' . $fd);
return false;
}
$server->push($fd, json_encode($data));
Log::info('Message sent to user via WebSocket', [
'user_id' => $userId,
'fd' => $fd,
'data' => $data
]);
return true;
}
/**
* Send message to multiple users
* 发送消息给多个用户
*
* @param array $userIds
* @param array $data
* @return array Array of user IDs who received the message
* @param array $userIds 用户 ID 数组
* @param array $data 消息数据
* @return array 成功发送的用户 ID 数组
*/
public function sendToUsers(array $userIds, array $data): array
{
@@ -106,247 +105,263 @@ class WebSocketService
}
/**
* Broadcast message to all connected clients
* 广播消息给所有用户
*
* @param array $data
* @param int|null $excludeUserId User ID to exclude from broadcast
* @return int Number of clients the message was sent to
* @param array $data 消息数据
* @param int|null $excludeUserId 要排除的用户 ID
* @return int 成功发送的用户数量
*/
public function broadcast(array $data, ?int $excludeUserId = null): int
{
$server = $this->getServer();
try {
$wsTable = $this->getWsTable();
$server = $this->getServer();
if (!$server) {
Log::warning('WebSocket server not available for broadcast');
return 0;
}
$message = json_encode($data);
$count = 0;
$wsTable = app('swoole')->wsTable;
$message = json_encode($data);
$count = 0;
foreach ($server->connections as $fd) {
if (!$server->isEstablished($fd)) {
continue;
}
// Check if we should exclude this user
if ($excludeUserId) {
$fdInfo = $wsTable->get('fd:' . $fd);
if ($fdInfo && $fdInfo['value'] == $excludeUserId) {
foreach ($wsTable as $key => $row) {
// 只处理用户映射（uid:*）
if (strpos($key, 'uid:') !== 0) {
continue;
}
$userId = (int)substr($key, 4); // 移除 'uid:' 前缀
$fd = (int)$row['value'];
// 跳过排除的用户
if ($excludeUserId && $userId == $excludeUserId) {
continue;
}
// 检查连接是否已建立并发送
if ($server->isEstablished($fd)) {
if ($server->push($fd, $message)) {
$count++;
}
} else {
// 删除过期连接
$wsTable->del('uid:' . $userId);
$wsTable->del('fd:' . $fd);
}
}
$server->push($fd, $message);
$count++;
Log::info('广播消息已发送', [
'exclude_user_id' => $excludeUserId,
'sent_to' => $count
]);
return $count;
} catch (\Exception $e) {
Log::error('广播消息失败', [
'exclude_user_id' => $excludeUserId,
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
]);
return 0;
}
Log::info('Broadcast sent via WebSocket', [
'data' => $data,
'exclude_user_id' => $excludeUserId,
'count' => $count
]);
return $count;
}
/**
* Send message to all subscribers of a channel
* 发送消息到频道
*
* @param string $channel
* @param array $data
* @return int Number of subscribers who received the message
* @param string $channel 频道名称
* @param array $data 消息数据
* @return int 成功发送的订阅者数量
*/
public function sendToChannel(string $channel, array $data): int
{
$server = $this->getServer();
try {
$wsTable = $this->getWsTable();
$server = $this->getServer();
if (!$server) {
Log::warning('WebSocket server not available for channel broadcast', ['channel' => $channel]);
$message = json_encode($data);
$count = 0;
$channelPrefix = 'channel:' . $channel . ':fd:';
foreach ($wsTable as $key => $row) {
// 只处理该频道的订阅
if (strpos($key, $channelPrefix) !== 0) {
continue;
}
$fd = (int)substr($key, strlen($channelPrefix));
// 检查连接是否已建立并发送
if ($server->isEstablished($fd)) {
if ($server->push($fd, $message)) {
$count++;
}
} else {
// 删除过期订阅
$wsTable->del($key);
}
}
Log::info('消息已发送到频道', [
'channel' => $channel,
'sent_to' => $count
]);
return $count;
} catch (\Exception $e) {
Log::error('发送消息到频道失败', [
'channel' => $channel,
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
]);
return 0;
}
$wsTable = app('swoole')->wsTable;
$count = 0;
$message = json_encode($data);
// Iterate through all connections and check if they're subscribed to the channel
foreach ($server->connections as $fd) {
if (!$server->isEstablished($fd)) {
continue;
}
$subscription = $wsTable->get('channel:' . $channel . ':fd:' . $fd);
if ($subscription) {
$server->push($fd, $message);
$count++;
}
}
Log::info('Channel message sent via WebSocket', [
'channel' => $channel,
'data' => $data,
'count' => $count
]);
return $count;
}
/**
* Get online user count
* 获取在线用户数量
*
* @return int
*/
public function getOnlineUserCount(): int
{
$server = $this->getServer();
try {
$wsTable = $this->getWsTable();
$count = 0;
if (!$server || !isset($server->wsTable)) {
foreach ($wsTable as $key => $row) {
if (strpos($key, 'uid:') === 0) {
$count++;
}
}
return $count;
} catch (\Exception $e) {
Log::error('获取在线用户数量失败', [
'error' => $e->getMessage()
]);
return 0;
}
// Count established connections
$count = 0;
foreach ($server->connections as $fd) {
if ($server->isEstablished($fd)) {
$count++;
}
}
return $count;
}
/**
* Check if a user is online
* 检查用户是否在线
*
* @param int $userId
* @param int $userId 用户 ID
* @return bool
*/
public function isUserOnline(int $userId): bool
{
$server = $this->getServer();
try {
$wsTable = $this->getWsTable();
$fdInfo = $wsTable->get('uid:' . $userId);
if (!$server) {
return false;
}
if ($fdInfo === false) {
return false;
}
$wsTable = app('swoole')->wsTable;
$server = $this->getServer();
$fd = (int)$fdInfo['value'];
$fdInfo = $wsTable->get('uid:' . $userId);
if (!$fdInfo || !$fdInfo['value']) {
return false;
}
$fd = (int)$fdInfo['value'];
return $server->isEstablished($fd);
}
/**
* Disconnect a user from WebSocket
*
* @param int $userId
* @return bool
*/
public function disconnectUser(int $userId): bool
{
$server = $this->getServer();
if (!$server) {
return false;
}
$wsTable = app('swoole')->wsTable;
$fdInfo = $wsTable->get('uid:' . $userId);
if (!$fdInfo || !$fdInfo['value']) {
return false;
}
$fd = (int)$fdInfo['value'];
if ($server->isEstablished($fd)) {
$server->push($fd, json_encode([
'type' => 'disconnect',
'data' => [
'message' => 'You have been disconnected',
'timestamp' => time()
]
]));
// Close the connection
$server->disconnect($fd);
// Clean up
$wsTable->del('uid:' . $userId);
$wsTable->del('fd:' . $fd);
Log::info('User disconnected from WebSocket by server', [
return $server->isEstablished($fd);
} catch (\Exception $e) {
Log::error('检查用户在线状态失败', [
'user_id' => $userId,
'fd' => $fd
'error' => $e->getMessage()
]);
return true;
return false;
}
return false;
}
/**
* Get all online user IDs
* 获取在线用户 ID 列表
*
* @return array
*/
public function getOnlineUserIds(): array
{
$server = $this->getServer();
try {
$wsTable = $this->getWsTable();
$userIds = [];
if (!$server) {
foreach ($wsTable as $key => $row) {
if (strpos($key, 'uid:') === 0) {
$userId = (int)substr($key, 4); // 移除 'uid:' 前缀
$userIds[] = $userId;
}
}
return $userIds;
} catch (\Exception $e) {
Log::error('获取在线用户 ID 列表失败', [
'error' => $e->getMessage()
]);
return [];
}
$wsTable = app('swoole')->wsTable;
$userIds = [];
foreach ($server->connections as $fd) {
if (!$server->isEstablished($fd)) {
continue;
}
$fdInfo = $wsTable->get('fd:' . $fd);
if ($fdInfo && $fdInfo['value']) {
$userIds[] = (int)$fdInfo['value'];
}
}
return array_unique($userIds);
}
/**
* Send system notification to all online users
* 断开用户 WebSocket 连接
*
* @param string $title
* @param string $message
* @param string $type
* @param array $extraData
* @return int
* @param int $userId 用户 ID
* @return bool
*/
public function sendSystemNotification(string $title, string $message, string $type = 'info', array $extraData = []): int
public function disconnectUser(int $userId): bool
{
try {
$wsTable = $this->getWsTable();
$server = $this->getServer();
// 获取用户的 fd
$fdInfo = $wsTable->get('uid:' . $userId);
if ($fdInfo === false) {
return false;
}
$fd = (int)$fdInfo['value'];
// 断开连接
$server->disconnect($fd);
// 删除映射
$wsTable->del('uid:' . $userId);
$wsTable->del('fd:' . $fd);
Log::info('用户已断开连接', [
'user_id' => $userId,
'fd' => $fd
]);
return true;
} catch (\Exception $e) {
Log::error('断开用户连接失败', [
'user_id' => $userId,
'error' => $e->getMessage()
]);
return false;
}
}
/**
* 发送系统通知
*
* @param string $title 标题
* @param string $message 消息内容
* @param string $type 类型
* @param array $extraData 额外数据
* @return int 成功发送的用户数量
*/
public function sendSystemNotification(
string $title,
string $message,
string $type = 'info',
array $extraData = []
): int {
$data = [
'type' => 'notification',
'data' => [
'title' => $title,
'message' => $message,
'type' => $type, // info, success, warning, error
'timestamp' => time(),
...$extraData
'type' => $type,
'data' => $extraData,
'timestamp' => time()
]
];
@@ -354,47 +369,57 @@ class WebSocketService
}
/**
* Send notification to specific users
* 发送通知给指定用户
*
* @param array $userIds
* @param string $title
* @param string $message
* @param string $type
* @param array $extraData
* @return array
* @param array $userIds 用户 ID 数组
* @param string $title 标题
* @param string $message 消息内容
* @param string $type 类型
* @param array $extraData 额外数据
* @return int 成功发送的用户数量
*/
public function sendNotificationToUsers(array $userIds, string $title, string $message, string $type = 'info', array $extraData = []): array
{
public function sendNotificationToUsers(
array $userIds,
string $title,
string $message,
string $type = 'info',
array $extraData = []
): int {
$data = [
'type' => 'notification',
'data' => [
'title' => $title,
'message' => $message,
'type' => $type,
'timestamp' => time(),
...$extraData
'data' => $extraData,
'timestamp' => time()
]
];
return $this->sendToUsers($userIds, $data);
$sentTo = $this->sendToUsers($userIds, $data);
return count($sentTo);
}
/**
* Push data update to specific users
* 推送数据更新
*
* @param array $userIds
* @param string $resourceType
* @param string $action
* @param array $data
* @return array
* @param array $userIds 用户 ID 数组
* @param string $resourceType 资源类型
* @param string $action 操作
* @param array $data 数据
* @return array 成功推送的用户 ID 数组
*/
public function pushDataUpdate(array $userIds, string $resourceType, string $action, array $data): array
{
public function pushDataUpdate(
array $userIds,
string $resourceType,
string $action,
array $data
): array {
$message = [
'type' => 'data_update',
'data' => [
'resource_type' => $resourceType, // e.g., 'user', 'order', 'product'
'action' => $action, // create, update, delete
'resource_type' => $resourceType,
'action' => $action,
'data' => $data,
'timestamp' => time()
]
@@ -404,16 +429,20 @@ class WebSocketService
}
/**
* Push data update to a channel
* 推送数据更新到频道
*
* @param string $channel
* @param string $resourceType
* @param string $action
* @param array $data
* @return int
* @param string $channel 频道名称
* @param string $resourceType 资源类型
* @param string $action 操作
* @param array $data 数据
* @return int 成功推送的订阅者数量
*/
public function pushDataUpdateToChannel(string $channel, string $resourceType, string $action, array $data): int
{
public function pushDataUpdateToChannel(
string $channel,
string $resourceType,
string $action,
array $data
): int {
$message = [
'type' => 'data_update',
'data' => [