add im即时通信

nb-admin/src/test/java/com/nb/admin/BusDeviceAlarmTest.java

@@ -1,6 +1,7 @@
 package com.nb.admin;
 
 import cn.hutool.bloomfilter.bitMap.BitMap;
+import com.baomidou.mybatisplus.core.toolkit.IdWorker;
 import com.nb.web.service.bus.controller.BusDeviceHistoryController;
 import com.nb.web.api.entity.BusDeviceAlarmEntity;
 import com.nb.web.api.entity.BusInfusionHistoryEntity;
@@ -13,6 +14,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.redisson.api.RBitSet;
+import org.redisson.api.RBloomFilter;
 import org.redisson.api.RSet;
 import org.redisson.api.RedissonClient;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -117,13 +119,9 @@ public class BusDeviceAlarmTest {
     private RedisTemplate redisTemplate;
     @Test
     public void redissonClientTest(){
-        RSet<Object> set = redissonClient.getSet("123");
-        System.out.println(set.size());
-        set.add("123");
-        System.out.println(set.size());
-        set.add("456");
-        Set<Object> objects = set.readAll();
-        System.out.println(objects);
+        RBitSet bitSet = redissonClient.getBitSet("123");
+        RBloomFilter<Object> bloomFilter = redissonClient.getBloomFilter("123");
+
     }
 
 

nb-common/ws-common/src/main/java/com/nb/common/websocket/PubResponse.java

@@ -15,16 +15,16 @@ import java.io.Serializable;
 @Data
 @AllArgsConstructor(staticName = "of")
 public class PubResponse implements Serializable {
-    private Long key;
+    private String key;
     private boolean success;
     private String errorMsg;
 
 
-    public static PubResponse success(Long key){
+    public static PubResponse success(String key){
         return PubResponse.of(key,true,"");
     }
 
-    public static PubResponse fail(Long key,String errorMsg){
+    public static PubResponse fail(String key,String errorMsg){
         return PubResponse.of(key,false,errorMsg);
     }
 }

nb-common/ws-common/src/main/java/com/nb/common/websocket/TopicMessage.java

@@ -12,7 +12,7 @@ import java.io.Serializable;
 public class TopicMessage implements Serializable {
     private Object message;
     private String param;
-    private Long key;
+    private String key;
 
 
     public static TopicMessage of(Object message,String param){
@@ -20,7 +20,7 @@ public class TopicMessage implements Serializable {
     }
 
 
-    public static TopicMessage of(Object message,String param,Long key){
+    public static TopicMessage of(Object message,String param,String key){
         return new TopicMessage(message,param,key);
     }
 

nb-common/ws-common/src/main/java/com/nb/common/websocket/filter/DefaultPubMsgFilter.java

@@ -22,7 +22,7 @@ public class DefaultPubMsgFilter implements PubMsgFilter {
     @Override
     public boolean doFilter(ChannelContext channelContext, MessagingRequest source) {
         if(source.getKey()!=null){
-            Tio.send(channelContext, WsResponse.fromText(JSONUtil.toJsonStr(PubResponse.fail(-1L,"发布消息时key不能为空")), WsPacket.CHARSET_NAME));
+            Tio.send(channelContext, WsResponse.fromText(JSONUtil.toJsonStr(PubResponse.fail("","发布消息时key不能为空")), WsPacket.CHARSET_NAME));
             return false;
         }
         return ObjectUtil.isNotNull(source.getPayload());

nb-common/ws-common/src/main/java/com/nb/common/websocket/msg/MessageResponse.java

@@ -19,5 +19,5 @@ public class MessageResponse implements Serializable {
     private String type;
     private String param;
     private Object payload;
-    private Long key;
+    private String key;
 }

nb-common/ws-common/src/main/java/com/nb/common/websocket/msg/MessagingRequest.java

@@ -10,7 +10,7 @@ public class MessagingRequest implements Serializable {
     /**
      * 唯一消息id
      */
-    private Long key;
+    private String key;
 
     /**
      * 心跳、设备或报警信息，例如ping、device,alarm,patient

nb-im/src/main/java/com/nb/im/controller/ImRoomController.java

@@ -13,6 +13,7 @@ import com.nb.common.crud.BaseService;
 import com.nb.common.crud.controller.BaseSaveController;
 import com.nb.core.exception.CustomException;
 import com.nb.core.result.R;
+import com.nb.im.controller.vo.RemarkReadVo;
 import com.nb.im.controller.vo.UnreadVo;
 import com.nb.im.entity.ImMsgEntity;
 import com.nb.im.entity.ImRoomEntity;
@@ -103,10 +104,16 @@ public class ImRoomController implements BaseSaveController<ImRoomEntity,String>
                 .lambda()
                 .eq(ImMsgEntity::getImRoomId,roomId)
                 .in(ImMsgEntity::getKey,keySet));
-        roomOperator.remarkUnreadMsgAll(userId);
         return R.success(result);
     }
 
+    @PostMapping
+    @ApiOperation("标记已读信息")
+    public R<Boolean> remarkRead(@RequestBody RemarkReadVo vo){
+        roomOperatorManager.getRoomOperator(vo.getRoomId())
+                .remarkUnreadMsg(vo.getUerId(),vo.getKey());
+        return R.success(true);
+    }
 
     @PostMapping("/read/unread")
     @ApiOperation("获取未读消息总数量")

nb-im/src/main/java/com/nb/im/controller/vo/RemarkReadVo.java

@@ -0,0 +1,31 @@
+package com.nb.im.controller.vo;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import javax.validation.constraints.NotNull;
+import java.util.*;
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName RemarkReadVo.java
+ * @Description TODO
+ * @createTime 2022年08月19日 10:24:00
+ */
+@Data
+@ApiModel("标记已读信息")
+public class RemarkReadVo {
+    @ApiModelProperty("聊天室id")
+    @NotNull(message = "聊天室id不能为空")
+    private String roomId;
+
+    @ApiModelProperty("已读消息key")
+    @NotNull(message = "消息key不能为空")
+    private List<Long> key;
+
+
+    @ApiModelProperty("用户id")
+    @NotNull(message = "用户id不能为空")
+    private String uerId;
+}

nb-im/src/main/java/com/nb/im/room/ImRoomOperator.java

@@ -12,7 +12,7 @@ import java.util.*;
  * @createTime 2022年08月18日 19:05:00
  */
 public class ImRoomOperator {
-    private RBitSet msgKeySet;
+    private RBloomFilter<String> msgKeyFilter;
     private RAtomicLong atomicLong;
     private final String id;
     private final RedissonClient redissonClient;
@@ -24,7 +24,7 @@ public class ImRoomOperator {
 
     public ImRoomOperator(String id, RedissonClient redissonClient, long maxSort,List<String> userIds) {
         this.id=id;
-        this.msgKeySet = redissonClient.getBitSet("im:msg-key."+id);
+        this.msgKeyFilter = redissonClient.getBloomFilter("im:msg-key."+id);
         this.atomicLong = redissonClient.getAtomicLong("im:incr." + id);
         this.redissonClient=redissonClient;
         userIds
@@ -47,12 +47,9 @@ public class ImRoomOperator {
      * @param key
      * @return void
      */
-    public void remarkUnreadMsg(String userId,Long key){
-        unReadMap.forEach((id,bitSet)->{
-            if(!userId.equals(id)){
-                bitSet.add(String.valueOf(key));
-            }
-        });
+    public void remarkUnreadMsg(String userId,List<Long> key){
+        unReadMap.get(userId)
+                .removeAll(key);
     }
 
 
@@ -113,12 +110,12 @@ public class ImRoomOperator {
      * @param key
      * @return boolean
      */
-    public boolean existMsg(Long key){
-        return  msgKeySet.get(key);
+    public boolean existMsg(String key){
+        return  msgKeyFilter.contains(key);
     }
 
     public void close(){
-        msgKeySet.deleteAsync();
+        msgKeyFilter.deleteAsync();
         atomicLong.deleteAsync();
         unReadMap.forEach((k,bitSet)->{
             if (bitSet.size()==0) {

nb-im/src/main/java/com/nb/im/ws/PubMsgInfo.java

@@ -22,7 +22,7 @@ import java.io.Serializable;
 @Validated
 public class PubMsgInfo implements Serializable {
     @ApiModelProperty("客户端生产的唯一消息key")
-    private Long key;
+    private String key;
 
     @ApiModelProperty("聊天室id")
     private String roomId;

nb-im/src/main/java/com/nb/im/ws/filter/MsgFormatFilter.java

@@ -26,7 +26,7 @@ public class MsgFormatFilter implements PubMsgFilter {
         String id = source.getId();
         if("im".equalsIgnoreCase(id)){
             if(ObjectUtil.isEmpty(source.getPayload())){
-                Tio.send(channelContext,WsResponse.fromText(JSONUtil.toJsonStr(PubResponse.fail(-1L,"发布消息不可为空")),WsPacket.CHARSET_NAME));
+                Tio.send(channelContext,WsResponse.fromText(JSONUtil.toJsonStr(PubResponse.fail("","发布消息不可为空")),WsPacket.CHARSET_NAME));
                 return false;
             }
             PubMsgInfo pubMsgInfo = JSONUtil.toBean(JSONUtil.toJsonStr(source.getPayload()), PubMsgInfo.class);