admin_consumer/app/Controller/WebSocketController.php

<?php

declare(strict_types=1);

namespace App\Controller;

use App\Amqp\Producer\MqProducer;
use App\JsonRpc\ChatServiceInterface;
use App\JsonRpc\UserServiceInterface;
use App\Service\Message\ReceiveHandleService;
use App\Service\RedisService;
use Hyperf\Amqp\Producer;
use Hyperf\Context\ApplicationContext as ContextApplicationContext;
use Hyperf\Contract\OnCloseInterface;
use Hyperf\Contract\OnMessageInterface;
use Hyperf\Contract\OnOpenInterface;
use Hyperf\Di\Annotation\Inject;
use Hyperf\Engine\WebSocket\Response;
use Phper666\JWTAuth\JWT;
use swoole\Server;

class WebSocketController implements OnMessageInterface, OnOpenInterface, OnCloseInterface
{

    #[Inject]
    protected JWT $jwt;
    /**
     * @var ChatServiceInterface
     */
    #[Inject]
    private $chatServiceClient;
    /**
     * @var UserServiceInterface
     */
    #[Inject]
    private $userServiceClient;

    /**
     * @Inject
     * @var ReceiveHandleService
     */
    protected $receiveHandle;
    protected $server;


    public function onMessage($server, $frame): void
    {
        //把数据推给前端
        $redisClient = new RedisService();
        $userId = $redisClient->findUser((string) $frame->fd);
        //存入队列
        $result = json_decode($frame->data, true);
        $result['user_id'] = $userId;

        // $show_id = $result['show_id'];

        var_dump($result, '-------------1----');
        $userInfo = $redisClient->getUserInfo((string) $userId);
        if ($userInfo) {
            $userInfoArr = json_decode($userInfo);
            $result['user_avatar'] = $userInfoArr['avatar'];
        } else {
            $userInfos = $this->userServiceClient->getUserInfo((int) $userId);
            $redisClient->setUserInfo($userId, $userInfos['data']);
            $result['user_avatar'] = $userInfos['data']['avatar'];
        }
        $myFriends = $redisClient->getUserFriends((string) $userId);
        $myFriendsArr = [];
        if ($myFriends) {
            $myFriendsArr = json_decode($myFriends);
        } else {
            $myFriends = $this->chatServiceClient->getFriendsList(['user_id' => $userId, 'status' => 2]);
            $redisClient->setUserFriends($userId, $myFriends['data']);
            $myFriendsArr = $myFriends['data'];
        }

        //推送给前台
        if (isset($result['type']) && in_array($result['type'], ['offer', 'answer', 'ice-candidate'])) {
            // 是视频通话信令，走特殊处理逻辑
            $this->handleVideoSignaling($server, $frame, $userId, $result);
            return;
        }
        //组装数据+头像
        if ($result['talk_type'] == 1) {
            //判断$result['receiver_id']是否是好友
            $myFriendsID = array_column($myFriendsArr, 'friend_id');
            if (!in_array($result['receiver_id'], $myFriendsID)) {
                $myFriends = $this->chatServiceClient->getFriendsList(['user_id' => $userId, 'status' => 2]);
                $redisClient->setUserFriends($userId, $myFriends['data']);
                $myFriendsArrdata = $myFriends['data'];
                $myFriendsArrID = array_column($myFriendsArrdata, 'friend_id');
                if (!in_array($result['receiver_id'], $myFriendsArrID)) {
                    $result['content'] = '您还不是好友，无法发送消息!';
                    $server->push((int) $frame->fd, json_encode($result));
                    return;
                }
            }
            //给自己推一条数据
            if ($server->isEstablished($frame->fd)) {
                $result['is_read'] = 1;
                $server->push((int) $frame->fd, json_encode($result));
                // 尝试连接
                try {
                    $saiddata = $result;
                    $saiddata['action'] = 'said';
                    $message = new MqProducer($saiddata);
                    $producer = ContextApplicationContext::getContainer()->get(Producer::class);
                    $re = $producer->produce($message);
                    var_dump('消息发送成功' . $frame->fd);
                } catch (\Exception $e) {
                    var_dump('消息发送失败: ' . $e->getMessage());
                    $this->retry($message);
                }
            } else {
                //给自己发一条未读消息
                try {
                    $saiddata = $result;
                    $saiddata['action'] = 'said';
                    $saiddata['is_read'] = 0;
                    $message = new MqProducer($saiddata);
                    $producer = ContextApplicationContext::getContainer()->get(Producer::class);
                    $re = $producer->produce($message);
                    var_dump('消息发送成功' . $frame->fd);
                } catch (\Exception $e) {
                    var_dump('消息发送失败: ' . $e->getMessage());
                    $this->retry($message);
                }
            }
            //给好友推送消息
            $fd = $redisClient->findFd((int) $result['receiver_id']);
            if ($server->isEstablished((int) $fd)) {
                $data = $result;

                $data['is_read'] = 0;
                $server->push((int) $fd, json_encode(value: $data));
                var_dump('消息给好友发送成功' . $fd);
                // 尝试连接
                try {
                    $chatdata = $result;
                    $chatdata['action'] = 'recieved';
                    $chatdata['receiver_id'] = $result['user_id'];
                    $chatdata['user_id'] = $result['receiver_id'];
                    $chatdata['is_read'] = 0;
                    $message = new MqProducer($chatdata);
                    $producer = ContextApplicationContext::getContainer()->get(Producer::class);
                    $re = $producer->produce($message);
                    var_dump('消息发送成功');
                } catch (\Exception $e) {
                    var_dump('消息发送失败: ' . $e->getMessage());
                    $this->retry($message);
                }
            } else {
                try {
                    $chatdata = $result;
                    $chatdata['action'] = 'recieved';
                    $chatdata['receiver_id'] = $result['user_id'];
                    $chatdata['user_id'] = $result['receiver_id'];
                    $chatdata['is_read'] = 0;
                    $message = new MqProducer($chatdata);
                    $producer = ContextApplicationContext::getContainer()->get(Producer::class);
                    $re = $producer->produce($message);
                    var_dump('消息发送成功');
                } catch (\Exception $e) {
                    var_dump('消息发送失败: ' . $e->getMessage());
                    $this->retry($message);
                }
            }
        } else if ($result['talk_type'] == 2) {
            //根据群找到 群用户，群发一遍消息
            $groupUserList = $this->chatServiceClient->getGroupMembers(['group_id' => $result['receiver_id']]);
            var_dump($groupUserList['data'], '----------------------------------');
            if ($groupUserList['data']) {
                $chatdata = $result;
                foreach ($groupUserList['data'] as $val) {
                    $fd = $redisClient->findFd((int) $val['user_id']);
                    if ($server->isEstablished((int) $fd)) {
                        var_dump($fd, $val['user_id']);
                        $server->push((int) $fd, json_encode($result));
                        try {
                            //分发说，只记录自己对自己
                            if ($result['user_id'] == $val['user_id']) {
                                $chatdata['receiver_id'] = $result['receiver_id'];
                                $chatdata['user_id'] = $result['user_id'];
                                $chatdata['group_receiver_id'] = $val['user_id'];
                                $chatdata['is_read'] = 1;
                                $chatdata['action'] = 'said';
                                $message = new MqProducer($chatdata);
                                $producer = ContextApplicationContext::getContainer()->get(Producer::class);
                                $re = $producer->produce($message);
                            }
                            //分发接收，不记录自己
                            if ($chatdata['user_id'] != $val['user_id']) {
                                $chatdata['receiver_id'] = $result['receiver_id'];
                                $chatdata['user_id'] = $val['user_id'];
                                $chatdata['group_receiver_id'] = $result['user_id'];
                                $chatdata['is_read'] = 0;
                                $chatdata['action'] = 'recieved';
                                $message = new MqProducer($chatdata);
                                $producer = ContextApplicationContext::getContainer()->get(Producer::class);
                                $re = $producer->produce($message);
                                var_dump('消息发送成功');
                            }
                        } catch (\Exception $e) {
                            var_dump('消息发送失败: ' . $e->getMessage());
                        }
                    } else {
                        try {
                            //分发说，只记录自己对自己
                            if ($result['user_id'] == $val['user_id']) {
                                $chatdata['receiver_id'] = $result['receiver_id'];
                                $chatdata['user_id'] = $result['user_id'];
                                $chatdata['group_receiver_id'] = $val['user_id'];
                                $chatdata['is_read'] = 0;
                                $chatdata['action'] = 'said';
                                $message = new MqProducer($chatdata);
                                $producer = ContextApplicationContext::getContainer()->get(Producer::class);
                                $re = $producer->produce($message);
                            }
                            //分发接收，不记录自己
                            if ($chatdata['user_id'] != $val['user_id']) {
                                $chatdata['receiver_id'] = $result['receiver_id'];
                                $chatdata['user_id'] = $val['user_id'];
                                $chatdata['group_receiver_id'] = $result['user_id'];
                                $chatdata['is_read'] = 0;
                                $chatdata['action'] = 'recieved';
                                $message = new MqProducer($chatdata);
                                $producer = ContextApplicationContext::getContainer()->get(Producer::class);
                                $re = $producer->produce($message);
                                var_dump('消息发送成功');
                            }
                        } catch (\Exception $e) {
                            var_dump('消息发送失败: ' . $e->getMessage());
                        }
                    }
                }
            }
        }
    }
    protected function handleVideoSignaling($server, $frame, $userId, $message)
    {
        $redisClient = new RedisService();
        $receiverId = $message['receiver_id'] ?? null;
        if (!$receiverId) {
            $server->push($frame->fd, json_encode(['error' => 'Missing receiver_id']));
            return;
        }
        // 获取接收方 fd
        $fd = $redisClient->findFd((int)$receiverId);

        // 判断接收者是否在线
        if ($server->isEstablished((int)$fd)) {
            // 转发消息给接收方
            $server->push((int)$fd, json_encode($message));
        } else {
            // 可选：缓存或通知对方不在线
            $server->push($frame->fd, json_encode([
                'type' => 'offline',
                'receiver_id' => $receiverId,
                'content' => '对方不在线'
            ]));
        }
    }
    public function retry($message): void
    {
        $maxRetries = 10; // 最大重试次数
        $retryCount = 0;
        while ($retryCount < $maxRetries) {
            try {
                $producer = ContextApplicationContext::getContainer()->get(Producer::class);
                $re = $producer->produce($message);
                var_dump('重试消息发送成功');
                break;
            } catch (\Exception $e) {
                $retryCount++;
                if ($retryCount >= $maxRetries) {
                    var_dump('达到最大重试次数，消息发送失败: ' . $e->getMessage());
                } else {
                    var_dump('第 ' . $retryCount . ' 次重试失败: ' . $e->getMessage());
                }
            }
        }
    }
    public function onClose($server, int $fd, int $reactorId): void
    {
        var_dump('closed::::::::::::::::::', $fd, "======", $reactorId, "+++++++++++");
        $redisClient = new RedisService();
        $userId = $redisClient->findUser((string) $fd);
        $redisClient->unbind((string) $fd, (int) $userId);
    }
    public function onOpen($server, $request): void
    {
        $token = $request->get['token'];
        $userInfo = $this->jwt->getClaimsByToken($token);
        $response = (new Response($server))->init($request);
        $fd = $response->getFd();
        $server->bind($fd, $userInfo['uid']);
        $redisClient = new RedisService();
        $redisClient->bind((string) $fd, $userInfo['uid']);
        $server->push($request->fd, json_encode([
            "event" => "connect",
            "content" => [
                "ping_interval" => 20,
                "ping_timeout" => 20 * 3,
                "content" => "连接成功",
                "fd" => $fd,
                "user_id" => $userInfo['uid'],
            ],
        ]));
    }
}