1、通信中间件去除上报消息测试类（去除注解） ; 2、remote模块，webSocket模块修改完善，增加机构orgTag控制； 3、从SSO模块获...

liurunyu

3 天以前 eb2d3ef0e4fa144d5a76823709a1ab2bc27a2ae9

1、通信中间件去除上报消息测试类（去除注解） ;
2、remote模块，webSocket模块修改完善，增加机构orgTag控制；
3、从SSO模块获取当前用户信息部分进行集成完善，以供其他部分共享应用。

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/aop/SsoAspect.java

@@ -14,10 +14,7 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.annotation.Order;
import org.springframework.http.*;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;

import java.lang.reflect.Method;
import java.util.Objects;
@@ -31,15 +28,8 @@
    @Value("${pipIrr.global.dev}")
    public String isDevStage ;//是否为开发阶段

    @Value("${pipIrr.sso.checkUrl}")
    public String ssoCheckUrl;

    private RestTemplate restTemplate;

    @Autowired
    public void setRestTemplate(RestTemplate restTemplate){
        this.restTemplate = restTemplate ;
    }
    private SsoCheck ssoCheck ;

    @Pointcut("@annotation(com.dy.common.aop.SsoAop)")
    public void ssoPointCut() {
@@ -55,7 +45,7 @@
            SsoAop aop = method.getAnnotation(SsoAop.class) ;
            if (Objects.nonNull(aop)){
                String token = UserTokenContext.get() ;
                Object rObj = this.check(token);
                Object rObj = this.ssoCheck.check(token);
                if(rObj != null){
                    if(rObj instanceof SsoVo ssoVo){
                        if(ssoVo.logined){
@@ -90,37 +80,5 @@
        }
    }

    /**
     * 调用SSO系统进行验证
     * @param token Header中的用户token
     * @return 返回对象
     */
    private Object check(String token){
        if(!StringUtils.isNullOrEmpty(ssoCheckUrl)){
            String url = UriComponentsBuilder.fromUriString(ssoCheckUrl)
                    .queryParam("token", token)
                    .build()
                    .toUriString();
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.parseMediaType("application/json;charset=UTF-8"));

            HttpEntity<?> httpEntity = new HttpEntity<>(headers);
            ResponseEntity<SsoVo> response = null;
            try {
                // 通过Get方式调用接口
                response = restTemplate.exchange(url, HttpMethod.GET, httpEntity, SsoVo.class);
            } catch (Exception e) {
                e.printStackTrace();
                return BaseResponseUtils.buildError("后端系统出错，访问单点登录系统异常");
            }
            if(response == null){
                return BaseResponseUtils.buildError("后端系统出错，访问单点登录系统异常");
            }else{
                return response.getBody();
            }
        }else {
            return BaseResponseUtils.buildError("后端系统出错，未得到SsoCheckUrl");
        }
    }

}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/aop/SsoCheck.java

New file
@@ -0,0 +1,100 @@
package com.dy.common.aop;

import com.dy.common.webUtil.BaseResponseUtils;
import com.mysql.cj.util.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.*;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;

/**
 * @Author: liurunyu
 * @Date: 2025/7/22 17:09
 * @Description
 */
@Component
public class SsoCheck {

    @Value("${pipIrr.sso.checkUrl}")
    private String ssoCheckUrl;

    private RestTemplate restTemplate;

    @Autowired
    public void setRestTemplate(RestTemplate restTemplate){
        this.restTemplate = restTemplate ;
    }

    /**
     * 调用SSO系统进行验证
     * @param token Header中的用户token
     * @return 返回对象
     */
    public Object check(String token){
        if(!StringUtils.isNullOrEmpty(ssoCheckUrl)){
            String url = UriComponentsBuilder.fromUriString(ssoCheckUrl)
                    .queryParam("token", token)
                    .build()
                    .toUriString();
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.parseMediaType("application/json;charset=UTF-8"));

            HttpEntity<?> httpEntity = new HttpEntity<>(headers);
            ResponseEntity<SsoVo> response = null;
            try {
                // 通过Get方式调用接口
                response = this.restTemplate.exchange(url, HttpMethod.GET, httpEntity, SsoVo.class);
            } catch (Exception e) {
                e.printStackTrace();
                return BaseResponseUtils.buildError("后端系统出错，访问单点登录系统异常");
            }
            if(response == null){
                return BaseResponseUtils.buildError("后端系统出错，访问单点登录系统异常");
            }else{
                return response.getBody();
            }
        }else {
            return BaseResponseUtils.buildError("后端系统出错，未得到SsoCheckUrl");
        }
    }


    /**
     * 调用SSO系统进行验证
     * @param token Header中的用户token
     * @param power 一个权限
     * @param allPower 多个权限
     * @param anyPower 多个权限
     * @return 返回对象
     */
    public Object check(String token, String power, String[] allPower, String[] anyPower){
        if(!StringUtils.isNullOrEmpty(ssoCheckUrl)){
            String url = UriComponentsBuilder.fromUriString(ssoCheckUrl)
                    .queryParam("token", token)
                    .queryParam("power", power)
                    .queryParam("allPower", (Object) allPower)
                    .queryParam("anyPower", (Object) anyPower)
                    .build()
                    .toUriString();
            HttpHeaders headers = new HttpHeaders();
            HttpEntity<?> httpEntity = new HttpEntity<>(headers);
            ResponseEntity<SsoVo> response = null;
            try {
                // 通过Get方式调用接口
                response = restTemplate.exchange(url, HttpMethod.GET, httpEntity, SsoVo.class);
            } catch (Exception e) {
                e.printStackTrace();
                return BaseResponseUtils.buildError("后端系统出错，访问单点登录系统异常");
            }
            if(response == null){
                return BaseResponseUtils.buildError("后端系统出错，访问单点登录系统异常");
            }else{
                return response.getBody();
            }
        }else {
            return BaseResponseUtils.buildError("后端系统出错，未得到SsoCheckUrl");
        }
    }
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/aop/SsoPowerAspect.java

@@ -14,13 +14,7 @@
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.annotation.Order;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.ResponseEntity;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.util.UriComponentsBuilder;

import java.lang.reflect.Method;
import java.util.Objects;
@@ -34,15 +28,8 @@
    @Value("${pipIrr.global.dev}")
    public String isDevStage ;//是否为开发阶段

    @Value("${pipIrr.sso.checkUrl}")
    public String SsoCheckUrl ;

    private RestTemplate restTemplate;

    @Autowired
    public void setRestTemplate(RestTemplate restTemplate){
        this.restTemplate = restTemplate ;
    }
    private SsoCheck ssoCheck ;

    @Pointcut("@annotation(com.dy.common.aop.SsoPowerAop)")
    public void ssoPowerPointCut() {
@@ -68,7 +55,7 @@
                    String[] allPower = aop.ifAllPower() ;
                    String[] anyPower = aop.ifAnyPower() ;
                    String token = UserTokenContext.get() ;
                    Object rObj = this.check(token, power, allPower, anyPower);
                    Object rObj = this.ssoCheck.check(token, power, allPower, anyPower);
                    if(rObj != null){
                        if(rObj instanceof SsoVo ssoVo){
                            if(ssoVo.logined){
@@ -104,41 +91,5 @@
        }
    }

    /**
     * 调用SSO系统进行验证
     * @param token Header中的用户token
     * @param power 一个权限
     * @param allPower 多个权限
     * @param anyPower 多个权限
     * @return 返回对象
     */
    private Object check(String token, String power, String[] allPower, String[] anyPower){
        if(!StringUtils.isNullOrEmpty(SsoCheckUrl)){
            String url = UriComponentsBuilder.fromUriString(SsoCheckUrl)
                    .queryParam("token", token)
                    .queryParam("power", power)
                    .queryParam("allPower", (Object) allPower)
                    .queryParam("anyPower", (Object) anyPower)
                    .build()
                    .toUriString();
            HttpHeaders headers = new HttpHeaders();
            HttpEntity<?> httpEntity = new HttpEntity<>(headers);
            ResponseEntity<SsoVo> response = null;
            try {
                // 通过Get方式调用接口
                response = restTemplate.exchange(url, HttpMethod.GET, httpEntity, SsoVo.class);
            } catch (Exception e) {
                e.printStackTrace();
                return BaseResponseUtils.buildError("后端系统出错，访问单点登录系统异常");
            }
            if(response == null){
                return BaseResponseUtils.buildError("后端系统出错，访问单点登录系统异常");
            }else{
                return response.getBody();
            }
        }else {
            return BaseResponseUtils.buildError("后端系统出错，未得到SsoCheckUrl");
        }
    }

}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/contant/Constant.java

@@ -7,6 +7,6 @@
    public static final int AspectOrderDataSource = 1 ;


    //public static final String UserTokenKeyInHeader = "token" ;
    public static final String UserTokenKeyInHeader = "Token" ;
    //public static final String TokenKeyInHeader = "token" ;
    public static final String TokenKeyInHeader = "Token" ;
}

 pipIrr-platform/pipIrr-common/src/main/java/com/dy/common/webFilter/UserTokenFilter.java

@@ -32,7 +32,7 @@
            // 如果不是非过滤项，执行过滤逻辑
            HttpServletRequest request = (HttpServletRequest) servletRequest;
            HttpServletResponse response = (HttpServletResponse) servletResponse;
            String token = request.getHeader(Constant.UserTokenKeyInHeader);
            String token = request.getHeader(Constant.TokenKeyInHeader);
            if(!StringUtils.isNullOrEmpty(token)){
                UserTokenContext.set(token);
                filterChain.doFilter(servletRequest, servletResponse);

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/PipIrrMwRtuApplication.java

@@ -10,7 +10,7 @@
import org.springframework.context.annotation.FilterType;
import org.springframework.scheduling.annotation.EnableScheduling;

@EnableScheduling
//@EnableScheduling
@SpringBootApplication
@EnableAspectJAutoProxy
@ComponentScan(basePackages = {"com.dy.common", "com.dy.pipIrrGlobal", "com.dy.rtuMw"},

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/server/msCenter/TestMsResource.java

@@ -10,11 +10,11 @@
 * @Description
 */
//使起作用，本类注解@Component，并在PipIrrMwRtuApplication注解@EnableScheduling
@Component
//@Component
public class TestMsResource {

    // 设置定时N秒一次
    @Scheduled(cron = "0/4 * * * * ?")
    //@Scheduled(cron = "0/4 * * * * ?")
    public void test1() throws Exception {
        MsObj msObj = new MsObj() ;
        setAttr4OpenCloseValve("开阀", msObj) ;
@@ -22,7 +22,7 @@
    }

    // 设置定时N秒一次
    @Scheduled(cron = "0/6 * * * * ?")
    //@Scheduled(cron = "0/6 * * * * ?")
    public void test2() throws Exception {
        MsObj msObj = new MsObj() ;
        setAttr4OpenCloseValve("关阀", msObj) ;
@@ -30,7 +30,7 @@
    }

    // 设置定时N秒一次
    @Scheduled(cron = "0/10 * * * * ?")
    //@Scheduled(cron = "0/10 * * * * ?")
    public void test4() throws Exception {
        MsObj msObj = new MsObj() ;
        setAttr4StateAlarm("阀态", msObj) ;
@@ -40,7 +40,7 @@
    }

    // 设置定时N秒一次
    @Scheduled(cron = "0/15 * * * * ?")
    //@Scheduled(cron = "0/15 * * * * ?")
    public void test3() throws Exception {
        MsObj msObj = new MsObj() ;
        setAttr4StateAlarm("阀态", msObj) ;

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/web/comResult/CommandResultDeal.java

@@ -48,7 +48,7 @@
            restTemplate.getMessageConverters().set(1,new StringHttpMessageConverter(StandardCharsets.UTF_8));
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.parseMediaType("application/json;charset=UTF-8"));
            headers.set(Constant.UserTokenKeyInHeader, ServerProperties.orgTag);
            headers.set(Constant.TokenKeyInHeader, ServerProperties.orgTag);
            HttpEntity<?> httpEntity = new HttpEntity<>(data, headers);
            ResponseEntity<WebResponseVo> response = null;
            try {
@@ -78,7 +78,7 @@
            restTemplate.getMessageConverters().set(1,new StringHttpMessageConverter(StandardCharsets.UTF_8));
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.parseMediaType("application/json;charset=UTF-8"));
            headers.set(Constant.UserTokenKeyInHeader, ServerProperties.orgTag);
            headers.set(Constant.TokenKeyInHeader, ServerProperties.orgTag);
            HttpEntity<?> httpEntity = new HttpEntity<>(subMsg, headers);
            ResponseEntity<WebResponseVo> response = null;
            try {

 pipIrr-platform/pipIrr-mw/pipIrr-mw-rtu/src/main/java/com/dy/rtuMw/web/webRequest/WebRequestDeal.java

@@ -47,7 +47,7 @@
            restTemplate.getMessageConverters().set(1,new StringHttpMessageConverter(StandardCharsets.UTF_8));
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.parseMediaType("application/json;charset=UTF-8"));
            headers.set(Constant.UserTokenKeyInHeader, ServerProperties.orgTag);
            headers.set(Constant.TokenKeyInHeader, ServerProperties.orgTag);
            HttpEntity<?> httpEntity = new HttpEntity<>(obj, headers);
            ResponseEntity<BaseResponse> response = null;
            try {

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/largeScreen/WebSocketHeartBeat.java

@@ -18,7 +18,7 @@
    // 设置定时十秒一次
    @Scheduled(cron = "0/10 * * * * ?")
    public void WsHeartBeat() throws Exception {
        WebSocketServer.sendMessage2AllClient(getHeartBeatMessage());
        WebSocketServer.sendMessage2AllClient(null, getHeartBeatMessage());
    }

    public static String getHeartBeatMessage() {

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/largeScreen/WebSocketServer.java

@@ -1,6 +1,9 @@
package com.dy.pipIrrRemote.largeScreen;

import com.dy.common.aop.SsoCheck;
import com.dy.common.aop.SsoVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;

@@ -18,7 +21,7 @@
 * 管理了（在嵌入式web Servlet环境中）
 */
@Slf4j
@ServerEndpoint("/websocket/ls/{id}")
@ServerEndpoint("/websocket/ls/{token}")
@Component
@Scope("prototype") // 非单例，每次请求都会创建新的实例
public class WebSocketServer {
@@ -31,32 +34,43 @@

    // 客户端连接会话，通过它给客户端发送数据
    private Session session;

    private String orgTag;

    // 客户端上线时刻
    //private String onLineDt ;
    // 客户端id
    private String id = "";
    // 客户端
    private String token = "";

    @Autowired
    private SsoCheck ssoCheck ;

    /**
     * 连接建立成功调用的方法
     * @param session websocket会话对象
     * @param id 客户端id
     * @param token 客户端id
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("id") String id) {
        this.session = session;
        //this.onLineDt = DateTime.yyyy_MM_dd_HH_mm_ss() ;
        this.id = id;
        if(this.id == null || this.id.length() == 0){
            this.id = "" + System.nanoTime() ;
        }
        this.sendMessage(WebSocketHeartBeat.getHeartBeatMessage());
    public void onOpen(Session session, @PathParam("token") String token) {
        if(this.token == null || this.token.length() == 0){
            this.session = session;
            //this.onLineDt = DateTime.yyyy_MM_dd_HH_mm_ss() ;
            this.token = token;
            this.sendMessage(WebSocketHeartBeat.getHeartBeatMessage());

        if (webSocketMap.containsKey(id)) {
            webSocketMap.remove(id);
            webSocketMap.put(id, this);
        } else {
            webSocketMap.put(id, this);
            WebSocketServer.addOnlineCount();
            if (webSocketMap.containsKey(token)) {
                webSocketMap.remove(token);
                webSocketMap.put(token, this);
            } else {
                webSocketMap.put(token, this);
                WebSocketServer.addOnlineCount();
            }
            Object rObj = this.ssoCheck.check(token);
            if(rObj != null) {
                if (rObj instanceof SsoVo ssoVo) {
                    this.orgTag = ssoVo.dataSourceName ;
                }
            }
        }
    }

@@ -65,12 +79,12 @@
     */
    @OnClose
    public void onClose() {
        if (webSocketMap.containsKey(id)) {
            webSocketMap.remove(id);
        if (webSocketMap.containsKey(token)) {
            webSocketMap.remove(token);
            //从set中删除
            WebSocketServer.subOnlineCount();
        }
        log.info("客户端:" + id + "，关闭了websocket");
        log.info("客户端:" + token + "，关闭了websocket");
    }

    /**
@@ -80,7 +94,7 @@
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("客户端:" + id + "，websocket报文:" + message);
        log.info("客户端:" + token + "，websocket报文:" + message);
    }

    /**
@@ -90,7 +104,7 @@
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("客户端:" + this.id + "，websocket会话异常，原因:" + error.getMessage());
        log.error("客户端:" + this.token + "，websocket会话异常，原因:" + error.getMessage());
    }

    /**
@@ -100,29 +114,35 @@
        try{
            this.session.getBasicRemote().sendText(message);
        }catch (Exception e){
            log.error("客户端:" + id + "，websocket网络发送数据异常", e);
            log.error("客户端:" + token + "，websocket网络发送数据异常", e);
        }
    }

    /**
     * 服务器主动群推送消息
     */
    public static void sendMessage2AllClient(String message) throws IOException {
        ConcurrentHashMap.KeySetView<String, WebSocketServer> ids = webSocketMap.keySet();
        for (String id : ids) {
            WebSocketServer webSocketServer = webSocketMap.get(id);
            webSocketServer.sendMessage(message);
    public static void sendMessage2AllClient(String orgTag, String message) throws IOException {
        ConcurrentHashMap.KeySetView<String, WebSocketServer> tokens = webSocketMap.keySet();
        for (String token : tokens) {
            WebSocketServer webSocketServer = webSocketMap.get(token);
            if(orgTag == null){
                webSocketServer.sendMessage(message);
            }else{
                if(webSocketServer.orgTag != null && webSocketServer.orgTag.equals(orgTag)){
                    webSocketServer.sendMessage(message);
                }
            }
        }
    }

    /**
     * 服务器指定客户端推送消息
     */
    public static void sendMessage2OneClient(String message, String id) throws IOException {
        if (message != null && message.length() != 0 && webSocketMap.containsKey(id)) {
            webSocketMap.get(id).sendMessage(message);
    public static void sendMessage2OneClient(String message, String token) throws IOException {
        if (message != null && message.length() != 0 && webSocketMap.containsKey(token)) {
            webSocketMap.get(token).sendMessage(message);
        } else {
            log.error("客户端" + id + "，不在线！");
            log.error("客户端" + token + "，不在线！");
        }
    }


 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/msCenter/CenterMsReceiveCtrl.java

@@ -46,8 +46,8 @@
    @PostMapping(path = "receive", consumes = MediaType.APPLICATION_JSON_VALUE)
    public BaseResponse<Boolean> receive(@RequestBody List<JSONObject> list, HttpServletRequest req, HttpServletResponse rep) {
        //通信中间件传过来的机构tag，以用于查找数据源
        String token = req.getHeader(Constant.UserTokenKeyInHeader);
        DataSourceContext.set(token);
        String orgTag = req.getHeader(Constant.TokenKeyInHeader);
        DataSourceContext.set(orgTag);

        if(list != null && list.size() > 0){
            for (JSONObject jo : list) {
@@ -69,7 +69,7 @@
                    }
                }
            }
            sendByWebSocket(list) ;
            sendByWebSocket(orgTag, list) ;
        }
        return null ;
    }
@@ -78,12 +78,12 @@
     * 通过websocket把消息推送出去，当前接收方是前端的大屏展示模块
     * @param list
     */
    private void sendByWebSocket(List<JSONObject> list){
    private void sendByWebSocket(String orgTag, List<JSONObject> list){
        WebSocketMessage vo = new WebSocketMessage() ;
        vo.type = WebSocketMessage.TYPE_JSON ;
        vo.content = list ;
        try {
            WebSocketServer.sendMessage2AllClient(JSON.toJSONString(vo));
            WebSocketServer.sendMessage2AllClient(orgTag, JSON.toJSONString(vo));
        }catch (Exception e){
            log.error("推送消息失败", e) ;
        }

 pipIrr-platform/pipIrr-web/pipIrr-web-remote/src/main/java/com/dy/pipIrrRemote/rtuUpgrage/RtuUpgradeStateReceiverCtrl.java

@@ -72,7 +72,7 @@
        info.ugRtuStateList = info.ugRtuStateList.stream().sorted(comparator).collect(Collectors.toList());

        //通信中间件传过来的机构tag，以用于查找数据源
        String token = req.getHeader(Constant.UserTokenKeyInHeader);
        String token = req.getHeader(Constant.TokenKeyInHeader);
        DataSourceContext.set(token);

        if(cache == null){