La123123.
/
im-backed
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

多终端同时在线后端改造（开发中）

master
xsx 3 years ago
parent
0bb3f686e0
commit
bf5bb70751
22 changed files with 331 additions and 290 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 15
      im-client/src/main/java/com/bx/imclient/IMClient.java
  2. 162
      im-client/src/main/java/com/bx/imclient/sender/IMSender.java
  3. 5
      im-commom/src/main/java/com/bx/imcommon/contant/RedisKey.java
  4. 43
      im-commom/src/main/java/com/bx/imcommon/model/IMGroupMessage.java
  5. 9
      im-commom/src/main/java/com/bx/imcommon/model/IMPrivateMessage.java
  6. 13
      im-commom/src/main/java/com/bx/imcommon/model/IMRecvInfo.java
  7. 30
      im-commom/src/main/java/com/bx/imcommon/model/IMUserInfo.java
  8. 18
      im-commom/src/main/java/com/bx/imcommon/model/SendResult.java
  9. 2
      im-platform/src/main/java/com/bx/implatform/contant/RedisKey.java
  10. 9
      im-platform/src/main/java/com/bx/implatform/listener/GroupMessageListener.java
  11. 28
      im-platform/src/main/java/com/bx/implatform/listener/PrivateMessageListener.java
  12. 67
      im-platform/src/main/java/com/bx/implatform/service/impl/GroupMessageServiceImpl.java
  13. 21
      im-platform/src/main/java/com/bx/implatform/service/impl/PrivateMessageServiceImpl.java
  14. 22
      im-platform/src/main/java/com/bx/implatform/service/impl/WebrtcServiceImpl.java
  15. 53
      im-server/src/main/java/com/bx/imserver/netty/processor/GroupMessageProcessor.java
  16. 18
      im-server/src/main/java/com/bx/imserver/netty/processor/PrivateMessageProcessor.java
  17. 1
      im-server/src/main/java/com/bx/imserver/task/AbstractPullMessageTask.java
  18. 13
      im-server/src/main/java/com/bx/imserver/task/PullUnreadGroupMessageTask.java
  19. 13
      im-server/src/main/java/com/bx/imserver/task/PullUnreadPrivateMessageTask.java
  20. 10
      im-ui/src/components/chat/ChatPrivateVideo.vue
  21. 52
      im-ui/src/components/chat/ChatVideoAcceptor.vue
  22. 5
      im-ui/src/view/Home.vue

15
im-client/src/main/java/com/bx/imclient/IMClient.java
View File

