hyperf-chat/app/Amqp/Consumer/ChatMessageConsumer.php

<?php

declare(strict_types=1);

namespace App\Amqp\Consumer;

use App\Model\UsersFriend;
use Hyperf\Amqp\Annotation\Consumer;
use Hyperf\Amqp\Result;
use Hyperf\Amqp\Message\ConsumerMessage;
use Hyperf\Redis\Redis;
use Hyperf\Amqp\Message\Type;
use Hyperf\Amqp\Builder\QueueBuilder;
use PhpAmqpLib\Message\AMQPMessage;
use App\Model\User;
use App\Helper\PushMessageHelper;
use App\Model\Chat\ChatRecord;
use App\Model\Chat\ChatRecordsCode;
use App\Model\Chat\ChatRecordsFile;
use App\Model\Chat\ChatRecordsInvite;
use App\Service\SocketFDService;
use App\Service\SocketRoomService;

/**
 * @Consumer(name="聊天消息消费者",enable=true)
 */
class ChatMessageConsumer extends ConsumerMessage
{
    /**
     * 交换机名称
     *
     * @var string
     */
    public $exchange = 'im.message.fanout';

    /**
     * 交换机类型
     *
     * @var string
     */
    public $type = Type::FANOUT;

    /**
     * 路由key
     *
     * @var string
     */
    public $routingKey = 'consumer:im:message';

    /**
     * @var SocketFDService
     */
    private $socketFDService;

    /**
     * @var SocketRoomService
     */
    private $socketRoomService;

    const EVENTS = [
        'event_talk' => 'onConsumeTalk',
        'event_keyboard' => 'onConsumeKeyboard',
        'event_online_status' => 'onConsumeOnlineStatus',
        'event_revoke_talk' => 'onConsumeRevokeTalk',
    ];

    /**
     * ChatMessageConsumer constructor.
     * @param SocketFDService $socketFDService
     * @param SocketRoomService $socketRoomService
     */
    public function __construct(SocketFDService $socketFDService, SocketRoomService $socketRoomService)
    {
        $this->socketFDService = $socketFDService;
        $this->socketRoomService = $socketRoomService;
        $this->setQueue('queue:im-message:' . SERVER_RUN_ID);
    }

    /**
     * 重写创建队列生成类
     *
     * 注释：设置自动删除队列
     *
     * @return QueueBuilder
     */
    public function getQueueBuilder(): QueueBuilder
    {
        return parent::getQueueBuilder()->setAutoDelete(true);
    }

    /**
     * 消费队列消息
     *
     * @param $data
     * @param AMQPMessage $message
     * @return string
     */
    public function consumeMessage($data, AMQPMessage $message): string
    {
        if (isset($data['event'])) {
            return $this->{self::EVENTS[$data['event']]}($data, $message);
        }

        return Result::ACK;
    }

