feat: 开始管理用户是否读取了消息

目前出现了BUG。我修改代码后。前端websocket收不到消息了。

src/main/java/com/youlai/system/model/entity/NoticeStatus.java

@@ -1,7 +1,11 @@
 package com.youlai.system.model.entity;
 
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
 import lombok.Getter;
 import lombok.Setter;
+
+import java.io.Serializable;
 import java.time.LocalDateTime;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.youlai.system.common.base.BaseEntity;
@@ -15,10 +19,16 @@ import com.youlai.system.common.base.BaseEntity;
 @Getter
 @Setter
 @TableName("sys_notice_status")
-public class NoticeStatus extends BaseEntity {
+public class NoticeStatus implements Serializable {
 
     private static final long serialVersionUID = 1L;
 
+    /**
+     * 主键ID
+     */
+    @TableId(type = IdType.AUTO)
+    private Long id;
+
     /**
      * 公共通知id
      */
@@ -26,7 +36,7 @@ public class NoticeStatus extends BaseEntity {
     /**
      * 用户id
      */
-    private Integer userId;
+    private Long userId;
     /**
      * 读取状态，0未读，1已读取
      */

src/main/java/com/youlai/system/service/WebsocketService.java

@@ -1,14 +1,17 @@
 package com.youlai.system.service;
 
+import java.util.Set;
+
 public interface WebsocketService {
 
    void addUser(String username);
 
    void removeUser(String username) ;
 
+   Set<String> getUsers();
    /**
     * 发送消息到前端
     * @param message
     */
-   void sendStringToFrontend(String message);
+   void sendStringToFrontend(String sender,String message);
 }

src/main/java/com/youlai/system/service/impl/NoticeServiceImpl.java

@@ -1,9 +1,15 @@
 package com.youlai.system.service.impl;
 
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.google.gson.*;
+import com.youlai.system.model.entity.NoticeStatus;
+import com.youlai.system.model.entity.SysUser;
 import com.youlai.system.security.util.SecurityUtils;
+import com.youlai.system.service.NoticeStatusService;
+import com.youlai.system.service.SysUserService;
 import com.youlai.system.service.WebsocketService;
+import jodd.util.StringUtil;
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;
 import com.baomidou.mybatisplus.core.metadata.IPage;
@@ -39,6 +45,11 @@ public class NoticeServiceImpl extends ServiceImpl<NoticeMapper, Notice> impleme
     private final NoticeConverter noticeConverter;
 
     private final WebsocketService webSocketServer;
+
+    private final NoticeStatusService noticeStatusService;
+
+    private final SysUserService sysUserService;
+
     private final Gson gson = new GsonBuilder()
             .registerTypeAdapter(LocalDateTime.class, (JsonSerializer<LocalDateTime>) (localDateTime, type, jsonSerializationContext) ->
                     new JsonPrimitive(localDateTime.format(DateTimeFormatter.ISO_LOCAL_DATE_TIME)))
@@ -47,18 +58,30 @@ public class NoticeServiceImpl extends ServiceImpl<NoticeMapper, Notice> impleme
             .create();
 
     private void sendWebSocketMsg(Notice notice) {
-        if(notice.getSendStatus() > 0){
+        if (notice.getSendStatus() > 0) {
             String jsonNotice = gson.toJson(noticeConverter.toVO(notice));
-             webSocketServer.sendStringToFrontend(jsonNotice);
+            webSocketServer.sendStringToFrontend(SecurityUtils.getUsername(), jsonNotice);
+            List<SysUser> list = sysUserService.list();
+            for (SysUser sysUser : list) {
+                NoticeStatus noticeStatus = noticeStatusService.getOne(new LambdaQueryWrapper<NoticeStatus>().eq(NoticeStatus::getUserId, sysUser.getId()).eq(NoticeStatus::getNoticeId, notice.getId()));
+                if (noticeStatus == null) {
+                    noticeStatus = new NoticeStatus();
+                    noticeStatus.setUserId(sysUser.getId());
+                    noticeStatus.setNoticeId(notice.getId());
+                    noticeStatus.setReadStatus(0L);
+                    noticeStatusService.save(noticeStatus);
+                }
+            }
+
         }
     }
 
     /**
-    * 获取通知公告分页列表
-    *
-    * @param queryParams 查询参数
-    * @return {@link IPage<NoticeVO>} 通知公告分页列表
-    */
+     * 获取通知公告分页列表
+     *
+     * @param queryParams 查询参数
+     * @return {@link IPage<NoticeVO>} 通知公告分页列表
+     */
     @Override
     public IPage<NoticeVO> getNoticePage(NoticeQuery queryParams) {
         Page<NoticeVO> pageVO = this.baseMapper.getNoticePage(
@@ -67,7 +90,7 @@ public class NoticeServiceImpl extends ServiceImpl<NoticeMapper, Notice> impleme
         );
         return pageVO;
     }
-    
+
     /**
      * 获取通知公告表单数据
      *
@@ -79,7 +102,7 @@ public class NoticeServiceImpl extends ServiceImpl<NoticeMapper, Notice> impleme
         Notice entity = this.getById(id);
         return noticeConverter.toForm(entity);
     }
-    
+
     /**
      * 新增通知公告
      *
@@ -94,31 +117,31 @@ public class NoticeServiceImpl extends ServiceImpl<NoticeMapper, Notice> impleme
         entity.setUpdateBy(SecurityUtils.getUserId());
         entity.setIsDelete(0);
         boolean result = this.save(entity);
-        if(result){
+        if (result) {
             sendWebSocketMsg(entity);
         }
         return result;
     }
-    
+
     /**
      * 更新通知公告
      *
-     * @param id   通知公告ID
+     * @param id       通知公告ID
      * @param formData 通知公告表单对象
      * @return
      */
     @Override
-    public boolean updateNotice(Long id,NoticeForm formData) {
+    public boolean updateNotice(Long id, NoticeForm formData) {
         Notice entity = noticeConverter.toEntity(formData);
         entity.setUpdateBy(SecurityUtils.getUserId());
         entity.setIsDelete(0);
         boolean result = this.updateById(entity);
-        if(result) {
+        if (result) {
             sendWebSocketMsg(entity);
         }
         return result;
     }
-    
+
     /**
      * 删除通知公告
      *

src/main/java/com/youlai/system/service/impl/WebsocketServiceImpl.java

@@ -1,6 +1,7 @@
 package com.youlai.system.service.impl;
 
 import com.youlai.system.event.UserConnectionEvent;
+import com.youlai.system.model.dto.ChatMessage;
 import com.youlai.system.service.WebsocketService;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -8,7 +9,9 @@ import org.springframework.context.event.EventListener;
 import org.springframework.messaging.simp.SimpMessagingTemplate;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
+import org.springframework.web.socket.messaging.SessionDisconnectEvent;
 
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -19,8 +22,14 @@ public class WebsocketServiceImpl implements WebsocketService {
 
     private final SimpMessagingTemplate messagingTemplate;
 
+    // 在线用户
     private final Set<String> onlineUsers = ConcurrentHashMap.newKeySet();
 
+    // 离线消息
+    private final Set<ChatMessage> offlineMessages = ConcurrentHashMap.newKeySet();
+
+    private final Map<ChatMessage,Set<String>> messageReceiptStatus = new ConcurrentHashMap<>();
+
     @Override
     public void addUser(String username) {
         onlineUsers.add(username);
@@ -31,12 +40,22 @@ public class WebsocketServiceImpl implements WebsocketService {
         onlineUsers.remove(username);
     }
 
+    @Override
+    public Set<String> getUsers() {
+        return onlineUsers;
+    }
+
     @EventListener
     public void handleUserConnectionEvent(UserConnectionEvent event) {
         String username = event.getUsername();
         if (event.isConnected()) {
             onlineUsers.add(username);
             log.info("User connected: {}", username);
+            // 发送离线消息
+            offlineMessages.forEach(message -> {
+                messagingTemplate.convertAndSendToUser(username, "/topic/chat", message);
+                messageReceiptStatus.computeIfAbsent(message, k -> ConcurrentHashMap.newKeySet()).add(username);
+            });
         } else {
             onlineUsers.remove(username);
             log.info("User disconnected: {}", username);
@@ -45,14 +64,31 @@ public class WebsocketServiceImpl implements WebsocketService {
         messagingTemplate.convertAndSend("/topic/onlineUserCount", onlineUsers.size());
     }
 
+    @EventListener
+    public void handleSessionDisconnect(SessionDisconnectEvent event) {
+        String username = event.getUser().getName();
+        onlineUsers.remove(username);
+    }
+
     @Scheduled(fixedRate = 5000)
     public void sendOnlineUserCount() {
         messagingTemplate.convertAndSend("/topic/onlineUserCount", onlineUsers.size());
     }
 
     @Override
-    public void sendStringToFrontend(String message) {
-        messagingTemplate.convertAndSend("/topic/chat", message);
-    }
+    public void sendStringToFrontend(String sender, String message) {
+        ChatMessage chatMessage = new ChatMessage(sender, message);
+        offlineMessages.add(chatMessage);
+        messageReceiptStatus.putIfAbsent(chatMessage, ConcurrentHashMap.newKeySet());
+        onlineUsers.forEach(receiver -> {
+            messagingTemplate.convertAndSendToUser(receiver, "/topic/chat", chatMessage);
+            messageReceiptStatus.get(chatMessage).add(receiver);
+        });
+        if(messageReceiptStatus.get(chatMessage).size() == onlineUsers.size()) {
+            //记录完成状态
+            offlineMessages.remove(chatMessage);//从离线消息中移除已发送的消息
+            messageReceiptStatus.remove(chatMessage);//从消息接收状态集合总移除已发送的消息
 
+        }
+    }
 }