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

初始化项目

Browse Source
This commit is contained in:
molong
2026-02-08 22:38:13 +08:00
commit 334d2c6312
201 changed files with 32724 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

451
app/Services/WebSocket/WebSocketHandler.php Normal file
View File

@@ -0,0 +1,451 @@
<?php
namespace App\Services\WebSocket;
use Hhxsv5\LaravelS\Swoole\WebSocketHandlerInterface;
use Swoole\Http\Request;
use Swoole\WebSocket\Frame;
use Swoole\WebSocket\Server;
use Illuminate\Support\Facades\Log;
use App\Services\Auth\UserOnlineService;
/**
* WebSocket Handler
*
* Handles WebSocket connections, messages, and disconnections
*/
class WebSocketHandler implements WebSocketHandlerInterface
{
/**
* @var UserOnlineService
*/
protected $userOnlineService;
/**
* WebSocketHandler constructor
*/
public function __construct()
{
$this->userOnlineService = app(UserOnlineService::class);
}
/**
* Handle WebSocket connection open event
*
* @param Server $server
* @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'] ?? '/';
Log::info('WebSocket connection opened', [
'fd' => $fd,
'path' => $path,
'ip' => $request->server['remote_addr'] ?? 'unknown'
]);
// 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
$server->wsTable->set('uid:' . $userId, [
'value' => $fd,
'expiry' => time() + 3600, // 1 hour expiry
]);
$server->wsTable->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([
'type' => 'error',
'data' => [
'message' => 'Authentication required. Please provide user_id and token.',
'code' => 401
]
]));
}
} catch (\Exception $e) {
Log::error('WebSocket onOpen error', [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
]);
}
}
/**
* Handle WebSocket message event
*
* @param Server $server
* @param Frame $frame
* @return void
*/
public function onMessage(Server $server, Frame $frame): void
{
try {
$fd = $frame->fd;
$data = $frame->data;
Log::info('WebSocket message received', [
'fd' => $fd,
'data' => $data,
'opcode' => $frame->opcode
]);
// Parse incoming message
$message = json_decode($data, true);
if (!$message) {
$server->push($fd, json_encode([
'type' => 'error',
'data' => [
'message' => 'Invalid JSON format',
'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()
]);
}
}
/**
* 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':
// Respond to ping with pong
$server->push($fd, json_encode([
'type' => 'pong',
'data' => [
'timestamp' => time()
]
]));
break;
case 'heartbeat':
// Handle heartbeat
$server->push($fd, json_encode([
'type' => 'heartbeat_ack',
'data' => [
'timestamp' => time()
]
]));
break;
case 'chat':
// Handle chat message
$this->handleChatMessage($server, $fd, $data);
break;
case 'broadcast':
// Handle broadcast message (admin only)
$this->handleBroadcast($server, $fd, $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 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 = $server->wsTable->get('uid:' . $toUserId);
if ($targetFd && $targetFd['value']) {
$server->push((int)$targetFd['value'], json_encode([
'type' => 'chat',
'data' => [
'from_user_id' => $server->wsTable->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 = $server->wsTable->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);
}
}
// 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
$server->wsTable->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
$server->wsTable->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
* @return void
*/
public function onClose(Server $server, $fd, $reactorId): void
{
try {
Log::info('WebSocket connection closed', [
'fd' => $fd,
'reactor_id' => $reactorId
]);
// Get user ID from wsTable
$userId = $server->wsTable->get('fd:' . $fd)['value'] ?? null;
if ($userId) {
// Remove user connection mapping
$server->wsTable->del('uid:' . $userId);
$server->wsTable->del('fd:' . $fd);
// Update user online status
$this->userOnlineService->updateUserOnlineStatus($userId, $fd, false);
Log::info('User disconnected from WebSocket', [
'user_id' => $userId,
'fd' => $fd
]);
}
// Clean up channel subscriptions
// Note: In production, you might want to iterate through all channel keys
// and remove the ones associated with this fd
} catch (\Exception $e) {
Log::error('WebSocket onClose error', [
'error' => $e->getMessage(),
'trace' => $e->getTraceAsString()
]);
}
}
}

402
app/Services/WebSocket/WebSocketService.php Normal file
View File

@@ -0,0 +1,402 @@
<?php
namespace App\Services\WebSocket;
use Illuminate\Support\Facades\Log;
use Swoole\WebSocket\Server;
/**
* WebSocket Service
*
* Provides helper functions for WebSocket operations
*/
class WebSocketService
{
/**
* Get Swoole WebSocket Server instance
*
* @return Server|null
*/
public function getServer(): ?Server
{
return app('swoole.server');
}
/**
* Send message to a specific user
*
* @param int $userId
* @param array $data
* @return bool
*/
public function sendToUser(int $userId, array $data): bool
{
$server = $this->getServer();
if (!$server || !isset($server->wsTable)) {
Log::warning('WebSocket server not available', ['user_id' => $userId]);
return false;
}
$fdInfo = $server->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
$server->wsTable->del('uid:' . $userId);
$server->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
*/
public function sendToUsers(array $userIds, array $data): array
{
$sentTo = [];
foreach ($userIds as $userId) {
if ($this->sendToUser($userId, $data)) {
$sentTo[] = $userId;
}
}
return $sentTo;
}
/**
* 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
*/
public function broadcast(array $data, ?int $excludeUserId = null): int
{
$server = $this->getServer();
if (!$server) {
Log::warning('WebSocket server not available for broadcast');
return 0;
}
$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 = $server->wsTable->get('fd:' . $fd);
if ($fdInfo && $fdInfo['value'] == $excludeUserId) {
continue;
}
}
$server->push($fd, $message);
$count++;
}
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
*/
public function sendToChannel(string $channel, array $data): int
{
$server = $this->getServer();
if (!$server || !isset($server->wsTable)) {
Log::warning('WebSocket server not available for channel broadcast', ['channel' => $channel]);
return 0;
}
$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 = $server->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();
if (!$server || !isset($server->wsTable)) {
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
* @return bool
*/
public function isUserOnline(int $userId): bool
{
$server = $this->getServer();
if (!$server || !isset($server->wsTable)) {
return false;
}
$fdInfo = $server->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 || !isset($server->wsTable)) {
return false;
}
$fdInfo = $server->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
$server->wsTable->del('uid:' . $userId);
$server->wsTable->del('fd:' . $fd);
Log::info('User disconnected from WebSocket by server', [
'user_id' => $userId,
'fd' => $fd
]);
return true;
}
return false;
}
/**
* Get all online user IDs
*
* @return array
*/
public function getOnlineUserIds(): array
{
$server = $this->getServer();
if (!$server || !isset($server->wsTable)) {
return [];
}
$userIds = [];
foreach ($server->connections as $fd) {
if (!$server->isEstablished($fd)) {
continue;
}
$fdInfo = $server->wsTable->get('fd:' . $fd);
if ($fdInfo && $fdInfo['value']) {
$userIds[] = (int)$fdInfo['value'];
}
}
return array_unique($userIds);
}
/**
* Send system notification to all online users
*
* @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
]
];
return $this->broadcast($data);
}
/**
* Send notification to specific users
*
* @param array $userIds
* @param string $title
* @param string $message
* @param string $type
* @param array $extraData
* @return array
*/
public function sendNotificationToUsers(array $userIds, string $title, string $message, string $type = 'info', array $extraData = []): array
{
$data = [
'type' => 'notification',
'data' => [
'title' => $title,
'message' => $message,
'type' => $type,
'timestamp' => time(),
...$extraData
]
];
return $this->sendToUsers($userIds, $data);
}
/**
* Push data update to specific users
*
* @param array $userIds
* @param string $resourceType
* @param string $action
* @param array $data
* @return 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
'data' => $data,
'timestamp' => time()
]
];
return $this->sendToUsers($userIds, $message);
}
/**
* Push data update to a channel
*
* @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
{
$message = [
'type' => 'data_update',
'data' => [
'resource_type' => $resourceType,
'action' => $action,
'data' => $data,
'timestamp' => time()
]
];
return $this->sendToChannel($channel, $message);
}
}