add 新增延迟队列判定

fix 修复部分页面Bug

coffee-admin/src/test/java/com/coffee/admin/BusClinicTest.java

@@ -1,5 +1,7 @@
 package com.coffee.admin;
 
+import cn.hutool.core.date.DateUtil;
+import cn.hutool.json.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.coffee.bus.controller.vo.ClinicStatsVo;
 import com.coffee.bus.entity.BusDeviceRunningEntity;
@@ -8,8 +10,14 @@ import com.coffee.bus.listener.event.bean.DeviceInfoEvent;
 import com.coffee.bus.service.LocalBusClinicService;
 import com.coffee.bus.service.LocalBusInfusionHistoryService;
 import com.coffee.bus.service.dto.ClinicStatsReturnResult;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.redisson.api.RBlockingQueue;
+import org.redisson.api.RDelayedQueue;
+import org.redisson.api.RedissonClient;
+import org.redisson.codec.JsonJacksonCodec;
+import org.redisson.codec.TypedJsonJacksonCodec;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.context.ApplicationContext;
@@ -17,6 +25,7 @@ import org.springframework.test.context.junit4.SpringRunner;
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @Author longsanlang
@@ -28,10 +37,12 @@ import java.util.Map;
 @SpringBootTest(classes = AdminApplication.class)
 public class BusClinicTest {
 
+    @Autowired
+    private RedissonClient redissonClient;
     @Autowired
     private LocalBusClinicService clinicService;
-
-
+    @Autowired
+    private ObjectMapper objectMapper;
     @Autowired
     private LocalBusInfusionHistoryService infusionHistoryService;
 
@@ -47,6 +58,19 @@ public class BusClinicTest {
         System.out.println(stats);
     }
 
+    @Test
+    public void redisson(){
+        RBlockingQueue<Object> blockingFairQueue = redissonClient.getBlockingQueue("delay_queue",new JsonJacksonCodec());
+        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
+        JSONObject jsonObject = new JSONObject();
+        jsonObject.putOpt(DateUtil.now(),"1");
+        System.out.println(jsonObject);
+
+        delayedQueue.offerAsync(jsonObject,60, TimeUnit.SECONDS);
+        while (true){
+
+        }
+    }
 
     @Test
     public void save(){

coffee-common/pom.xml

@@ -116,5 +116,16 @@
             <groupId>jakarta.websocket</groupId>
             <artifactId>jakarta.websocket-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.redisson</groupId>
+            <artifactId>redisson-spring-boot-starter</artifactId>
+            <optional>true</optional>
+            <scope>compile</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.jboss.marshalling</groupId>
+            <artifactId>jboss-marshalling</artifactId>
+        </dependency>
     </dependencies>
 </project>

coffee-common/src/main/java/com/coffee/common/cache/ClusterConfigStorage.java

@@ -22,7 +22,7 @@ import java.util.concurrent.TimeUnit;
 public class ClusterConfigStorage implements ConfigStorage {
     private final RMap<String, Object>  cacheMap;
     public ClusterConfigStorage(RedissonClient redissonClient,String name){
-        cacheMap = SpringUtil.getBean(RedissonClient.class).getMap(name);
+        cacheMap = redissonClient.getMap(name);
     }
 
     @Override

coffee-common/src/main/java/com/coffee/common/config/websocket/DefaultMessageListener.java

@@ -7,6 +7,8 @@ import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import org.redisson.api.RPatternTopic;
 import org.redisson.api.listener.PatternMessageListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
 import org.tio.core.ChannelContext;
 import org.tio.core.Tio;
 import org.tio.core.utils.TioUtils;
@@ -35,7 +37,7 @@ public class DefaultMessageListener implements PatternMessageListener<TopicMessa
             String json = null;
             try {
                 json = objectMapper.writeValueAsString(MessageResponse.of(id, "result", msg.getParam(),
-                        JSONUtil.parse(objectMapper.writeValueAsString(msg.getMessage()))));
+                        msg.getMessage()));
                 Tio.send(channelContext, WsResponse.fromText(json, WsPacket.CHARSET_NAME));
             } catch (JsonProcessingException e) {
                 log.error("ws消息订阅,解析失败,message:【】", JSONUtil.toJsonStr(msg));

coffee-common/src/main/java/com/coffee/common/config/websocket/MessageResponse.java

@@ -19,5 +19,5 @@ public class MessageResponse implements Serializable {
     private String id;
     private String type;
     private String param;
-    private JSON payload;
+    private Object payload;
 }

coffee-common/src/main/java/com/coffee/common/util/RedissonUtil.java

@@ -1,16 +1,21 @@
 package com.coffee.common.util;
 
-import cn.hutool.json.JSONUtil;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.redisson.api.RPatternTopic;
-import org.redisson.api.RTopic;
-import org.redisson.api.RedissonClient;
+import cn.hutool.core.date.DateUtil;
+import io.netty.channel.DefaultEventLoop;
+import io.netty.channel.EventLoopGroup;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import org.redisson.Redisson;
+import org.redisson.api.*;
 import org.redisson.client.codec.Codec;
-import org.redisson.codec.JsonJacksonCodec;
+import org.redisson.config.Config;
+import org.redisson.config.SingleServerConfig;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
-
-import javax.annotation.PostConstruct;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 
 /**
  * @author lifang
@@ -21,16 +26,17 @@ import javax.annotation.PostConstruct;
  */
 @Component
 public class RedissonUtil {
+
     @Autowired
+    @Getter
     private RedissonClient redissonClient;
-    @Autowired
-    private ObjectMapper objectMapper;
-    private JsonJacksonCodec jsonObjectCodec;
 
-    @PostConstruct
-    public void init(){
-        jsonObjectCodec=new JsonJacksonCodec(objectMapper);
-    }
+    private static Map<String,RDelayedQueue<Object>> delayedQueueMap=new ConcurrentHashMap<>();
+
+    private static Map<String,RBlockingDeque<Object>> blockingDequeMap=new ConcurrentHashMap<>();
+
+
+    private EventLoopGroup eventLoopGroup=new DefaultEventLoop();
 
     public RPatternTopic getPatternTopic(String pattern){
         return redissonClient.getPatternTopic(pattern);
@@ -43,4 +49,48 @@ public class RedissonUtil {
     public RTopic getTopic(String pattern, Codec codec){
         return redissonClient.getTopic(pattern,codec);
     }
+
+    public RDelayedQueue getDelayedQueue(String name, Consumer<Object> consumer){
+        name="queue:"+name;
+        synchronized (name){
+            if(!blockingDequeMap.containsKey(name)){
+                RBlockingDeque<Object> value = redissonClient.getBlockingDeque(name);
+                delayedQueueMap.putIfAbsent(name,redissonClient.getDelayedQueue(value));
+                value.subscribeOnElements(i->{
+                    //开启新的线程消费，不可阻塞该异步线程
+                    eventLoopGroup.submit(()->consumer.accept(i));
+                    value.popAsync();
+                });
+            }
+            return delayedQueueMap.get(name);
+        }
+    }
+
+
+
+
+    public static void main(String[] args) {
+        Config config = new Config();
+        SingleServerConfig singleServerConfig = config.useSingleServer();
+        singleServerConfig.setAddress("redis://192.168.100.32:9736");
+        singleServerConfig.setPassword("6E6985E1F7CB40F24A\\.")
+                .setDatabase(3);
+//        config.setTransportMode(TransportMode.EPOLL);
+
+        RedissonClient redisson = Redisson.create(config);
+        RBlockingQueue<Object> blockingQueue = redisson.getBlockingQueue("dest_queue1");
+        RDelayedQueue<Object> delayedQueue = redisson.getDelayedQueue(blockingQueue);
+        blockingQueue.subscribeOnElements(str->{
+            String now = DateUtil.now();
+            System.out.println(str);
+        });
+
+//        for(int i=1;i<=5;i++) {
+//            JSONObject jsonObject = new JSONObject().putOpt(DateUtil.now(), i);
+//            // 向阻塞队列放入数据
+//            delayedQueue.offerAsync(jsonObject, 3, TimeUnit.SECONDS);
+//        }
+
+
+    }
 }

coffee-framework/src/main/java/com/coffee/framework/config/WebAppMvcConfig.java

@@ -3,20 +3,15 @@ package com.coffee.framework.config;
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.util.StrUtil;
 import com.coffee.common.config.BooleanToIntegerSerializer;
+import com.fasterxml.jackson.core.JsonGenerator;
 import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.DeserializationContext;
-import com.fasterxml.jackson.databind.DeserializationFeature;
-import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.*;
 import com.fasterxml.jackson.databind.deser.std.StdScalarDeserializer;
-import com.fasterxml.jackson.databind.deser.std.StringDeserializer;
 import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.fasterxml.jackson.databind.ser.std.StringSerializer;
-import com.fasterxml.jackson.databind.ser.std.ToStringSerializer;
-import org.python.antlr.ast.Str;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Profile;
 import org.springframework.format.Formatter;
 import org.springframework.format.FormatterRegistry;
 import org.springframework.http.converter.HttpMessageConverter;
@@ -30,7 +25,7 @@ import java.io.IOException;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.*;
- 
+
 /**
  * @Description TODO
  * @Classname Knife4jWebMvcConfig
@@ -40,13 +35,22 @@ import java.util.*;
 @Configuration
 //@Profile("dev")
 public class WebAppMvcConfig implements WebMvcConfigurer {
+    @Autowired
+    private ObjectMapper objectMapper;
     @Autowired
     private List<HandlerInterceptor> interceptors;
     @Override
     public void extendMessageConverters(List<HttpMessageConverter<?>> converters) {
         MappingJackson2HttpMessageConverter converter = new MappingJackson2HttpMessageConverter();
-        ObjectMapper objectMapper = converter.getObjectMapper();
+//        ObjectMapper objectMapper = converter.getObjectMapper();
         // 时间格式化
+        SerializerProvider serializerProvider = objectMapper.getSerializerProvider();
+        serializerProvider.setNullValueSerializer(new JsonSerializer<Object>() {
+            @Override
+            public void serialize(Object value, JsonGenerator gen, SerializerProvider serializers) throws IOException {
+                gen.writeNull();
+            }
+        });
         objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
         objectMapper.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"));
 
@@ -72,13 +76,13 @@ public class WebAppMvcConfig implements WebMvcConfigurer {
 
     @Override
     public void addResourceHandlers(ResourceHandlerRegistry registry) {
- 
+
         registry.addResourceHandler("/**").addResourceLocations("classpath:/static/");
- 
+
         /** 配置knife4j 显示文档 */
         registry.addResourceHandler("doc.html")
                 .addResourceLocations("classpath:/META-INF/resources/");
- 
+
         /**
          * 配置swagger-ui显示文档
          */

coffee-system/src/main/java/com/coffee/bus/config/HospitalConfigHandler.java

@@ -13,9 +13,11 @@ import java.util.Map;
  * @Description 医院配置处理器
  * @createTime 2022年05月18日 10:33:00
  */
-public interface HospitalConfigHandler<T> {
+public interface HospitalConfigHandler<E,T> {
 
-    ConfigEnum getType();
+    String getName();
+
+    String getId();
 
     String getDescription();
 
@@ -23,10 +25,10 @@ public interface HospitalConfigHandler<T> {
      * 描述: 设置配置参数
      * @author lifang
      * @date 2022/5/18 10:34
-     * @param map
+     * @param config
      * @return void
      */
-    void setConfig(Map<String,Object> map);
+    void setConfig(E config);
 
     /**
      * 描述: 处理传入数据

coffee-system/src/main/java/com/coffee/bus/config/HospitalFunctionAnalConfigHandler.java

@@ -0,0 +1,238 @@
+package com.coffee.bus.config;
+
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
+import com.coffee.bus.config.bean.FunctionAnalConfig;
+import com.coffee.bus.entity.BusDeviceRunningEntity;
+import com.coffee.bus.entity.BusInfusionHistoryEntity;
+import com.coffee.bus.registry.device.DeviceRegistry;
+import com.coffee.bus.service.LocalBusDeviceRunningService;
+import com.coffee.bus.service.LocalBusInfusionHistoryService;
+import com.coffee.bus.utils.WsPublishUtils;
+import com.coffee.common.cache.ClusterConfigStorage;
+import com.coffee.common.util.RedissonUtil;
+import lombok.Builder;
+import lombok.Data;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RDelayedQueue;
+
+import java.io.Serializable;
+import java.util.Date;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName HospitalFunctionAnalConfigHandler.java
+ * @Description 镇痛不足检测
+ * @createTime 2022年05月18日 11:25:00
+ */
+@Slf4j
+public class HospitalFunctionAnalConfigHandler extends ClusterConfigStorage implements HospitalConfigHandler<FunctionAnalConfig,BusDeviceRunningEntity> {
+
+    private final LocalBusDeviceRunningService deviceRunningService;
+
+    private final LocalBusInfusionHistoryService infusionHistoryService;
+
+    private final RedissonUtil redissonUtil;
+
+    private final DeviceRegistry deviceRegistry;
+    @Getter
+    private String name;
+
+    public HospitalFunctionAnalConfigHandler(String name, LocalBusDeviceRunningService deviceRunningService, LocalBusInfusionHistoryService infusionHistoryService, RedissonUtil redissonUtil, DeviceRegistry deviceRegistry, WsPublishUtils wsPublishUtils) {
+        super(redissonUtil.getRedissonClient(), name);
+        this.name=name;
+        this.deviceRunningService = deviceRunningService;
+        this.infusionHistoryService = infusionHistoryService;
+        this.redissonUtil = redissonUtil;
+        this.deviceRegistry = deviceRegistry;
+        this.wsPublishUtils = wsPublishUtils;
+    }
+
+    private WsPublishUtils wsPublishUtils;
+
+
+
+    @Override
+    public String getId() {
+        return "insufficient";
+    }
+
+    @Override
+    public String getDescription() {
+        return "进行镇痛不足判定、不在服务区判定、低输注判定";
+    }
+
+
+    @Override
+    public void setConfig(FunctionAnalConfig config) {
+        this.setConfig("config",config);
+    }
+
+    @Override
+    public void handler(BusDeviceRunningEntity source) {
+        FunctionAnalConfig config = this.getConfig("config").as(FunctionAnalConfig.class);
+        if(config==null){
+            log.warn("id:{},配置名称:{}，不存在",getId(),name);
+            return;
+        }
+        handlerAnal(source,config);
+    }
+
+    /**
+     * 描述: 镇痛不足处理
+     * 在时间窗口中判断pca是否达到一定次数
+     * @author lifang
+     * @date 2022/5/18 14:05
+     * @param source
+     * @param analConfig
+     * @return void
+     */
+    private void handlerAnal(BusDeviceRunningEntity source,FunctionAnalConfig analConfig){
+        RDelayedQueue delayedQueue = redissonUtil.getDelayedQueue(getId() + "-" + source.getTenantId() + "-" + source.getDeviceId(), e -> {
+            if(e instanceof AnalEntity){
+                this.judgeAnalPoor((AnalEntity) e);
+            }
+        });
+        //如果有新的输注产生,则清空延迟队列
+        if(source.isNewInfusion()){
+            //镇痛不足设置为false
+            source.setWarnAnalgesicPoor(false);
+            delayedQueue.clear();
+        }
+        Integer insufficientTime = analConfig.getInsufficientTime();
+        Integer insufficientCount = analConfig.getInsufficientCount();
+        Boolean valid = analConfig.getValid();
+        if(valid==null||(insufficientTime==null||insufficientTime<=0)||
+                (insufficientCount==null||insufficientCount<=0)){
+            return;
+        }
+
+        AnalEntity anal = AnalEntity.builder()
+                .deviceId(source.getDeviceId())
+                .pcaValidCount(source.getPcaValidCount())
+                .pcaInvalidCount(source.getPcaInvalidCount())
+                .timeout(analConfig.getInsufficientTime())
+                .unit(TimeUnit.SECONDS)
+                .threshold(analConfig.getInsufficientCount())
+                .judgeByValid(analConfig.getValid())
+                .tenantId(source.getTenantId())
+                .timestamp(new Date())
+                .build();
+
+        delayedQueue.offer(anal,anal.getTimeout(),anal.getUnit());
+    }
+
+
+    /**
+     * 描述: 判断延迟任务是否触发镇痛不足
+     * @author lifang
+     * @date 2022/5/19 9:14
+     * @param
+     * @return void
+     */
+    private void judgeAnalPoor(AnalEntity anal){
+        //镇痛消失延迟队列
+        RDelayedQueue noneAnalDelayedQueue = redissonUtil.getDelayedQueue("none-" + getId()+"-" + anal.getTenantId()+"-" + anal.getDeviceId(), e -> {
+            if(e instanceof NoneAnalEntity){
+                handleNoneAnal((NoneAnalEntity) e);
+            }
+        });
+        String deviceId = anal.getDeviceId();
+        BusDeviceRunningEntity runningInfo = deviceRunningService.getOne(new QueryWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getDeviceId, deviceId).eq(BusDeviceRunningEntity::getTenantId, anal.getTenantId()));
+        if(runningInfo==null){
+            //泵已换绑医院，无需再处理
+            return;
+        }
+        //PCA差值
+        int subCount=0;
+        if(anal.isJudgeByValid()){
+            subCount=Math.subtractExact(runningInfo.getPcaValidCount(), Optional.ofNullable(anal.getPcaValidCount()).orElse(0));
+        }else {
+            subCount=Math.subtractExact(runningInfo.getPcaInvalidCount(), Optional.ofNullable(anal.getPcaInvalidCount()).orElse(0));
+        }
+        if(subCount>anal.getThreshold()){
+            log.info("设备：{}镇痛不足",deviceId);
+            //触发阈值,设置为镇痛不足
+            if(!Boolean.TRUE.equals(runningInfo.getWarnAnalgesicPoor())){
+                deviceRunningService.update(new UpdateWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getId,runningInfo.getId())
+                        .set(BusDeviceRunningEntity::getWarnAnalgesicPoor,true));
+                infusionHistoryService.update(new UpdateWrapper<BusInfusionHistoryEntity>().lambda().eq(BusInfusionHistoryEntity::getId,runningInfo.getInfusionId())
+                        .set(BusInfusionHistoryEntity::getWarnAnalgesicPoor,true));
+                //报警/提醒缓存重置
+                deviceRegistry.getOperator(deviceId).setAlarmOrWarn(null);
+                if(Boolean.TRUE.equals(runningInfo.getMaster())){
+                    //发布推送
+                    wsPublishUtils.publishPatientMonitor(runningInfo.getPatientCode(),runningInfo.getTenantId());
+                }
+                //设置镇痛不足后,设置镇痛消失延迟时间
+            }
+            noneAnalDelayedQueue.clear();
+        }else {
+            //没有触发阈值,发入镇痛消失处理
+            FunctionAnalConfig config = this.getConfig("config").as(FunctionAnalConfig.class);
+            NoneAnalEntity noneAnal = NoneAnalEntity.builder()
+                    .deviceId(anal.getDeviceId())
+                    .tenantId(anal.getTenantId())
+                    .timeout(config.getDisappearTime())
+                    .unit(TimeUnit.SECONDS)
+                    .timestamp(new Date())
+                    .build();
+            noneAnalDelayedQueue.offer(noneAnal,noneAnal.getTimeout(),noneAnal.getUnit());
+        }
+    }
+
+
+    private void handleNoneAnal(NoneAnalEntity noneAnal){
+        String deviceId = noneAnal.getDeviceId();
+        BusDeviceRunningEntity runningInfo = deviceRunningService.getOne(new QueryWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getDeviceId, deviceId).eq(BusDeviceRunningEntity::getTenantId, noneAnal.getTenantId()));
+        if(runningInfo==null){
+            //泵已换绑医院，无需再处理
+            return;
+        }
+        //将设备由镇痛不足取消成功
+        if (deviceRunningService.update(new UpdateWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getId, runningInfo.getId())
+                .eq(BusDeviceRunningEntity::getWarnAnalgesicPoor, true)
+                .set(BusDeviceRunningEntity::getWarnAnalgesicPoor, false))) {
+            log.info("设备：{}取消镇痛不足",deviceId);
+            infusionHistoryService.update(new UpdateWrapper<BusInfusionHistoryEntity>().lambda().eq(BusInfusionHistoryEntity::getId,runningInfo.getInfusionId())
+                    .set(BusInfusionHistoryEntity::getWarnAnalgesicPoor,false));
+            deviceRegistry.getOperator(deviceId).setAlarmOrWarn(null);
+            if(Boolean.TRUE.equals(runningInfo.getMaster())){
+                //发布推送
+                wsPublishUtils.publishPatientMonitor(runningInfo.getPatientCode(),runningInfo.getTenantId());
+            }
+        }
+
+
+
+    }
+    @Data
+    @Builder
+    static class AnalEntity implements Serializable {
+        private String deviceId;
+        private Integer pcaValidCount;
+        private Integer pcaInvalidCount;
+        private Integer timeout;
+        private TimeUnit unit;
+        private Integer threshold;
+        private boolean judgeByValid;
+        private String tenantId;
+        private Date timestamp;
+    }
+
+    @Data
+    @Builder
+    static class NoneAnalEntity implements Serializable {
+        private String deviceId;
+        private Integer timeout;
+        private TimeUnit unit;
+        private String tenantId;
+        private Date timestamp;
+    }
+
+}

coffee-system/src/main/java/com/coffee/bus/config/HospitalFunctionExtraConfigHandler.java

@@ -0,0 +1,216 @@
+package com.coffee.bus.config;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
+import com.coffee.bus.config.bean.FunctionExtraConfig;
+import com.coffee.bus.entity.BusDeviceRunningEntity;
+import com.coffee.bus.entity.BusInfusionHistoryEntity;
+import com.coffee.bus.enums.DeviceStatusEnum;
+import com.coffee.bus.enums.DeviceTypeEnum;
+import com.coffee.bus.registry.device.DeviceRegistry;
+import com.coffee.bus.service.LocalBusDeviceRunningService;
+import com.coffee.bus.service.LocalBusInfusionHistoryService;
+import com.coffee.bus.utils.WsPublishUtils;
+import com.coffee.common.cache.ClusterConfigStorage;
+import com.coffee.common.util.RedissonUtil;
+import lombok.Builder;
+import lombok.Data;
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RDelayedQueue;
+import org.redisson.api.RedissonClient;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.util.Date;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName HospitalFunctionLowInfusionConfigHandler.java
+ * @Description 低输注判断处理
+ * 不在服务区处理：当设备下线时，且最后一个包不为关机包，则开启延迟任务进行不在服务区处理判断
+ * 低输注处理：当设备持续发送输注下限时，开启低输注延迟任务，当输注>输注下限时，撤销所有延迟任务
+ * @createTime 2022年05月19日 15:13:00
+ */
+@Slf4j
+public class HospitalFunctionExtraConfigHandler  extends ClusterConfigStorage implements HospitalConfigHandler<FunctionExtraConfig, BusDeviceRunningEntity>{
+
+    private RedissonUtil redissonUtil;
+
+    private LocalBusDeviceRunningService deviceRunningService;
+
+    private LocalBusInfusionHistoryService infusionHistoryService;
+
+    private DeviceRegistry deviceRegistry;
+
+    private WsPublishUtils wsPublishUtils;
+    @Getter
+    private String name;
+
+    public HospitalFunctionExtraConfigHandler(RedissonClient redissonClient, String name) {
+        super(redissonClient, name);
+    }
+
+    @Override
+    public String getId() {
+        return "extra";
+    }
+
+    @Override
+    public String getDescription() {
+        return "医院其他配置处理,不在服务区设置、低输注设置";
+    }
+
+    @Override
+    public void setConfig(FunctionExtraConfig config) {
+        this.setConfig("config",config);
+    }
+
+    @Override
+    public void handler(BusDeviceRunningEntity source) {
+        FunctionExtraConfig config = this.getConfig("config").as(FunctionExtraConfig.class);
+        if(config==null){
+            log.warn("id:{},配置名称:{}，不存在",getId(),name);
+            return;
+        }
+        if (config.getNoSignal()!=null&&config.getNoSignal()>0) {
+            judgeNoSignal(source,config.getNoSignal());
+        }
+        if (config.getLowInfusion()!=null&&config.getLowInfusion()>0) {
+            judgeLowInfusion(source,config.getLowInfusion());
+        }
+    }
+
+    /**
+     * 描述: 不在服务区处理
+     * @author lifang
+     * @date 2022/5/19 15:52
+     * @param source
+     * @param interval
+     * @return void
+     */
+    private void judgeNoSignal(BusDeviceRunningEntity source,int interval){
+        RDelayedQueue delayedQueue = redissonUtil.getDelayedQueue(getId() +"-signal"+ "-" + source.getTenantId() + "-" + source.getDeviceId(), e -> {
+            if(e instanceof  NoSignalEntity){
+                this.handleNoSignal((NoSignalEntity) e);
+            }
+        });
+        delayedQueue.clear();
+
+    }
+
+    /**
+     * 描述: 不在服务区处理
+     * @author lifang
+     * @date 2022/5/19 16:02
+     * @param source
+     * @return void
+     */
+    private void handleNoSignal(NoSignalEntity source){
+        String deviceId = source.getDeviceId();
+        String tenantId = source.getTenantId();
+        BusDeviceRunningEntity runningInfo = deviceRunningService.getOne(new QueryWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getDeviceId, deviceId).eq(BusDeviceRunningEntity::getTenantId, tenantId));
+        if (runningInfo == null) {
+            //泵已换绑医院，无需再处理
+            return;
+        }
+        deviceRunningService.update(new UpdateWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getId, runningInfo.getId())
+                .set(BusDeviceRunningEntity::getRunState, DeviceStatusEnum.NoSignal));
+        infusionHistoryService.update(new UpdateWrapper<BusInfusionHistoryEntity>().lambda().eq(BusInfusionHistoryEntity::getId, runningInfo.getInfusionId())
+                .set(BusInfusionHistoryEntity::getRunState, DeviceStatusEnum.NoSignal));
+        //报警/提醒缓存重置
+        deviceRegistry.getOperator(deviceId)
+                .setStatus(DeviceStatusEnum.NoSignal);
+        if (Boolean.TRUE.equals(runningInfo.getMaster())) {
+            //发布推送
+            wsPublishUtils.publishPatientMonitor(runningInfo.getPatientCode(), runningInfo.getTenantId());
+        }
+    }
+
+
+    /**
+     * 描述: 低输注判断
+     * @author lifang
+     * @date 2022/5/19 15:52
+     * @param source
+     * @param interval
+     * @return void
+     */
+    private void judgeLowInfusion(BusDeviceRunningEntity source,int interval){
+        RDelayedQueue delayedQueue = redissonUtil.getDelayedQueue(getId() +"-signal"+ "-" + source.getTenantId() + "-" + source.getDeviceId(), e -> {
+            if(e instanceof  LowInfusionEntity){
+                handleLowInfusion((LowInfusionEntity)e);
+            }
+        });
+        //仅智能泵存在
+        if(DeviceTypeEnum.intelligent.equals(source.getType())){
+            if (source.getFlowDownLimit()!=null&&source.getFlowDownLimit().compareTo(BigDecimal.valueOf(0.1))<1) {
+                //输注量超过阈值，进行低输注判定
+                LowInfusionEntity lowInfusionEntity = LowInfusionEntity.builder()
+                        .deviceId(source.getDeviceId())
+                        .timeout(interval)
+                        .unit(TimeUnit.MINUTES)
+                        .tenantId(source.getTenantId())
+                        .timestamp(new Date())
+                        .build();
+                delayedQueue.offer(lowInfusionEntity,lowInfusionEntity.getTimeout(),lowInfusionEntity.getUnit());
+            }
+
+
+
+        }else {
+            //输注量未超过阈值，取消低输注状态
+            source.setWarnLowBattery(false);
+            delayedQueue.clear();
+        }
+    }
+
+    /**
+     * 描述: 低输注状态处理
+     * @author lifang
+     * @date 2022/5/19 16:08
+     * @param source
+     * @return void
+     */
+    private void handleLowInfusion(LowInfusionEntity source) {
+        String deviceId = source.getDeviceId();
+        String tenantId = source.getTenantId();
+        BusDeviceRunningEntity runningInfo = deviceRunningService.getOne(new QueryWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getDeviceId, deviceId).eq(BusDeviceRunningEntity::getTenantId, tenantId));
+        if (runningInfo == null) {
+            //泵已换绑医院，无需再处理
+            return;
+        }
+        deviceRunningService.update(new UpdateWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getId, runningInfo.getId())
+                .set(BusDeviceRunningEntity::getWarnLowBattery, true));
+        infusionHistoryService.update(new UpdateWrapper<BusInfusionHistoryEntity>().lambda().eq(BusInfusionHistoryEntity::getId, runningInfo.getInfusionId())
+                .set(BusInfusionHistoryEntity::getWarnLowBattery, true));
+        //报警/提醒缓存重置
+        deviceRegistry.getOperator(deviceId).setAlarmOrWarn(null);
+        if (Boolean.TRUE.equals(runningInfo.getMaster())) {
+            //发布推送
+            wsPublishUtils.publishPatientMonitor(runningInfo.getPatientCode(), runningInfo.getTenantId());
+        }
+    }
+
+    @Data
+    @Builder
+    static class NoSignalEntity implements Serializable {
+        private String deviceId;
+        private Integer timeout;
+        private TimeUnit unit;
+        private String tenantId;
+        private Date timestamp;
+    }
+
+    @Data
+    @Builder
+    static class LowInfusionEntity implements Serializable {
+        private String deviceId;
+        private Integer timeout;
+        private TimeUnit unit;
+        private String tenantId;
+        private Date timestamp;
+    }
+}