    /**
     * 对话聊天消息
     *
     * @param array $data 队列消息
     * @param AMQPMessage $message
     * @return string
     */
    public function onConsumeTalk(array $data, AMQPMessage $message)
    {
        $redis = container()->get(Redis::class);

        //[加锁]防止消息重复消费
        $lockName = sprintf('ws:message-lock:%s:%s', SERVER_RUN_ID, $data['uuid']);
        if (!$redis->rawCommand('SET', $lockName, 1, 'NX', 'EX', 60)) {
            return Result::ACK;
        }

        $source = $data['data']['source'];

        $fids = $this->socketFDService->findUserFds($data['data']['sender']);
        if ($source == 1) {// 私聊
            $fids = array_merge($fids, $this->socketFDService->findUserFds($data['data']['receive']));
        } else if ($source == 2) {//群聊
            $userIds = $this->socketRoomService->getRoomMembers(strval($data['data']['receive']));
            foreach ($userIds as $uid) {
                $fids = array_merge($fids, $this->socketFDService->findUserFds(intval($uid)));
            }
        }

        // 去重
        $fids = array_unique($fids);
        if (empty($fids)) {
            return Result::ACK;
        }

        /**
         * @var ChatRecord
         */
        $result = ChatRecord::leftJoin('users', 'users.id', '=', 'chat_records.user_id')
            ->where('chat_records.id', $data['data']['record_id'])
            ->first([
                'chat_records.id',
                'chat_records.source',
                'chat_records.msg_type',
                'chat_records.user_id',
                'chat_records.receive_id',
                'chat_records.content',
                'chat_records.is_revoke',
                'chat_records.created_at',

                'users.nickname',
                'users.avatar as avatar',
            ]);

        if (!$result) {
            return Result::ACK;
        }

        $file = [];
        $code_block = [];
        $forward = [];
        $invite = [];
        switch ($result->msg_type) {
            case 2://文件消息
                $file = ChatRecordsFile::where('record_id', $result->id)->first(['id', 'record_id', 'user_id', 'file_source', 'file_type', 'save_type', 'original_name', 'file_suffix', 'file_size', 'save_dir']);
                $file = $file ? $file->toArray() : [];
                if ($file) {
                    $file['file_url'] = get_media_url($file['save_dir']);
                }
                break;
            case 3://入群消息/退群消息
                $notifyInfo = ChatRecordsInvite::where('record_id', $result->id)->first([
                    'record_id', 'type', 'operate_user_id', 'user_ids'
                ]);

                $userInfo = User::where('id', $notifyInfo->operate_user_id)->first(['nickname', 'id']);
                $membersIds = explode(',', $notifyInfo->user_ids);

                $invite = [
                    'type' => $notifyInfo->type,
                    'operate_user' => ['id' => $userInfo->id, 'nickname' => $userInfo->nickname],
                    'users' => User::select('id', 'nickname')->whereIn('id', $membersIds)->get()->toArray()
                ];

                unset($notifyInfo, $userInfo, $membersIds);
                break;
            case 4://会话记录消息

                break;
            case 5://代码块消息
                $code_block = ChatRecordsCode::where('record_id', $result->id)->first(['record_id', 'code_lang', 'code']);
                $code_block = $code_block ? $code_block->toArray() : [];
                break;
        }

        $msg = [
            'send_user' => $data['data']['sender'],
            'receive_user' => $data['data']['receive'],
            'source_type' => $data['data']['source'],
            'data' => PushMessageHelper::formatTalkMsg([
                'id' => $result->id,
                'msg_type' => $result->msg_type,
                'source' => $result->source,
                'avatar' => $result->avatar,
                'nickname' => $result->nickname,
                "user_id" => $result->user_id,
                "receive_id" => $result->receive_id,
                "created_at" => $result->created_at,
                "content" => $result->content,
                "file" => $file,
                "code_block" => $code_block,
                'forward' => $forward,
                'invite' => $invite
            ])
        ];

        $server = server();
        foreach ($fids as $fd) {
            $fd = intval($fd);
            if ($server->exist($fd)) {
                $server->push($fd, json_encode(['event_talk', $msg]));
            }
        }

        return Result::ACK;
    }

    /**
     * 键盘输入事件消息
     *
     * @param array $data 队列消息
     * @param AMQPMessage $message
     * @return string
     */
    public function onConsumeKeyboard(array $data, AMQPMessage $message)
    {
        $fds = $this->socketFDService->findUserFds($data['data']['receive_user']);
        $server = server();
        foreach ($fds as $fd) {
            $fd = intval($fd);
            if ($server->exist($fd)) {
                $server->push($fd, json_encode(['event_keyboard', $data['data']]));
            }
        }

        return Result::ACK;
    }

    /**
     * 用户上线或下线消息
     *
     * @param array $data 队列消息
     * @param AMQPMessage $message
     * @return string
     */
    public function onConsumeOnlineStatus(array $data, AMQPMessage $message)
    {
        $user_id = $data['data']['user_id'];
        $friends = UsersFriend::getFriendIds(intval($user_id));

        $fds = [];
        foreach ($friends as $friend_id) {
            $fds = array_merge($fds, $this->socketFDService->findUserFds(intval($friend_id)));
        }

        $fds = array_unique($fds);
        $server = server();
        foreach ($fds as $fd) {
            $fd = intval($fd);
            if ($server->exist($fd)) {
                $server->push($fd, json_encode(['event_online_status', $data['data']]));
            }
        }

        return Result::ACK;
    }

    /**
     * 撤销聊天消息
     *
     * @param array $data 队列消息
     * @param AMQPMessage $message
     * @return string
     */
    public function onConsumeRevokeTalk(array $data, AMQPMessage $message)
    {
        $record = ChatRecord::where('id', $data['data']['record_id'])->first(['id', 'source', 'user_id', 'receive_id']);

        $fds = [];
        if ($record->source == 1) {
            $fds = array_merge($fds, $this->socketFDService->findUserFds($record->user_id));
            $fds = array_merge($fds, $this->socketFDService->findUserFds($record->receive_id));
        } else if ($record->source == 2) {
            $userIds = $this->socketRoomService->getRoomMembers(strval($record->receive_id));
            foreach ($userIds as $uid) {
                $fds = array_merge($fds, $this->socketFDService->findUserFds(intval($uid)));
            }
        }

        $fds = array_unique($fds);
        $server = server();
        foreach ($fds as $fd) {
            $fd = intval($fd);
            if ($server->exist($fd)) {
                $server->push($fd, json_encode(['event_revoke_talk', [
                    'record_id' => $record->id,
                    'source' => $record->source,
                    'user_id' => $record->user_id,
                    'receive_id' => $record->receive_id,
                ]]));
            }
        }

        return Result::ACK;
    }
}