feat(websocket): 实现 websocket广播功能

qfrjava

2025-04-18 fc3cf8628d46b02a7c86c865889954fa6edb6c9a

feat(websocket): 实现 websocket广播功能

- 新增 SMTWebSocketHandler 组件，实现广播消息到所有连接客户端的功能
- 在 SMTWebSocketConfig 中注入 SMTWebSocketHandler，并完成 websocket路由配置
- 修改 checkForTicketStatusChanges 定时任务，添加对新订单的处理逻辑- 在 SMTAIServerApp 中添加 webSocketApp 方法，用于发送消息到 websocket
- 更新 SMTJavaAIControl 中的 createOrder 方法，增加 oname 和 otype 字段
- 修改 SMTJsonFlowNodeScript，添加获取当前日期、UUID 和 websocket 广播的方法

 JAVA/SMTAIServer/src/main/java/com/smtaiserver/smtaiserver/control/SMTJavaAIControl.java

@@ -295,53 +295,53 @@
        

        return tranReq.returnJson(jsonWr);

    }

	
    public ModelAndView createOrder(SMTAIServerRequest tranReq) throws Exception

    {

        String title = tranReq.convParamToString("title", true);

        String wstypeid = tranReq.convParamToString("wstypeid", false);

        String wstypename = tranReq.convParamToString("wstypename", false);

        String urgencylevel = tranReq.convParamToString("urgencylevel", false);

        String shape = tranReq.convParamToString("shape", false);

        String deadline = tranReq.convParamToString("deadline", false);

        String stepstatus = tranReq.convParamToString("stepstatus", false);

        String originatetypetext = tranReq.convParamToString("originatetypetext", false);

        String content = tranReq.convParamToString("content", false);

        String address = tranReq.convParamToString("address", false);

        String stepid = tranReq.convParamToString("stepid", false);

        String org = tranReq.convParamToString("org", false);

        String deptname = tranReq.convParamToString("deptname", false);

        String dealtime = tranReq.convParamToString("dealtime", false);

        String username = tranReq.convParamToString("username", false);

        String contactname = tranReq.convParamToString("contactname", false);

        String contactphone = tranReq.convParamToString("contactphone", false);

        String acceptuserid = tranReq.convParamToString("acceptuserid", false);

        String acceptusername = tranReq.convParamToString("acceptusername", false);

        String acceptuseraccount = tranReq.convParamToString("acceptuseraccount", false);

        String dealuserid = tranReq.convParamToString("dealuserid", false);

        String dealusername = tranReq.convParamToString("dealusername", false);

        String dealuseraccount = tranReq.convParamToString("dealuseraccount", false);

        String flowstatus = tranReq.convParamToString("flowstatus", false);

        String attachment = tranReq.convParamToString("attachment", false);

        String creator = tranReq.convParamToString("creator", false);

        String createtime = tranReq.convParamToString("createtime", false);

        String isplan = tranReq.convParamToString("isplan", false);

        String originateid = tranReq.convParamToString("originateid", false);

        String appointstarttime = tranReq.convParamToString("appointstarttime", false);

        String appointendtime = tranReq.convParamToString("appointendtime", false);

        String parentid = tranReq.convParamToString("parentid", false);

        String relateids = tranReq.convParamToString("relateids", false);



            SMTDatabase db = SMTAIServerApp.getApp().allocDatabase();

            try

            {

  public ModelAndView createOrder(SMTAIServerRequest tranReq) throws Exception {

    String title = tranReq.convParamToString("title", true);

    String wstypeid = tranReq.convParamToString("wstypeid", false);

    String wstypename = tranReq.convParamToString("wstypename", false);

    String urgencylevel = tranReq.convParamToString("urgencylevel", false);

    String shape = tranReq.convParamToString("shape", false);

    String deadline = tranReq.convParamToString("deadline", false);

    String stepstatus = tranReq.convParamToString("stepstatus", false);

    String originatetypetext = tranReq.convParamToString("originatetypetext", false);

    String content = tranReq.convParamToString("content", false);

    String address = tranReq.convParamToString("address", false);

    String stepid = tranReq.convParamToString("stepid", false);

    String org = tranReq.convParamToString("org", false);

    String deptname = tranReq.convParamToString("deptname", false);

    String dealtime = tranReq.convParamToString("dealtime", false);

    String username = tranReq.convParamToString("username", false);

    String contactname = tranReq.convParamToString("contactname", false);

    String contactphone = tranReq.convParamToString("contactphone", false);

    String acceptuserid = tranReq.convParamToString("acceptuserid", false);

    String acceptusername = tranReq.convParamToString("acceptusername", false);

    String acceptuseraccount = tranReq.convParamToString("acceptuseraccount", false);

    String dealuserid = tranReq.convParamToString("dealuserid", false);

    String dealusername = tranReq.convParamToString("dealusername", false);

    String dealuseraccount = tranReq.convParamToString("dealuseraccount", false);

    String flowstatus = tranReq.convParamToString("flowstatus", false);

    String attachment = tranReq.convParamToString("attachment", false);

    String creator = tranReq.convParamToString("creator", false);

    String createtime = tranReq.convParamToString("createtime", false);

    String isplan = tranReq.convParamToString("isplan", false);

    String originateid = tranReq.convParamToString("originateid", false);

    String appointstarttime = tranReq.convParamToString("appointstarttime", false);

    String appointendtime = tranReq.convParamToString("appointendtime", false);

    String parentid = tranReq.convParamToString("parentid", false);

    String relateids = tranReq.convParamToString("relateids", false);

    String oname = tranReq.convParamToString("oname", false);

    String otype = tranReq.convParamToString("otype", false);



    SMTDatabase db = SMTAIServerApp.getApp().allocDatabase();

    try {

      db.executeSQL(

          "INSERT INTO work_order_list("

              + "order_id, title, wstypeid, wstypename, urgencylevel, shape, deadline, stepstatus, "

              + "originatetypetext, content, address, stepid, org, deptname, dealtime, username, contactname, "

              + "originatetypetext, content, address, step_id, org, deptname, dealtime, username, contactname, "

              + "contactphone, acceptuserid, acceptusername, acceptuseraccount, dealuserid, dealusername, "

              + "dealuseraccount, flowstatus, attachment, creator, createtime, isplan, originateid, "

              + "appointstarttime, appointendtime, parentid, relateids,is_notify"

              + "appointstarttime, appointendtime, parentid, relateids,is_notify,oname,otype"

              + ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,?)",

          new Object[] {

            SMTStatic.newUUID(),

@@ -378,14 +378,16 @@
            appointendtime,

            parentid,

            relateids,

            "N"

            "N",

            oname,

            otype

          });

            }catch (Exception e){

                throw new Exception("数据库错误", e);

            }

    } catch (Exception e) {

      throw new Exception("数据库错误", e);

    }



            return tranReq.returnJsonState(true, null, null);

    }

    return tranReq.returnJsonState(true, null, null);

  }



  public ModelAndView updateOrderStatus(SMTAIServerRequest tranReq) throws Exception {

    String id = tranReq.convParamToString("id", true);


 JAVA/SMTAIServer/src/main/java/com/smtaiserver/smtaiserver/core/SMTAIServerApp.java

@@ -22,6 +22,7 @@
import com.smtaiserver.smtaiserver.javaai.qwen.SMTQwenAgentManager;

import com.smtaiserver.smtaiserver.javaai.qwen.SMTQwenApp;

import com.smtaiserver.smtaiserver.javaai.sse.SMTSSEBroadcastChat;

import com.smtaiserver.smtaiserver.web.SMTWebSocketHandler;

import com.smtservlet.core.SMTApp;

import com.smtservlet.core.SMTApp.SMTEhCacheManagerInitialize;

import com.smtservlet.util.Json;

@@ -42,6 +43,7 @@
import java.util.Set;

import java.util.regex.Matcher;

import java.util.regex.Pattern;

import javax.annotation.Resource;

import javax.servlet.http.HttpSessionEvent;

import net.sf.ehcache.Cache;

import net.sf.ehcache.CacheManager;

@@ -144,7 +146,8 @@
    private static Pattern                    _patGlobalMacro = Pattern.compile("\\{\\{\\{([\\w\\.]+)\\}\\}\\}");

    private static Logger                     _logger = LogManager.getLogger(SMTQwenApp.class);

    private static Pattern                     _patIsNumber = Pattern.compile("^[1-9]\\d*$");

	
    @Resource

    private  SMTWebSocketHandler webSocketHandler;

    private static String[]                    _encacheIdList = {

        "GlobalConfig",

        "MetricsDefMap",

@@ -212,7 +215,11 @@
            return chat;

        }

    }

	
    public void webSocketApp(String jsonString)

    {

        webSocketHandler.broadcast(jsonString);

    }



    public void setServiceEncache(SMTAIServerEncache serverEncache)

    {

        _serverEncache = serverEncache;


 JAVA/SMTAIServer/src/main/java/com/smtaiserver/smtaiserver/javaai/jsonflow/node/SMTJsonFlowNodeScript.java

@@ -16,6 +16,7 @@
import com.smtservlet.util.SMTJsonWriter;

import com.smtservlet.util.SMTStatic;

import java.util.ArrayList;

import java.util.Date;

import java.util.List;

import java.util.Map.Entry;

import org.mozilla.javascript.Context;

@@ -37,6 +38,17 @@
        public String getGlobalConfig(String key, String defaultValue) throws Exception {

          String globalConfig = (String) SMTAIServerApp.getApp().getGlobalConfig(key, defaultValue);

          return globalConfig;

        }

        public String getNowDate() throws Exception {

            return SMTStatic.toString(new Date());

        }

        public String getUUID() throws Exception {

            return  SMTStatic.newUUID();

        }

        public  void webSocketBroadcast(NativeObject nativeObject) throws Exception {

            SMTJsonWriter jsonWr = new SMTJsonWriter(false);

            SMTAIServerApp.convJSToJsonWriter(nativeObject, jsonWr);

            SMTAIServerApp.getApp().webSocketApp(jsonWr.toString());

        }



        public Object getArg(String key)

@@ -303,7 +315,7 @@
            SMTDatabase db = SMTAIServerApp.getApp().allocDatabase();

            return querySQL(db, sql, nvParams);

        }

		


        public NativeArray querySQL(String dsId, String sql, NativeObject nvParams) throws Exception

        {

            SMTDatabase db = SMTAIServerApp.getApp().getDataSource(dsId).allocDatabase();


 JAVA/SMTAIServer/src/main/java/com/smtaiserver/smtaiserver/scheduledTasks/checkForTicketStatusChanges.java

@@ -2,11 +2,13 @@

import com.smtaiserver.smtaiserver.core.SMTAIServerApp;
import com.smtaiserver.smtaiserver.database.SMTDatabase;
import com.smtaiserver.smtaiserver.web.SMTWebSocketHandler;
import java.util.Calendar;
import java.util.Date;
import lombok.extern.log4j.Log4j2;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@@ -20,10 +22,11 @@
@Log4j2
@Component
public class checkForTicketStatusChanges {
  private static Logger _logger = LogManager.getLogger(checkForTicketStatusChanges.class);
  private static final Logger _logger = LogManager.getLogger(checkForTicketStatusChanges.class);

  // 一秒一次
  @Scheduled(cron = "0/1 * * * * ?")
  @Autowired private SMTWebSocketHandler webSocketHandler;

  @Scheduled(cron = "0/10 * * * * ?") // 一秒一次
  public void checkForTicketStatusChanges() throws Exception {
    _logger.info("start checkForTicketStatusChanges task");
    SMTDatabase db = SMTAIServerApp.getApp().allocDatabase();
@@ -32,13 +35,17 @@
    cal.setTime(date);
    cal.add(Calendar.MINUTE, -5);
    Date time = cal.getTime();

    try {
      SMTDatabase.DBRecords dbRecords =
          db.querySQL(
              "select * from work_order_list where is_notify =? and  status_update_time > ?",
              "select order_id from work_order_list where is_notify =? and  status_update_time > ?",
              new Object[] {"N", time});
      _logger.info("查询到的数量：{}",dbRecords.getRowCount());
      dbRecords.getRecords().forEach(record -> {});

      _logger.info("查询到的数量：{}", dbRecords.getRowCount());
        for (SMTDatabase.DBRecord rec : dbRecords.getRecords()) {
//          SMTJavaAIControl.
        }

    } catch (Exception e) {
      _logger.error("checkForTicketStatusChanges error", e);

 JAVA/SMTAIServer/src/main/java/com/smtaiserver/smtaiserver/web/SMTWebSocketConfig.java

@@ -1,26 +1,27 @@
package com.smtaiserver.smtaiserver.web;



import org.springframework.context.annotation.Bean;

import org.springframework.context.annotation.Configuration;

import org.springframework.web.socket.config.annotation.EnableWebSocket;

import org.springframework.web.socket.config.annotation.WebSocketConfigurer;

import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;

import org.springframework.web.socket.handler.TextWebSocketHandler;

import org.springframework.web.socket.server.support.HttpSessionHandshakeInterceptor;



@Configuration

@EnableWebSocket

public class SMTWebSocketConfig implements WebSocketConfigurer

{

    @Bean

    public TextWebSocketHandler webSocketHandler() {

        return new SMTWebSocketHandler();





    private final SMTWebSocketHandler webSocketHandler;



    public SMTWebSocketConfig(SMTWebSocketHandler webSocketHandler) {

        this.webSocketHandler = webSocketHandler;

    }



    @Override

    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {

         registry.addHandler(webSocketHandler(), "/ws")

             .addInterceptors(new HttpSessionHandshakeInterceptor())

             .setAllowedOrigins("*");

        registry.addHandler(webSocketHandler, "/ws")

                .addInterceptors(new HttpSessionHandshakeInterceptor())

                .setAllowedOrigins("*");

    }

}


 JAVA/SMTAIServer/src/main/java/com/smtaiserver/smtaiserver/web/SMTWebSocketHandler.java

@@ -3,12 +3,13 @@
import java.util.Collections;

import java.util.HashSet;

import java.util.Set;



import org.springframework.stereotype.Component;

import org.springframework.web.socket.CloseStatus;

import org.springframework.web.socket.TextMessage;

import org.springframework.web.socket.WebSocketSession;

import org.springframework.web.socket.handler.TextWebSocketHandler;



@Component

public class SMTWebSocketHandler extends TextWebSocketHandler {



    private final Set<WebSocketSession> sessions = Collections.synchronizedSet(new HashSet<>());

@@ -37,4 +38,18 @@
        sessions.remove(session);

        System.out.println("Client disconnected: " + session.getId());

    }



    public void broadcast(String message) {

        synchronized (sessions) {

            for (WebSocketSession session : sessions) {

                if (session.isOpen()) {

                    try {

                        session.sendMessage(new TextMessage(message));

                    } catch (Exception e) {

                        e.printStackTrace();

                    }

                }

            }

        }

    }

}