coffee-system/src/main/java/com/coffee/bus/config/HospitalFunctionOtherConfigHandler.java

@@ -1,46 +0,0 @@
-package com.coffee.bus.config;
-
-import cn.hutool.json.JSONUtil;
-import com.coffee.bus.entity.BusDeviceRunningEntity;
-import com.coffee.bus.enums.ConfigEnum;
-import com.coffee.bus.service.dto.function.FunctionOtherConfig;
-import com.coffee.common.cache.ClusterConfigStorage;
-
-import org.redisson.api.RedissonClient;
-import org.springframework.data.redis.core.RedisTemplate;
-
-import java.util.Map;
-
-/**
- * @author lifang
- * @version 1.0.0
- * @ClassName HospitalFunctionOtherConfigHandler.java
- * @Description 其他配置
- * @createTime 2022年05月18日 11:25:00
- */
-public class HospitalFunctionOtherConfigHandler extends ClusterConfigStorage implements HospitalConfigHandler<BusDeviceRunningEntity> {
-
-    public HospitalFunctionOtherConfigHandler(RedissonClient redissonClient, String name) {
-        super(redissonClient, name);
-    }
-
-    @Override
-    public ConfigEnum getType() {
-        return ConfigEnum.other;
-    }
-
-    @Override
-    public String getDescription() {
-        return "进行镇痛不足判定、不在服务区判定、低输注判定";
-    }
-
-    @Override
-    public void setConfig(Map<String, Object> map) {
-        JSONUtil.toBean(JSONUtil.toJsonStr(map), FunctionOtherConfig.class);
-    }
-
-    @Override
-    public void handler(BusDeviceRunningEntity source) {
-
-    }
-}

