mirror of
https://gitee.com/zhijiantianya/ruoyi-vue-pro.git
synced 2026-07-08 22:40:32 +08:00
refactor: 扁平化 IM WebSocket 通知推送 API
- 将 WebSocket 推送入口统一为 userId/userIds + conversationType + contentType + payload - 移除业务侧 ImNotificationWebSocketDTO 构造和无会话专用发送入口 - 收敛私聊、群聊、频道、好友、加群申请、RTC 通知调用路径 - 精简 ImNotificationWebSocketDTO，仅保留统一外壳字段 - 保留群消息 payload 的 receiptStatus、readCount、receiverUserIds - 更新相关单元测试，覆盖群消息通知 payload 字段
This commit is contained in:
YunaiV
@ -2,11 +2,11 @@ package cn.iocoder.yudao.module.im.mq.consumer.friend;
|
||||
|
||||
import cn.hutool.core.collection.CollUtil;
|
||||
import cn.iocoder.yudao.module.im.dal.dataobject.friend.ImFriendDO;
|
||||
import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.dto.ImPrivateMessageDTO;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.dto.notification.friend.FriendInfoUpdatedNotification;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.notification.friend.FriendInfoUpdatedNotification;
|
||||
import cn.iocoder.yudao.module.system.api.message.user.AdminUserProfileUpdateMessage;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
@ -52,8 +52,9 @@ public class AdminUserProfileUpdateConsumer {
|
||||
try {
|
||||
FriendInfoUpdatedNotification payload = (FriendInfoUpdatedNotification) new FriendInfoUpdatedNotification()
|
||||
.setOperatorUserId(userId).setFriendUserId(userId);
|
||||
websocketService.sendPrivateMessageAsync(friend.getFriendUserId(), ImPrivateMessageDTO.ofFriendNotification(
|
||||
ImMessageTypeEnum.FRIEND_INFO_UPDATED.getType(), userId, friend.getFriendUserId(), payload));
|
||||
websocketService.sendNotificationAsync(friend.getFriendUserId(),
|
||||
ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.FRIEND_INFO_UPDATED.getType(), payload);
|
||||
successCount++;
|
||||
} catch (Exception e) {
|
||||
log.warn("[onMessage][userId({}) friendUserId({}) 推送失败]",
|
||||
|
||||
@ -14,7 +14,8 @@ import cn.iocoder.yudao.module.im.dal.dataobject.friend.ImFriendRequestDO;
|
||||
import cn.iocoder.yudao.module.im.dal.mysql.friend.ImFriendRequestMapper;
|
||||
import cn.iocoder.yudao.module.im.enums.friend.ImFriendRequestHandleResultEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.friend.ImFriendStateEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.framework.config.ImProperties;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.notification.friend.FriendRequestApprovedNotification;
|
||||
@ -104,8 +105,8 @@ public class ImFriendRequestServiceImpl implements ImFriendRequestService {
|
||||
if (fromUser != null) {
|
||||
payload.setFromNickname(fromUser.getNickname()).setFromAvatar(fromUser.getAvatar());
|
||||
}
|
||||
websocketService.sendPrivateMessageAsync(toUserId, ImPrivateMessageDTO.ofFriendNotification(
|
||||
ImMessageTypeEnum.FRIEND_REQUEST_RECEIVED.getType(), fromUserId, toUserId, payload));
|
||||
websocketService.sendNotificationAsync(toUserId, ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.FRIEND_REQUEST_RECEIVED.getType(), payload);
|
||||
|
||||
// 4. 全局自动通过开关：注册 afterCommit 回调，事务提交后再走同意流程
|
||||
// 回调内 try/catch 兜底 —— afterCommit 异常会被 Spring 静默吞掉，否则同意失败时申请方永远等不到 APPROVED
|
||||
@ -191,8 +192,8 @@ public class ImFriendRequestServiceImpl implements ImFriendRequestService {
|
||||
FriendRequestApprovedNotification payload = (FriendRequestApprovedNotification)
|
||||
new FriendRequestApprovedNotification().setRequestId(request.getId())
|
||||
.setOperatorUserId(userId).setFriendUserId(userId);
|
||||
websocketService.sendPrivateMessageAsync(request.getFromUserId(), ImPrivateMessageDTO.ofFriendNotification(
|
||||
ImMessageTypeEnum.FRIEND_REQUEST_APPROVED.getType(), userId, request.getFromUserId(), payload));
|
||||
websocketService.sendNotificationAsync(request.getFromUserId(), ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.FRIEND_REQUEST_APPROVED.getType(), payload);
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -216,8 +217,8 @@ public class ImFriendRequestServiceImpl implements ImFriendRequestService {
|
||||
new FriendRequestRejectedNotification().setRequestId(request.getId())
|
||||
.setHandleContent(handleContent)
|
||||
.setOperatorUserId(userId).setFriendUserId(userId);
|
||||
websocketService.sendPrivateMessageAsync(request.getFromUserId(), ImPrivateMessageDTO.ofFriendNotification(
|
||||
ImMessageTypeEnum.FRIEND_REQUEST_REJECTED.getType(), userId, request.getFromUserId(), payload));
|
||||
websocketService.sendNotificationAsync(request.getFromUserId(), ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.FRIEND_REQUEST_REJECTED.getType(), payload);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@ -11,12 +11,12 @@ import cn.iocoder.yudao.module.im.dal.dataobject.friend.ImFriendDO;
|
||||
import cn.iocoder.yudao.module.im.dal.dataobject.friend.ImFriendRequestDO;
|
||||
import cn.iocoder.yudao.module.im.dal.mysql.friend.ImFriendMapper;
|
||||
import cn.iocoder.yudao.module.im.enums.friend.ImFriendStateEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.service.message.ImPrivateMessageService;
|
||||
import cn.iocoder.yudao.module.im.service.message.dto.ImPrivateMessageSendDTO;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.dto.ImPrivateMessageDTO;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.dto.notification.friend.*;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.notification.friend.*;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.cache.annotation.CacheEvict;
|
||||
@ -160,8 +160,8 @@ public class ImFriendServiceImpl implements ImFriendService {
|
||||
FriendUpdateNotification payload = (FriendUpdateNotification) new FriendUpdateNotification()
|
||||
.setDisplayName(reqVO.getDisplayName()).setSilent(reqVO.getSilent()).setPinned(reqVO.getPinned())
|
||||
.setOperatorUserId(userId).setFriendUserId(reqVO.getFriendUserId());
|
||||
websocketService.sendPrivateMessageAsync(userId, ImPrivateMessageDTO.ofFriendNotification(
|
||||
ImMessageTypeEnum.FRIEND_UPDATE.getType(), userId, userId, payload));
|
||||
websocketService.sendNotificationAsync(userId, ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.FRIEND_UPDATE.getType(), payload);
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -183,7 +183,7 @@ public class ImFriendServiceImpl implements ImFriendService {
|
||||
FriendAddNotification payload = (FriendAddNotification) new FriendAddNotification()
|
||||
.setOperatorUserId(fromUserId).setFriendUserId(toUserId);
|
||||
privateMessageService.sendPrivateMessage(fromUserId, new ImPrivateMessageSendDTO()
|
||||
.setReceiverId(toUserId).setType(ImMessageTypeEnum.FRIEND_ADD.getType()).setContent(payload));
|
||||
.setReceiverId(toUserId).setType(ImContentTypeEnum.FRIEND_ADD.getType()).setContent(payload));
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -202,7 +202,7 @@ public class ImFriendServiceImpl implements ImFriendService {
|
||||
FriendAddNotification payload = (FriendAddNotification) new FriendAddNotification()
|
||||
.setOperatorUserId(friendUserId).setFriendUserId(friendUserId);
|
||||
privateMessageService.sendPrivateMessage(userId, new ImPrivateMessageSendDTO()
|
||||
.setReceiverId(friendUserId).setType(ImMessageTypeEnum.FRIEND_ADD.getType())
|
||||
.setReceiverId(friendUserId).setType(ImContentTypeEnum.FRIEND_ADD.getType())
|
||||
.setContent(payload).setPersistent(false));
|
||||
}
|
||||
|
||||
@ -224,7 +224,7 @@ public class ImFriendServiceImpl implements ImFriendService {
|
||||
FriendDeleteNotification payload = ((FriendDeleteNotification) new FriendDeleteNotification()
|
||||
.setOperatorUserId(userId).setFriendUserId(friendUserId)).setClear(clear);
|
||||
privateMessageService.sendPrivateMessage(userId, new ImPrivateMessageSendDTO()
|
||||
.setReceiverId(friendUserId).setType(ImMessageTypeEnum.FRIEND_DELETE.getType())
|
||||
.setReceiverId(friendUserId).setType(ImContentTypeEnum.FRIEND_DELETE.getType())
|
||||
.setContent(payload).setPersistent(false));
|
||||
}
|
||||
|
||||
@ -250,8 +250,8 @@ public class ImFriendServiceImpl implements ImFriendService {
|
||||
// 3. 推 FRIEND_BLOCK 给 A 多端
|
||||
FriendBlockNotification payload = (FriendBlockNotification) new FriendBlockNotification()
|
||||
.setOperatorUserId(userId).setFriendUserId(friendUserId);
|
||||
websocketService.sendPrivateMessageAsync(userId, ImPrivateMessageDTO.ofFriendNotification(
|
||||
ImMessageTypeEnum.FRIEND_BLOCK.getType(), userId, userId, payload));
|
||||
websocketService.sendNotificationAsync(userId, ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.FRIEND_BLOCK.getType(), payload);
|
||||
}
|
||||
|
||||
@Override
|
||||
@ -276,8 +276,8 @@ public class ImFriendServiceImpl implements ImFriendService {
|
||||
// 3. 推 FRIEND_UNBLOCK 给 A 多端
|
||||
FriendUnblockNotification payload = (FriendUnblockNotification) new FriendUnblockNotification()
|
||||
.setOperatorUserId(userId).setFriendUserId(friendUserId);
|
||||
websocketService.sendPrivateMessageAsync(userId, ImPrivateMessageDTO.ofFriendNotification(
|
||||
ImMessageTypeEnum.FRIEND_UNBLOCK.getType(), userId, userId, payload));
|
||||
websocketService.sendNotificationAsync(userId, ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.FRIEND_UNBLOCK.getType(), payload);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@ -13,7 +13,8 @@ import cn.iocoder.yudao.module.im.dal.mysql.group.ImGroupRequestMapper;
|
||||
import cn.iocoder.yudao.module.im.enums.group.ImGroupAddSourceEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.group.ImGroupMemberRoleEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.group.ImGroupRequestHandleResultEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.message.ImMessageTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.service.message.ImGroupMessageService;
|
||||
import cn.iocoder.yudao.module.im.service.message.dto.ImGroupMessageSendDTO;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
|
||||
@ -100,8 +101,8 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
|
||||
AdminUserRespDTO applyUser = adminUserApi.getUser(userId);
|
||||
GroupRequestReceivedNotification payload = buildRequestNotification(group, request, applyUser);
|
||||
for (Long receiverUserId : getGroupMemberListByOwnerAndAdminUserIds(group)) {
|
||||
websocketService.sendPrivateMessageAsync(receiverUserId, ImPrivateMessageDTO.ofGroupNotification(
|
||||
ImMessageTypeEnum.GROUP_REQUEST_RECEIVED.getType(), userId, receiverUserId, payload));
|
||||
websocketService.sendNotificationAsync(receiverUserId, ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.GROUP_REQUEST_RECEIVED.getType(), payload);
|
||||
}
|
||||
return request;
|
||||
}
|
||||
@ -200,8 +201,8 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
|
||||
AdminUserRespDTO applyUser = userMap.get(request.getUserId());
|
||||
GroupRequestReceivedNotification payload = buildRequestNotification(group, request, applyUser);
|
||||
for (Long receiverUserId : ownerAndAdmins) {
|
||||
websocketService.sendPrivateMessageAsync(receiverUserId, ImPrivateMessageDTO.ofGroupNotification(
|
||||
ImMessageTypeEnum.GROUP_REQUEST_RECEIVED.getType(), inviterUserId, receiverUserId, payload));
|
||||
websocketService.sendNotificationAsync(receiverUserId, ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.GROUP_REQUEST_RECEIVED.getType(), payload);
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -405,8 +406,8 @@ public class ImGroupRequestServiceImpl implements ImGroupRequestService {
|
||||
Set<Long> receivers = new LinkedHashSet<>(getGroupMemberListByOwnerAndAdminUserIds(group));
|
||||
receivers.add(applicantUserId);
|
||||
for (Long receiverUserId : receivers) {
|
||||
websocketService.sendPrivateMessageAsync(receiverUserId, ImPrivateMessageDTO.ofGroupNotification(
|
||||
messageType, operatorUserId, receiverUserId, payload));
|
||||
websocketService.sendNotificationAsync(receiverUserId, ImConversationTypeEnum.NONE.getType(),
|
||||
messageType, payload);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -97,7 +97,8 @@ public class ImChannelMessageServiceImpl implements ImChannelMessageService {
|
||||
*/
|
||||
@Async
|
||||
public void readChannelMessageEvent(Long userId, Long channelId, Long readId) {
|
||||
webSocketService.sendChannelMessageAsync(userId, ImChannelMessageDTO.ofRead(channelId, readId));
|
||||
webSocketService.sendNotificationAsync(userId, ImConversationTypeEnum.CHANNEL.getType(),
|
||||
ImContentTypeEnum.READ.getType(), ImMessageReadNotification.ofChannel(channelId, readId));
|
||||
}
|
||||
|
||||
private ImChannelMessageServiceImpl getSelf() {
|
||||
@ -121,11 +122,12 @@ public class ImChannelMessageServiceImpl implements ImChannelMessageService {
|
||||
channelMessageMapper.insert(message);
|
||||
|
||||
// 3. 异步推 WebSocket：指定用户走点对点；全员（receiverUserIds 为空）走广播
|
||||
ImChannelMessageDTO dto = ImChannelMessageDTO.ofSend(message);
|
||||
ImChannelMessageNotification dto = ImChannelMessageNotification.ofSend(message);
|
||||
if (CollUtil.isNotEmpty(reqVO.getReceiverUserIds())) {
|
||||
webSocketService.sendChannelMessageAsync(reqVO.getReceiverUserIds(), dto);
|
||||
webSocketService.sendNotificationAsync(reqVO.getReceiverUserIds(), ImConversationTypeEnum.CHANNEL.getType(),
|
||||
dto.getType(), dto);
|
||||
} else {
|
||||
webSocketService.broadcastChannelMessageAsync(dto);
|
||||
webSocketService.broadcastNotificationAsync(ImConversationTypeEnum.CHANNEL.getType(), dto.getType(), dto);
|
||||
}
|
||||
return message.getId();
|
||||
}
|
||||
|
||||
@ -118,7 +118,9 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
|
||||
}
|
||||
|
||||
// 3. WebSocket 异步推送（快照内可见成员 + 发送方多端同步）
|
||||
imWebSocketService.sendGroupMessageAsync(receiverUserIds, ImGroupMessageDTO.ofSend(message));
|
||||
ImGroupMessageNotification notification = ImGroupMessageNotification.ofSend(message);
|
||||
imWebSocketService.sendNotificationAsync(receiverUserIds, ImConversationTypeEnum.GROUP.getType(),
|
||||
notification.getType(), notification);
|
||||
return message;
|
||||
}
|
||||
|
||||
@ -149,7 +151,9 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
|
||||
}
|
||||
|
||||
// 2. WebSocket 异步推送
|
||||
imWebSocketService.sendGroupMessageAsync(receiverUserIds, ImGroupMessageDTO.ofSend(message));
|
||||
ImGroupMessageNotification notification = ImGroupMessageNotification.ofSend(message);
|
||||
imWebSocketService.sendNotificationAsync(receiverUserIds, ImConversationTypeEnum.GROUP.getType(),
|
||||
notification.getType(), notification);
|
||||
return message;
|
||||
}
|
||||
|
||||
@ -347,8 +351,9 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
|
||||
@SuppressWarnings("DataFlowIssue")
|
||||
public void readGroupMessageEvent(Long userId, Long groupId, Long prevMaxMessageId, Long newMaxMessageId) {
|
||||
// 1. 发送 READ 事件给自己的其他终端（多端同步）
|
||||
imWebSocketService.sendGroupMessageAsync(userId,
|
||||
ImGroupMessageDTO.ofRead(userId, groupId, newMaxMessageId));
|
||||
imWebSocketService.sendNotificationAsync(userId, ImConversationTypeEnum.GROUP.getType(),
|
||||
ImContentTypeEnum.READ.getType(),
|
||||
ImMessageReadNotification.ofGroup(userId, groupId, newMaxMessageId));
|
||||
|
||||
// 2. 刷新 (prevMaxMessageId, newMaxMessageId] 区间内的待回执消息
|
||||
List<ImGroupMessageDO> pendingMessages = groupMessageMapper.selectListByGroupIdAndPendingReceipt(
|
||||
@ -378,8 +383,9 @@ public class ImGroupMessageServiceImpl implements ImGroupMessageService {
|
||||
}
|
||||
|
||||
// 2.2 发送 RECEIPT 事件给消息发送方（只有 ta 关心已读进度）
|
||||
imWebSocketService.sendGroupMessageAsync(message.getSenderId(),
|
||||
ImGroupMessageDTO.ofReceipt(message.getId(), groupId, readCount, newReceiptStatus));
|
||||
imWebSocketService.sendNotificationAsync(message.getSenderId(), ImConversationTypeEnum.GROUP.getType(),
|
||||
ImContentTypeEnum.RECEIPT.getType(),
|
||||
ImMessageReceiptNotification.ofGroup(message.getId(), groupId, readCount, newReceiptStatus));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -103,9 +103,11 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
|
||||
}
|
||||
|
||||
// 3. WebSocket 异步推送：接收方 + 发送方多端同步
|
||||
ImPrivateMessageDTO websocketMessage = ImPrivateMessageDTO.ofSend(message);
|
||||
imWebSocketService.sendPrivateMessageAsync(message.getReceiverId(), websocketMessage);
|
||||
imWebSocketService.sendPrivateMessageAsync(senderId, websocketMessage);
|
||||
ImPrivateMessageNotification notification = ImPrivateMessageNotification.ofSend(message);
|
||||
imWebSocketService.sendNotificationAsync(message.getReceiverId(), ImConversationTypeEnum.PRIVATE.getType(),
|
||||
notification.getType(), notification);
|
||||
imWebSocketService.sendNotificationAsync(senderId, ImConversationTypeEnum.PRIVATE.getType(),
|
||||
notification.getType(), notification);
|
||||
return message;
|
||||
}
|
||||
|
||||
@ -131,11 +133,13 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
|
||||
}
|
||||
|
||||
// 2. WebSocket 异步推送：双向（默认）；单边语义（persistent=false）下仅推 sender 多端，对方不感知
|
||||
ImPrivateMessageDTO websocketMessage = ImPrivateMessageDTO.ofSend(message);
|
||||
ImPrivateMessageNotification notification = ImPrivateMessageNotification.ofSend(message);
|
||||
if (persistent) {
|
||||
imWebSocketService.sendPrivateMessageAsync(dto.getReceiverId(), websocketMessage);
|
||||
imWebSocketService.sendNotificationAsync(dto.getReceiverId(), ImConversationTypeEnum.PRIVATE.getType(),
|
||||
notification.getType(), notification);
|
||||
}
|
||||
imWebSocketService.sendPrivateMessageAsync(senderId, websocketMessage);
|
||||
imWebSocketService.sendNotificationAsync(senderId, ImConversationTypeEnum.PRIVATE.getType(),
|
||||
notification.getType(), notification);
|
||||
return message;
|
||||
}
|
||||
|
||||
@ -256,10 +260,11 @@ public class ImPrivateMessageServiceImpl implements ImPrivateMessageService {
|
||||
}
|
||||
|
||||
// 4. 异步发送 READ + RECEIPT 事件（已读位置以前端上报为准，与多端 / 对方 UI 显示一致）
|
||||
imWebSocketService.sendPrivateMessageAsync(userId,
|
||||
ImPrivateMessageDTO.ofRead(userId, receiverId, messageId));
|
||||
imWebSocketService.sendPrivateMessageAsync(receiverId,
|
||||
ImPrivateMessageDTO.ofReceipt(userId, receiverId, messageId));
|
||||
imWebSocketService.sendNotificationAsync(userId, ImConversationTypeEnum.PRIVATE.getType(),
|
||||
ImContentTypeEnum.READ.getType(), ImMessageReadNotification.ofPrivate(userId, receiverId, messageId));
|
||||
imWebSocketService.sendNotificationAsync(receiverId, ImConversationTypeEnum.PRIVATE.getType(),
|
||||
ImContentTypeEnum.RECEIPT.getType(),
|
||||
ImMessageReceiptNotification.ofPrivate(userId, receiverId, messageId));
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@ -56,7 +56,7 @@ import static cn.iocoder.yudao.module.im.enums.ErrorCodeConstants.*;
|
||||
* 并发幂等：同好友对 / 同群活跃唯一性走 {@link ImRtcCallLockRedisDAO} 分布式锁 + 锁内 SELECT 兜底；webhook 兜底走条件 UPDATE；
|
||||
* <p>
|
||||
* 推送通道分流：
|
||||
* 1601 RTC_CALL（INVITING / JOINED / REJECTED / NO_ANSWER / LEFT 子类型）→ {@link ImWebSocketService#sendPrivateMessageAsync} 仅推参与方；
|
||||
* 1601 RTC_CALL（INVITING / JOINED / REJECTED / NO_ANSWER / LEFT 子类型）→ {@link ImWebSocketService#sendNotificationAsync} 仅推参与方；
|
||||
* 1602 / 1603 PARTICIPANT_CONNECTED / DISCONNECTED → {@link ImWebSocketService} 推参与方 + 群通话场景广播全群；
|
||||
* 1610 RTC_CALL_START + 1611 RTC_CALL_END → {@link ImPrivateMessageService} / {@link ImGroupMessageService} 入消息流当聊天 tip
|
||||
* （START 仅群通话；两者分别在 invite / cancel(leave) 事务里 INSERT，自增 id 自然保证顺序）
|
||||
@ -1001,8 +1001,8 @@ public class ImRtcCallServiceImpl implements ImRtcCallService {
|
||||
String token = signToken(inviteeId, resolveDisplayName(invitee, inviteeId), call.getRoom());
|
||||
ImRtcCallNotification payload = ImRtcCallNotification.ofInvite(
|
||||
call, inviter, imProperties.getRtc().getLivekitUrl(), token, inviteeIds);
|
||||
webSocketService.sendPrivateMessageAsync(inviteeId, ImPrivateMessageDTO.ofRtcNotification(
|
||||
ImMessageTypeEnum.RTC_CALL.getType(), call.getInviterUserId(), inviteeId, payload));
|
||||
webSocketService.sendNotificationAsync(inviteeId, ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.RTC_CALL.getType(), payload);
|
||||
}
|
||||
|
||||
/**
|
||||
@ -1017,8 +1017,8 @@ public class ImRtcCallServiceImpl implements ImRtcCallService {
|
||||
AdminUserRespDTO operator = operatorUserId != null ? adminUserApi.getUser(operatorUserId) : null;
|
||||
ImRtcCallNotification payload = ImRtcCallNotification.ofReject(call, operatorUserId, operator);
|
||||
for (Long receiverUserId : getCallAudienceUserIdList(call)) {
|
||||
webSocketService.sendPrivateMessageAsync(receiverUserId, ImPrivateMessageDTO.ofRtcNotification(
|
||||
ImMessageTypeEnum.RTC_CALL.getType(), operatorUserId, receiverUserId, payload));
|
||||
webSocketService.sendNotificationAsync(receiverUserId, ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.RTC_CALL.getType(), payload);
|
||||
}
|
||||
}
|
||||
|
||||
@ -1034,8 +1034,8 @@ public class ImRtcCallServiceImpl implements ImRtcCallService {
|
||||
private void pushCallNoAnswerNotification(ImRtcCallDO call, Long operatorUserId, AdminUserRespDTO operator) {
|
||||
ImRtcCallNotification payload = ImRtcCallNotification.ofNoAnswer(call, operatorUserId, operator);
|
||||
for (Long receiverUserId : getCallAudienceUserIdList(call)) {
|
||||
webSocketService.sendPrivateMessageAsync(receiverUserId, ImPrivateMessageDTO.ofRtcNotification(
|
||||
ImMessageTypeEnum.RTC_CALL.getType(), operatorUserId, receiverUserId, payload));
|
||||
webSocketService.sendNotificationAsync(receiverUserId, ImConversationTypeEnum.NONE.getType(),
|
||||
ImContentTypeEnum.RTC_CALL.getType(), payload);
|
||||
}
|
||||
}
|
||||
|
||||
@ -1074,8 +1074,7 @@ public class ImRtcCallServiceImpl implements ImRtcCallService {
|
||||
if (CollUtil.isEmpty(receivers)) {
|
||||
return;
|
||||
}
|
||||
ImPrivateMessageDTO dto = ImPrivateMessageDTO.ofRtcNotification(type, actorUserId, null, payload);
|
||||
webSocketService.sendPrivateMessageAsync(receivers, dto);
|
||||
webSocketService.sendNotificationAsync(receivers, ImConversationTypeEnum.NONE.getType(), type, payload);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@ -34,55 +34,24 @@ public class ImWebSocketServiceImpl implements ImWebSocketService {
|
||||
private WebSocketSenderApi webSocketSenderApi;
|
||||
|
||||
@Override
|
||||
public void sendPrivateMessageAsync(Collection<Long> userIds, ImPrivateMessageDTO dto) {
|
||||
// 说明：通过 executeAfterTransaction 保证事务提交后再推送，避免客户端收到消息后查询数据库时事务尚未提交
|
||||
// 通过 getSelf() 获取 Spring 代理对象调用 @Async 方法，确保异步 AOP 生效（直接 this 调用会绕过代理）
|
||||
executeAfterTransaction(() -> getSelf().doSendPrivateMessage(userIds, dto));
|
||||
public void sendNotificationAsync(Collection<Long> userIds, Integer conversationType, Integer contentType,
|
||||
Object payload) {
|
||||
ImNotificationWebSocketDTO notification = buildNotification(conversationType, contentType, payload);
|
||||
executeAfterTransaction(() -> getSelf().doSendNotification(userIds, notification));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sendGroupMessageAsync(Collection<Long> userIds, ImGroupMessageDTO dto) {
|
||||
executeAfterTransaction(() -> getSelf().doSendGroupMessage(userIds, dto));
|
||||
public void broadcastNotificationAsync(Integer conversationType, Integer contentType, Object payload) {
|
||||
ImNotificationWebSocketDTO notification = buildNotification(conversationType, contentType, payload);
|
||||
executeAfterTransaction(() -> getSelf().doBroadcastNotification(notification));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sendChannelMessageAsync(Collection<Long> userIds, ImChannelMessageDTO dto) {
|
||||
executeAfterTransaction(() -> getSelf().doSendChannelMessage(userIds, dto));
|
||||
}
|
||||
|
||||
@Override
|
||||
public void broadcastChannelMessageAsync(ImChannelMessageDTO dto) {
|
||||
executeAfterTransaction(() -> getSelf().doBroadcastChannelMessage(dto));
|
||||
}
|
||||
|
||||
/**
|
||||
* 异步发送私聊 WebSocket 消息；多收件人共享同一 dto，避免按收件人重复注册 afterCommit 回调
|
||||
*/
|
||||
@Async
|
||||
public void doSendPrivateMessage(Collection<Long> userIds, ImPrivateMessageDTO dto) {
|
||||
for (Long userId : getDistinctUserIds(userIds)) {
|
||||
try {
|
||||
webSocketSenderApi.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
|
||||
ImPrivateMessageDTO.TYPE, dto);
|
||||
} catch (Exception e) {
|
||||
log.error("[doSendPrivateMessage][userId({}) dto({}) 发送失败]", userId, dto, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 异步发送群聊 WebSocket 消息
|
||||
*/
|
||||
@Async
|
||||
public void doSendGroupMessage(Collection<Long> userIds, ImGroupMessageDTO dto) {
|
||||
for (Long userId : getDistinctUserIds(userIds)) {
|
||||
try {
|
||||
webSocketSenderApi.sendObject(UserTypeEnum.ADMIN.getValue(), userId,
|
||||
ImGroupMessageDTO.TYPE, dto);
|
||||
} catch (Exception e) {
|
||||
log.error("[doSendGroupMessage][userId({}) dto({}) 发送失败]", userId, dto, e);
|
||||
}
|
||||
}
|
||||
private static ImNotificationWebSocketDTO buildNotification(Integer conversationType, Integer contentType,
|
||||
Object payload) {
|
||||
return new ImNotificationWebSocketDTO()
|
||||
.setConversationType(conversationType)
|
||||
.setContentType(contentType)
|
||||
.setPayload(payload);
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
@ -112,7 +112,7 @@ public class ImFriendRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertNull(result);
|
||||
verify(friendService).silentReAddFriend(eq(1L), eq(2L), eq("老张"), eq(1));
|
||||
verify(friendRequestMapper, never()).insert(any(ImFriendRequestDO.class));
|
||||
verify(websocketService, never()).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService, never()).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -157,7 +157,7 @@ public class ImFriendRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertEquals(ImFriendRequestHandleResultEnum.UNHANDLED.getResult(), saved.getHandleResult());
|
||||
assertSame(saved, result);
|
||||
// 断言：推送给接收方
|
||||
verify(websocketService).sendPrivateMessageAsync(eq(2L), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService).sendNotificationAsync(eq(2L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -210,7 +210,7 @@ public class ImFriendRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
any(LocalDateTime.class));
|
||||
assertEquals(100L, result.getId());
|
||||
assertEquals(ImFriendRequestHandleResultEnum.UNHANDLED.getResult(), result.getHandleResult());
|
||||
verify(websocketService).sendPrivateMessageAsync(eq(2L), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService).sendNotificationAsync(eq(2L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
// ========== agreeFriendRequest ==========
|
||||
@ -230,7 +230,7 @@ public class ImFriendRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
// 断言：双向建立好友 + 推 APPROVED 给发起方
|
||||
verify(adminUserApi).validateUserList(ListUtil.of(1L, 2L));
|
||||
verify(friendService).becomeFriends(request);
|
||||
verify(websocketService).sendPrivateMessageAsync(eq(1L), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService).sendNotificationAsync(eq(1L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -305,7 +305,7 @@ public class ImFriendRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
eq(ImFriendRequestHandleResultEnum.UNHANDLED.getResult()), captor.capture());
|
||||
assertEquals(ImFriendRequestHandleResultEnum.REFUSED.getResult(), captor.getValue().getHandleResult());
|
||||
assertEquals("不认识", captor.getValue().getHandleContent());
|
||||
verify(websocketService).sendPrivateMessageAsync(eq(1L), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService).sendNotificationAsync(eq(1L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -320,7 +320,7 @@ public class ImFriendRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
ServiceException exception = assertThrows(ServiceException.class,
|
||||
() -> friendRequestService.refuseFriendRequest(2L, 100L, "x"));
|
||||
assertEquals(FRIEND_REQUEST_HANDLED.getCode(), exception.getCode());
|
||||
verify(websocketService, never()).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService, never()).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
// ========== getMyFriendRequestList ==========
|
||||
|
||||
@ -68,7 +68,7 @@ public class ImFriendServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertEquals(100L, captor.getValue().getId());
|
||||
assertTrue(captor.getValue().getSilent());
|
||||
// 断言：推送了好友更新通知
|
||||
verify(imWebSocketService).sendPrivateMessageAsync(eq(1L), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(1L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -100,7 +100,7 @@ public class ImFriendServiceImplTest extends BaseMockitoUnitTest {
|
||||
() -> friendService.updateFriend(1L, reqVO));
|
||||
assertEquals(FRIEND_NOT_FRIEND.getCode(), exception.getCode());
|
||||
verify(imFriendMapper, never()).updateById(any(ImFriendDO.class));
|
||||
verify(imWebSocketService, never()).sendPrivateMessageAsync(any(Long.class), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService, never()).sendNotificationAsync(any(Long.class), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -123,7 +123,7 @@ public class ImFriendServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertEquals("老张", captor.getValue().getDisplayName());
|
||||
assertNull(captor.getValue().getSilent());
|
||||
// 断言：推送好友更新通知
|
||||
verify(imWebSocketService).sendPrivateMessageAsync(eq(1L), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(1L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -138,7 +138,7 @@ public class ImFriendServiceImplTest extends BaseMockitoUnitTest {
|
||||
// 断言：没查记录、没触发 SQL 更新 / 没发 WebSocket 推送
|
||||
verify(imFriendMapper, never()).selectByUserIdAndFriendUserId(anyLong(), anyLong());
|
||||
verify(imFriendMapper, never()).updateById(any(ImFriendDO.class));
|
||||
verify(imWebSocketService, never()).sendPrivateMessageAsync(any(Long.class), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService, never()).sendNotificationAsync(any(Long.class), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
// ========== addFriend0（内部方法，被 becomeFriends / silentReAddFriend 调用） ==========
|
||||
|
||||
@ -116,7 +116,7 @@ public class ImGroupRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
verify(groupMemberService, never()).addGroupMember(anyLong(), anyLong(), anyInt(), anyInt(), anyLong());
|
||||
verify(groupRequestMapper).insert(any(ImGroupRequestDO.class));
|
||||
// 1503 推送给 owner(99) + admin(98)，去重后两条
|
||||
verify(websocketService, times(2)).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService, times(2)).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -149,7 +149,7 @@ public class ImGroupRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
eq(ImGroupAddSourceEnum.SEARCH.getSource()), any());
|
||||
assertEquals(50L, result.getId());
|
||||
assertEquals(ImGroupRequestHandleResultEnum.UNHANDLED.getResult(), result.getHandleResult());
|
||||
verify(websocketService).sendPrivateMessageAsync(eq(99L), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService).sendNotificationAsync(eq(99L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -208,7 +208,7 @@ public class ImGroupRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
verify(groupMessageService).sendGroupMessage(eq(1L), dtoCaptor.capture());
|
||||
assertEquals(ImContentTypeEnum.GROUP_MEMBER_ENTER.getType(), dtoCaptor.getValue().getType());
|
||||
// 1505 推送给申请人 + owner，去重后两条
|
||||
verify(websocketService, times(2)).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService, times(2)).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -278,7 +278,7 @@ public class ImGroupRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
// 不写群成员；推 1506 给申请人 + 群主（同一人 1L 时去重为 1 + 申请人 2L = 2 条）
|
||||
verify(groupMemberService, never()).addGroupMember(anyLong(), anyLong(), anyInt(), any(), any());
|
||||
verify(websocketService, times(2)).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService, times(2)).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
// ==================== createInviteRequestList ====================
|
||||
@ -305,7 +305,7 @@ public class ImGroupRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
// 断言：插入 2 条 + 推 1503 给 owner（每条 1 帧）共 2 帧
|
||||
ArgumentCaptor<ImGroupRequestDO> captor = ArgumentCaptor.forClass(ImGroupRequestDO.class);
|
||||
verify(groupRequestMapper, times(2)).insert(captor.capture());
|
||||
verify(websocketService, times(2)).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService, times(2)).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
// 断言：每条记录 inviterUserId=1 + addSource=INVITE，避免审批通过后回写群成员留痕的来源为空 / 脏带旧值
|
||||
Collection<ImGroupRequestDO> inserted = captor.getAllValues();
|
||||
assertEquals(2, inserted.size());
|
||||
@ -337,7 +337,7 @@ public class ImGroupRequestServiceImplTest extends BaseMockitoUnitTest {
|
||||
// 断言：复用并重置旧邀请申请
|
||||
verify(groupRequestMapper).updateInviteByIdReset(eq(50L), eq(1L),
|
||||
eq(ImGroupAddSourceEnum.INVITE.getSource()), any());
|
||||
verify(websocketService).sendPrivateMessageAsync(eq(99L), any(ImPrivateMessageDTO.class));
|
||||
verify(websocketService).sendNotificationAsync(eq(99L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
private AdminUserRespDTO buildUser(Long id, String nickname) {
|
||||
|
||||
@ -342,7 +342,7 @@ public class ImGroupServiceImplTest extends BaseMockitoUnitTest {
|
||||
verify(groupMemberService).addGroupMembers(eq(10L), anyCollection(),
|
||||
eq(ImGroupAddSourceEnum.INVITE.getSource()), eq(1L));
|
||||
verify(groupRequestService, never()).createInviteRequestList(anyLong(), anyLong(), anyCollection());
|
||||
verify(webSocketService, never()).sendGroupMessageAsync(anyCollection(), any(ImGroupMessageDTO.class));
|
||||
verify(webSocketService, never()).sendNotificationAsync(anyCollection(), anyInt(), anyInt(), any());
|
||||
ArgumentCaptor<ImGroupMessageSendDTO> dtoCaptor = ArgumentCaptor.forClass(ImGroupMessageSendDTO.class);
|
||||
verify(groupMessageService).sendGroupMessage(eq(1L), anyCollection(), dtoCaptor.capture());
|
||||
assertEquals(ImContentTypeEnum.GROUP_MEMBER_INVITE.getType(), dtoCaptor.getValue().getType());
|
||||
@ -526,7 +526,7 @@ public class ImGroupServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
// 断言：不会触发添加、不会推送
|
||||
verify(groupMemberService, never()).addGroupMembers(anyLong(), anyCollection());
|
||||
verify(webSocketService, never()).sendGroupMessageAsync(anyCollection(), any(ImGroupMessageDTO.class));
|
||||
verify(webSocketService, never()).sendNotificationAsync(anyCollection(), anyInt(), anyInt(), any());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -88,7 +88,7 @@ public class ImChannelMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
channelMessageService.readChannelMessages(1L, 10L, 100L);
|
||||
|
||||
// 断言：读位置未前进 → 不推 READ 事件
|
||||
verify(webSocketService, never()).sendChannelMessageAsync(anyLong(), any());
|
||||
verify(webSocketService, never()).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -116,9 +116,16 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertEquals(ImMessageReceiptStatusEnum.PENDING.getStatus(), result.getReceiptStatus());
|
||||
|
||||
// 验证推送给 3 个群成员（含发送者自己，用于多端同步）
|
||||
verify(imWebSocketService).sendGroupMessageAsync(argThat((Collection<Long> ids) ->
|
||||
ArgumentCaptor<ImGroupMessageNotification> payloadCaptor =
|
||||
ArgumentCaptor.forClass(ImGroupMessageNotification.class);
|
||||
verify(imWebSocketService).sendNotificationAsync(argThat((Collection<Long> ids) ->
|
||||
ids.size() == 3 && ids.contains(1L) && ids.contains(2L) && ids.contains(3L)),
|
||||
any(ImGroupMessageDTO.class));
|
||||
eq(ImConversationTypeEnum.GROUP.getType()), eq(ImContentTypeEnum.TEXT.getType()),
|
||||
payloadCaptor.capture());
|
||||
ImGroupMessageNotification payload = payloadCaptor.getValue();
|
||||
assertEquals(ImMessageReceiptStatusEnum.PENDING.getStatus(), payload.getReceiptStatus());
|
||||
assertEquals(0, payload.getReadCount());
|
||||
assertEquals(List.of(1L, 2L, 3L), payload.getReceiverUserIds());
|
||||
}
|
||||
}
|
||||
|
||||
@ -152,9 +159,9 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
// 断言：固化快照必含发送者，推送目标补回发送者 = {1,2,3}
|
||||
assertTrue(result.getReceiverUserIds().contains(1L), "发送者必须在 receiver_user_ids 快照内");
|
||||
verify(imWebSocketService).sendGroupMessageAsync(argThat((Collection<Long> ids) ->
|
||||
verify(imWebSocketService).sendNotificationAsync(argThat((Collection<Long> ids) ->
|
||||
ids.size() == 3 && ids.contains(1L) && ids.contains(2L) && ids.contains(3L)),
|
||||
any(ImGroupMessageDTO.class));
|
||||
anyInt(), anyInt(), any());
|
||||
}
|
||||
}
|
||||
|
||||
@ -211,7 +218,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
// 被引用原消息：定向给 {1,2}（对发送人 1 可见，但不覆盖广播受众 {1,2,3}）
|
||||
ImGroupMessageDO original = ImGroupMessageDO.builder().id(500L).groupId(10L).senderId(2L)
|
||||
.status(ImMessageStatusEnum.NORMAL.getStatus()).receiverUserIds(List.of(1L, 2L))
|
||||
.type(ImMessageTypeEnum.TEXT.getType()).content("{\"content\":\"密\"}")
|
||||
.type(ImContentTypeEnum.TEXT.getType()).content("{\"content\":\"密\"}")
|
||||
.sendTime(LocalDateTime.now()).build();
|
||||
when(groupMessageMapper.selectById(500L)).thenReturn(original);
|
||||
|
||||
@ -237,7 +244,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
when(groupMemberService.getActiveGroupMemberUserIdsByGroupId(10L)).thenReturn(List.of(1L, 2L, 3L));
|
||||
ImGroupMessageDO original = ImGroupMessageDO.builder().id(500L).groupId(10L).senderId(2L)
|
||||
.status(ImMessageStatusEnum.NORMAL.getStatus()).receiverUserIds(List.of(1L, 2L, 3L))
|
||||
.type(ImMessageTypeEnum.TEXT.getType()).content("{\"content\":\"公开\"}")
|
||||
.type(ImContentTypeEnum.TEXT.getType()).content("{\"content\":\"公开\"}")
|
||||
.sendTime(LocalDateTime.now()).build();
|
||||
when(groupMessageMapper.selectById(500L)).thenReturn(original);
|
||||
when(groupMessageMapper.insert(any(ImGroupMessageDO.class))).thenAnswer(invocation -> {
|
||||
@ -390,9 +397,9 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
verify(groupMessageMapper).updateById(any(ImGroupMessageDO.class));
|
||||
verify(groupMessageMapper).insert(any(ImGroupMessageDO.class));
|
||||
// 验证：给 2 个活跃成员推送撤回提示
|
||||
verify(imWebSocketService).sendGroupMessageAsync(argThat((Collection<Long> ids) ->
|
||||
verify(imWebSocketService).sendNotificationAsync(argThat((Collection<Long> ids) ->
|
||||
ids.size() == 2 && ids.contains(1L) && ids.contains(2L)),
|
||||
any(ImGroupMessageDTO.class));
|
||||
anyInt(), anyInt(), any());
|
||||
}
|
||||
}
|
||||
|
||||
@ -588,7 +595,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertEquals(MESSAGE_SENSITIVE_WORD_BLOCKED.getCode(), exception.getCode());
|
||||
// 断言：不入库、不推送
|
||||
verify(groupMessageMapper, never()).insert(any(ImGroupMessageDO.class));
|
||||
verify(imWebSocketService, never()).sendGroupMessageAsync(anyCollection(), any(ImGroupMessageDTO.class));
|
||||
verify(imWebSocketService, never()).sendNotificationAsync(anyCollection(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
// ========== 撤回：补充边界 ==========
|
||||
@ -649,7 +656,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
// 断言：已读位置更新 + READ 事件
|
||||
verify(conversationReadService).updateConversationReadPosition(1L, ImConversationTypeEnum.GROUP.getType(), 10L, 100L);
|
||||
verify(imWebSocketService).sendGroupMessageAsync(eq(1L), any(ImGroupMessageDTO.class));
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(1L), anyInt(), anyInt(), any());
|
||||
}
|
||||
}
|
||||
|
||||
@ -664,7 +671,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertEquals(MESSAGE_GROUP_READ_DISABLED.getCode(), exception.getCode());
|
||||
// 断言：已读位置不写、不推送
|
||||
verify(conversationReadService, never()).updateConversationReadPosition(anyLong(), anyInt(), anyLong(), anyLong());
|
||||
verify(imWebSocketService, never()).sendGroupMessageAsync(anyLong(), any(ImGroupMessageDTO.class));
|
||||
verify(imWebSocketService, never()).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -726,7 +733,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
// 断言：调用了 CAS 更新但读位置未前进 → 不推送
|
||||
verify(conversationReadService).updateConversationReadPosition(1L, ImConversationTypeEnum.GROUP.getType(), 10L, 100L);
|
||||
verify(imWebSocketService, never()).sendGroupMessageAsync(anyLong(), any(ImGroupMessageDTO.class));
|
||||
verify(imWebSocketService, never()).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -787,7 +794,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertEquals(100L, captor.getValue().getId());
|
||||
assertEquals(ImMessageReceiptStatusEnum.DONE.getStatus(), captor.getValue().getReceiptStatus());
|
||||
// 断言：给消息发送方（5）推送 RECEIPT 事件
|
||||
verify(imWebSocketService).sendGroupMessageAsync(eq(5L), any(ImGroupMessageDTO.class));
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(5L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -824,7 +831,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
// 断言：回执状态保持 PENDING，不更新 DB
|
||||
verify(groupMessageMapper, never()).updateById(any(ImGroupMessageDO.class));
|
||||
// 仍然推送 RECEIPT 事件给发送方（携带已读人数）
|
||||
verify(imWebSocketService).sendGroupMessageAsync(eq(5L), any(ImGroupMessageDTO.class));
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(5L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
// ========== sendGroupMessage(senderId, dto)：receiverUserIds 定向过滤 ==========
|
||||
@ -833,17 +840,17 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
public void testSendGroupMessage_noReceiverUserIds_broadcastsToAllMembers() {
|
||||
// 准备：无定向 → 应发给所有 ENABLE 成员（含发送者自己，多端同步）
|
||||
ImGroupMessageSendDTO dto = new ImGroupMessageSendDTO()
|
||||
.setGroupId(10L).setType(ImMessageTypeEnum.RECALL.getType())
|
||||
.setGroupId(10L).setType(ImContentTypeEnum.RECALL.getType())
|
||||
.setContent("{\"messageId\":1}");
|
||||
when(groupMemberService.getActiveGroupMemberUserIdsByGroupId(10L))
|
||||
.thenReturn(List.of(1L, 2L, 3L, 4L));
|
||||
|
||||
groupMessageService.sendGroupMessage(1L, dto);
|
||||
|
||||
verify(imWebSocketService).sendGroupMessageAsync(argThat((Collection<Long> ids) ->
|
||||
verify(imWebSocketService).sendNotificationAsync(argThat((Collection<Long> ids) ->
|
||||
ids.size() == 4 && ids.contains(1L) && ids.contains(2L)
|
||||
&& ids.contains(3L) && ids.contains(4L)),
|
||||
any(ImGroupMessageDTO.class));
|
||||
anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -851,7 +858,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
// 准备：定向给 {2,3}，发送者 1；群成员 {1,2,3,4}
|
||||
// 预期推送目标：{1,2,3}（发送者自己多端同步 + 定向列表）；4 被过滤
|
||||
ImGroupMessageSendDTO dto = new ImGroupMessageSendDTO()
|
||||
.setGroupId(10L).setType(ImMessageTypeEnum.RECALL.getType())
|
||||
.setGroupId(10L).setType(ImContentTypeEnum.RECALL.getType())
|
||||
.setContent("{\"messageId\":1}")
|
||||
.setReceiverUserIds(List.of(2L, 3L));
|
||||
when(groupMemberService.getActiveGroupMemberUserIdsByGroupId(10L))
|
||||
@ -859,10 +866,10 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
groupMessageService.sendGroupMessage(1L, dto);
|
||||
|
||||
verify(imWebSocketService).sendGroupMessageAsync(argThat((Collection<Long> ids) ->
|
||||
verify(imWebSocketService).sendNotificationAsync(argThat((Collection<Long> ids) ->
|
||||
ids.size() == 3 && ids.contains(1L) && ids.contains(2L)
|
||||
&& ids.contains(3L) && !ids.contains(4L)),
|
||||
any(ImGroupMessageDTO.class));
|
||||
anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -870,7 +877,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
// 边界：发送者不在群成员缓存里（理论上应先被 validateMemberInGroup 挡住，这里纯防御）
|
||||
// 预期：定向用户 + 发送者自己（始终纳入快照保证多端同步）；非定向成员 4 被过滤
|
||||
ImGroupMessageSendDTO dto = new ImGroupMessageSendDTO()
|
||||
.setGroupId(10L).setType(ImMessageTypeEnum.RECALL.getType())
|
||||
.setGroupId(10L).setType(ImContentTypeEnum.RECALL.getType())
|
||||
.setContent("{\"messageId\":1}")
|
||||
.setReceiverUserIds(List.of(2L, 3L));
|
||||
when(groupMemberService.getActiveGroupMemberUserIdsByGroupId(10L))
|
||||
@ -878,10 +885,10 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
groupMessageService.sendGroupMessage(99L, dto);
|
||||
|
||||
verify(imWebSocketService).sendGroupMessageAsync(argThat((Collection<Long> ids) ->
|
||||
verify(imWebSocketService).sendNotificationAsync(argThat((Collection<Long> ids) ->
|
||||
ids.size() == 3 && ids.contains(2L) && ids.contains(3L)
|
||||
&& ids.contains(99L) && !ids.contains(4L)),
|
||||
any(ImGroupMessageDTO.class));
|
||||
anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
// ========== sendGroupMessage(senderId, dto)：helper 行为 ==========
|
||||
@ -935,7 +942,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
groupMessageService.sendGroupMessage(1L, dto);
|
||||
|
||||
verify(groupMessageMapper, never()).insert(any(ImGroupMessageDO.class));
|
||||
verify(imWebSocketService).sendGroupMessageAsync(anyCollection(), any(ImGroupMessageDTO.class));
|
||||
verify(imWebSocketService).sendNotificationAsync(anyCollection(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -949,7 +956,7 @@ public class ImGroupMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
// 断言：不读取 active members，按调用方 targets 推送
|
||||
verify(groupMemberService, never()).getActiveGroupMemberUserIdsByGroupId(anyLong());
|
||||
verify(imWebSocketService).sendGroupMessageAsync(eq(targets), any(ImGroupMessageDTO.class));
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(targets), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
// ========== getGroupMessageList ==========
|
||||
|
||||
@ -7,6 +7,7 @@ import cn.iocoder.yudao.module.im.controller.admin.message.vo.privates.ImPrivate
|
||||
import cn.iocoder.yudao.module.im.controller.admin.message.vo.privates.ImPrivateMessageSendReqVO;
|
||||
import cn.iocoder.yudao.module.im.dal.dataobject.message.ImPrivateMessageDO;
|
||||
import cn.iocoder.yudao.module.im.dal.mysql.message.ImPrivateMessageMapper;
|
||||
import cn.iocoder.yudao.module.im.enums.ImConversationTypeEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.message.ImMessageReceiptStatusEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.message.ImMessageStatusEnum;
|
||||
import cn.iocoder.yudao.module.im.enums.ImContentTypeEnum;
|
||||
@ -16,8 +17,10 @@ import cn.iocoder.yudao.module.im.service.friend.ImFriendService;
|
||||
import cn.iocoder.yudao.module.im.service.sensitiveword.ImSensitiveWordService;
|
||||
import cn.iocoder.yudao.module.im.service.message.dto.ImPrivateMessageSendDTO;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.ImWebSocketService;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.dto.ImPrivateMessageDTO;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.dto.message.RecallMessage;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImMessageReadNotification;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImMessageReceiptNotification;
|
||||
import cn.iocoder.yudao.module.im.service.websocket.notification.message.ImPrivateMessageNotification;
|
||||
import cn.iocoder.yudao.module.im.dal.dataobject.message.content.RecallMessage;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.InjectMocks;
|
||||
@ -61,7 +64,7 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
ImPrivateMessageSendReqVO reqVO = new ImPrivateMessageSendReqVO();
|
||||
reqVO.setClientMessageId("test-uuid-001");
|
||||
reqVO.setReceiverId(2L);
|
||||
reqVO.setType(ImMessageTypeEnum.TEXT.getType());
|
||||
reqVO.setType(ImContentTypeEnum.TEXT.getType());
|
||||
reqVO.setContent("{\"content\":\"你好\"}");
|
||||
return reqVO;
|
||||
}
|
||||
@ -87,7 +90,7 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertNotNull(result);
|
||||
assertEquals(1L, result.getSenderId());
|
||||
assertEquals(2L, result.getReceiverId());
|
||||
assertEquals(ImMessageTypeEnum.TEXT.getType(), result.getType());
|
||||
assertEquals(ImContentTypeEnum.TEXT.getType(), result.getType());
|
||||
assertEquals(ImMessageStatusEnum.NORMAL.getStatus(), result.getStatus());
|
||||
assertEquals(ImMessageReceiptStatusEnum.PENDING.getStatus(), result.getReceiptStatus(),
|
||||
"用户私聊消息默认需要回执（PENDING）");
|
||||
@ -98,8 +101,8 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
verify(sensitiveWordService).validateText(reqVO.getContent());
|
||||
verify(privateMessageMapper).insert(any(ImPrivateMessageDO.class));
|
||||
// 验证推送给接收方和发送方
|
||||
verify(imWebSocketService).sendPrivateMessageAsync(eq(2L), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService).sendPrivateMessageAsync(eq(1L), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(2L), anyInt(), anyInt(), any());
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(1L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -187,22 +190,25 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
// 断言：发送了 READ + RECEIPT 事件，payload 字段正确
|
||||
ArgumentCaptor<Long> userCaptor = ArgumentCaptor.forClass(Long.class);
|
||||
ArgumentCaptor<ImPrivateMessageDTO> contentCaptor = ArgumentCaptor.forClass(ImPrivateMessageDTO.class);
|
||||
verify(imWebSocketService, times(2)).sendPrivateMessageAsync(
|
||||
userCaptor.capture(), contentCaptor.capture());
|
||||
ArgumentCaptor<Integer> contentTypeCaptor = ArgumentCaptor.forClass(Integer.class);
|
||||
ArgumentCaptor<Object> payloadCaptor = ArgumentCaptor.forClass(Object.class);
|
||||
verify(imWebSocketService, times(2)).sendNotificationAsync(
|
||||
userCaptor.capture(), eq(ImConversationTypeEnum.PRIVATE.getType()),
|
||||
contentTypeCaptor.capture(), payloadCaptor.capture());
|
||||
|
||||
// 第一次：发给自己的 READ 事件
|
||||
assertEquals(1L, userCaptor.getAllValues().get(0));
|
||||
ImPrivateMessageDTO readPayload = contentCaptor.getAllValues().get(0);
|
||||
assertEquals(ImMessageTypeEnum.READ.getType(), readPayload.getType());
|
||||
assertEquals(ImContentTypeEnum.READ.getType(), contentTypeCaptor.getAllValues().get(0));
|
||||
ImMessageReadNotification readPayload = (ImMessageReadNotification) payloadCaptor.getAllValues().get(0);
|
||||
assertEquals(1L, readPayload.getSenderId());
|
||||
assertEquals(2L, readPayload.getReceiverId());
|
||||
assertEquals(5L, readPayload.getId(), "READ id 应为前端上报的 messageId");
|
||||
|
||||
// 第二次：发给对方的 RECEIPT 事件
|
||||
assertEquals(2L, userCaptor.getAllValues().get(1));
|
||||
ImPrivateMessageDTO receiptPayload = contentCaptor.getAllValues().get(1);
|
||||
assertEquals(ImMessageTypeEnum.RECEIPT.getType(), receiptPayload.getType());
|
||||
assertEquals(ImContentTypeEnum.RECEIPT.getType(), contentTypeCaptor.getAllValues().get(1));
|
||||
ImMessageReceiptNotification receiptPayload =
|
||||
(ImMessageReceiptNotification) payloadCaptor.getAllValues().get(1);
|
||||
assertEquals(5L, receiptPayload.getId(), "RECEIPT id 应为前端上报的 messageId");
|
||||
}
|
||||
|
||||
@ -228,7 +234,7 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
verify(privateMessageMapper).updateById(any(ImPrivateMessageDO.class));
|
||||
verify(privateMessageMapper).insert(any(ImPrivateMessageDO.class));
|
||||
// 验证推送了消息（给接收方和发送方）
|
||||
verify(imWebSocketService, times(2)).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService, times(2)).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -304,7 +310,7 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertEquals(MESSAGE_SENSITIVE_WORD_BLOCKED.getCode(), exception.getCode());
|
||||
// 断言：不入库、不推送
|
||||
verify(privateMessageMapper, never()).insert(any(ImPrivateMessageDO.class));
|
||||
verify(imWebSocketService, never()).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService, never()).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -319,7 +325,7 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
// 断言：不更新消息状态、不推送
|
||||
verify(privateMessageMapper, never()).updateBySenderIdAndReceiverIdAndIdLeAndReceiptStatus(
|
||||
anyLong(), anyLong(), anyLong(), anyInt(), any(ImPrivateMessageDO.class));
|
||||
verify(imWebSocketService, never()).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService, never()).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -332,7 +338,7 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
privateMessageService.readPrivateMessages(1L, 2L, 5L);
|
||||
|
||||
// 断言：读位置没前进，不推送 READ / RECEIPT
|
||||
verify(imWebSocketService, never()).sendPrivateMessageAsync(anyLong(), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService, never()).sendNotificationAsync(anyLong(), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
// ========== getMaxReadMessageId 测试 ==========
|
||||
@ -397,8 +403,8 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
assertNotNull(message.getClientMessageId());
|
||||
assertNotNull(message.getSendTime());
|
||||
// 断言：sender + receiver 双端推送
|
||||
verify(imWebSocketService).sendPrivateMessageAsync(eq(1L), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService).sendPrivateMessageAsync(eq(2L), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(1L), anyInt(), anyInt(), any());
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(2L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
@Test
|
||||
@ -410,8 +416,8 @@ public class ImPrivateMessageServiceImplTest extends BaseMockitoUnitTest {
|
||||
privateMessageService.sendPrivateMessage(1L, dto);
|
||||
|
||||
verify(privateMessageMapper, never()).insert(any(ImPrivateMessageDO.class));
|
||||
verify(imWebSocketService).sendPrivateMessageAsync(eq(1L), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService, never()).sendPrivateMessageAsync(eq(2L), any(ImPrivateMessageDTO.class));
|
||||
verify(imWebSocketService).sendNotificationAsync(eq(1L), anyInt(), anyInt(), any());
|
||||
verify(imWebSocketService, never()).sendNotificationAsync(eq(2L), anyInt(), anyInt(), any());
|
||||
}
|
||||
|
||||
// ========== getPrivateMessageList ==========
|
||||
|
||||
@ -143,7 +143,7 @@ public class ImRtcCallServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
// 断言：成功 1 个；NO_ANSWER 信令推到主叫；不触发 endSession
|
||||
assertEquals(1, result);
|
||||
verify(webSocketService).sendPrivateMessageAsync(eq(200L), any(ImPrivateMessageDTO.class));
|
||||
verify(webSocketService).sendNotificationAsync(eq(200L), anyInt(), anyInt(), any());
|
||||
verify(rtcCallMapper, never()).updateByIdAndStatusIn(any(), anyCollection(), any());
|
||||
}
|
||||
|
||||
@ -238,7 +238,7 @@ public class ImRtcCallServiceImplTest extends BaseMockitoUnitTest {
|
||||
rtcCallService.noAnswerCallCheck(100L, "r1");
|
||||
|
||||
// 断言：NO_ANSWER 信令推到主叫 200L；不触发 endSession
|
||||
verify(webSocketService).sendPrivateMessageAsync(eq(200L), any(ImPrivateMessageDTO.class));
|
||||
verify(webSocketService).sendNotificationAsync(eq(200L), anyInt(), anyInt(), any());
|
||||
verify(rtcCallMapper, never()).updateByIdAndStatusIn(any(), anyCollection(), any());
|
||||
}
|
||||
|
||||
|
||||
@ -57,10 +57,12 @@ public class ImWebSocketServiceImplTest extends BaseMockitoUnitTest {
|
||||
.thenReturn(imWebSocketService);
|
||||
|
||||
// 准备
|
||||
ImPrivateMessageDTO dto = new ImPrivateMessageDTO().setSenderId(1L).setReceiverId(2L);
|
||||
ImPrivateMessageNotification dto = new ImPrivateMessageNotification().setSenderId(1L).setReceiverId(2L)
|
||||
.setType(ImContentTypeEnum.TEXT.getType());
|
||||
|
||||
// 调用：无事务，应立即发送
|
||||
imWebSocketService.sendPrivateMessageAsync(2L, dto);
|
||||
imWebSocketService.sendNotificationAsync(2L, ImConversationTypeEnum.PRIVATE.getType(),
|
||||
ImContentTypeEnum.TEXT.getType(), dto);
|
||||
|
||||
// 断言
|
||||
verify(webSocketSenderApi).sendObject(
|
||||
@ -79,10 +81,12 @@ public class ImWebSocketServiceImplTest extends BaseMockitoUnitTest {
|
||||
// 准备：开启事务同步
|
||||
TransactionSynchronizationManager.initSynchronization();
|
||||
try {
|
||||
ImPrivateMessageDTO dto = new ImPrivateMessageDTO().setSenderId(1L).setReceiverId(2L);
|
||||
ImPrivateMessageNotification dto = new ImPrivateMessageNotification().setSenderId(1L).setReceiverId(2L)
|
||||
.setType(ImContentTypeEnum.TEXT.getType());
|
||||
|
||||
// 调用
|
||||
imWebSocketService.sendPrivateMessageAsync(2L, dto);
|
||||
imWebSocketService.sendNotificationAsync(2L, ImConversationTypeEnum.PRIVATE.getType(),
|
||||
ImContentTypeEnum.TEXT.getType(), dto);
|
||||
|
||||
// 断言：事务未提交，未推送
|
||||
verify(webSocketSenderApi, never()).sendObject(anyInt(), anyLong(), anyString(), any());
|
||||
@ -115,8 +119,10 @@ public class ImWebSocketServiceImplTest extends BaseMockitoUnitTest {
|
||||
ImGroupMessageNotification dto = new ImGroupMessageNotification();
|
||||
dto.setGroupId(10L);
|
||||
dto.setSenderId(1L);
|
||||
dto.setType(ImContentTypeEnum.TEXT.getType());
|
||||
|
||||
imWebSocketService.sendGroupMessageAsync(ListUtil.of(1L, 2L, 3L), dto);
|
||||
imWebSocketService.sendNotificationAsync(ListUtil.of(1L, 2L, 3L),
|
||||
ImConversationTypeEnum.GROUP.getType(), ImContentTypeEnum.TEXT.getType(), dto);
|
||||
|
||||
verify(webSocketSenderApi).sendObject(
|
||||
eq(UserTypeEnum.ADMIN.getValue()), eq(1L), eq(ImNotificationWebSocketDTO.TYPE),
|
||||
@ -145,7 +151,8 @@ public class ImWebSocketServiceImplTest extends BaseMockitoUnitTest {
|
||||
doThrow(new RuntimeException("user offline"))
|
||||
.when(webSocketSenderApi).sendObject(anyInt(), eq(1L), anyString(), any());
|
||||
|
||||
imWebSocketService.sendGroupMessageAsync(ListUtil.of(1L, 2L, 3L), dto);
|
||||
imWebSocketService.sendNotificationAsync(ListUtil.of(1L, 2L, 3L),
|
||||
ImConversationTypeEnum.GROUP.getType(), ImContentTypeEnum.TEXT.getType(), dto);
|
||||
|
||||
// 2L 和 3L 也都被推送
|
||||
verify(webSocketSenderApi).sendObject(anyInt(), eq(2L), anyString(), any());
|
||||
@ -154,8 +161,12 @@ public class ImWebSocketServiceImplTest extends BaseMockitoUnitTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDoSendGroupMessage_emptyUserIds_noSend() {
|
||||
ImGroupMessageDTO dto = new ImGroupMessageDTO().setGroupId(10L);
|
||||
public void testDoSendNotification_emptyUserIds_noSend() {
|
||||
ImGroupMessageNotification dto = new ImGroupMessageNotification();
|
||||
dto.setGroupId(10L);
|
||||
dto.setType(ImContentTypeEnum.TEXT.getType());
|
||||
ImNotificationWebSocketDTO notification = buildNotification(ImConversationTypeEnum.GROUP,
|
||||
ImContentTypeEnum.TEXT, dto);
|
||||
|
||||
imWebSocketService.doSendNotification(Collections.emptyList(), notification);
|
||||
imWebSocketService.doSendNotification(null, notification);
|
||||
@ -164,8 +175,12 @@ public class ImWebSocketServiceImplTest extends BaseMockitoUnitTest {
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDoSendGroupMessage_distinctUserIds() {
|
||||
ImGroupMessageDTO dto = new ImGroupMessageDTO().setGroupId(10L);
|
||||
public void testDoSendNotification_distinctUserIds() {
|
||||
ImGroupMessageNotification dto = new ImGroupMessageNotification();
|
||||
dto.setGroupId(10L);
|
||||
dto.setType(ImContentTypeEnum.TEXT.getType());
|
||||
ImNotificationWebSocketDTO notification = buildNotification(ImConversationTypeEnum.GROUP,
|
||||
ImContentTypeEnum.TEXT, dto);
|
||||
|
||||
imWebSocketService.doSendNotification(Arrays.asList(1L, 2L, 1L, null), notification);
|
||||
|
||||
@ -187,12 +202,14 @@ public class ImWebSocketServiceImplTest extends BaseMockitoUnitTest {
|
||||
.thenReturn(imWebSocketService);
|
||||
|
||||
// 准备：sender 抛异常
|
||||
ImPrivateMessageDTO dto = new ImPrivateMessageDTO().setSenderId(1L).setReceiverId(2L);
|
||||
ImPrivateMessageNotification dto = new ImPrivateMessageNotification().setSenderId(1L).setReceiverId(2L)
|
||||
.setType(ImContentTypeEnum.TEXT.getType());
|
||||
doThrow(new RuntimeException("user offline"))
|
||||
.when(webSocketSenderApi).sendObject(anyInt(), anyLong(), anyString(), any());
|
||||
|
||||
// 调用：异常应被吞掉，不向上抛
|
||||
imWebSocketService.sendPrivateMessageAsync(2L, dto);
|
||||
imWebSocketService.sendNotificationAsync(2L, ImConversationTypeEnum.PRIVATE.getType(),
|
||||
ImContentTypeEnum.TEXT.getType(), dto);
|
||||
|
||||
verify(webSocketSenderApi).sendObject(anyInt(), eq(2L), anyString(), any());
|
||||
}
|
||||
@ -206,12 +223,34 @@ public class ImWebSocketServiceImplTest extends BaseMockitoUnitTest {
|
||||
|
||||
ImGroupMessageNotification dto = new ImGroupMessageNotification();
|
||||
dto.setGroupId(10L);
|
||||
dto.setType(ImContentTypeEnum.TEXT.getType());
|
||||
|
||||
imWebSocketService.sendGroupMessageAsync(42L, dto);
|
||||
imWebSocketService.sendNotificationAsync(42L, ImConversationTypeEnum.GROUP.getType(),
|
||||
ImContentTypeEnum.TEXT.getType(), dto);
|
||||
|
||||
verify(webSocketSenderApi).sendObject(
|
||||
eq(UserTypeEnum.ADMIN.getValue()), eq(42L), eq(ImGroupMessageDTO.TYPE), eq(dto));
|
||||
eq(UserTypeEnum.ADMIN.getValue()), eq(42L), eq(ImNotificationWebSocketDTO.TYPE),
|
||||
argThat(actual -> isNotification(actual, ImConversationTypeEnum.GROUP,
|
||||
ImContentTypeEnum.TEXT, dto)));
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean isNotification(Object object, ImConversationTypeEnum conversationType,
|
||||
ImContentTypeEnum contentType, Object payload) {
|
||||
if (!(object instanceof ImNotificationWebSocketDTO notification)) {
|
||||
return false;
|
||||
}
|
||||
return conversationType.getType().equals(notification.getConversationType())
|
||||
&& contentType.getType().equals(notification.getContentType())
|
||||
&& payload == notification.getPayload();
|
||||
}
|
||||
|
||||
private static ImNotificationWebSocketDTO buildNotification(ImConversationTypeEnum conversationType,
|
||||
ImContentTypeEnum contentType, Object payload) {
|
||||
return new ImNotificationWebSocketDTO()
|
||||
.setConversationType(conversationType.getType())
|
||||
.setContentType(contentType.getType())
|
||||
.setPayload(payload);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user