408 lines
13 KiB
PHP
408 lines
13 KiB
PHP
<?php
|
|
declare(strict_types=1);
|
|
/**
|
|
* This is my open source code, please do not use it for commercial applications.
|
|
* For the full copyright and license information,
|
|
* please view the LICENSE file that was distributed with this source code
|
|
*
|
|
* @author Yuandong<837215079@qq.com>
|
|
* @link https://github.com/gzydong/hyperf-chat
|
|
*/
|
|
|
|
namespace App\Amqp\Consumer;
|
|
|
|
use App\Constants\TalkMsgType;
|
|
use App\Constants\TalkType;
|
|
use Hyperf\Amqp\Annotation\Consumer;
|
|
use Hyperf\Amqp\Result;
|
|
use Hyperf\Amqp\Message\ConsumerMessage;
|
|
use Hyperf\Amqp\Message\Type;
|
|
use Hyperf\Amqp\Builder\QueueBuilder;
|
|
use PhpAmqpLib\Message\AMQPMessage;
|
|
use App\Model\User;
|
|
use App\Model\UsersFriend;
|
|
use App\Model\Chat\TalkRecords;
|
|
use App\Model\Chat\TalkRecordsCode;
|
|
use App\Model\Chat\TalkRecordsFile;
|
|
use App\Model\Chat\TalkRecordsInvite;
|
|
use App\Model\Chat\TalkRecordsForward;
|
|
use App\Model\Group\Group;
|
|
use App\Service\SocketClientService;
|
|
use App\Constants\SocketConstants;
|
|
use App\Cache\Repository\LockRedis;
|
|
use App\Cache\SocketRoom;
|
|
|
|
/**
|
|
* 消息推送消费者队列
|
|
* @Consumer(name="ConsumerChat",enable=true)
|
|
*/
|
|
class ChatMessageConsumer extends ConsumerMessage
|
|
{
|
|
/**
|
|
* 交换机名称
|
|
*
|
|
* @var string
|
|
*/
|
|
public $exchange = SocketConstants::CONSUMER_MESSAGE_EXCHANGE;
|
|
|
|
/**
|
|
* 交换机类型
|
|
*
|
|
* @var string
|
|
*/
|
|
public $type = Type::FANOUT;
|
|
|
|
/**
|
|
* 路由key
|
|
*
|
|
* @var string
|
|
*/
|
|
public $routingKey = 'consumer:im:message';
|
|
|
|
/**
|
|
* @var SocketClientService
|
|
*/
|
|
private $socketClientService;
|
|
|
|
/**
|
|
* 消息事件与回调事件绑定
|
|
*
|
|
* @var array
|
|
*/
|
|
const EVENTS = [
|
|
// 聊天消息事件
|
|
SocketConstants::EVENT_TALK => 'onConsumeTalk',
|
|
|
|
// 键盘输入事件
|
|
SocketConstants::EVENT_KEYBOARD => 'onConsumeKeyboard',
|
|
|
|
// 用户在线状态事件
|
|
SocketConstants::EVENT_ONLINE_STATUS => 'onConsumeOnlineStatus',
|
|
|
|
// 聊天消息推送事件
|
|
SocketConstants::EVENT_REVOKE_TALK => 'onConsumeRevokeTalk',
|
|
|
|
// 好友申请相关事件
|
|
SocketConstants::EVENT_FRIEND_APPLY => 'onConsumeFriendApply'
|
|
];
|
|
|
|
/**
|
|
* ChatMessageConsumer constructor.
|
|
*
|
|
* @param SocketClientService $socketClientService
|
|
*/
|
|
public function __construct(SocketClientService $socketClientService)
|
|
{
|
|
$this->socketClientService = $socketClientService;
|
|
|
|
// 动态设置 Rabbit MQ 消费队列名称
|
|
$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'])) {
|
|
// [加锁]防止消息重复消费
|
|
$lockName = sprintf('ws-message:%s-%s', SERVER_RUN_ID, $data['uuid']);
|
|
if (!LockRedis::getInstance()->lock($lockName, 60)) {
|
|
return Result::ACK;
|
|
}
|
|
|
|
// 调用对应事件绑定的回调方法
|
|
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): string
|
|
{
|
|
$talk_type = $data['data']['talk_type'];
|
|
$sender_id = $data['data']['sender_id'];
|
|
$receiver_id = $data['data']['receiver_id'];
|
|
$record_id = $data['data']['record_id'];
|
|
|
|
$fds = [];
|
|
$groupInfo = null;
|
|
|
|
if ($talk_type == TalkType::PRIVATE_CHAT) {
|
|
$fds = array_merge(
|
|
$this->socketClientService->findUserFds($sender_id),
|
|
$this->socketClientService->findUserFds($receiver_id)
|
|
);
|
|
} else if ($talk_type == TalkType::GROUP_CHAT) {
|
|
foreach (SocketRoom::getInstance()->getRoomMembers(strval($receiver_id)) as $uid) {
|
|
$fds = array_merge($fds, $this->socketClientService->findUserFds(intval($uid)));
|
|
}
|
|
|
|
$groupInfo = Group::where('id', $receiver_id)->first(['group_name', 'avatar']);
|
|
}
|
|
|
|
// 客户端ID去重
|
|
if (!$fds = array_unique($fds)) {
|
|
return Result::ACK;
|
|
}
|
|
|
|
$result = TalkRecords::leftJoin('users', 'users.id', '=', 'talk_records.user_id')
|
|
->where('talk_records.id', $record_id)
|
|
->first([
|
|
'talk_records.id',
|
|
'talk_records.talk_type',
|
|
'talk_records.msg_type',
|
|
'talk_records.user_id',
|
|
'talk_records.receiver_id',
|
|
'talk_records.content',
|
|
'talk_records.is_revoke',
|
|
'talk_records.created_at',
|
|
'users.nickname',
|
|
'users.avatar',
|
|
]);
|
|
|
|
if (!$result) return Result::ACK;
|
|
|
|
$file = $code_block = $forward = $invite = [];
|
|
|
|
switch ($result->msg_type) {
|
|
case TalkMsgType::FILE_MESSAGE:
|
|
$file = TalkRecordsFile::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() : [];
|
|
$file && $file['file_url'] = get_media_url($file['save_dir']);
|
|
break;
|
|
|
|
case TalkMsgType::FORWARD_MESSAGE:
|
|
$forward = ['num' => 0, 'list' => []];
|
|
$forwardInfo = TalkRecordsForward::where('record_id', $result->id)->first(['records_id', 'text']);
|
|
if ($forwardInfo) {
|
|
$forward = [
|
|
'num' => count(parse_ids($forwardInfo->records_id)),
|
|
'list' => json_decode($forwardInfo->text, true) ?? []
|
|
];
|
|
}
|
|
break;
|
|
|
|
case TalkMsgType::CODE_MESSAGE:
|
|
$code_block = TalkRecordsCode::where('record_id', $result->id)->first(['record_id', 'code_lang', 'code']);
|
|
$code_block = $code_block ? $code_block->toArray() : [];
|
|
break;
|
|
|
|
case TalkMsgType::GROUP_INVITE_MESSAGE:
|
|
$notifyInfo = TalkRecordsInvite::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']);
|
|
$invite = [
|
|
'type' => $notifyInfo->type,
|
|
'operate_user' => ['id' => $userInfo->id, 'nickname' => $userInfo->nickname],
|
|
'users' => User::whereIn('id', parse_ids($notifyInfo->user_ids))->get(['id', 'nickname'])->toArray()
|
|
];
|
|
|
|
unset($notifyInfo, $userInfo);
|
|
break;
|
|
}
|
|
|
|
$notify = [
|
|
'sender_id' => $sender_id,
|
|
'receiver_id' => $receiver_id,
|
|
'talk_type' => $talk_type,
|
|
'data' => $this->formatTalkMessage([
|
|
'id' => $result->id,
|
|
'talk_type' => $result->talk_type,
|
|
'msg_type' => $result->msg_type,
|
|
"user_id" => $result->user_id,
|
|
"receiver_id" => $result->receiver_id,
|
|
'avatar' => $result->avatar,
|
|
'nickname' => $result->nickname,
|
|
'group_name' => $groupInfo ? $groupInfo->group_name : '',
|
|
'group_avatar' => $groupInfo ? $groupInfo->avatar : '',
|
|
"created_at" => $result->created_at,
|
|
"content" => $result->content,
|
|
"file" => $file,
|
|
"code_block" => $code_block,
|
|
'forward' => $forward,
|
|
'invite' => $invite
|
|
])
|
|
];
|
|
|
|
$this->socketPushNotify($fds, json_encode([SocketConstants::EVENT_TALK, $notify]));
|
|
|
|
return Result::ACK;
|
|
}
|
|
|
|
/**
|
|
* 键盘输入事件消息
|
|
*
|
|
* @param array $data 队列消息
|
|
* @param AMQPMessage $message
|
|
* @return string
|
|
*/
|
|
public function onConsumeKeyboard(array $data, AMQPMessage $message): string
|
|
{
|
|
$fds = $this->socketClientService->findUserFds($data['data']['receiver_id']);
|
|
|
|
$this->socketPushNotify($fds, json_encode([SocketConstants::EVENT_KEYBOARD, $data['data']]));
|
|
|
|
return Result::ACK;
|
|
}
|
|
|
|
/**
|
|
* 用户上线或下线消息
|
|
*
|
|
* @param array $data 队列消息
|
|
* @param AMQPMessage $message
|
|
* @return string
|
|
*/
|
|
public function onConsumeOnlineStatus(array $data, AMQPMessage $message): string
|
|
{
|
|
$user_id = (int)$data['data']['user_id'];
|
|
$status = (int)$data['data']['status'];
|
|
|
|
$fds = [];
|
|
foreach (UsersFriend::getFriendIds($user_id) as $friend_id) {
|
|
$fds = array_merge($fds, $this->socketClientService->findUserFds(intval($friend_id)));
|
|
}
|
|
|
|
$this->socketPushNotify(array_unique($fds), json_encode([
|
|
SocketConstants::EVENT_ONLINE_STATUS, [
|
|
'user_id' => $user_id,
|
|
'status' => $status
|
|
]
|
|
]));
|
|
|
|
return Result::ACK;
|
|
}
|
|
|
|
/**
|
|
* 撤销聊天消息
|
|
*
|
|
* @param array $data 队列消息
|
|
* @param AMQPMessage $message
|
|
* @return string
|
|
*/
|
|
public function onConsumeRevokeTalk(array $data, AMQPMessage $message): string
|
|
{
|
|
$record = TalkRecords::where('id', $data['data']['record_id'])->first(['id', 'talk_type', 'user_id', 'receiver_id']);
|
|
|
|
$fds = [];
|
|
if ($record->talk_type == TalkType::PRIVATE_CHAT) {
|
|
$fds = array_merge($fds, $this->socketClientService->findUserFds($record->user_id));
|
|
$fds = array_merge($fds, $this->socketClientService->findUserFds($record->receiver_id));
|
|
} else if ($record->talk_type == TalkType::GROUP_CHAT) {
|
|
$userIds = SocketRoom::getInstance()->getRoomMembers(strval($record->receiver_id));
|
|
foreach ($userIds as $uid) {
|
|
$fds = array_merge($fds, $this->socketClientService->findUserFds((int)$uid));
|
|
}
|
|
}
|
|
|
|
$fds = array_unique($fds);
|
|
$this->socketPushNotify($fds, json_encode([SocketConstants::EVENT_REVOKE_TALK, [
|
|
'talk_type' => $record->talk_type,
|
|
'sender_id' => $record->user_id,
|
|
'receiver_id' => $record->receiver_id,
|
|
'record_id' => $record->id,
|
|
]]));
|
|
|
|
return Result::ACK;
|
|
}
|
|
|
|
/**
|
|
* 好友申请消息
|
|
*
|
|
* @param array $data 队列消息
|
|
* @param AMQPMessage $message
|
|
* @return string
|
|
*/
|
|
public function onConsumeFriendApply(array $data, AMQPMessage $message): string
|
|
{
|
|
$fds = $this->socketClientService->findUserFds($data['data']['receive']);
|
|
|
|
$this->socketPushNotify(array_unique($fds), json_encode([
|
|
SocketConstants::EVENT_FRIEND_APPLY,
|
|
$data['data']
|
|
]));
|
|
|
|
return Result::ACK;
|
|
}
|
|
|
|
/**
|
|
* WebSocket 消息推送
|
|
*
|
|
* @param $fds
|
|
* @param $message
|
|
*/
|
|
private function socketPushNotify($fds, $message)
|
|
{
|
|
$server = server();
|
|
foreach ($fds as $fd) {
|
|
$server->exist(intval($fd)) && $server->push(intval($fd), $message);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* 格式化对话的消息体
|
|
*
|
|
* @param array $data 对话的消息
|
|
* @return array
|
|
*/
|
|
private function formatTalkMessage(array $data): array
|
|
{
|
|
$message = [
|
|
"id" => 0, // 消息记录ID
|
|
"talk_type" => 1, // 消息来源[1:好友私信;2:群聊]
|
|
"msg_type" => 1, // 消息类型
|
|
"user_id" => 0, // 发送者用户ID
|
|
"receiver_id" => 0, // 接收者ID[好友ID或群ID]
|
|
|
|
// 发送消息人的信息
|
|
"nickname" => "",// 用户昵称
|
|
"avatar" => "",// 用户头像
|
|
"group_name" => "",// 群组名称
|
|
"group_avatar" => "",// 群组头像
|
|
|
|
// 不同的消息类型
|
|
"file" => [],
|
|
"code_block" => [],
|
|
"forward" => [],
|
|
"invite" => [],
|
|
|
|
// 消息创建时间
|
|
"content" => '',// 文本消息
|
|
"created_at" => "",
|
|
|
|
// 消息属性
|
|
"is_revoke" => 0, // 消息是否撤销
|
|
];
|
|
|
|
return array_merge($message, array_intersect_key($data, $message));
|
|
}
|
|
}
|