添加ws仓库设备

jyzb-api/src/main/java/com/junmp/jyzb/api/util/WsWarehouseReq.java

+package com.junmp.jyzb.api.util;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
+
+import java.util.List;
+
+
+@Data
+public class WsWarehouseReq {
+    @JsonProperty("deviceId")
+    private String deviceId;
+    @JsonProperty("onlineState")
+    private String onlineState;
+    @JsonProperty("orgId")
+    private String orgId;
+    @JsonProperty("warehouseId")
+    private String warehouseId;
+
+    @JsonProperty("boxList")
+    private List<warehouseDevs> warehouseDevs;
+
+
+    @Data
+    // 内部类表示box
+    public static class warehouseDevs {
+        @JsonProperty("devId")
+        private String devId;
+
+        @JsonProperty("devIP")
+        private int devIP;
+
+        @JsonProperty("devName")
+        private String devName;
+        @JsonProperty("devState")
+        private String devState;
+        @JsonProperty("devType")
+        private String devType;
+    }
+}
\ No newline at end of file

jyzb-biz/src/main/java/com/junmp/jyzb/cache/DeviceRedisCache.java

+package com.junmp.jyzb.cache;
+
+import com.junmp.jyzb.api.constant.JYZBConstant;
+import com.junmp.v2.cache.AbstractRedisCache;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+@Component
+public class DeviceRedisCache extends AbstractRedisCache<String> {
+
+    public DeviceRedisCache(RedisTemplate<String, String> redisTemplate) {
+        super(redisTemplate);
+    }
+
+    @Override
+    public String getCommonKeyPrefix() {
+        return JYZBConstant.JYZB_CACHE_PREFIX;
+    }
+}

jyzb-biz/src/main/java/com/junmp/jyzb/service/impl/CabinetServiceImpl.java

@@ -44,35 +44,10 @@ public class CabinetServiceImpl extends ServiceImpl<CabinetMapper, Cabinet> impl
     private PubOrgService pubOrgService;
     @Resource
     private CabinetBoxService cabinetBoxService;
-    @Resource
-    private PolicemanService policemanService;
-
-    @Resource
-    private CabinetBoxPoliceService cabinetBoxPoliceService;
 
-    @Resource
-    private CabinetBoxPoliceMapper cabinetBoxPoliceMapper;
     @Autowired
     private RabbitMQSendMsg MQ;
 
