feat: 停止当前聊天输出接口功能完善

servers/boe-server-all/src/main/java/com/xboe/config/MqConfig.java

@@ -0,0 +1,34 @@
+package com.xboe.config;
+
+import org.apache.activemq.command.ActiveMQTopic;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jms.annotation.EnableJms;
+import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
+import org.springframework.jms.config.JmsListenerContainerFactory;
+
+import javax.jms.ConnectionFactory;
+import javax.jms.Topic;
+
+@EnableJms
+@Configuration
+public class MqConfig {
+
+    /**
+     * 配置topic
+     */
+    @Bean
+    public Topic broadcastTopic() {
+        return new ActiveMQTopic("broadcast.session.termination");
+    }
+
+    // 配置JmsListenerContainerFactory为发布/订阅模式
+    @Bean
+    public JmsListenerContainerFactory<?> jmsListenerContainerFactory(ConnectionFactory connectionFactory) {
+        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
+        factory.setConnectionFactory(connectionFactory);
+        factory.setPubSubDomain(true); // 设置为发布/订阅模式
+        factory.setSubscriptionDurable(false); // 非持久订阅
+        return factory;
+    }
+}

servers/boe-server-all/src/main/java/com/xboe/module/boecase/mq/BroadcastMessageConsumer.java

@@ -0,0 +1,28 @@
+package com.xboe.module.boecase.mq;
+
+
+import com.xboe.module.boecase.service.ICaseAiChatService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jms.annotation.JmsListener;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+public class BroadcastMessageConsumer {
+
+    @Autowired
+    private ICaseAiChatService iCaseAiChatService;
+
+    /**
+     * 接收会话终止广播消息
+     *
+     * @param conversationId 会话ID
+     */
+    @JmsListener(destination = "broadcast.session.termination",
+            containerFactory = "jmsListenerContainerFactory")
+    public void receiveSessionTerminationBroadcastMessage(String conversationId) {
+        log.info("收到会话终止广播消息：{}", conversationId);
+        iCaseAiChatService.eventSourceCancel(conversationId);
+    }
+}

servers/boe-server-all/src/main/java/com/xboe/module/boecase/service/ICaseAiChatService.java

@@ -66,6 +66,13 @@ public interface ICaseAiChatService {
      */
     boolean stopChatOutput(String conversationId);
 
+    /**
+     * 取消eventSource
+     *
+     * @param conversationId 会话ID
+     */
+    void eventSourceCancel(String conversationId);
+
     /**
      * 消息反馈保存
      * likeStatus: 踩/赞

servers/boe-server-all/src/main/java/com/xboe/module/boecase/service/impl/CaseAiChatServiceImpl.java

@@ -10,6 +10,7 @@ import com.xboe.module.boecase.dao.CaseAiConversationsDao;
 import com.xboe.module.boecase.dao.CaseDocumentLogDao;
 import com.xboe.module.boecase.dao.CasesDao;
 import com.xboe.module.boecase.dto.CaseAiChatDto;
+import com.xboe.module.boecase.entity.AiChatConversationData;
 import com.xboe.module.boecase.dto.CaseAiMsgLikeDto;
 import com.xboe.module.boecase.entity.CaseAiConversations;
 import com.xboe.module.boecase.entity.CaseDocumentLog;
@@ -20,7 +21,6 @@ import com.xboe.module.boecase.service.ICaseAiChatService;
 import com.xboe.module.boecase.service.IElasticSearchIndexService;
 import com.xboe.module.boecase.vo.CaseAiMessageVo;
 import com.xboe.module.boecase.vo.CaseReferVo;
-import com.xboe.module.boecase.entity.AiChatConversationData;
 import com.xboe.module.boecase.vo.ConversationExcelVo;
 import com.xboe.system.organization.vo.OrgSimpleVo;
 import com.xboe.system.user.service.IUserService;
@@ -45,7 +45,11 @@ import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.jms.JmsException;
+import org.springframework.jms.annotation.JmsListener;
+import org.springframework.jms.core.JmsTemplate;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -55,7 +59,6 @@ import javax.servlet.http.HttpServletResponse;
 import java.io.File;
 import java.io.IOException;
 import java.io.OutputStream;
-import java.net.SocketTimeoutException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -100,6 +103,12 @@ public class CaseAiChatServiceImpl implements ICaseAiChatService {
     @Autowired
     private CasesDao casesDao;
 
+    @Autowired
+    private JmsTemplate jmsTemplate;
+
+    @Value("${activemq.topic.name}")
+    private String topicName;
+
     // 用于存储会话ID与EventSource的映射关系，以便能够中断特定会话
     private final Map<String, EventSource> conversationEventSourceMap = new ConcurrentHashMap<>();
 
@@ -760,6 +769,22 @@ public class CaseAiChatServiceImpl implements ICaseAiChatService {
 
     @Override
     public boolean stopChatOutput(String conversationId) {
+        log.info("收到停止会话 {} 的指令", conversationId);
+        // 发送广播消息，通知中断连接
+        try {
+            jmsTemplate.convertAndSend(topicName, conversationId);
+        } catch (JmsException e) {
+            log.error("发送停止会话 {} 输出时发生异常", conversationId, e);
+            return false;
+        }
+        return true;
+    }
+
+    /**
+     * 取消eventSource
+     */
+    @Override
+    public void eventSourceCancel(String conversationId) {
         EventSource eventSource = conversationEventSourceMap.get(conversationId);
         if (eventSource != null) {
             try {
@@ -767,16 +792,13 @@ public class CaseAiChatServiceImpl implements ICaseAiChatService {
                 eventSource.cancel();
                 // 注意：cancel()会触发onFailure回调，在onFailure中会清理资源
                 log.info("成功发送停止会话 {} 的指令", conversationId);
-                return true;
             } catch (Exception e) {
                 log.error("停止会话 {} 输出时发生异常", conversationId, e);
                 // 即使出现异常，也从Map中移除，避免内存泄漏
                 conversationEventSourceMap.remove(conversationId);
-                return false;
             }
         } else {
             log.info("未找到会话 {} 对应的事件源，可能已经完成或不存在", conversationId);
-            return true;
         }
     }