@ -1,14 +1,12 @@
package com.bx.imclient; package com.bx.imclient;
import com.bx.imclient.listener.MessageListenerMulticaster;
import com.bx.imclient.sender.IMSender; import com.bx.imclient.sender.IMSender;
import com.bx.imcommon.model.GroupMessageInfo; import com.bx.imcommon.model.IMGroupMessage;
import com.bx.imcommon.model.IMPrivateMessage; import com.bx.imcommon.model.IMPrivateMessage;
import com.bx.imcommon.model.PrivateMessageInfo;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Configuration;
import java.util.List;
@Configuration @Configuration
public class IMClient { public class IMClient {
@ -30,18 +28,17 @@ public class IMClient {
* *
* @param message 私有消息 * @param message 私有消息
*/ */
public void sendPrivateMessage(IMPrivateMessage message){ public void sendPrivateMessage(IMPrivateMessage<?> message){
imSender.sendPrivateMessage(message); imSender.sendPrivateMessage(message);
} }
/** /**
* 发送群聊消息（发送结果通过MessageListener接收） * 发送群聊消息（发送结果通过MessageListener接收）
* *
* @param recvIds 群聊用户id列表 * @param message 群聊消息
* @param messageInfo 消息体，将转成json发送到客户端
*/ */
public void sendGroupMessage(List<Long> recvIds, GroupMessageInfo... messageInfo){ public void sendGroupMessage(IMGroupMessage<?> message){
imSender.sendGroupMessage(recvIds,messageInfo); imSender.sendGroupMessage(message);
} }

162
im-client/src/main/java/com/bx/imclient/sender/IMSender.java
View File

@ -22,29 +22,26 @@ public class IMSender {
@Autowired @Autowired
@Qualifier("IMRedisTemplate") @Qualifier("IMRedisTemplate")
private RedisTemplate redisTemplate; private RedisTemplate<String, Object> redisTemplate;
@Autowired @Autowired
private MessageListenerMulticaster listenerMulticaster; private MessageListenerMulticaster listenerMulticaster;
public void sendPrivateMessage(IMPrivateMessage<?> message) { public void sendPrivateMessage(IMPrivateMessage<?> message) {
List<Integer> terminals = message.getRecvTerminals(); for (Integer terminal : message.getRecvTerminals()) {
for (Integer terminal : terminals) {
// 获取对方连接的channelId // 获取对方连接的channelId
String key = String.join(":",RedisKey.IM_USER_SERVER_ID, message.getRecvId().toString(), terminal.toString()); String key = String.join(":",RedisKey.IM_USER_SERVER_ID, message.getRecvId().toString(), terminal.toString());
Integer serverId = (Integer) redisTemplate.opsForValue().get(key); Integer serverId = (Integer) redisTemplate.opsForValue().get(key);
// 如果对方在线，将数据存储至redis，等待拉取推送 // 如果对方在线，将数据存储至redis，等待拉取推送
if (serverId != null) { if (serverId != null) {
IMRecvInfo[] recvInfos = new IMRecvInfo[message.getDatas().size()]; IMRecvInfo[] recvInfos = new IMRecvInfo[message.getDatas().size()];
String sendKey = RedisKey.IM_UNREAD_PRIVATE_QUEUE + serverId; String sendKey = String.join(":",RedisKey.IM_UNREAD_PRIVATE_QUEUE,serverId.toString());
for (int i = 0; i < message.getDatas().size(); i++) { for (int i = 0; i < message.getDatas().size(); i++) {
IMRecvInfo recvInfo = new IMRecvInfo(); IMRecvInfo recvInfo = new IMRecvInfo();
recvInfo.setCmd(IMCmdType.PRIVATE_MESSAGE.code()); recvInfo.setCmd(IMCmdType.PRIVATE_MESSAGE.code());
recvInfo.setRecvTerminal(terminal);
recvInfo.setSendResult(message.getSendResult()); recvInfo.setSendResult(message.getSendResult());
List<Long> recvIds = new LinkedList<>(); recvInfo.setSender(message.getSender());
recvIds.add(message.getRecvId()); recvInfo.setReceivers(Collections.singletonList(new IMUserInfo(message.getRecvId(), terminal)));
recvInfo.setRecvIds(recvIds);
recvInfo.setData(message.getDatas().get(i)); recvInfo.setData(message.getDatas().get(i));
recvInfos[i]=recvInfo; recvInfos[i]=recvInfo;
} }
@ -53,8 +50,8 @@ public class IMSender {
// 回复消息状态 // 回复消息状态
for (int i = 0; i < message.getDatas().size(); i++) { for (int i = 0; i < message.getDatas().size(); i++) {
SendResult result = new SendResult(); SendResult result = new SendResult();
result.setRecvId(message.getRecvId()); result.setSender(message.getSender());
result.setRecvTerminal(terminal); result.setReceiver(new IMUserInfo(message.getRecvId(), terminal));
result.setCode(IMSendCode.NOT_ONLINE.code()); result.setCode(IMSendCode.NOT_ONLINE.code());
result.setData(message.getDatas().get(i)); result.setData(message.getDatas().get(i));
listenerMulticaster.multicast(IMListenerType.PRIVATE_MESSAGE, result); listenerMulticaster.multicast(IMListenerType.PRIVATE_MESSAGE, result);
@ -62,78 +59,95 @@ public class IMSender {
} }
// 推送给自己的其他终端 // 推送给自己的其他终端
if (message.getSendToSelf() && !message.getSendTerminal().equals(terminal)) { if (message.getSendToSelf() && !message.getSender().getTerminal().equals(terminal)) {
// 获取终端连接的channelId // 获取终端连接的channelId
key = String.join(":",RedisKey.IM_USER_SERVER_ID, message.getSendId().toString(), terminal.toString()); key = String.join(":",RedisKey.IM_USER_SERVER_ID, message.getSender().getId().toString(), terminal.toString());
serverId = (Integer) redisTemplate.opsForValue().get(key); serverId = (Integer) redisTemplate.opsForValue().get(key);
// 如果终端在线，将数据存储至redis，等待拉取推送 // 如果终端在线，将数据存储至redis，等待拉取推送
if (serverId != null) { if (serverId != null) {
String sendKey = RedisKey.IM_UNREAD_PRIVATE_QUEUE + serverId; String sendKey = String.join(":",RedisKey.IM_UNREAD_PRIVATE_QUEUE, serverId.toString());
IMRecvInfo[] recvInfos = new IMRecvInfo[message.getDatas().size()]; IMRecvInfo[] recvInfos = new IMRecvInfo[message.getDatas().size()];
for (int i = 0; i < message.getDatas().size(); i++) { for (int i = 0; i < message.getDatas().size(); i++) {
IMRecvInfo recvInfo = new IMRecvInfo(); IMRecvInfo recvInfo = new IMRecvInfo();
recvInfo.setCmd(IMCmdType.PRIVATE_MESSAGE.code()); // 自己的消息不需要回推消息结果
recvInfo.setRecvTerminal(terminal); recvInfo.setSendResult(false);
// 自己的消息不需要回推消息结果 recvInfo.setCmd(IMCmdType.PRIVATE_MESSAGE.code());
recvInfo.setSendResult(false); recvInfo.setSender(message.getSender());
LinkedList<Long> recvIds = new LinkedList<>(); recvInfo.setReceivers(Collections.singletonList(new IMUserInfo(message.getSender().getId(),terminal)));
recvIds.add(message.getSendId()); recvInfo.setData(message.getDatas().get(i));
recvInfo.setRecvIds(recvIds); recvInfos[i]=recvInfo;
recvInfo.setData(message.getDatas().get(i)); }
recvInfos[i]=recvInfo; redisTemplate.opsForList().rightPushAll(sendKey, recvInfos);
} }
redisTemplate.opsForList().rightPushAll(sendKey, recvInfos);
} }
}
} }
} }
public void sendGroupMessage(List<Long> recvIds, GroupMessageInfo... messageInfos) { public void sendGroupMessage(IMGroupMessage<?> message) {
// 根据群聊每个成员所连的IM-server，进行分组 // 根据群聊每个成员所连的IM-server，进行分组
List<Long> offLineIds = Collections.synchronizedList(new LinkedList<Long>()); List<IMUserInfo> offLineUsers = Collections.synchronizedList(new LinkedList<>());
Map<Integer, List<Long>> serverMap = new ConcurrentHashMap<>(); // 格式:map<服务器id,list<接收方>>
recvIds.parallelStream().forEach(id -> { Map<Integer, List<IMUserInfo>> serverMap = new ConcurrentHashMap<>();
String key = RedisKey.IM_USER_SERVER_ID + id; for (Integer terminal : message.getRecvTerminals()) {
Integer serverId = (Integer) redisTemplate.opsForValue().get(key); message.getRecvIds().parallelStream().forEach(id -> {
if (serverId != null) { String key = String.join(":",RedisKey.IM_USER_SERVER_ID, id.toString(),terminal.toString());
// 此处需要加锁，否则list可以会被覆盖 Integer serverId = (Integer)redisTemplate.opsForValue().get(key);
synchronized (serverMap) { if (serverId != null) {
if (serverMap.containsKey(serverId)) { List<IMUserInfo> list = serverMap.computeIfAbsent(serverId,o->Collections.synchronizedList(new LinkedList<>()));
serverMap.get(serverId).add(id); list.add(new IMUserInfo(id,terminal));
} else { } else {
List<Long> list = Collections.synchronizedList(new LinkedList<Long>()); // 加入离线列表
list.add(id); offLineUsers.add(new IMUserInfo(id,terminal));
serverMap.put(serverId, list);
}
} }
} else { });
offLineIds.add(id); }
}
});
// 逐个server发送 // 逐个server发送
// for (Map.Entry<Integer, List<Long>> entry : serverMap.entrySet()) { for (Map.Entry<Integer, List<IMUserInfo>> entry : serverMap.entrySet()) {
// IMRecvInfo[] recvInfos = new IMRecvInfo[messageInfos.length]; IMRecvInfo recvInfo = new IMRecvInfo();
// for (int i = 0; i < messageInfos.length; i++) { recvInfo.setCmd(IMCmdType.GROUP_MESSAGE.code());
// IMRecvInfo<GroupMessageInfo> recvInfo = new IMRecvInfo<>(); recvInfo.setReceivers(new LinkedList<>(entry.getValue()));
// recvInfo.setCmd(IMCmdType.GROUP_MESSAGE.code()); recvInfo.setSender(message.getSender());
// recvInfo.setRecvIds(new LinkedList<>(entry.getValue())); recvInfo.setSendResult(message.getSendResult());
// recvInfo.setData(messageInfos[i]); recvInfo.setData(message.getData());
// recvInfos[i] = recvInfo; // 推送至队列
// } String key = String.join(":",RedisKey.IM_UNREAD_GROUP_QUEUE,entry.getKey().toString());
// String key = RedisKey.IM_UNREAD_GROUP_QUEUE + entry.getKey(); redisTemplate.opsForList().rightPush(key, recvInfo);
// redisTemplate.opsForList().rightPushAll(key, recvInfos); }
// } // 对离线用户回复消息状态
// // 不在线的用户，回复消息状态 if(message.getSendResult()){
// for (GroupMessageInfo messageInfo : messageInfos) { for (IMUserInfo offLineUser : offLineUsers) {
// for (Long id : offLineIds) { SendResult result = new SendResult();
// // 回复消息状态 result.setSender(message.getSender());
// SendResult result = new SendResult(); result.setReceiver(offLineUser);
// result.setMessageInfo(messageInfo); result.setCode(IMSendCode.NOT_ONLINE.code());
// result.setRecvId(id); result.setData(message.getData());
// result.setCode(IMSendCode.NOT_ONLINE); listenerMulticaster.multicast(IMListenerType.GROUP_MESSAGE, result);
// listenerMulticaster.multicast(IMListenerType.GROUP_MESSAGE, result); }
// } }
// }
// 推送给自己的其他终端
if (message.getSendToSelf()) {
for (Integer terminal : message.getRecvTerminals()) {
if(terminal.equals(message.getSender().getTerminal())){
continue;
}
// 获取终端连接的channelId
String key = String.join(":",RedisKey.IM_USER_SERVER_ID, message.getSender().getId().toString(), terminal.toString());
Integer serverId = (Integer) redisTemplate.opsForValue().get(key);
// 如果终端在线，将数据存储至redis，等待拉取推送
if (serverId != null) {
IMRecvInfo recvInfo = new IMRecvInfo();
recvInfo.setCmd(IMCmdType.GROUP_MESSAGE.code());
recvInfo.setSender(message.getSender());
recvInfo.setReceivers(Collections.singletonList(new IMUserInfo(message.getSender().getId(),terminal)));
// 自己的消息不需要回推消息结果
recvInfo.setSendResult(false);
recvInfo.setData(message.getData());
String sendKey = String.join(":",RedisKey.IM_UNREAD_GROUP_QUEUE,serverId.toString());
redisTemplate.opsForList().rightPush(sendKey, recvInfo);
}
}
}
} }
public Boolean isOnline(Long userId) { public Boolean isOnline(Long userId) {

5
im-commom/src/main/java/com/bx/imcommon/contant/RedisKey.java
View File

@ -7,13 +7,12 @@ public class RedisKey {
// 用户ID所连接的IM-server的ID // 用户ID所连接的IM-server的ID
public final static String IM_USER_SERVER_ID = "im:user:server_id"; public final static String IM_USER_SERVER_ID = "im:user:server_id";
// 未读私聊消息队列 // 未读私聊消息队列
public final static String IM_UNREAD_PRIVATE_QUEUE = "im:unread:private:"; public final static String IM_UNREAD_PRIVATE_QUEUE = "im:unread:private";
// 未读群聊消息队列 // 未读群聊消息队列
public final static String IM_UNREAD_GROUP_QUEUE = "im:unread:group:"; public final static String IM_UNREAD_GROUP_QUEUE = "im:unread:group";
// 私聊消息发送结果队列 // 私聊消息发送结果队列
public final static String IM_RESULT_PRIVATE_QUEUE = "im:result:private"; public final static String IM_RESULT_PRIVATE_QUEUE = "im:result:private";
// 群聊消息发送结果队列 // 群聊消息发送结果队列
public final static String IM_RESULT_GROUP_QUEUE = "im:result:group"; public final static String IM_RESULT_GROUP_QUEUE = "im:result:group";
} }

43
im-commom/src/main/java/com/bx/imcommon/model/IMGroupMessage.java
View File

@ -0,0 +1,43 @@
package com.bx.imcommon.model;
import com.bx.imcommon.enums.IMTerminalType;
import lombok.Data;
import java.util.List;
@Data
public class IMGroupMessage<T> {
/**
* 发送方
*/
private IMUserInfo sender;
/**
* 接收者id列表(群成员列表)
*/
private List<Long> recvIds;
/**
* 接收者终端类型,默认全部
*/
private List<Integer> recvTerminals = IMTerminalType.codes();
/**
* 是否发送给自己的其他终端,默认true
*/
private Boolean sendToSelf = true;
/**
* 是否需要回推发送结果,默认true
*/
private Boolean sendResult = true;
/**
* 消息内容
*/
private T data;
}

9
im-commom/src/main/java/com/bx/imcommon/model/IMPrivateMessage.java
View File

@ -10,14 +10,9 @@ import java.util.List;
public class IMPrivateMessage<T> { public class IMPrivateMessage<T> {
/** /**
* 发送者id * 发送方
*/ */
private Long sendId; private IMUserInfo sender;
/**
* 发送者终端类型 IMTerminalType
*/
private Integer sendTerminal;
/** /**
* 接收者id * 接收者id

13
im-commom/src/main/java/com/bx/imcommon/model/IMRecvInfo.java
View File

@ -13,19 +13,14 @@ public class IMRecvInfo {
private Integer cmd; private Integer cmd;
/* /*
* 发送者id * 发送方
*/ */
private Long sendId; private IMUserInfo sender;
/* /*
* 接收终端类型 IMTerminalType * 接收方用户列表
*/ */
private Integer recvTerminal; List<IMUserInfo> receivers;
/*
* 接收者id列表
*/
private List<Long> recvIds;
/* /*
* 是否需要回调发送结果 * 是否需要回调发送结果

30
im-commom/src/main/java/com/bx/imcommon/model/IMUserInfo.java
View File

@ -0,0 +1,30 @@
package com.bx.imcommon.model;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.List;
/**
* @author: 谢绍许
* @date: 2023-09-24 09:23:11
* @version: 1.0
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class IMUserInfo {
/**
* 用户id
*/
private Long id;
/**
* 用户终端类型 IMTerminalType
*/
private Integer terminal;
}

18
im-commom/src/main/java/com/bx/imcommon/model/SendResult.java
View File

@ -1,21 +1,19 @@
package com.bx.imcommon.model; package com.bx.imcommon.model;
import com.bx.imcommon.enums.IMSendCode;
import com.bx.imcommon.enums.IMTerminalType;
import lombok.Data; import lombok.Data;
@Data @Data
public class SendResult<T> { public class SendResult {
/* /**
* 接收者id * 发送方
*/ */
private Long recvId; private IMUserInfo sender;
/* /**
* 接收者终端类型 IMTerminalType * 接收方
*/ */
private Integer recvTerminal; private IMUserInfo receiver;
/* /*
* 发送状态 IMCmdType * 发送状态 IMCmdType
@ -25,6 +23,6 @@ public class SendResult<T> {
/* /*
* 消息内容 * 消息内容
*/ */
private T data; private Object data;
} }

2
im-platform/src/main/java/com/bx/implatform/contant/RedisKey.java
View File

@ -3,7 +3,7 @@ package com.bx.implatform.contant;
public class RedisKey { public class RedisKey {
// 已读群聊消息位置(已读最大id) // 已读群聊消息位置(已读最大id)
public final static String IM_GROUP_READED_POSITION = "im:readed:group:position:"; public final static String IM_GROUP_READED_POSITION = "im:readed:group:position";
// webrtc 会话信息 // webrtc 会话信息
public final static String IM_WEBRTC_SESSION = "im:webrtc:session"; public final static String IM_WEBRTC_SESSION = "im:webrtc:session";
// 缓存前缀 // 缓存前缀

9
im-platform/src/main/java/com/bx/implatform/listener/GroupMessageListener.java
View File

@ -23,14 +23,9 @@ public class GroupMessageListener implements MessageListener {
@Override @Override
public void process(SendResult result){ public void process(SendResult result){
GroupMessageInfo messageInfo = (GroupMessageInfo) result.getData(); GroupMessageInfo messageInfo = (GroupMessageInfo) result.getData();
// 提示类数据不记录
if(messageInfo.getType().equals(MessageType.TIP)){
return;
}
// 保存该用户已拉取的最大消息id // 保存该用户已拉取的最大消息id
if(result.getCode().equals(IMSendCode.SUCCESS)) { if(result.getCode().equals(IMSendCode.SUCCESS.code())) {
String key = RedisKey.IM_GROUP_READED_POSITION + messageInfo.getGroupId() + ":" + result.getRecvId(); String key = String.join(":",RedisKey.IM_GROUP_READED_POSITION,messageInfo.getGroupId().toString(),result.getReceiver().getId().toString());
redisTemplate.opsForValue().set(key, messageInfo.getId()); redisTemplate.opsForValue().set(key, messageInfo.getId());
} }
} }

28
im-platform/src/main/java/com/bx/implatform/listener/PrivateMessageListener.java
View File

@ -35,32 +35,6 @@ public class PrivateMessageListener implements MessageListener {
public void process(SendResult result){ public void process(SendResult result){
PrivateMessageInfo messageInfo = (PrivateMessageInfo) result.getData(); PrivateMessageInfo messageInfo = (PrivateMessageInfo) result.getData();
IMSendCode resultCode = IMSendCode.fromCode(result.getCode()); IMSendCode resultCode = IMSendCode.fromCode(result.getCode());
// 提示类数据不记录
if(messageInfo.getType().equals(MessageType.TIP.code())){
return;
}
// 视频通话信令不记录
if(messageInfo.getType() >= MessageType.RTC_CALL.code() && messageInfo.getType()< MessageType.RTC_CANDIDATE.code()){
// // 通知用户呼叫失败了
// if(messageInfo.getType().equals(MessageType.RTC_CALL.code())
// && !resultCode.equals(IMSendCode.SUCCESS)){
// PrivateMessageInfo msgInfo = new PrivateMessageInfo();
// msgInfo.setRecvId(messageInfo.getSendId());
// msgInfo.setSendId(messageInfo.getRecvId());
// msgInfo.setType(MessageType.RTC_FAILED.code());
// msgInfo.setContent(resultCode.description());
// msgInfo.setSendTime(new Date());
//
// IMPrivateMessage sendMessage = new IMPrivateMessage();
// sendMessage.setSendId(messageInfo.getSendId());
// sendMessage.setRecvId(messageInfo.getRecvId());
// sendMessage.setSendTerminal(result.getRecvTerminal());
// sendMessage.setSendToSelf(false);
// sendMessage.setDatas(Arrays.asList(messageInfo));
// imClient.sendPrivateMessage(sendMessage);
// }
return;
}
// 更新消息状态,这里只处理成功消息，失败的消息继续保持未读状态 // 更新消息状态,这里只处理成功消息，失败的消息继续保持未读状态
if(resultCode.equals(IMSendCode.SUCCESS)){ if(resultCode.equals(IMSendCode.SUCCESS)){
UpdateWrapper<PrivateMessage> updateWrapper = new UpdateWrapper<>(); UpdateWrapper<PrivateMessage> updateWrapper = new UpdateWrapper<>();
@ -68,7 +42,7 @@ public class PrivateMessageListener implements MessageListener {
.eq(PrivateMessage::getStatus, MessageStatus.UNREAD.code()) .eq(PrivateMessage::getStatus, MessageStatus.UNREAD.code())
.set(PrivateMessage::getStatus, MessageStatus.ALREADY_READ.code()); .set(PrivateMessage::getStatus, MessageStatus.ALREADY_READ.code());
privateMessageService.update(updateWrapper); privateMessageService.update(updateWrapper);
log.info("消息已读，消息id:{}，发送者:{},接收者:{}",messageInfo.getId(),messageInfo.getSendId(),messageInfo.getRecvId()); log.info("消息已读，消息id:{}，发送者:{},接收者:{},终端:{}",messageInfo.getId(),result.getSender().getId(),result.getReceiver().getId(),result.getReceiver().getTerminal());
} }
} }

67
im-platform/src/main/java/com/bx/implatform/service/impl/GroupMessageServiceImpl.java
View File

@ -5,6 +5,8 @@ import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.bx.imclient.IMClient; import com.bx.imclient.IMClient;
import com.bx.imcommon.contant.Constant; import com.bx.imcommon.contant.Constant;
import com.bx.imcommon.model.GroupMessageInfo; import com.bx.imcommon.model.GroupMessageInfo;
import com.bx.imcommon.model.IMGroupMessage;
import com.bx.imcommon.model.IMUserInfo;
import com.bx.implatform.contant.RedisKey; import com.bx.implatform.contant.RedisKey;
import com.bx.implatform.entity.Group; import com.bx.implatform.entity.Group;
import com.bx.implatform.entity.GroupMember; import com.bx.implatform.entity.GroupMember;
@ -18,6 +20,7 @@ import com.bx.implatform.service.IGroupMemberService;
import com.bx.implatform.service.IGroupMessageService; import com.bx.implatform.service.IGroupMessageService;
import com.bx.implatform.service.IGroupService; import com.bx.implatform.service.IGroupService;
import com.bx.implatform.session.SessionContext; import com.bx.implatform.session.SessionContext;
import com.bx.implatform.session.UserSession;
import com.bx.implatform.util.BeanUtils; import com.bx.implatform.util.BeanUtils;
import com.bx.implatform.vo.GroupMessageVO; import com.bx.implatform.vo.GroupMessageVO;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -56,7 +59,7 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
*/ */
@Override @Override
public Long sendMessage(GroupMessageVO vo) { public Long sendMessage(GroupMessageVO vo) {
Long userId = SessionContext.getSession().getUserId(); UserSession session = SessionContext.getSession();
Group group = groupService.getById(vo.getGroupId()); Group group = groupService.getById(vo.getGroupId());
if(group == null){ if(group == null){
throw new GlobalException(ResultCode.PROGRAM_ERROR,"群聊不存在"); throw new GlobalException(ResultCode.PROGRAM_ERROR,"群聊不存在");
@ -66,20 +69,24 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
} }
// 判断是否在群里 // 判断是否在群里
List<Long> userIds = groupMemberService.findUserIdsByGroupId(group.getId()); List<Long> userIds = groupMemberService.findUserIdsByGroupId(group.getId());
if(!userIds.contains(userId)){ if(!userIds.contains(session.getUserId())){
throw new GlobalException(ResultCode.PROGRAM_ERROR,"您已不在群聊里面，无法发送消息"); throw new GlobalException(ResultCode.PROGRAM_ERROR,"您已不在群聊里面，无法发送消息");
} }
// 保存消息 // 保存消息
GroupMessage msg = BeanUtils.copyProperties(vo, GroupMessage.class); GroupMessage msg = BeanUtils.copyProperties(vo, GroupMessage.class);
msg.setSendId(userId); msg.setSendId(session.getUserId());
msg.setSendTime(new Date()); msg.setSendTime(new Date());
this.save(msg); this.save(msg);
// 不用发给自己 // 不用发给自己
userIds = userIds.stream().filter(id->userId.equals(id)).collect(Collectors.toList()); userIds = userIds.stream().filter(id->!session.getUserId().equals(id)).collect(Collectors.toList());
// 群发 // 群发
GroupMessageInfo msgInfo = BeanUtils.copyProperties(msg, GroupMessageInfo.class); GroupMessageInfo msgInfo = BeanUtils.copyProperties(msg, GroupMessageInfo.class);
imClient.sendGroupMessage(userIds,msgInfo); IMGroupMessage sendMessage = new IMGroupMessage();
log.info("发送群聊消息，发送id:{},群聊id:{},内容:{}",userId,vo.getGroupId(),vo.getContent()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setRecvIds(userIds);
sendMessage.setData(msgInfo);
imClient.sendGroupMessage(sendMessage);
log.info("发送群聊消息，发送id:{},群聊id:{},内容:{}",session.getUserId(),vo.getGroupId(),vo.getContent());
return msg.getId(); return msg.getId();
} }
@ -93,19 +100,19 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
*/ */
@Override @Override
public void recallMessage(Long id) { public void recallMessage(Long id) {
Long userId = SessionContext.getSession().getUserId(); UserSession session = SessionContext.getSession();
GroupMessage msg = this.getById(id); GroupMessage msg = this.getById(id);
if(msg == null){ if(msg == null){
throw new GlobalException(ResultCode.PROGRAM_ERROR,"消息不存在"); throw new GlobalException(ResultCode.PROGRAM_ERROR,"消息不存在");
} }
if(!msg.getSendId().equals(userId)){ if(!msg.getSendId().equals(session.getUserId())){
throw new GlobalException(ResultCode.PROGRAM_ERROR,"这条消息不是由您发送,无法撤回"); throw new GlobalException(ResultCode.PROGRAM_ERROR,"这条消息不是由您发送,无法撤回");
} }
if(System.currentTimeMillis() - msg.getSendTime().getTime() > Constant.ALLOW_RECALL_SECOND * 1000){ if(System.currentTimeMillis() - msg.getSendTime().getTime() > Constant.ALLOW_RECALL_SECOND * 1000){
throw new GlobalException(ResultCode.PROGRAM_ERROR,"消息已发送超过5分钟，无法撤回"); throw new GlobalException(ResultCode.PROGRAM_ERROR,"消息已发送超过5分钟，无法撤回");
} }
// 判断是否在群里 // 判断是否在群里
GroupMember member = groupMemberService.findByGroupAndUserId(msg.getGroupId(),userId); GroupMember member = groupMemberService.findByGroupAndUserId(msg.getGroupId(),session.getUserId());
if(member == null){ if(member == null){
throw new GlobalException(ResultCode.PROGRAM_ERROR,"您已不在群聊里面，无法撤回消息"); throw new GlobalException(ResultCode.PROGRAM_ERROR,"您已不在群聊里面，无法撤回消息");
} }
@ -115,14 +122,20 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
// 群发 // 群发
List<Long> userIds = groupMemberService.findUserIdsByGroupId(msg.getGroupId()); List<Long> userIds = groupMemberService.findUserIdsByGroupId(msg.getGroupId());
// 不用发给自己 // 不用发给自己
userIds = userIds.stream().filter(uid->userId.equals(uid)).collect(Collectors.toList()); userIds = userIds.stream().filter(uid->!session.getUserId().equals(uid)).collect(Collectors.toList());
GroupMessageInfo msgInfo = BeanUtils.copyProperties(msg, GroupMessageInfo.class); GroupMessageInfo msgInfo = BeanUtils.copyProperties(msg, GroupMessageInfo.class);
msgInfo.setType(MessageType.TIP.code()); msgInfo.setType(MessageType.TIP.code());
String content = String.format("'%s'撤回了一条消息",member.getAliasName()); String content = String.format("'%s'撤回了一条消息",member.getAliasName());
msgInfo.setContent(content); msgInfo.setContent(content);
msgInfo.setSendTime(new Date()); msgInfo.setSendTime(new Date());
imClient.sendGroupMessage(userIds,msgInfo);
log.info("撤回群聊消息，发送id:{},群聊id:{},内容:{}",userId,msg.getGroupId(),msg.getContent()); IMGroupMessage sendMessage = new IMGroupMessage();
sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setRecvIds(userIds);
sendMessage.setData(msgInfo);
sendMessage.setSendResult(false);
imClient.sendGroupMessage(sendMessage);
log.info("撤回群聊消息，发送id:{},群聊id:{},内容:{}",session.getUserId(),msg.getGroupId(),msg.getContent());
} }
@ -133,18 +146,16 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
*/ */
@Override @Override
public void pullUnreadMessage() { public void pullUnreadMessage() {
Long userId = SessionContext.getSession().getUserId(); UserSession session = SessionContext.getSession();
List<Long> recvIds = new LinkedList(); List<GroupMember> members = groupMemberService.findByUserId(session.getUserId());
recvIds.add(userId);
List<GroupMember> members = groupMemberService.findByUserId(userId);
for(GroupMember member:members){ for(GroupMember member:members){
// 获取群聊已读的最大消息id，只推送未读消息 // 获取群聊已读的最大消息id，只推送未读消息
String key = RedisKey.IM_GROUP_READED_POSITION + member.getGroupId()+":"+userId; String key = String.join(":",RedisKey.IM_GROUP_READED_POSITION,member.getGroupId().toString(),session.getUserId().toString());
Integer maxReadedId = (Integer)redisTemplate.opsForValue().get(key); Integer maxReadedId = (Integer)redisTemplate.opsForValue().get(key);
QueryWrapper<GroupMessage> wrapper = new QueryWrapper(); QueryWrapper<GroupMessage> wrapper = new QueryWrapper();
wrapper.lambda().eq(GroupMessage::getGroupId,member.getGroupId()) wrapper.lambda().eq(GroupMessage::getGroupId,member.getGroupId())
.gt(GroupMessage::getSendTime,member.getCreatedTime()) .gt(GroupMessage::getSendTime,member.getCreatedTime())
.ne(GroupMessage::getSendId, userId) .ne(GroupMessage::getSendId, session.getUserId())
.ne(GroupMessage::getStatus, MessageStatus.RECALL.code()); .ne(GroupMessage::getStatus, MessageStatus.RECALL.code());
if(maxReadedId!=null){ if(maxReadedId!=null){
wrapper.lambda().gt(GroupMessage::getId,maxReadedId); wrapper.lambda().gt(GroupMessage::getId,maxReadedId);
@ -154,15 +165,19 @@ public class GroupMessageServiceImpl extends ServiceImpl<GroupMessageMapper, Gro
if(messages.isEmpty()){ if(messages.isEmpty()){
continue; continue;
} }
// 组装消息，准备推送 // 推送
List<GroupMessageInfo> messageInfos = messages.stream().map(m->{ for (GroupMessage message:messages ){
GroupMessageInfo msgInfo = BeanUtils.copyProperties(m, GroupMessageInfo.class); GroupMessageInfo msgInfo = BeanUtils.copyProperties(message, GroupMessageInfo.class);
return msgInfo; IMGroupMessage sendMessage = new IMGroupMessage();
}).collect(Collectors.toList()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
// 只推给自己当前终端
sendMessage.setRecvIds(Collections.singletonList(session.getUserId()));
sendMessage.setRecvTerminals(Collections.singletonList(session.getTerminal()));
sendMessage.setData(msgInfo);
imClient.sendGroupMessage(sendMessage);
}
// 发送消息 // 发送消息
GroupMessageInfo[] infoArr = messageInfos.toArray(new GroupMessageInfo[messageInfos.size()]); log.info("拉取未读群聊消息，用户id:{},群聊id:{},数量:{}",session.getUserId(),member.getGroupId(),messages.size());
imClient.sendGroupMessage(Collections.singletonList(userId), infoArr);
log.info("拉取未读群聊消息，用户id:{},群聊id:{},数量:{}",userId,member.getGroupId(),messageInfos.size());
} }
} }

21
im-platform/src/main/java/com/bx/implatform/service/impl/PrivateMessageServiceImpl.java
View File

@ -4,8 +4,8 @@ import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.bx.imclient.IMClient; import com.bx.imclient.IMClient;
import com.bx.imcommon.contant.Constant; import com.bx.imcommon.contant.Constant;
import com.bx.imcommon.enums.IMTerminalType;
import com.bx.imcommon.model.IMPrivateMessage; import com.bx.imcommon.model.IMPrivateMessage;
import com.bx.imcommon.model.IMUserInfo;
import com.bx.imcommon.model.PrivateMessageInfo; import com.bx.imcommon.model.PrivateMessageInfo;
import com.bx.implatform.entity.PrivateMessage; import com.bx.implatform.entity.PrivateMessage;
import com.bx.implatform.enums.MessageStatus; import com.bx.implatform.enums.MessageStatus;
@ -22,6 +22,7 @@ import com.bx.implatform.vo.PrivateMessageVO;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; import org.springframework.stereotype.Service;
import java.util.Collections; import java.util.Collections;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
@ -58,9 +59,8 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
// 推送消息 // 推送消息
PrivateMessageInfo msgInfo = BeanUtils.copyProperties(msg, PrivateMessageInfo.class); PrivateMessageInfo msgInfo = BeanUtils.copyProperties(msg, PrivateMessageInfo.class);
IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>(); IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>();
sendMessage.setSendId(msgInfo.getSendId()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setRecvId(msgInfo.getRecvId()); sendMessage.setRecvId(msgInfo.getRecvId());
sendMessage.setSendTerminal(session.getTerminal());
sendMessage.setSendToSelf(true); sendMessage.setSendToSelf(true);
sendMessage.setDatas(Collections.singletonList(msgInfo)); sendMessage.setDatas(Collections.singletonList(msgInfo));
imClient.sendPrivateMessage(sendMessage); imClient.sendPrivateMessage(sendMessage);
@ -96,9 +96,8 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
msgInfo.setContent("对方撤回了一条消息"); msgInfo.setContent("对方撤回了一条消息");
IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>(); IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>();
sendMessage.setSendId(msgInfo.getSendId()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setRecvId(msgInfo.getRecvId()); sendMessage.setRecvId(msgInfo.getRecvId());
sendMessage.setSendTerminal(session.getTerminal());
sendMessage.setSendToSelf(true); sendMessage.setSendToSelf(true);
sendMessage.setDatas(Collections.singletonList(msgInfo)); sendMessage.setDatas(Collections.singletonList(msgInfo));
imClient.sendPrivateMessage(sendMessage); imClient.sendPrivateMessage(sendMessage);
@ -142,14 +141,14 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
*/ */
@Override @Override
public void pullUnreadMessage() { public void pullUnreadMessage() {
UserSession session = SessionContext.getSession();
// 获取当前连接的channelId // 获取当前连接的channelId
Long userId = SessionContext.getSession().getUserId(); if (!imClient.isOnline(session.getUserId())) {
if (!imClient.isOnline(userId)) {
throw new GlobalException(ResultCode.PROGRAM_ERROR, "用户未建立连接"); throw new GlobalException(ResultCode.PROGRAM_ERROR, "用户未建立连接");
} }
// 获取当前用户所有未读消息 // 获取当前用户所有未读消息
QueryWrapper<PrivateMessage> queryWrapper = new QueryWrapper<>(); QueryWrapper<PrivateMessage> queryWrapper = new QueryWrapper<>();
queryWrapper.lambda().eq(PrivateMessage::getRecvId, userId) queryWrapper.lambda().eq(PrivateMessage::getRecvId, session.getUserId())
.eq(PrivateMessage::getStatus, MessageStatus.UNREAD); .eq(PrivateMessage::getStatus, MessageStatus.UNREAD);
List<PrivateMessage> messages = this.list(queryWrapper); List<PrivateMessage> messages = this.list(queryWrapper);
// 上传至redis，等待推送 // 上传至redis，等待推送
@ -157,11 +156,13 @@ public class PrivateMessageServiceImpl extends ServiceImpl<PrivateMessageMapper,
List<PrivateMessageInfo> messageInfos = messages.stream().map(m -> BeanUtils.copyProperties(m, PrivateMessageInfo.class)).collect(Collectors.toList()); List<PrivateMessageInfo> messageInfos = messages.stream().map(m -> BeanUtils.copyProperties(m, PrivateMessageInfo.class)).collect(Collectors.toList());
// 推送消息 // 推送消息
IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>(); IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>();
sendMessage.setRecvId(userId); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setRecvId(session.getUserId());
sendMessage.setRecvTerminals(Collections.singletonList(session.getTerminal()));
sendMessage.setSendToSelf(false); sendMessage.setSendToSelf(false);
sendMessage.setDatas(messageInfos); sendMessage.setDatas(messageInfos);
imClient.sendPrivateMessage(sendMessage); imClient.sendPrivateMessage(sendMessage);
log.info("拉取未读私聊消息，用户id:{},数量:{}", userId, messageInfos.size()); log.info("拉取未读私聊消息，用户id:{},数量:{}", session.getUserId(), messageInfos.size());
} }
} }
} }

22
im-platform/src/main/java/com/bx/implatform/service/impl/WebrtcServiceImpl.java
View File

@ -2,6 +2,7 @@ package com.bx.implatform.service.impl;
import com.bx.imclient.IMClient; import com.bx.imclient.IMClient;
import com.bx.imcommon.model.IMPrivateMessage; import com.bx.imcommon.model.IMPrivateMessage;
import com.bx.imcommon.model.IMUserInfo;
import com.bx.imcommon.model.PrivateMessageInfo; import com.bx.imcommon.model.PrivateMessageInfo;
import com.bx.implatform.config.ICEServer; import com.bx.implatform.config.ICEServer;
import com.bx.implatform.config.ICEServerConfig; import com.bx.implatform.config.ICEServerConfig;
@ -54,8 +55,7 @@ public class WebrtcServiceImpl implements IWebrtcService {
messageInfo.setContent(offer); messageInfo.setContent(offer);
IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>(); IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>();
sendMessage.setSendId(session.getUserId()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setSendTerminal(session.getTerminal());
sendMessage.setRecvId(uid); sendMessage.setRecvId(uid);
sendMessage.setSendToSelf(false); sendMessage.setSendToSelf(false);
sendMessage.setSendResult(false); sendMessage.setSendResult(false);
@ -82,8 +82,7 @@ public class WebrtcServiceImpl implements IWebrtcService {
messageInfo.setContent(answer); messageInfo.setContent(answer);
IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>(); IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>();
sendMessage.setSendId(session.getUserId()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setSendTerminal(session.getTerminal());
sendMessage.setRecvId(uid); sendMessage.setRecvId(uid);
// 告知其他终端已经接受会话,中止呼叫 // 告知其他终端已经接受会话,中止呼叫
sendMessage.setSendToSelf(true); sendMessage.setSendToSelf(true);
@ -107,8 +106,7 @@ public class WebrtcServiceImpl implements IWebrtcService {
messageInfo.setSendId(session.getUserId()); messageInfo.setSendId(session.getUserId());
IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>(); IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>();
sendMessage.setSendId(session.getUserId()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setSendTerminal(session.getTerminal());
sendMessage.setRecvId(uid); sendMessage.setRecvId(uid);
// 告知其他终端已经拒绝会话,中止呼叫 // 告知其他终端已经拒绝会话,中止呼叫
sendMessage.setSendToSelf(true); sendMessage.setSendToSelf(true);
@ -130,8 +128,7 @@ public class WebrtcServiceImpl implements IWebrtcService {
messageInfo.setSendId(session.getUserId()); messageInfo.setSendId(session.getUserId());
IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>(); IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>();
sendMessage.setSendId(session.getUserId()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setSendTerminal(session.getTerminal());
sendMessage.setRecvId(uid); sendMessage.setRecvId(uid);
sendMessage.setSendToSelf(false); sendMessage.setSendToSelf(false);
sendMessage.setSendResult(false); sendMessage.setSendResult(false);
@ -154,8 +151,7 @@ public class WebrtcServiceImpl implements IWebrtcService {
messageInfo.setSendId(session.getUserId()); messageInfo.setSendId(session.getUserId());
IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>(); IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>();
sendMessage.setSendId(session.getUserId()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setSendTerminal(session.getTerminal());
sendMessage.setRecvId(uid); sendMessage.setRecvId(uid);
// 告知其他终端已经会话失败,中止呼叫 // 告知其他终端已经会话失败,中止呼叫
sendMessage.setSendToSelf(true); sendMessage.setSendToSelf(true);
@ -181,8 +177,7 @@ public class WebrtcServiceImpl implements IWebrtcService {
messageInfo.setSendId(session.getUserId()); messageInfo.setSendId(session.getUserId());
IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>(); IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>();
sendMessage.setSendId(session.getUserId()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setSendTerminal(session.getTerminal());
sendMessage.setRecvId(uid); sendMessage.setRecvId(uid);
sendMessage.setSendToSelf(false); sendMessage.setSendToSelf(false);
sendMessage.setSendResult(false); sendMessage.setSendResult(false);
@ -206,8 +201,7 @@ public class WebrtcServiceImpl implements IWebrtcService {
messageInfo.setContent(candidate); messageInfo.setContent(candidate);
IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>(); IMPrivateMessage<PrivateMessageInfo> sendMessage = new IMPrivateMessage<>();
sendMessage.setSendId(session.getUserId()); sendMessage.setSender(new IMUserInfo(session.getUserId(),session.getTerminal()));
sendMessage.setSendTerminal(session.getTerminal());
sendMessage.setRecvId(uid); sendMessage.setRecvId(uid);
sendMessage.setSendToSelf(false); sendMessage.setSendToSelf(false);
sendMessage.setSendResult(false); sendMessage.setSendResult(false);

53
im-server/src/main/java/com/bx/imserver/netty/processor/GroupMessageProcessor.java
View File

@ -5,6 +5,7 @@ import com.bx.imcommon.enums.IMCmdType;
import com.bx.imcommon.enums.IMSendCode; import com.bx.imcommon.enums.IMSendCode;
import com.bx.imcommon.model.IMRecvInfo; import com.bx.imcommon.model.IMRecvInfo;
import com.bx.imcommon.model.IMSendInfo; import com.bx.imcommon.model.IMSendInfo;
import com.bx.imcommon.model.IMUserInfo;
import com.bx.imcommon.model.SendResult; import com.bx.imcommon.model.SendResult;
import com.bx.imserver.netty.UserChannelCtxMap; import com.bx.imserver.netty.UserChannelCtxMap;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
@ -26,47 +27,45 @@ public class GroupMessageProcessor extends AbstractMessageProcessor<IMRecvInfo>
@Async @Async
@Override @Override
public void process(IMRecvInfo recvInfo) { public void process(IMRecvInfo recvInfo) {
Object data = recvInfo.getData(); IMUserInfo sender = recvInfo.getSender();
List<Long> recvIds = recvInfo.getRecvIds(); List<IMUserInfo> receivers = recvInfo.getReceivers();
log.info("接收到群消息，发送者:{},接收id:{}，内容:{}",recvInfo.getSendId(),recvIds,data); log.info("接收到群消息，发送者:{},接收用户数量:{}，内容:{}",sender.getId(),receivers.size(),recvInfo.getData());
for(Long recvId:recvIds){ for(IMUserInfo receiver:receivers){
try { try {
ChannelHandlerContext channelCtx = UserChannelCtxMap.getChannelCtx(recvId,recvInfo.getRecvTerminal()); ChannelHandlerContext channelCtx = UserChannelCtxMap.getChannelCtx(receiver.getId(),receiver.getTerminal());
if(channelCtx != null){ if(channelCtx != null){
// 推送消息到用户 // 推送消息到用户
IMSendInfo sendInfo = new IMSendInfo(); IMSendInfo sendInfo = new IMSendInfo();
sendInfo.setCmd(IMCmdType.GROUP_MESSAGE.code()); sendInfo.setCmd(IMCmdType.GROUP_MESSAGE.code());
sendInfo.setData(data); sendInfo.setData(recvInfo.getData());
channelCtx.channel().writeAndFlush(sendInfo); channelCtx.channel().writeAndFlush(sendInfo);
// 消息发送成功确认 // 消息发送成功确认
String key = RedisKey.IM_RESULT_GROUP_QUEUE; sendResult(recvInfo,receiver,IMSendCode.SUCCESS);
SendResult sendResult = new SendResult();
sendResult.setRecvId(recvId);
sendResult.setCode(IMSendCode.SUCCESS.code());
sendResult.setData(data);
redisTemplate.opsForList().rightPush(key,sendResult);
}else { }else {
// 消息发送失败确认 // 消息发送成功确认
String key = RedisKey.IM_RESULT_GROUP_QUEUE; sendResult(recvInfo,receiver,IMSendCode.NOT_FIND_CHANNEL);
SendResult sendResult = new SendResult(); log.error("未找到channel,发送者:{},接收id:{}，内容:{}",sender.getId(),receiver.getId(),recvInfo.getData());
sendResult.setRecvId(recvId);
sendResult.setCode(IMSendCode.NOT_FIND_CHANNEL.code());
sendResult.setData(data);
redisTemplate.opsForList().rightPush(key,sendResult);
log.error("未找到WS连接,发送者:{},接收id:{}，内容:{}",recvInfo.getSendId(),recvId,data);
} }
}catch (Exception e){ }catch (Exception e){
// 消息发送失败确认 // 消息发送失败确认
String key = RedisKey.IM_RESULT_GROUP_QUEUE; sendResult(recvInfo,receiver,IMSendCode.UNKONW_ERROR);
SendResult sendResult = new SendResult(); log.error("发送消息异常,发送者:{},接收id:{}，内容:{}",sender.getId(),receiver.getId(),recvInfo.getData());
sendResult.setRecvId(recvId);
sendResult.setCode(IMSendCode.UNKONW_ERROR.code());
sendResult.setData(data);
redisTemplate.opsForList().rightPush(key,sendResult);
log.error("发送消息异常,发送者:{},接收id:{}，内容:{}",recvInfo.getSendId(),recvId,data);
} }
} }
} }
private void sendResult(IMRecvInfo recvInfo,IMUserInfo receiver,IMSendCode sendCode){
if(recvInfo.getSendResult()) {
SendResult result = new SendResult();
result.setSender(recvInfo.getSender());
result.setReceiver(receiver);
result.setCode(sendCode.code());
result.setData(recvInfo.getData());
// 推送到结果队列
String key = RedisKey.IM_RESULT_GROUP_QUEUE;
redisTemplate.opsForList().rightPush(key, result);
}
}
} }

18
im-server/src/main/java/com/bx/imserver/netty/processor/PrivateMessageProcessor.java
View File

@ -5,6 +5,7 @@ import com.bx.imcommon.enums.IMCmdType;
import com.bx.imcommon.enums.IMSendCode; import com.bx.imcommon.enums.IMSendCode;
import com.bx.imcommon.model.IMRecvInfo; import com.bx.imcommon.model.IMRecvInfo;
import com.bx.imcommon.model.IMSendInfo; import com.bx.imcommon.model.IMSendInfo;
import com.bx.imcommon.model.IMUserInfo;
import com.bx.imcommon.model.SendResult; import com.bx.imcommon.model.SendResult;
import com.bx.imserver.netty.UserChannelCtxMap; import com.bx.imserver.netty.UserChannelCtxMap;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelHandlerContext;
@ -22,10 +23,11 @@ public class PrivateMessageProcessor extends AbstractMessageProcessor<IMRecvInfo
@Override @Override
public void process(IMRecvInfo recvInfo) { public void process(IMRecvInfo recvInfo) {
Long recvId = recvInfo.getRecvIds().get(0); IMUserInfo sender = recvInfo.getSender();
log.info("接收到消息，发送者:{},接收者:{}，内容:{}",recvInfo.getSendId(),recvId,recvInfo.getData()); IMUserInfo receiver = recvInfo.getReceivers().get(0);
log.info("接收到消息，发送者:{},接收者:{}，内容:{}",sender.getId(),receiver.getId(),recvInfo.getData());
try{ try{
ChannelHandlerContext channelCtx = UserChannelCtxMap.getChannelCtx(recvId,recvInfo.getRecvTerminal()); ChannelHandlerContext channelCtx = UserChannelCtxMap.getChannelCtx(receiver.getId(),receiver.getTerminal());
if(channelCtx != null ){ if(channelCtx != null ){
// 推送消息到用户 // 推送消息到用户
IMSendInfo sendInfo = new IMSendInfo(); IMSendInfo sendInfo = new IMSendInfo();
@ -37,23 +39,25 @@ public class PrivateMessageProcessor extends AbstractMessageProcessor<IMRecvInfo
}else{ }else{
// 消息推送失败确认 // 消息推送失败确认
sendResult(recvInfo,IMSendCode.NOT_FIND_CHANNEL); sendResult(recvInfo,IMSendCode.NOT_FIND_CHANNEL);
log.error("未找到WS连接，发送者:{},接收者:{}，内容:{}",recvInfo.getSendId(),recvId,recvInfo.getData()); log.error("未找到channel，发送者:{},接收者:{}，内容:{}",sender.getId(),receiver.getId(),recvInfo.getData());
} }
}catch (Exception e){ }catch (Exception e){
// 消息推送失败确认 // 消息推送失败确认
sendResult(recvInfo,IMSendCode.UNKONW_ERROR); sendResult(recvInfo,IMSendCode.UNKONW_ERROR);
log.error("发送异常，发送者:{},接收者:{}，内容:{}",recvInfo.getSendId(),recvId,recvInfo.getData(),e); log.error("发送异常，发送者:{},接收者:{}，内容:{}",sender.getId(),receiver.getId(),recvInfo.getData(),e);
} }
} }
private void sendResult(IMRecvInfo recvInfo,IMSendCode sendCode){ private void sendResult(IMRecvInfo recvInfo,IMSendCode sendCode){
if(recvInfo.getSendResult()) { if(recvInfo.getSendResult()) {
String key = RedisKey.IM_RESULT_PRIVATE_QUEUE;
SendResult result = new SendResult(); SendResult result = new SendResult();
result.setRecvId(recvInfo.getRecvIds().get(0)); result.setSender(recvInfo.getSender());
result.setReceiver(recvInfo.getReceivers().get(0));
result.setCode(sendCode.code()); result.setCode(sendCode.code());
result.setData(recvInfo.getData()); result.setData(recvInfo.getData());
// 推送到结果队列
String key = RedisKey.IM_RESULT_PRIVATE_QUEUE;
redisTemplate.opsForList().rightPush(key, result); redisTemplate.opsForList().rightPush(key, result);
} }
} }

1
im-server/src/main/java/com/bx/imserver/task/AbstractPullMessageTask.java
View File

@ -40,7 +40,6 @@ public abstract class AbstractPullMessageTask {
if (serverGroup.isReady()) { if (serverGroup.isReady()) {
pullMessage(); pullMessage();
} }
Thread.sleep(100);
} catch (Exception e) { } catch (Exception e) {
log.error("任务调度异常", e); log.error("任务调度异常", e);
Thread.sleep(200); Thread.sleep(200);

13
im-server/src/main/java/com/bx/imserver/task/PullUnreadGroupMessageTask.java
View File

@ -10,8 +10,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
import java.util.List;
@Slf4j @Slf4j
@Component @Component
@ -23,16 +22,12 @@ public class PullUnreadGroupMessageTask extends AbstractPullMessageTask {
@Override @Override
public void pullMessage() { public void pullMessage() {
// 从redis拉取未读消息 // 从redis拉取未读消息
String key = RedisKey.IM_UNREAD_GROUP_QUEUE + IMServerGroup.serverId; String key = String.join(":",RedisKey.IM_UNREAD_GROUP_QUEUE,IMServerGroup.serverId+"");
List recvInfos = redisTemplate.opsForList().range(key,0,-1); IMRecvInfo recvInfo = (IMRecvInfo)redisTemplate.opsForList().leftPop(key,10, TimeUnit.SECONDS);
for(Object o: recvInfos){ if(recvInfo != null){
redisTemplate.opsForList().leftPop(key);
IMRecvInfo recvInfo = (IMRecvInfo)o;
AbstractMessageProcessor processor = ProcessorFactory.createProcessor(IMCmdType.GROUP_MESSAGE); AbstractMessageProcessor processor = ProcessorFactory.createProcessor(IMCmdType.GROUP_MESSAGE);
processor.process(recvInfo); processor.process(recvInfo);
} }
} }
} }

13
im-server/src/main/java/com/bx/imserver/task/PullUnreadPrivateMessageTask.java
View File

@ -11,9 +11,7 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate; import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
import java.util.List;
@Slf4j @Slf4j
@Component @Component
@ -26,14 +24,11 @@ public class PullUnreadPrivateMessageTask extends AbstractPullMessageTask {
@Override @Override
public void pullMessage() { public void pullMessage() {
// 从redis拉取未读消息 // 从redis拉取未读消息
String key = RedisKey.IM_UNREAD_PRIVATE_QUEUE + IMServerGroup.serverId; String key = String.join(":",RedisKey.IM_UNREAD_PRIVATE_QUEUE ,IMServerGroup.serverId+"");
List recvInfos = redisTemplate.opsForList().range(key,0,-1); IMRecvInfo recvInfo = (IMRecvInfo)redisTemplate.opsForList().leftPop(key,10, TimeUnit.SECONDS);
for(Object o: recvInfos){ if(recvInfo != null) {
redisTemplate.opsForList().leftPop(key);
IMRecvInfo recvInfo = (IMRecvInfo)o;
AbstractMessageProcessor processor = ProcessorFactory.createProcessor(IMCmdType.PRIVATE_MESSAGE); AbstractMessageProcessor processor = ProcessorFactory.createProcessor(IMCmdType.PRIVATE_MESSAGE);
processor.process(recvInfo); processor.process(recvInfo);
} }
} }

10
im-ui/src/components/chat/ChatPrivateVideo.vue
View File

@ -143,7 +143,7 @@
}, },
handleMessage(msg) { handleMessage(msg) {
if (msg.type == this.$enums.MESSAGE_TYPE.RTC_ACCEPT) { if (msg.type == this.$enums.MESSAGE_TYPE.RTC_ACCEPT) {
if (msg.sendId == this.$store.state.userStore.userInfo.id) { if (msg.selfSend) {
// 我在其他终端接受了对方的通话 // 我在其他终端接受了对方的通话
this.$message.success("已在其他设备接听"); this.$message.success("已在其他设备接听");
this.close(); this.close();
@ -163,8 +163,7 @@
} }
} else if (msg.type == this.$enums.MESSAGE_TYPE.RTC_REJECT) { } else if (msg.type == this.$enums.MESSAGE_TYPE.RTC_REJECT) {
console.log(msg) if (msg.selfSend) {
if (msg.sendId == this.$store.state.userStore.userInfo.id) {
// 我在其他终端拒绝了对方的通话 // 我在其他终端拒绝了对方的通话
this.$message.success("已在其他设备拒绝通话"); this.$message.success("已在其他设备拒绝通话");
this.close(); this.close();
@ -195,9 +194,8 @@
this.loading = true; this.loading = true;
this.state = 'CONNECTING'; this.state = 'CONNECTING';
this.audio.play(); this.audio.play();
}); })
}, },(error) => {
(error) => {
this.$message.error(error); this.$message.error(error);
}); });

52
im-ui/src/components/chat/ChatVideoAcceptor.vue
View File

@ -7,8 +7,8 @@
{{friend.nickName}} 请求和您进行视频通话... {{friend.nickName}} 请求和您进行视频通话...
</div> </div>
<div class="acceptor-btn-group"> <div class="acceptor-btn-group">
<div class="icon iconfont icon-phone-accept accept" @click="accpet()"></div> <div class="icon iconfont icon-phone-accept accept" @click="accpet()"></div>
<div class="icon iconfont icon-phone-reject reject" @click="reject()"></div> <div class="icon iconfont icon-phone-reject reject" @click="reject()"></div>
</div> </div>
</div> </div>
</template> </template>
@ -18,76 +18,75 @@
export default { export default {
name: "videoAcceptor", name: "videoAcceptor",
components:{HeadImage}, components: {
HeadImage
},
props: { props: {
friend:{ friend: {
type: Object type: Object
} }
}, },
data(){ data() {
return { return {
offer:{}, offer: {},
audio: new Audio() audio: new Audio()
} }
}, },
methods:{ methods: {
accpet(){ accpet() {
let info ={ let info = {
friend: this.friend, friend: this.friend,
master: false, master: false,
offer: this.offer offer: this.offer
} }
this.$store.commit("showChatPrivateVideoBox",info); this.$store.commit("showChatPrivateVideoBox", info);
this.close(); this.close();
}, },
reject(){ reject() {
this.$http({ this.$http({
url: `/webrtc/private/reject?uid=${this.friend.id}`, url: `/webrtc/private/reject?uid=${this.friend.id}`,
method: 'post' method: 'post'
}) })
this.close(); this.close();
}, },
failed(reason){ failed(reason) {
this.$http({ this.$http({
url: `/webrtc/private/failed?uid=${this.friend.id}&reason=${reason}`, url: `/webrtc/private/failed?uid=${this.friend.id}&reason=${reason}`,
method: 'post' method: 'post'
}) })
this.close(); this.close();
}, },
onCall(msgInfo){ onCall(msgInfo) {
console.log("onCall")
this.offer = JSON.parse(msgInfo.content); this.offer = JSON.parse(msgInfo.content);
if(this.$store.state.userStore.state == this.$enums.USER_STATE.BUSY){ if (this.$store.state.userStore.state == this.$enums.USER_STATE.BUSY) {
this.failed("对方正忙,暂时无法接听"); this.failed("对方正忙,暂时无法接听");
return; return;
} }
// 超时未接听 // 超时未接听
this.timer && clearTimeout(this.timer); this.timer && clearTimeout(this.timer);
this.timer = setTimeout(()=>{ this.timer = setTimeout(() => {
this.failed("对方未接听"); this.failed("对方未接听");
},30000) }, 30000)
this.audio.play(); this.audio.play();
}, },
onCancel(){ onCancel() {
// 取消视频通话请求 // 取消视频通话请求
this.$message.success("对方取消了呼叫"); this.$message.success("对方取消了呼叫");
this.close(); this.close();
}, },
handleMessage(msgInfo){ handleMessage(msgInfo) {
if(msgInfo.type == this.$enums.MESSAGE_TYPE.RTC_CALL){ if (msgInfo.type == this.$enums.MESSAGE_TYPE.RTC_CALL) {
this.onCall(msgInfo); this.onCall(msgInfo);
}else if(msgInfo.type == this.$enums.MESSAGE_TYPE.RTC_CANCEL){ } else if (msgInfo.type == this.$enums.MESSAGE_TYPE.RTC_CANCEL) {
this.onCancel(); this.onCancel();
} }
}, },
close(){ close() {
this.timer && clearTimeout(this.timer); this.timer && clearTimeout(this.timer);
this.audio.pause(); this.audio.pause();
this.$emit("close"); this.$emit("close");
}, },
initAudio(){ initAudio() {
let url = require(`@/assets/audio/call.wav`); let url = require(`@/assets/audio/call.wav`);
this.audio.src = url; this.audio.src = url;
this.audio.loop = true; this.audio.loop = true;
@ -99,7 +98,6 @@
} }
} }
</script> </script>
<style scoped lang="scss"> <style scoped lang="scss">
@ -122,9 +120,11 @@
.icon { .icon {
font-size: 50px; font-size: 50px;
cursor: pointer; cursor: pointer;
&.accept { &.accept {
color: green; color: green;
} }
&.reject { &.reject {
color: red; color: red;
} }

5
im-ui/src/view/Home.vue
View File

@ -87,6 +87,8 @@
this.pullUnreadMessage(); this.pullUnreadMessage();
}); });
this.$wsApi.onmessage((cmd, msgInfo) => { this.$wsApi.onmessage((cmd, msgInfo) => {
// 标记这条消息是不是自己发的
msgInfo.selfSend = msgInfo.sendId==this.$store.state.userStore.userInfo.id;
if (cmd == 2) { if (cmd == 2) {
// 异地登录，强制下线 // 异地登录，强制下线
this.$message.error("您已在其他地方登陆，将被强制下线"); this.$message.error("您已在其他地方登陆，将被强制下线");
@ -116,7 +118,6 @@
}, },
handlePrivateMessage(msg) { handlePrivateMessage(msg) {
// 好友列表存在好友信息，直接插入私聊消息 // 好友列表存在好友信息，直接插入私聊消息
msg.selfSend = msg.sendId==this.$store.state.userStore.userInfo.id;
let friendId = msg.selfSend?msg.recvId:msg.sendId; let friendId = msg.selfSend?msg.recvId:msg.sendId;
let friend = this.$store.state.friendStore.friends.find((f) => f.id == friendId); let friend = this.$store.state.friendStore.friends.find((f) => f.id == friendId);
if (friend) { if (friend) {
@ -191,7 +192,7 @@
// 插入消息 // 插入消息
this.$store.commit("insertMessage", msg); this.$store.commit("insertMessage", msg);
// 播放提示音 // 播放提示音
this.playAudioTip(); !msg.selfSend && this.playAudioTip();
}, },
handleExit() { handleExit() {
this.$wsApi.closeWebSocket(); this.$wsApi.closeWebSocket();

Write Preview
Loading…
Cancel
Save