coffee-system/src/main/java/com/coffee/bus/service/dto/function/FunctionAnalConfig.java → coffee-system/src/main/java/com/coffee/bus/config/bean/FunctionAnalConfig.java

@@ -1,9 +1,11 @@
-package com.coffee.bus.service.dto.function;
+package com.coffee.bus.config.bean;
 
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
+
 /**
  * @author lifang
  * @version 1.0.0
@@ -13,7 +15,7 @@ import lombok.Data;
  */
 @ApiModel("镇痛不足配置")
 @Data
-public class FunctionAnalConfig {
+public class FunctionAnalConfig implements Serializable {
     @ApiModelProperty("镇痛不足提醒消失时间")
     private Integer disappearTime;
     @ApiModelProperty("镇痛不足判定次数")

coffee-system/src/main/java/com/coffee/bus/service/dto/function/FunctionAutoUndoConfig.java → coffee-system/src/main/java/com/coffee/bus/config/bean/FunctionAutoUndoConfig.java

@@ -1,9 +1,11 @@
-package com.coffee.bus.service.dto.function;
+package com.coffee.bus.config.bean;
 
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
+
 /**
  * @author lifang
  * @version 1.0.0
@@ -13,7 +15,7 @@ import lombok.Data;
  */
 @ApiModel("自动撤泵配置")
 @Data
-public class FunctionAutoUndoConfig {
+public class FunctionAutoUndoConfig implements Serializable {
     @ApiModelProperty("是否开启自动撤泵")
     private boolean enable;
     @ApiModelProperty("不在服务区撤泵时间间隔")

coffee-system/src/main/java/com/coffee/bus/service/dto/function/FunctionExtraConfig.java → coffee-system/src/main/java/com/coffee/bus/config/bean/FunctionExtraConfig.java

@@ -1,7 +1,10 @@
-package com.coffee.bus.service.dto.function;
+package com.coffee.bus.config.bean;
 
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import java.io.Serializable;
 
 /**
  * @author lifang
@@ -11,7 +14,8 @@ import io.swagger.annotations.ApiModelProperty;
  * @createTime 2022年05月18日 10:22:00
  */
 @ApiModel("其他配置中的其他配置")
-public class FunctionExtraConfig {
+@Data
+public class FunctionExtraConfig implements Serializable {
     @ApiModelProperty("低输注时间设置")
     private Integer lowInfusion;
     @ApiModelProperty("不在服务区时间设置")

coffee-system/src/main/java/com/coffee/bus/config/bean/FunctionFinishMonitorConfig.java

@@ -0,0 +1,28 @@
+package com.coffee.bus.config.bean;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName FunctionUndoConfig.java
+ * @Description TODO
+ * @createTime 2022年05月18日 08:33:00
+ */
+@ApiModel("结束管理功能配置")
+@Data
+public class FunctionFinishMonitorConfig implements Serializable {
+    @ApiModelProperty("是否开启自动结束管理")
+    private boolean enable;
+    @ApiModelProperty("关机到结束时间间隔")
+    private Integer shutDownInterval;
+    @ApiModelProperty("自动撤泵人姓名")
+    private String undoBy;
+    @ApiModelProperty("自动撤泵人是否为麻醉医生")
+    private boolean noSignalInterval;
+
+}

coffee-system/src/main/java/com/coffee/bus/service/dto/function/FunctionManualUndoConfig.java → coffee-system/src/main/java/com/coffee/bus/config/bean/FunctionManualUndoConfig.java

@@ -1,9 +1,11 @@
-package com.coffee.bus.service.dto.function;
+package com.coffee.bus.config.bean;
 
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
+
 /**
  * @author lifang
  * @version 1.0.0
@@ -13,7 +15,7 @@ import lombok.Data;
  */
 @ApiModel("手动撤泵配置")
 @Data
-public class FunctionManualUndoConfig {
+public class FunctionManualUndoConfig implements Serializable {
     @ApiModelProperty("是否开启手动撤泵配置")
     private boolean enable;
     @ApiModelProperty("撤泵人为空检测")

coffee-system/src/main/java/com/coffee/bus/service/dto/function/FunctionOtherConfig.java → coffee-system/src/main/java/com/coffee/bus/config/bean/FunctionOtherConfig.java

@@ -1,9 +1,11 @@
-package com.coffee.bus.service.dto.function;
+package com.coffee.bus.config.bean;
 
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 
+import java.io.Serializable;
+
 /**
  * @author lifang
  * @version 1.0.0
@@ -13,7 +15,7 @@ import lombok.Data;
  */
 @Data
 @ApiModel("其他配置")
-public class FunctionOtherConfig {
+public class FunctionOtherConfig implements Serializable {
     @ApiModelProperty("其他配置")
     private FunctionExtraConfig other;
     @ApiModelProperty("镇痛不足配置")

coffee-system/src/main/java/com/coffee/bus/service/dto/function/FunctionUndoConfig.java → coffee-system/src/main/java/com/coffee/bus/config/bean/FunctionUndoConfig.java

@@ -1,8 +1,10 @@
-package com.coffee.bus.service.dto.function;
+package com.coffee.bus.config.bean;
 
 import io.swagger.annotations.ApiModel;
 import lombok.Data;
 
+import java.io.Serializable;
+
 /**
  * @author lifang
  * @version 1.0.0
@@ -12,7 +14,7 @@ import lombok.Data;
  */
 @ApiModel("撤泵功能配置")
 @Data
-public class FunctionUndoConfig {
+public class FunctionUndoConfig implements Serializable {
     private FunctionAutoUndoConfig auto;
 
     private FunctionManualUndoConfig manual;

coffee-system/src/main/java/com/coffee/bus/controller/BusPatientController.java

@@ -89,7 +89,7 @@ public class BusPatientController  implements BaseQueryController<BusPatientEnti
     @ApiOperation(value = "判断给定病号中是否可以结束临床管理",notes = "病患当前绑定主设备必须要在关机、不在服务器、待机中才能结束管理,权限【无】")
     @ApiResponses({
             @ApiResponse(code = 500,message = "没有选择病号"),
-            @ApiResponse(code =200 ,message = "若存在不可以结束的临床信息，将会把该临床的主泵信息返回",response = BusDeviceRunningEntity.class)
+            @ApiResponse(code =200 ,message = "若存在不可以结束的临床信息，返回1",response = BusDeviceRunningEntity.class)
     })
     public R<Boolean> judgeFinished(@RequestBody List<String> patientCodes){
         if(CollectionUtil.isEmpty(patientCodes)){
@@ -101,7 +101,8 @@ public class BusPatientController  implements BaseQueryController<BusPatientEnti
                 .select(BusDeviceRunningEntity::getDeviceId,BusDeviceRunningEntity::getPatientCode,BusDeviceRunningEntity::getRunState,BusDeviceRunningEntity::getAlarm)
                 .eq(BusDeviceRunningEntity::getMaster,true)
                 .eq(BusDeviceRunningEntity::getMonitorType,true)
-                .ne(BusDeviceRunningEntity::getRunState, DeviceStatusEnum.Shutdown)
+                .nested(i->i.ne(BusDeviceRunningEntity::getRunState,DeviceStatusEnum.Shutdown)
+                        .ne(BusDeviceRunningEntity::getRunState,DeviceStatusEnum.NoSignal))
                 .in(BusDeviceRunningEntity::getPatientCode,patientCodes))));
     }
 

coffee-system/src/main/java/com/coffee/bus/entity/BusDeviceEntity.java

@@ -50,14 +50,6 @@ public class BusDeviceEntity extends TenantGenericEntity<String,String> {
     @Length(max = 255,message = "设备类型不得超过255个字符")
     private DeviceTypeEnum type;
 
-    @ApiModelProperty(value = "运行状态",readOnly = true)
-    @TableField(javaType = true,updateStrategy = FieldStrategy.IGNORED)
-    private DeviceStatusEnum runState;
-
-    @ApiModelProperty(value = "报警信息",readOnly = true)
-    @TableField(javaType = true,updateStrategy = FieldStrategy.IGNORED)
-    private DeviceAlarmEnum alarm;
-
     @ApiModelProperty(value = "第三方平台返回配置")
     @TableField(typeHandler = FastjsonTypeHandler.class,updateStrategy = FieldStrategy.DEFAULT)
     private AliIotConfig config;

coffee-system/src/main/java/com/coffee/bus/entity/BusDeviceHistoryEntity.java

@@ -81,6 +81,7 @@ public class BusDeviceHistoryEntity extends CommonDeviceParam<String,String> {
         entity.setPcaValidCount(running.getPcaValidCount());
         entity.setPcaInvalidCount(running.getPcaInvalidCount());
         entity.setPcaTotalCount(running.getPcaTotalCount());
+        entity.setDataNumber(running.getDataNumber());
         /**
          * 脉冲泵参数
          */

coffee-system/src/main/java/com/coffee/bus/entity/BusDeviceRunningEntity.java

@@ -83,6 +83,13 @@ public class BusDeviceRunningEntity extends CommonDeviceParam<String,String> {
     @JsonIgnoreProperties
     private String infusionModifyId;
 
+
+    /**
+     * 当前运行状态是否开启了新的输注,只在设备上传解析过程中使用，其他地方无用处
+     */
+    @TableField(exist = false)
+    @JsonIgnoreProperties
+    private boolean newInfusion;
     /**
      * @author 龙三郎
      * 根据阿里云传回数据的items，设置输注的属性

coffee-system/src/main/java/com/coffee/bus/entity/BusInfusionHistoryEntity.java

@@ -111,7 +111,7 @@ public class BusInfusionHistoryEntity extends CommonDeviceParam<String,String> {
         entity.setPcaValidCount(running.getPcaValidCount());
         entity.setPcaInvalidCount(running.getPcaInvalidCount());
         entity.setPcaTotalCount(running.getPcaTotalCount());
-
+        entity.setDataNumber(running.getDataNumber());
         /**
          * 脉冲泵参数
          */

coffee-system/src/main/java/com/coffee/bus/entity/BusInfusionModifyEntity.java

@@ -176,6 +176,7 @@ public class BusInfusionModifyEntity extends TenantGenericEntity<String,String>
         entity.setPcaValidCount(running.getPcaValidCount());
         entity.setPcaInvalidCount(running.getPcaInvalidCount());
         entity.setPcaTotalCount(running.getPcaTotalCount());
+        entity.setDataNumber(running.getDataNumber());
         /**
          * 脉冲泵参数
          */

coffee-system/src/main/java/com/coffee/bus/his/strategy/HisStrategyEnum.java

@@ -14,6 +14,7 @@ import lombok.Getter;
 @AllArgsConstructor
 @Getter
 public enum  HisStrategyEnum  implements IEnum<Integer> {
+    NONE(0,"无his对接"),
     ALL(1,"拉取所有的病人数据"),
     PART(2,"拉取部分病人数据(例、最新三个月)"),
     NEW(3,"只拉取最新的病人数据"),

coffee-system/src/main/java/com/coffee/bus/listener/event/bean/DeviceLifecycleEvent.java

@@ -0,0 +1,28 @@
+package com.coffee.bus.listener.event.bean;
+
+import lombok.Getter;
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName DeviceLifecycleEvent.java
+ * @Description 设备上/下线事件
+ * @createTime 2022年05月19日 15:27:00
+ */
+
+public class DeviceLifecycleEvent extends ApplicationEvent {
+    @Getter
+    private final String deviceId;
+
+    /*
+     * 设备是否上线
+     */
+    @Getter
+    private final boolean online;
+    public DeviceLifecycleEvent(Object source,String deviceId,boolean online) {
+        super(source);
+        this.deviceId=deviceId;
+        this.online=online;
+    }
+}

coffee-system/src/main/java/com/coffee/bus/registry/device/ClusterDeviceOperator.java

@@ -125,12 +125,12 @@ public class ClusterDeviceOperator implements DeviceOperator<DeviceCacheInfo> {
     }
 
     @Override
-    public void setAlarm(String alarm) {
+    public void setAlarmOrWarn(String alarm) {
         configStorage.setConfig(DeviceKeyConstant.ALARM,alarm);
     }
 
     @Override
-    public String getAlarm() {
+    public String getAlarmOrWarn() {
         return getValue(DeviceKeyConstant.ALARM).asString();
     }
 

coffee-system/src/main/java/com/coffee/bus/registry/device/ClusterDeviceRegistry.java

@@ -67,7 +67,7 @@ public class ClusterDeviceRegistry implements DeviceRegistry {
                     }
                 }
                 if (BusDeviceAlarmEntity.alarmOrWarn(runningInfo)) {
-                    deviceOperator.setAlarm(BusDeviceAlarmEntity.parseRunning(runningInfo).signParm(DeviceInfoListener.sign));
+                    deviceOperator.setAlarmOrWarn(BusDeviceAlarmEntity.parseRunning(runningInfo).signParm(DeviceInfoListener.sign));
                 }
                 deviceOperator.setMaster(runningInfo.getMaster());
                 deviceOperator.setStartTime(runningInfo.getStartTime());

coffee-system/src/main/java/com/coffee/bus/registry/device/DeviceOperator.java

@@ -139,16 +139,16 @@ public interface DeviceOperator<T> extends Operator<T> {
     String getInfusionId();
 
     /**
-     * 设置当前设备报警信息
+     * 设置当前设备报警（提醒）信息
      * @param alarm
      */
-    void setAlarm(String alarm);
+    void setAlarmOrWarn(String alarm);
 
     /**
-     * 获取当前设备报警信息
+     * 获取当前设备报警（提醒）信息
      * @return
      */
-    String getAlarm();
+    String getAlarmOrWarn();
 
     /**
      * 设置当前设备输注参数

coffee-system/src/main/java/com/coffee/bus/registry/patient/ClusterPatientOperator.java

@@ -1,7 +1,5 @@
 package com.coffee.bus.registry.patient;
 
-import cn.hutool.core.collection.CollectionUtil;
-import cn.hutool.json.JSONUtil;
 import com.coffee.bus.registry.constant.PatientKeyConstant;
 import com.coffee.bus.registry.patient.bean.DeviceTimeSmallInfo;
 import com.coffee.bus.registry.patient.bean.PatientCacheInfo;
@@ -9,7 +7,6 @@ import com.coffee.common.cache.ConfigStorage;
 import com.coffee.common.enums.SexEnum;
 
 import java.util.*;
-import java.util.stream.Collectors;
 
 /**
  * @author lifang

coffee-system/src/main/java/com/coffee/bus/service/LocalBusPatientService.java

@@ -38,6 +38,7 @@ import org.springframework.transaction.annotation.Transactional;
 import org.springframework.web.context.request.async.DeferredResult;
 
 import java.time.LocalDateTime;
+import java.time.ZoneOffset;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
@@ -288,15 +289,15 @@ public class LocalBusPatientService extends BaseService<BusPatientMapper, BusPat
                 result.setToday(result.getToday()+1);
                 return;
             }
-            if (includeTimes(startTime, 1)) {
+            if (includeTimes(startTime, -1)) {
                 result.setOneDay(result.getOneDay()+1);
                 return;
             }
-            if (includeTimes(startTime, 2)) {
+            if (includeTimes(startTime, -2)) {
                 result.setTwoDay(result.getTwoDay()+1);
                 return;
             }
-            if (includeTimes(startTime, 3)) {
+            if (includeTimes(startTime, -3)) {
                 result.setThreeDay(result.getThreeDay()+1);
                 return;
             }
@@ -321,7 +322,7 @@ public class LocalBusPatientService extends BaseService<BusPatientMapper, BusPat
         LocalDateTime dateTime = LocalDateTime.now().plusDays(offset);
         LocalDateTime beginTime = LocalDateTimeUtil.beginOfDay(dateTime);
         LocalDateTime endTime = LocalDateTimeUtil.endOfDay(dateTime);
-        return beginTime.getSecond()<time.getSeconds()
-                && time.getSeconds()<endTime.getSecond();
+        return beginTime.toInstant(ZoneOffset.of("+8")).toEpochMilli()<time.getTime()
+                && time.getTime()<endTime.toInstant(ZoneOffset.of("+8")).toEpochMilli();
     }
 }

coffee-system/src/main/java/com/coffee/bus/service/dto/ClinicStatsReturnResult.java

@@ -40,6 +40,9 @@ public class ClinicStatsReturnResult implements Serializable {
             return result;
         }
         queryResults.forEach(queryResult -> {
+            if(queryResult.getUploadTime()==null){
+                return;
+            }
             String uploadTime = DateUtil.format(queryResult.getUploadTime(), DatePattern.NORM_DATETIME_PATTERN);
             List<BigDecimal> continueDoses = Optional.ofNullable(result.getContinueDose()).orElse(new ArrayList<>());
             List<BigDecimal> appendDoses = Optional.ofNullable(result.getAppendDose()).orElse(new ArrayList<>());

coffee-system/src/main/java/com/coffee/bus/service/dto/PatientDeviceNoneResult.java

@@ -19,6 +19,9 @@ import java.io.Serializable;
 @Data
 @ApiModel("病人无设备返回数据")
 public class PatientDeviceNoneResult  implements Serializable {
+    @ApiModelProperty("临床id")
+    private String clinicId;
+
     @ApiModelProperty("病患名称")
     private String name;
 

coffee-system/src/main/java/com/coffee/bus/utils/WsPublishUtils.java

@@ -62,6 +62,7 @@ public class WsPublishUtils implements Serializable{
     public void publishPatientMonitor(String patientCode,String tenantId){
         Assert.hasText(tenantId,"医院id不能为空");
         PatientMonitorResult message = patientService.lookMonitorByPatientCode(patientCode, tenantId);
+        message.handleWarn();
         this.publish(WebSocketConstant.getPatientMonitor(null, patientCode, tenantId).getTopic(),
                 TopicMessage.of(message,patientCode)
         );

coffee-system/src/main/java/com/coffee/bus/websocket/listener/DeviceInfoListener.java

@@ -12,6 +12,8 @@ import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
 import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
 import com.baomidou.mybatisplus.core.toolkit.IdWorker;
+import com.coffee.bus.config.HospitalFunctionAnalConfigHandler;
+import com.coffee.bus.config.bean.FunctionAnalConfig;
 import com.coffee.bus.entity.*;
 import com.coffee.bus.enums.PatientAlarmEnum;
 import com.coffee.bus.registry.device.DeviceRegistry;
@@ -24,8 +26,10 @@ import com.coffee.bus.registry.patient.bean.DeviceTimeSmallInfo;
 import com.coffee.bus.registry.patient.bean.PatientCacheInfo;
 import com.coffee.bus.service.*;
 import com.coffee.bus.utils.WsPublishUtils;
+import com.coffee.common.util.RedissonUtil;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RedissonClient;
 import org.springframework.context.event.EventListener;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
@@ -70,6 +74,10 @@ public class DeviceInfoListener {
     private final WsPublishUtils wsPublishUtils;
 
     private final LocalBusDeviceService deviceService;
+
+    private final RedissonUtil redissonUtil;
+
+
     /**
      * 监听上传的数据信息，
      * 若设备详情发生变化，则及时通知相应的ws通道
@@ -86,7 +94,7 @@ public class DeviceInfoListener {
             /****************处理泵数据 todo 后续交由上游处理****************/
             BusDeviceRunningEntity device = infoEvent.getContent();
             device.setMonitorType(true);
-            device.setMaster(true);
+            device.setMaster(false);
             //1、判断该设备是否已和医院绑定并开启使用
             String deviceId = device.getDeviceId();
             device.setUploadTime(device.getUploadTime()==null?new Date():device.getUploadTime());
@@ -105,13 +113,14 @@ public class DeviceInfoListener {
             }
 
             log.info("接收到设备数据:[{}]", JSONUtil.toJsonStr(infoEvent.getContent()));
-            AtomicBoolean newInfusion=new AtomicBoolean(false);
             //缓存操作
             List<Supplier<?>> cacheOperation=new ArrayList<>();
             //处理设备运行数据,     判断是否为注册后第一次开机，判断是否为新的输注信息
-            boolean first = handleRunningInfo(device, deviceOperator,newInfusion,cacheOperation);
+            boolean first = handleRunningInfo(device, deviceOperator,cacheOperation);
+
+
             //处理输注参数
-            handleInfusion(device, deviceOperator, newInfusion, cacheOperation);
+            handleInfusion(device, deviceOperator, cacheOperation);
             //处理历史运行数据
             BusDeviceHistoryEntity history=handleRunningHistory(device);
             //处理报警、提醒信息
@@ -155,10 +164,9 @@ public class DeviceInfoListener {
      * 设备运行数据处理，返回是否为第一次接受数据消息
      * @param device 接收到的设备信息
      * @param deviceOperator 设备缓存信息操作符
-     * @param newInfusion 是否为新的输注
      * @return 是否为第一次接收数据消息
      */
-    private boolean handleRunningInfo( BusDeviceRunningEntity device,DeviceOperator<DeviceCacheInfo> deviceOperator,AtomicBoolean newInfusion,List<Supplier<?>> suppliers){
+    private boolean handleRunningInfo( BusDeviceRunningEntity device,DeviceOperator<DeviceCacheInfo> deviceOperator,List<Supplier<?>> suppliers){
         //判断此条数据的分包标识是否发生了改变
         String originClassify = deviceOperator.getClassification();
         String classification = device.getClassification();
@@ -167,7 +175,7 @@ public class DeviceInfoListener {
         //todo 这部分操作交由上游处理
         device.setTenantId(deviceOperator.getTenantId());
         if(!classification.equals(originClassify)){
-            newInfusion.set(true);
+            device.setNewInfusion(true);
             //启动新的输注，则撤泵标识取消
             device.setIsUndo(false);
             //分包标识发生了改变，设备开机时间重新计算
@@ -182,15 +190,14 @@ public class DeviceInfoListener {
             first=true;
         }else {
             device.setId(usingId);
-            if(!newInfusion.get()){
+            if(!device.isNewInfusion()){
                 device.setStartTime(deviceOperator.getStartTime());
             }
         }
         suppliers.add(()->{
             //更新设备缓存数据
-            if(newInfusion.get()){
+            if(device.isNewInfusion()){
                 deviceOperator.setUndo(false);
-//                patientOperator.setFinished(false);
             }
             deviceOperator.setClassification(device.getClassification());
 
@@ -210,7 +217,7 @@ public class DeviceInfoListener {
      * @param device 接收到的设备数据
      * @return 是否发生了换泵操作
      */
-    private void handlePatient(BusDeviceRunningEntity device, AtomicBoolean newInfusion,List<Supplier<?>> suppliers){
+    private void handlePatient(BusDeviceRunningEntity device,List<Supplier<?>> suppliers){
         String deviceId = device.getDeviceId();
         DeviceOperator<DeviceCacheInfo> currentDeviceOperator = deviceRegistry.getOperator(deviceId);
         PatientOperator<PatientCacheInfo> currentPatientOperator = patientRegistry.getOperator(device.getTenantId(), device.getPatientCode());
@@ -221,7 +228,7 @@ public class DeviceInfoListener {
             log.error("病号：【{}】临床发生由无泵转为有泵",device.getPatientCode());
             patientService.update(new UpdateWrapper<BusPatientEntity>().lambda().eq(BusPatientEntity::getCode,device.getPatientCode())
                     .eq(BusPatientEntity::getTenantId,device.getTenantId())
-                    .set(newInfusion.get(),BusPatientEntity::getInfusionId,device.getInfusionId())
+                    .set(device.isNewInfusion(),BusPatientEntity::getInfusionId,device.getInfusionId())
                     .set(BusPatientEntity::getAlarm, null)
             );
             suppliers.add(()->{
@@ -235,10 +242,10 @@ public class DeviceInfoListener {
                 return null;
             });
         }else {
-            if(newInfusion.get()){
+            if(device.isNewInfusion()){
                 patientService.update(new UpdateWrapper<BusPatientEntity>().lambda().eq(BusPatientEntity::getCode,device.getPatientCode())
                         .eq(BusPatientEntity::getTenantId,device.getTenantId())
-                        .set(newInfusion.get(),BusPatientEntity::getInfusionId,device.getInfusionId()));
+                        .set(device.isNewInfusion(),BusPatientEntity::getInfusionId,device.getInfusionId()));
             }
         }
         //更新泵所绑定当前病人缓存信息
@@ -364,13 +371,12 @@ public class DeviceInfoListener {
      * 处理输注历史记录信息
      * @param device 设备当前运行数据
      * @param deviceOperator 设备数据操作符
-     * @param newInfusion 是否创建新的输注记录
      * @param cacheOperation 缓存操作
      */
-    private BusInfusionHistoryEntity handleInfusionHistory(BusDeviceRunningEntity device, DeviceOperator<DeviceCacheInfo> deviceOperator, AtomicBoolean newInfusion, List<Supplier<?>> cacheOperation) {
+    private BusInfusionHistoryEntity handleInfusionHistory(BusDeviceRunningEntity device, DeviceOperator<DeviceCacheInfo> deviceOperator, List<Supplier<?>> cacheOperation) {
         BusInfusionHistoryEntity infusionHistory = BusInfusionHistoryEntity.parseRunningInfo(device);
         String originInfusionId = deviceOperator.getInfusionId();
-        if (newInfusion.get()) {
+        if (device.isNewInfusion()) {
             //检测数据库是否存在该输注信息
             infusionHistory.setId(String.valueOf(IdWorker.getId()));
             infusionHistoryService.update(new UpdateWrapper<BusInfusionHistoryEntity>().lambda()
@@ -394,18 +400,17 @@ public class DeviceInfoListener {
      * 处理输注信息
      * @param device 设备运行数据
      * @param deviceOperator 设备操作符
-     * @param  newInfusion 是否为新的输注记录
      * @param cacheOperation 缓存操作
      */
-    private BusInfusionHistoryEntity handleInfusion(BusDeviceRunningEntity device, DeviceOperator<DeviceCacheInfo> deviceOperator, AtomicBoolean newInfusion, List<Supplier<?>> cacheOperation) {
+    private BusInfusionHistoryEntity handleInfusion(BusDeviceRunningEntity device, DeviceOperator<DeviceCacheInfo> deviceOperator, List<Supplier<?>> cacheOperation) {
         PatientOperator<PatientCacheInfo> operator = patientRegistry.getOperator(device.getTenantId(), device.getPatientCode(),MapUtil.of("startTime", device.getStartTime()));
         device.setClinicId(operator.getClinicId());
 
-        BusInfusionHistoryEntity infusionHistory = handleInfusionHistory(device, deviceOperator, newInfusion, cacheOperation);
+        BusInfusionHistoryEntity infusionHistory = handleInfusionHistory(device, deviceOperator, cacheOperation);
         BusInfusionModifyEntity busInfusionModify = BusInfusionModifyEntity.parseRunningInfo(device);
         String signHex = busInfusionModify.signParam(sign);
         String infusionParam = deviceOperator.getInfusionParam();
-
+        device.setMaster(true);
         //输注参数已发生变化
         if(ObjectUtil.notEqual(signHex,infusionParam)){
             infusionModifyService.save(busInfusionModify);
@@ -415,8 +420,8 @@ public class DeviceInfoListener {
 
             if(Boolean.TRUE.equals(patientOperator.getFinished())){
                 //若临床已结束，则只有新的输注能够进行泵冲突处理
-                if(newInfusion.get()){
-                    handlePatient(device,newInfusion,cacheOperation);
+                if(device.isNewInfusion()){
+                    handlePatient(device,cacheOperation);
                     clinicService
                             .update(new UpdateWrapper<BusClinicEntity>().lambda().eq(BusClinicEntity::getId,patientOperator.getClinicId())
                                     .set(BusClinicEntity::getFinished,false));
@@ -427,7 +432,7 @@ public class DeviceInfoListener {
                     });
                 }
             }else {
-                handlePatient(device,newInfusion,cacheOperation);
+                handlePatient(device,cacheOperation);
             }
             cacheOperation.add(()->{
                 deviceOperator.setInfusionParam(signHex);
@@ -473,7 +478,7 @@ public class DeviceInfoListener {
     private void handleAlarmOrWarn(BusDeviceHistoryEntity history, DeviceOperator deviceOperator, List<Supplier<?>> cacheOperation){
         if (BusDeviceAlarmEntity.alarmOrWarn(history)) {
             //获取上一状态下的报警信息
-            String lastAlarmSign = deviceOperator.getAlarm();
+            String lastAlarmSign = deviceOperator.getAlarmOrWarn();
             BusDeviceAlarmEntity alarm = BusDeviceAlarmEntity.parseHistory(history);
             String alarmSign = alarm.signParm(sign);
 
@@ -506,14 +511,14 @@ public class DeviceInfoListener {
                 if(change!=0){
                     alarmService.save(alarm);
                     cacheOperation.add(()->{
-                        deviceOperator.setAlarm(alarmSign);
+                        deviceOperator.setAlarmOrWarn(alarmSign);
                         return null;
                     });
                 }
             }
         }else {
             cacheOperation.add(()->{
-                deviceOperator.setAlarm(null);
+                deviceOperator.setAlarmOrWarn(null);
                 return null;
             });
         }

coffee-system/src/main/resources/mapper/bus/BusClinicMapper.xml

@@ -62,7 +62,7 @@
             </if>
         </where>)
         as i
-        left JOIN (select * from bus_device_history
+        JOIN (select * from bus_device_history
         <where>
             <if test="query.infusionModifyIds != null and query.infusionModifyIds.size > 0">
                 and infusion_modify_id in

coffee-system/src/main/resources/mapper/bus/BusDeviceHistoryMapper.xml

@@ -8,6 +8,7 @@
          bus_device_history  as h
         join (select * from bus_infusion_history where clinic_id=#{query.clinicId} and device_id=#{query.deviceId}) as i
         on h.infusion_id=i.id
+        order by h.upload_time
     </select>
 
 

coffee-system/src/main/resources/mapper/bus/BusPatientMapper.xml

@@ -76,6 +76,7 @@
     </resultMap>
 
     <resultMap id="deviceNone" type="com.coffee.bus.service.dto.PatientDeviceNoneResult">
+        <result column="clinic_id" property="clinicId"/>
         <result column="name" property="name"/>
         <result column="gender" property="gender"/>
         <result column="code" property="code"/>
@@ -118,6 +119,7 @@
     <select id="selectNoneDevice" resultMap="deviceNone">
         SELECT
          p.code as code,
+         c.id as clinic_id,
          c.patient_gender as gender,
          c.`patient_name` as name,
          c.patient_age as age,
@@ -133,9 +135,11 @@
          c.asa as asa,
          c.formula as formula,
          c.finished as finished,
+         c.monitor_start_time as monitor_start_time,
          c.config_person as config_person
         FROM (SELECT `name`,gender,CODE,tenant_id FROM bus_patient WHERE bus_patient.`alarm`=2 ) AS p
-        left join  (SELECT * FROM bus_clinic WHERE finished=0 ) AS c ON  c.`patient_code`=p.code;
+        left join  (SELECT * FROM bus_clinic WHERE finished=0 ) AS c ON  c.`patient_code`=p.code
+        order by c.monitor_start_time desc;
     </select>
 
     <select id="selectMonitor" resultMap="monitorResult" parameterType="com.coffee.bus.service.dto.PatientMonitorQuery">
@@ -266,7 +270,7 @@
         </where>
         ) as i on p.infusion_id=i.id
         join (select device_id,alias from bus_device_running) as r on r.device_id=i.device_id
-        left join (select * from bus_clinic
+        join (select * from bus_clinic
         <where>
             <if test="query.timeRange != null and query.timeRange.size >0">
                 and monitor_start_time &gt; #{query.timeRange[0]} and  monitor_start_time &lt; #{query.timeRange[1]}

pom.xml

@@ -37,7 +37,7 @@
         <tio.version>3.6.0.v20200315-RELEASE</tio.version>
         <jython.version>2.7.1</jython.version>
         <knife4j.version>2.0.7</knife4j.version>
-        <redisson.version>3.13.6</redisson.version>
+        <redisson.version>3.17.0</redisson.version>
     </properties>
 
     <modules>
@@ -211,6 +211,11 @@
                 <version>${coffee.project.version}</version>
             </dependency>
 
+            <dependency>
+                <groupId>org.jboss.marshalling</groupId>
+                <artifactId>jboss-marshalling</artifactId>
+                <version>2.0.9.Final</version>
+            </dependency>
             <!-- redisson配置 -->
             <dependency>
                 <groupId>org.redisson</groupId>