-    private List<Long> getAllOrgId(CabinetReq req, String includeLowerLevel) {
-        List<Long> allOrg = new ArrayList<>();
-        if (includeLowerLevel.equals("false")) {
-            allOrg.add(req.getOrgId());
-        } else if (includeLowerLevel.equals("true")) {
-            //查询某组织机构的本级及下级
-            allOrg = pubOrgService.getLowerOrg(req.getOrgId());
-        }
-        return allOrg;
-    }
-
-    private List<Cabinet> getAllCabinetByOrg( List<Long> allOrgId) {
-        List<Cabinet> allCabinet = new ArrayList<>();
-        allCabinet = cabinetMapper.getAllCabinetByOrgList(allOrgId);
-        return allCabinet;
-    }
-
-
     /**
      * 添加单警柜
      *

jyzb-biz/src/main/java/com/junmp/jyzb/websocket/WarehouseWebSocketServer.java

@@ -8,10 +8,18 @@ import com.alibaba.fastjson2.JSON;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.junmp.iot.server.ws.WebSocketServerListener;
+import com.junmp.iot.server.ws.impl.DefaultWebSocketServerProtocol;
+import com.junmp.junmpProcess.redis.FlowRedisCache;
+import com.junmp.jyzb.api.util.WsCabinetReq;
+import com.junmp.jyzb.api.util.WsWarehouseReq;
+import com.junmp.jyzb.cache.DeviceRedisCache;
 import com.junmp.jyzb.cache.OnlineRedisCache;
 import com.junmp.jyzb.entity.Cabinet;
+import com.junmp.jyzb.entity.CabinetBox;
 import com.junmp.jyzb.entity.Temp.OnlineState;
 import com.junmp.jyzb.service.CabinetService;
+import com.junmp.jyzb.utils.DateTimeUtil;
 import com.junmp.v2.ws.api.enums.ClientMsgType;
 import com.junmp.v2.ws.api.enums.ServerMsgType;
 import com.junmp.v2.ws.api.enums.SysMsgTypeEnum;
@@ -22,6 +30,8 @@ import com.junmp.v2.ws.channel.WsSessionContainer;
 import com.junmp.v2.ws.message.WsMessageHandle;
 import com.junmp.v2.ws.session.WsSessionHandle;
 import io.jsonwebtoken.JwtException;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;
@@ -32,164 +42,61 @@ import javax.websocket.server.PathParam;
 import javax.websocket.server.ServerEndpoint;
 import java.io.IOException;
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 
 
 @Component
-@ServerEndpoint("/warehouse/{data}")
-public class WarehouseWebSocketServer {
-    private static final Logger log = LoggerFactory.getLogger(com.junmp.v2.ws.server.ScreenWebSocketServer.class);
+@Slf4j
+public class WarehouseWebSocketServer implements WebSocketServerListener {
+//    private static final Logger log = LoggerFactory.getLogger(com.junmp.v2.ws.server.ScreenWebSocketServer.class);
     @Resource
     public CabinetService cabinetService;
     @Resource
-    private OnlineRedisCache onlineRedisCache;
+    private DeviceRedisCache redisCache;
 
-    public WarehouseWebSocketServer() {
+    @Override
+    public String uri() {
+        return "/device/ws";
     }
 
-    @OnOpen
-    public void onOpen(Session session, @PathParam("data") String data) {
-        String userId = null;
-
+    @SneakyThrows
+    @Override
+    public void onText(DefaultWebSocketServerProtocol protocol) {
         try {
-            // 获取当前时间的时间戳
-            Instant now = Instant.now();
-
-            // 转换为long类型
-            long timestamp = now.toEpochMilli();
-
-                Cabinet cabinet=cabinetService.getOne(new LambdaQueryWrapper<Cabinet>()
-                        .eq(Cabinet::getCabinetNum, data));
-                cabinet.setOnlineState(1);
-                cabinet.setUpdateTime(DateTime.now());
-//                cabinetService.updateById(cabinet);
-//                //将数据放入redis
-//                onlineRedisCache.addSessionValue(session.getId(),type,typeId,timestamp);
-
-
-        } catch (JwtException var15) {
-            log.error("建立连接异常>>" + var15.getMessage());
 
-            try {
-                session.close();
-                log.warn("因连接失败，会话自动关闭");
-            } catch (IOException var14) {
-                var14.printStackTrace();
-                log.error("会话关闭异常>>" + var14.getMessage());
-            }
-        }
-
-        WsSessionContainer wsContainer = new WsSessionContainer(session);
-        WsSession<WsSessionContainer> socketSession = new WsSession();
-        WsMsgDto replyMsg = new WsMsgDto();
-        boolean var13 = false;
 
-        try {
-            var13 = true;
-            replyMsg.setServerMsgType(ServerMsgType.REPLY_MSG_TYPE.getCode());
-//            replyMsg.setToUserId(userId);
-            replyMsg.setData(session.getId());
-            socketSession.setSessionId(session.getId());
-            socketSession.setUserId(userId);
-            socketSession.setWsApi(wsContainer);
-            socketSession.setToken(data);
-//            socketSession.setData();
-            socketSession.setConnectionTime(System.currentTimeMillis());
-            WsSessionHandle.addWsSession(socketSession);
-            var13 = false;
-        } finally {
-            if (var13) {
-                wsContainer.writeAndFlush(replyMsg);
-                WsCallbackApi wsCallbackApi = WsMessageHandle.getWsCallback(SysMsgTypeEnum.LISTENER_ON_OPEN.getCode());
-                if (ObjectUtil.isNotEmpty(wsCallbackApi)) {
-                    wsCallbackApi.callback(SysMsgTypeEnum.LISTENER_ON_OPEN.getCode(), (Object)null, socketSession);
-                }
+            String text = protocol.readText();
+            ObjectMapper objectMapper = new ObjectMapper();
+            WsWarehouseReq deviceData = objectMapper.readValue(text, WsWarehouseReq.class);
+            deviceData.setOnlineState("true");
+            String channel =  protocol.requestMessage().getChannelId();
+            String jsonData=objectMapper.writeValueAsString(deviceData);
+            redisCache.put(channel,jsonData+":"+deviceData.getOrgId()+":"+deviceData.getWarehouseId());
 
-                log.info("已建立连接");
-            }
+            protocol.response("true");
         }
-
-        wsContainer.writeAndFlush(replyMsg);
-        WsCallbackApi wsCallbackApi = WsMessageHandle.getWsCallback(SysMsgTypeEnum.LISTENER_ON_OPEN.getCode());
-        if (ObjectUtil.isNotEmpty(wsCallbackApi)) {
-            wsCallbackApi.callback(SysMsgTypeEnum.LISTENER_ON_OPEN.getCode(), (Object)null, socketSession);
+        catch (Exception e)
+        {
+            protocol.response("false");
         }
-
-        log.info("已建立连接");
     }
 
-    @OnMessage
-    public void onMessage(String message, Session channel) {
-        WsMsgDto wsMsgDto = (WsMsgDto) JSON.parseObject(message, WsMsgDto.class);
-        WsSession<WsSessionContainer> wsSession = WsSessionHandle.getSessionBySessionId(channel.getId());
-        if (ObjectUtil.isNotEmpty(wsSession) && ClientMsgType.USER_HEART.getCode().equals(wsMsgDto.getClientMsgType())) {
-            log.info(StrUtil.format("来自与客户端[{}]的心跳消息。", new Object[]{wsMsgDto.getFromUserId()}));
-            if (ObjectUtil.isNotEmpty(wsSession)) {
-                wsSession.setLastActiveTime(System.currentTimeMillis());
-            }
+    @Override
+    public void onClose(DefaultWebSocketServerProtocol protocol) {
 
-        } else if (!ObjectUtil.isEmpty(wsMsgDto.getFromUserId())) {
-            if (ObjectUtil.isNotEmpty(wsSession)) {
-                wsSession.setLastActiveTime(System.currentTimeMillis());
-                WsCallbackApi callbackApi;
-                if (ClientMsgType.OPEN_AI_TYPE.getCode().equals(wsMsgDto.getClientMsgType())) {
-                    log.info(StrUtil.format("来自与客户端[{}]调用OpenAi的消息。", new Object[]{wsMsgDto.getFromUserId()}));
-                    callbackApi = (WsCallbackApi) SpringUtil.getBean(WsCallbackApi.class);
-                    if (ObjectUtil.isNull(callbackApi)) {
-                        log.warn("回调WsCallbackApi未注册.");
-                    } else {
-                        callbackApi.callback(wsMsgDto.getClientMsgType(), wsMsgDto.getData(), wsSession);
-                    }
-                } else {
-                    callbackApi = WsMessageHandle.getWsCallback(wsMsgDto.getClientMsgType());
-                    if (ObjectUtil.isNotEmpty(callbackApi)) {
-                        callbackApi.callback(wsMsgDto.getClientMsgType(), wsMsgDto, wsSession);
-                    } else {
-                        log.warn("未知的消息类型");
-                        channel.getAsyncRemote().sendText("{\"serverMsgType\":\"未知的消息类型\"}");
-                    }
-                }
-            }
-
-        }
+        String channel= protocol.requestMessage().getChannelId();
+        String key= redisCache.get("*:" + channel);
+        redisCache.remove(key);
+        WebSocketServerListener.super.onClose(protocol);
     }
 
-    @OnClose
-    public void onClose(Session session) {
-        try {
-            WsSession<WsSessionContainer> wsSession = WsSessionHandle.getSessionBySessionId(session.getId());
-            WsCallbackApi wsCallbackApi = WsMessageHandle.getWsCallback(SysMsgTypeEnum.LISTENER_ON_CLOSE.getCode());
-            if (ObjectUtil.isNotEmpty(wsSession) && ObjectUtil.isNotEmpty(wsCallbackApi)) {
-                OnlineState os=onlineRedisCache.getDataValue(wsSession.getSessionId());
-                String type= os.getType();
-                String typeId=os.getTypeId();
-                if (type.equals("cabinet"))//警柜断开连接
-                {
-                    Cabinet cabinet=cabinetService.getOne(new LambdaQueryWrapper<Cabinet>()
-                            .eq(Cabinet::getCabinetNum, typeId));
-                    cabinet.setOnlineState(0);
-                    cabinet.setUpdateTime(DateTime.now());
-                    cabinetService.updateById(cabinet);
-                    onlineRedisCache.remove(wsSession.getSessionId());
-                } else if (type.equals("warehouse")) {//仓库主机已连接
-
-                }
-                wsCallbackApi.callback(SysMsgTypeEnum.LISTENER_ON_CLOSE.getCode(), (Object)null, wsSession);
-            }
-
-        } finally {
-            WsSessionHandle.delById(session.getId());
-            log.warn("会话已关闭");
-        }
-
+    @Override
+    public void onBinary(DefaultWebSocketServerProtocol protocol) {
+        WebSocketServerListener.super.onBinary(protocol);
     }
 
-    @OnError
-    public void onError(Session session, Throwable error) {
-        WsSession<WsSessionContainer> wsSession = WsSessionHandle.getSessionBySessionId(session.getId());
-        WsCallbackApi wsCallbackApi = WsMessageHandle.getWsCallback(SysMsgTypeEnum.LISTENER_ON_ERROR.getCode());
-        if (ObjectUtil.isNotEmpty(wsCallbackApi)) {
-            wsCallbackApi.callback(SysMsgTypeEnum.LISTENER_ON_ERROR.getCode(), (Object)null, wsSession);
-        }
 
-    }
 }
\ No newline at end of file

jyzb-biz/src/test/java/com/junmp/police/test/Base64Test.java

-package com.junmp.police.test;
-
-import org.apache.tomcat.util.digester.DocumentProperties;
-
-import java.nio.charset.StandardCharsets;
-import java.util.Base64;
-
-public class Base64Test {
-
-    public static void main(String args[]){
-//        String str = "abc";
-//        byte[] val = str.getBytes(StandardCharsets.UTF_8);
-//        String base64EncodedChunk = Base64.getEncoder().encodeToString(val);
-//        System.out.println(base64EncodedChunk);
-    }
-}

jyzb-boot/src/main/resources/application-local.yml

@@ -10,7 +10,7 @@ thing:
   server:
     websocket:
       start: true
-
+      port: 10040
 # 配置第三方请求
 junmp:
   swagger:

jyzb-boot/src/main/resources/application-prod.yml

@@ -6,7 +6,11 @@ easy-es:
   address: 192.168.3.121:9200
   username:
   password:
-
+thing:
+  server:
+    websocket:
+      start: true
+      port: 10030
 # 配置第三方请求
 junmp:
   swagger:

jyzb-process/src/main/java/com/junmp/junmpProcess/service/impl/FlowTaskServiceImpl.java

@@ -210,6 +210,7 @@ public class FlowTaskServiceImpl extends FlowServiceFactory implements IFlowTask
         }
             String objJson= redisCache.get(processInstanceId);
             redisCache.remove(processInstanceId);
+
             List<MessageSendReq> msgs= JSONObject.parseArray(objJson,MessageSendReq.class);
             if (CollectionUtil.isNotEmpty(msgs)){
                 msgs.forEach(msg->