|
|
@ -143,115 +143,6 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro |
|
|
return msgInfo; |
|
|
return msgInfo; |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@Override |
|
|
|
|
|
public void pullOfflineMessage(Long minId) { |
|
|
|
|
|
UserSession session = SessionContext.getSession(); |
|
|
|
|
|
if (!imClient.isOnline(session.getUserId())) { |
|
|
|
|
|
throw new GlobalException("网络连接失败，无法拉取离线消息"); |
|
|
|
|
|
} |
|
|
|
|
|
// 查询用户加入的群组
|
|
|
|
|
|
List<GroupMember> members = groupMemberService.findByUserId(session.getUserId()); |
|
|
|
|
|
Map<Long, GroupMember> groupMemberMap = CollStreamUtil.toIdentityMap(members, GroupMember::getGroupId); |
|
|
|
|
|
Set<Long> groupIds = groupMemberMap.keySet(); |
|
|
|
|
|
if (CollectionUtil.isEmpty(groupIds)) { |
|
|
|
|
|
// 关闭加载中标志
|
|
|
|
|
|
this.sendLoadingMessage(false, session); |
|
|
|
|
|
return; |
|
|
|
|
|
} |
|
|
|
|
|
// 只拉最近一个月
|
|
|
|
|
|
Date minDate = DateUtils.addMonths(new Date(), -1); |
|
|
|
|
|
LambdaQueryWrapper<GroupMessage> wrapper = Wrappers.lambdaQuery(); |
|
|
|
|
|
wrapper.gt(GroupMessage::getId, minId); |
|
|
|
|
|
wrapper.gt(GroupMessage::getSendTime, minDate); |
|
|
|
|
|
wrapper.in(GroupMessage::getGroupId, groupIds); |
|
|
|
|
|
wrapper.orderByDesc(GroupMessage::getId); |
|
|
|
|
|
wrapper.last("limit 50000"); |
|
|
|
|
|
List<GroupMessage> messages = this.list(wrapper); |
|
|
|
|
|
// 通过群聊对消息进行分组
|
|
|
|
|
|
Map<Long, List<GroupMessage>> messageGroupMap = |
|
|
|
|
|
messages.stream().collect(Collectors.groupingBy(GroupMessage::getGroupId)); |
|
|
|
|
|
// 退群前的消息
|
|
|
|
|
|
List<GroupMember> quitMembers = groupMemberService.findQuitInMonth(session.getUserId()); |
|
|
|
|
|
for (GroupMember quitMember : quitMembers) { |
|
|
|
|
|
wrapper = Wrappers.lambdaQuery(); |
|
|
|
|
|
wrapper.gt(GroupMessage::getId, minId); |
|
|
|
|
|
wrapper.between(GroupMessage::getSendTime, minDate, quitMember.getQuitTime()); |
|
|
|
|
|
wrapper.eq(GroupMessage::getGroupId, quitMember.getGroupId()); |
|
|
|
|
|
wrapper.ne(GroupMessage::getStatus, MessageStatus.RECALL.code()); |
|
|
|
|
|
wrapper.orderByDesc(GroupMessage::getId); |
|
|
|
|
|
List<GroupMessage> groupMessages = this.list(wrapper); |
|
|
|
|
|
messageGroupMap.put(quitMember.getGroupId(), groupMessages); |
|
|
|
|
|
groupMemberMap.put(quitMember.getGroupId(), quitMember); |
|
|
|
|
|
} |
|
|
|
|
|
EXECUTOR.execute(() -> { |
|
|
|
|
|
// 开启加载中标志
|
|
|
|
|
|
this.sendLoadingMessage(true, session); |
|
|
|
|
|
// 推送消息
|
|
|
|
|
|
int sendCount = 0; |
|
|
|
|
|
for (Map.Entry<Long, List<GroupMessage>> entry : messageGroupMap.entrySet()) { |
|
|
|
|
|
Long groupId = entry.getKey(); |
|
|
|
|
|
List<GroupMessage> groupMessages = entry.getValue(); |
|
|
|
|
|
// 第一次拉取时,一个群最多推送3000条消息，防止前端接收能力溢出导致卡顿
|
|
|
|
|
|
List<GroupMessage> sendMessages = groupMessages; |
|
|
|
|
|
if (minId <= 0 && groupMessages.size() > 3000) { |
|
|
|
|
|
sendMessages = groupMessages.subList(0, 3000); |
|
|
|
|
|
} |
|
|
|
|
|
// id从小到大排序
|
|
|
|
|
|
CollectionUtil.reverse(sendMessages); |
|
|
|
|
|
// 填充消息状态
|
|
|
|
|
|
String key = StrUtil.join(":", RedisKey.IM_GROUP_READED_POSITION, groupId); |
|
|
|
|
|
Object o = redisTemplate.opsForHash().get(key, session.getUserId().toString()); |
|
|
|
|
|
long readedMaxId = Objects.isNull(o) ? -1 : Long.parseLong(o.toString()); |
|
|
|
|
|
Map<Object, Object> maxIdMap = null; |
|
|
|
|
|
for (GroupMessage m : sendMessages) { |
|
|
|
|
|
// 推送过程如果用户下线了，则不再推送
|
|
|
|
|
|
if (!imClient.isOnline(session.getUserId(), IMTerminalType.fromCode(session.getTerminal()))) { |
|
|
|
|
|
log.info("用户已下线，停止推送离线群聊消息,用户id:{}", session.getUserId()); |
|
|
|
|
|
return; |
|
|
|
|
|
} |
|
|
|
|
|
// 排除加群之前的消息
|
|
|
|
|
|
GroupMember member = groupMemberMap.get(m.getGroupId()); |
|
|
|
|
|
if (DateUtil.compare(member.getCreatedTime(), m.getSendTime()) > 0) { |
|
|
|
|
|
continue; |
|
|
|
|
|
} |
|
|
|
|
|
// 排除不需要接收的消息
|
|
|
|
|
|
List<String> recvIds = CommaTextUtils.asList(m.getRecvIds()); |
|
|
|
|
|
if (!recvIds.isEmpty() && !recvIds.contains(session.getUserId().toString())) { |
|
|
|
|
|
continue; |
|
|
|
|
|
} |
|
|
|
|
|
// 组装vo
|
|
|
|
|
|
GroupMessageVO vo = BeanUtils.copyProperties(m, GroupMessageVO.class); |
|
|
|
|
|
// 被@用户列表
|
|
|
|
|
|
List<String> atIds = CommaTextUtils.asList(m.getAtUserIds()); |
|
|
|
|
|
vo.setAtUserIds(atIds.stream().map(Long::parseLong).collect(Collectors.toList())); |
|
|
|
|
|
// 填充状态
|
|
|
|
|
|
vo.setStatus(readedMaxId >= m.getId() ? MessageStatus.READED.code() : MessageStatus.PENDING.code()); |
|
|
|
|
|
// 针对回执消息填充已读人数
|
|
|
|
|
|
if (m.getReceipt()) { |
|
|
|
|
|
if (Objects.isNull(maxIdMap)) { |
|
|
|
|
|
maxIdMap = redisTemplate.opsForHash().entries(key); |
|
|
|
|
|
} |
|
|
|
|
|
int count = getReadedUserIds(maxIdMap, m.getId(), m.getSendId()).size(); |
|
|
|
|
|
vo.setReadedCount(count); |
|
|
|
|
|
} |
|
|
|
|
|
// 推送
|
|
|
|
|
|
IMGroupMessage<GroupMessageVO> sendMessage = new IMGroupMessage<>(); |
|
|
|
|
|
sendMessage.setSender(new IMUserInfo(m.getSendId(), IMTerminalType.WEB.code())); |
|
|
|
|
|
sendMessage.setRecvIds(Arrays.asList(session.getUserId())); |
|
|
|
|
|
sendMessage.setRecvTerminals(Arrays.asList(session.getTerminal())); |
|
|
|
|
|
sendMessage.setSendResult(false); |
|
|
|
|
|
sendMessage.setSendToSelf(false); |
|
|
|
|
|
sendMessage.setData(vo); |
|
|
|
|
|
imClient.sendGroupMessage(sendMessage); |
|
|
|
|
|
sendCount++; |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
// 关闭加载中标志
|
|
|
|
|
|
this.sendLoadingMessage(false, session); |
|
|
|
|
|
log.info("拉取离线群聊消息,用户id:{},数量:{}", session.getUserId(), sendCount++); |
|
|
|
|
|
}); |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
public List<GroupMessageVO> loadOffineMessage(Long minId) { |
|
|
public List<GroupMessageVO> loadOffineMessage(Long minId) { |
|
|
UserSession session = SessionContext.getSession(); |
|
|
UserSession session = SessionContext.getSession(); |
|
|
|
xsx
2 months ago