feat(activity): 实现随机任务开始发送 WebSocket通知功能

src/main/java/com/wangxiaolu/promotion/websocket/TemporaryActivityTaskClockType.java → src/main/java/com/wangxiaolu/promotion/enums/activity/TemporaryActivityTaskClockMessageType.java

-package com.wangxiaolu.promotion.websocket;
+package com.wangxiaolu.promotion.enums.activity;
 
-public enum TemporaryActivityTaskClockType {
 
-    RANDOM_TASK_TART("randomTaskStart","随机任务开始"),
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.Getter;
+
+@Getter
+@AllArgsConstructor
+public enum TemporaryActivityTaskClockMessageType {
+
+
+    RANDOM_TASK_START("randomTaskStart","随机任务开始通知"),
     RANDOM_TASK_CLOCK("randomTaskClock","随机任务打卡"),
     POS_CLOCK("posClock","Pos机任务打卡"),
     TASK_CLOCK_QUERY("taskClockQuery","任务信息查询"),
@@ -13,19 +21,6 @@ public enum TemporaryActivityTaskClockType {
     private String name;
 
 
-    TemporaryActivityTaskClockType(String type, String name) {
-        this.type = type;
-        this.name = name;
-    }
-
 
-    public String getName(String  type) {
-        for (TemporaryActivityTaskClockType value : values()) {
-            if (value.type.equals(type)) {
-                return value.name;
-            }
-        }
-        return "";
-    }
 
 }

src/main/java/com/wangxiaolu/promotion/websocket/MessageBean.java

-package com.wangxiaolu.promotion.websocket;
-
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-public class MessageBean {
-    @JsonProperty("msgType")
-    private String type;
-    @JsonProperty("data")
-    private String content;
-
-    // Getter and Setter
-    public String getType() {
-        return type;
-    }
-
-    public void setType(String type) {
-        this.type = type;
-    }
-
-    public String getContent() {
-        return content;
-    }
-
-    public void setContent(String content) {
-        this.content = content;
-    }
-}

src/main/java/com/wangxiaolu/promotion/websocket/TemporaryActivityTaskClockSocketHandler.java

 package com.wangxiaolu.promotion.websocket;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.wangxiaolu.promotion.websocket.pojo.MessageBean;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.springframework.web.socket.CloseStatus;
 import org.springframework.web.socket.TextMessage;
+import org.springframework.web.socket.WebSocketMessage;
 import org.springframework.web.socket.WebSocketSession;
 import org.springframework.web.socket.handler.TextWebSocketHandler;
 import java.io.IOException;
@@ -67,10 +69,10 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle
         MessageBean messageBean = objectMapper.readValue(payload, MessageBean.class);
         handleMessageType(messageBean);
         // 业务消息处理（示例：广播消息）
-        broadcast("用户[" + userId + "]：" + payload);
+//        broadcast("用户[" + userId + "]：" + payload);
     }
     private void handleMessageType(MessageBean messageBean) {
-        switch (messageBean.getType()) {
+        switch (messageBean.getMsgType()) {
             case "ACTIVITY_START":
 //                startActivity(messageBean.getContent());
                 break;
@@ -78,7 +80,7 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle
 //                endActivity(messageBean.getContent());
                 break;
             default:
-                log.warn("未知消息类型: {}", messageBean.getType());
+                log.warn("未知消息类型: {}", messageBean.getMsgType());
         }
     }
     /**
@@ -128,10 +130,23 @@ public class TemporaryActivityTaskClockSocketHandler extends TextWebSocketHandle
     /**
      * 向指定用户发送消息
      */
-    public void sendToUser(String userId, String message) throws IOException {
+    public Integer sendToUser(String userId, String message) throws IOException {
         WebSocketSession session = sessions.get(userId);
         if (session != null && session.isOpen()) {
+//            String jsonMessage = objectMapper.writeValueAsString(message);
             session.sendMessage(new TextMessage(message));
+            return 1;
         }
+        return 0;
+    }
+
+    public Integer sendToUser(String userId, MessageBean messageBean) throws IOException {
+        WebSocketSession session = sessions.get(userId);
+        if (session != null && session.isOpen()) {
+            String jsonMessage = objectMapper.writeValueAsString(messageBean);
+            session.sendMessage(new TextMessage(jsonMessage));
+            return 1;
+        }
+        return 0;
     }
 }

src/main/java/com/wangxiaolu/promotion/websocket/pojo/MessageBean.java

+package com.wangxiaolu.promotion.websocket.pojo;
+
+import cn.hutool.json.JSONObject;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
+
+
+/**
+ * 消息bean
+ * @author : lvbencai
+ * @date : 2025/11/13
+ */
+@Data
+public class MessageBean {
+    @JsonProperty("msgType")
+    private String msgType;
+    @JsonProperty("data")
+    private Object data;
+
+
+}

src/main/java/com/wangxiaolu/promotion/xxljobtask/ActivityStautsHandler.java

@@ -6,18 +6,23 @@ import cn.hutool.core.date.DateUtil;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.wangxiaolu.promotion.domain.activity.mapper.entity.TemporaryActivityTaskClockDO;
 import com.wangxiaolu.promotion.enums.activity.ActivityClockTaskStatus;
+import com.wangxiaolu.promotion.enums.activity.ActivityPhotoType;
+import com.wangxiaolu.promotion.enums.activity.TemporaryActivityTaskClockMessageType;
 import com.wangxiaolu.promotion.pojo.user.dto.WxTemporaryInfoDto;
 import com.wangxiaolu.promotion.service.activity.manage.ActivityTypeQueryService;
 import com.wangxiaolu.promotion.service.activity.manage.EmployeeCoreTemporaryInfoService;
 import com.wangxiaolu.promotion.service.activity.temporary.TemporaryActivityTaskClockService;
 import com.wangxiaolu.promotion.service.activity.temporary.impl.SubscribeMessageService;
 import com.wangxiaolu.promotion.utils.DateUtils;
+import com.wangxiaolu.promotion.websocket.TemporaryActivityTaskClockSocketHandler;
+import com.wangxiaolu.promotion.websocket.pojo.MessageBean;
 import com.xxl.job.core.handler.annotation.XxlJob;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.poi.ss.formula.functions.T;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import java.io.IOException;
 import java.text.SimpleDateFormat;
 import java.util.Date;
 import java.util.HashMap;
@@ -39,7 +44,8 @@ public class ActivityStautsHandler {
     private TemporaryActivityTaskClockService taskClockService;
     @Autowired
     private SubscribeMessageService messageService;
-
+    @Autowired
+    private TemporaryActivityTaskClockSocketHandler taskClockSocketHandler;
     @Autowired
     private EmployeeCoreTemporaryInfoService temporaryInfoService;
 
@@ -47,20 +53,33 @@ public class ActivityStautsHandler {
     public void sendSubscribeMessage() {
 
         DateTime dateTime = DateUtil.offsetMinute(new Date(), 10);
-
         // 1. 查询订单信息和用户订阅记录
         List<TemporaryActivityTaskClockDO> list = taskClockService.list(new LambdaQueryWrapper<TemporaryActivityTaskClockDO>()
-                .eq(TemporaryActivityTaskClockDO::getTaskStatus, ActivityClockTaskStatus.TO_BE_START)
-//                .eq(TemporaryActivityTaskClockDO::getSubscribeStatus, 1)
-//                .eq(TemporaryActivityTaskClockDO::getIsSendSubscribe, 0)
-                .le(TemporaryActivityTaskClockDO::getRequiredlockTime, dateTime)
-                .ge(TemporaryActivityTaskClockDO::getRequiredlockTime, new Date()));
+                .eq(TemporaryActivityTaskClockDO::getTaskStatus, ActivityClockTaskStatus.TO_BE_START.getType())
+                .eq(TemporaryActivityTaskClockDO::getTaskType, ActivityPhotoType.RANDOM_TASK.getType())
+//                .le(TemporaryActivityTaskClockDO::getRequiredlockTime, dateTime)
+//                .ge(TemporaryActivityTaskClockDO::getRequiredlockTime, new Date())
+                );
 
         for (TemporaryActivityTaskClockDO taskClockDO : list) {
             if (taskClockDO.getSubscribeStatus() != 1) {
+
+                // 发送websocket 通知用户
+                try {
+//                    taskClockSocketHandler.handleMessage(taskClockDO.getTemporaryId().toString(), "订阅通知发送成功");
+//                    taskClockSocketHandler.sendToUser(taskClockDO.getTemporaryId().toString(), "订阅通知发送成功");
+                    MessageBean msgBean = new MessageBean();
+                    msgBean.setMsgType(TemporaryActivityTaskClockMessageType.RANDOM_TASK_START.getType());
+                    msgBean.setData(taskClockDO);
+                    taskClockSocketHandler.sendToUser(taskClockDO.getTemporaryId().toString(), msgBean);
                     log.info("用户{}未订阅活动打卡通知", taskClockDO.getTemporaryName());
-                taskClockDO.setIsSendSubscribe(1);
-                taskClockService.updateById(taskClockDO);
+//                    taskClockDO.setIsSendSubscribe(1);
+//                    taskClockService.updateById(taskClockDO);
+                } catch (IOException e) {
+                    // 添加异常处理
+                    log.error("发送Websocket订阅通知失败", e);
+                    throw new RuntimeException(e);
+                }
                 continue;
             }
 
@@ -100,60 +119,6 @@ public class ActivityStautsHandler {
 
     }
 
-    @XxlJob("sendWebsocketMessage")
-    public void sendWebsocketMessage() {
-
-        DateTime dateTime = DateUtil.offsetMinute(new Date(), 10);
-        /**
-         * 10分钟后结束打卡的，（已经进行中5分钟了）
-         * 发送websocket消息
-         */
-
-        // 1. 查询订单信息和用户订阅记录
-        List<TemporaryActivityTaskClockDO> list = taskClockService.list(new LambdaQueryWrapper<TemporaryActivityTaskClockDO>()
-                .eq(TemporaryActivityTaskClockDO::getTaskStatus, ActivityClockTaskStatus.STARTING)
-                .eq(TemporaryActivityTaskClockDO::getSubscribeStatus, 1)
-                .eq(TemporaryActivityTaskClockDO::getIsSendSubscribe, 0)
-                .le(TemporaryActivityTaskClockDO::getRequiredlockTime, dateTime)
-                .ge(TemporaryActivityTaskClockDO::getRequiredlockTime, new Date()));
-
-        for (TemporaryActivityTaskClockDO taskClockDO : list) {
-
-            // 2. 构建模板数据（根据小程序订阅模板的字段定义）
-            Map<String, String> data = new HashMap<>();
-            // 模板中的字段1
-            data.put("thing1", "随机任务已开启");
-            // 模板中的字段2
-            data.put("time2", new SimpleDateFormat("yyyy-MM-dd HH:mm").format(new Date()));
-            // 模板中的字段3
-            data.put("requiredlockTime", DateUtil.format(taskClockDO.getRequiredlockTime(), "yyyy-MM-dd HH:mm"));
-
-            // 3. 发送通知
-            // 获取openid
-            WxTemporaryInfoDto wxTemporaryInfoDto = temporaryInfoService.selectById(taskClockDO.getTemporaryId());
-
-            String openid = wxTemporaryInfoDto.getOpenId();
-            // 3. 发送通知
-            String templateId = SUBSCRIBE_MESSAGE_TEMPLATE_ID;
-            boolean success = messageService.sendSubscribeMessage(
-                    openid,
-                    templateId,
-                    // 跳转页面
-                    "/pages/order/detail?",
-                    data
-            );
-            if (success) {
-                log.info("促销员{}订阅通知发送成功", taskClockDO.getTemporaryName());
-                taskClockDO.setSubscribeTime(new Date());
-            }
-            // 修改状态 待开始 -> 进行中
-            taskClockDO.setTaskStatus(ActivityClockTaskStatus.STARTING.getType());
-
-            taskClockService.updateById(taskClockDO);
-
-        }
-
-    }
 
 
 }