add 新增病人主题监听

fix 修复病人缓存数据转换bug

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

@@ -12,8 +12,10 @@ import com.coffee.common.config.websocket.handler.TopicWrapper;
  */
 public class WebSocketConstant {
     public static final String MONITOR_TIME_COUNT ="monitor-time-count";
+
     public static final String MONITOR_STATE_COUNT ="monitor-state-count";
-    public static final String DEVICE_CONFLICT = "device-conflict";
+
+    public static final String PATIENT_ADD ="patient-add";
     /**
      * 病人监控订阅
      */
@@ -44,24 +46,22 @@ public class WebSocketConstant {
     }
 
     /**
-     * 获取 临床设备冲突主题
+     * 获取 病人监护变化主题
      * @param productName
      * @param param
      * @param tenantId
      * @return
      */
-    public static TopicWrapper getDeviceConflictTopic(String productName,String param,String tenantId){
-        return getTopic(DEVICE_CONFLICT,productName,param,tenantId);
+    public static TopicWrapper getPatientMonitor(String productName,String param,String tenantId){
+        return getTopic(PATIENT_MONITOR,productName,param,tenantId);
     }
 
     /**
-     * 获取 病人监护变化主题
-     * @param productName
-     * @param param
-     * @param tenantId
+     * 获取 病人新增主题
+     * @param tenantId 设备所属医院
      * @return
      */
-    public static TopicWrapper getPatientMonitor(String productName,String param,String tenantId){
-        return getTopic(PATIENT_MONITOR,productName,param,tenantId);
+    public static TopicWrapper getPatientAdd(String tenantId){
+        return getTopic(PATIENT_ADD,null,null,tenantId);
     }
 }

coffee-common/src/main/java/com/coffee/common/crud/BaseService.java

@@ -74,6 +74,16 @@ public abstract class BaseService<M extends BaseMapper<E>, E,PK extends Serializ
         return false;
     }
 
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public boolean save(E entity) {
+        if (super.save(entity)) {
+            postSave(entity);
+            return true;
+        }
+        return false;
+    }
+
     /**
      * @Author lifang
      * @Date 10:05 2022/3/14

coffee-system/src/main/java/com/coffee/bus/enums/DeviceAlarmEnum.java

@@ -32,11 +32,7 @@ public enum DeviceAlarmEnum  implements IEnum<Integer> {
     LowBattery(7,"电量耗尽报警"),
     OutOfControl(8,"电机失控报警"),
     Machine(9,"机械故障"),
-    NoSignal(10,"不在服务区"),
-
-
-
-    ;
+    NoSignal(10,"不在服务区");
 
 
     @Getter

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

@@ -2,6 +2,7 @@ package com.coffee.bus.registry.patient;
 
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.util.StrUtil;
+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;
@@ -10,6 +11,7 @@ import com.coffee.common.enums.SexEnum;
 import lombok.AllArgsConstructor;
 
 import java.util.*;
+import java.util.stream.Collectors;
 
 /**
  * @author lifang
@@ -18,10 +20,13 @@ import java.util.*;
  * @Description 病号操作符
  * @createTime 2022年04月07日 15:25:00
  */
-@AllArgsConstructor
 public class ClusterPatientOperator implements PatientOperator<PatientCacheInfo> {
     private final ConfigStorage configStorage;
 
+    public ClusterPatientOperator(ConfigStorage configStorage) {
+        this.configStorage = configStorage;
+    }
+
     @Override
     public ConfigStorage getConfig() {
         return configStorage;
@@ -212,7 +217,14 @@ public class ClusterPatientOperator implements PatientOperator<PatientCacheInfo>
 
     @Override
     public Set<DeviceTimeSmallInfo> getAllDevice() {
-        return getValue(PatientKeyConstant.DEVICES).as(HashSet.class);
+        Set<Map<String,Object>> set = getValue(PatientKeyConstant.DEVICES).as(HashSet.class);
+        if(CollectionUtil.isNotEmpty(set)){
+            return set.stream()
+                    .map(JSONUtil::toJsonStr)
+                    .map(json-> JSONUtil.toBean(json,DeviceTimeSmallInfo.class))
+                    .collect(Collectors.toSet());
+        }
+        return new HashSet<>();
     }
 
     @Override
@@ -246,4 +258,8 @@ public class ClusterPatientOperator implements PatientOperator<PatientCacheInfo>
     public List<String> getAllKeys(){
         return Arrays.asList("code","gender","name","tenantId","clinicId","startTime","finished","bindDeviceId","devices");
     }
+
+    //该病人是否为新增病人
+    private boolean newPatient;
+
 }

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

@@ -1,13 +1,17 @@
 package com.coffee.bus.registry.patient;
 
+import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.coffee.bus.entity.BusClinicEntity;
+import com.coffee.bus.entity.BusDeviceRunningEntity;
 import com.coffee.bus.entity.BusInfusionHistoryEntity;
 import com.coffee.bus.entity.BusPatientEntity;
 import com.coffee.bus.registry.RegistryConstant;
+import com.coffee.bus.registry.patient.bean.DeviceTimeSmallInfo;
 import com.coffee.bus.registry.patient.bean.PatientCacheInfo;
 import com.coffee.bus.service.LocalBusClinicService;
+import com.coffee.bus.service.LocalBusDeviceRunningService;
 import com.coffee.bus.service.LocalBusInfusionHistoryService;
 import com.coffee.bus.service.LocalBusPatientService;
 import com.coffee.common.cache.manager.ClusterConfigStorageManager;
@@ -17,7 +21,11 @@ import org.springframework.dao.DuplicateKeyException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
-import java.util.Map;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
 
 /**
  * @author lifang
@@ -34,6 +42,7 @@ public class ClusterPatientRegistry implements PatientRegistry {
     private final ClusterConfigStorageManager configStorageManager;
     private final LocalBusClinicService clinicService;
     private final LocalBusInfusionHistoryService infusionService;
+    private final LocalBusDeviceRunningService deviceRunningService;
     @Override
     @Transactional(rollbackFor = Exception.class)
     public PatientOperator<PatientCacheInfo> getOperator(String hospitalId, String patientCode) {
@@ -41,7 +50,7 @@ public class ClusterPatientRegistry implements PatientRegistry {
             return null;
         }
         String key=getId()+hospitalId+":"+patientCode;
-        ClusterPatientOperator patientOperator = new ClusterPatientOperator( configStorageManager.getStorage(key));
+        ClusterPatientOperator patientOperator = new ClusterPatientOperator(configStorageManager.getStorage(key));
         String code = patientOperator.getCode();
         if(StrUtil.isNullOrUndefined(code)){
             //将新数据存入数据库
@@ -67,8 +76,8 @@ public class ClusterPatientRegistry implements PatientRegistry {
             //不存在临床或临床已结束
             if(clinic==null||Boolean.TRUE.equals(clinic.getFinished())){
                 clinicService.asyncFromHis(hospitalId,patientCode);
-                patient.setClinicId("-1");
-                patientOperator.setClinicId("-1");
+                patient.setClinicId("");
+                patientOperator.setClinicId("");
             }else {
                 //填充临床信息
                 patient.setName(clinic.getName());
@@ -88,6 +97,19 @@ public class ClusterPatientRegistry implements PatientRegistry {
                 patient.setMonitorStartTime(infusion.getStartTime());
                 patient.setMonitorEndTime(infusion.getUndoTime());
             }
+            //填充主副泵信息
+            List<BusDeviceRunningEntity> runningList = deviceRunningService.list(new QueryWrapper<BusDeviceRunningEntity>().lambda().eq(BusDeviceRunningEntity::getPatientCode, patientCode).eq(BusDeviceRunningEntity::getTenantId, hospitalId));
+            if (CollectionUtil.isNotEmpty(runningList)) {
+                Set<DeviceTimeSmallInfo> allDevice = runningList.stream().map(running ->
+                        DeviceTimeSmallInfo.of(running.getDeviceId(), running.getStartTime())
+                )
+                        .collect(Collectors.toSet());
+                Optional<BusDeviceRunningEntity> first = runningList.stream().filter(running -> Boolean.TRUE.equals(running.getMaster())).findFirst();
+                if(first.isPresent()){
+                    patientOperator.setBindDeviceId(first.get().getDeviceId());
+                }
+                patientOperator.setAllDevice(allDevice);
+            }
             if(insert){
                 try {
                     log.info("医院[{}]新增病号数据[{}]",hospitalId,patientCode);

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

@@ -165,6 +165,4 @@ public interface PatientOperator<T> extends Operator<T> {
      * 情空病号泵信息
      */
     void clearDevice();
-
-
 }

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

@@ -3,6 +3,8 @@ package com.coffee.bus.service;
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.date.LocalDateTimeUtil;
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.core.metadata.IPage;
@@ -17,6 +19,8 @@ import com.coffee.bus.mapper.BusPatientMapper;
 import com.coffee.bus.registry.patient.PatientOperator;
 import com.coffee.bus.registry.patient.PatientRegistry;
 import com.coffee.bus.registry.patient.bean.PatientCacheInfo;
+import com.coffee.bus.utils.WsPublishUtils;
+import com.coffee.common.config.websocket.WebSocketConstant;
 import com.coffee.common.crud.BaseService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -27,6 +31,10 @@ import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author lifang
@@ -45,6 +53,13 @@ public class LocalBusPatientService extends BaseService<BusPatientMapper, BusPat
     @Lazy
     private PatientRegistry patientRegistry;
 
+    @Autowired
+    private WsPublishUtils wsPublishUtils;
+
+    private ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
+
+
+
     @Override
     public void validateBeforeSave(BusPatientEntity entity) {
 
@@ -60,7 +75,16 @@ public class LocalBusPatientService extends BaseService<BusPatientMapper, BusPat
 
     }
 
-
+    @Override
+    public void postSave(BusPatientEntity entity) {
+        super.postSave(entity);
+        //新增病人后推送主题,延迟推送，保证处理逻辑已全部完成
+        executorService.schedule(()->{
+                    wsPublishUtils.publish(WebSocketConstant.getPatientAdd(entity.getTenantId()).getTopic(),
+                            new JSONObject().putOpt("patientCode",entity.getCode()));
+                }
+        ,3, TimeUnit.SECONDS);
+    }
 
     /**
      * 异步获取病人信息 todo

coffee-system/src/main/java/com/coffee/bus/websocket/MonitorTimeCountHandler.java

@@ -17,7 +17,7 @@ public class MonitorTimeCountHandler extends Subscribe {
 
     @Override
     public String getId() {
-        return WebSocketConstant.MONITOR_STATE_COUNT;
+        return WebSocketConstant.MONITOR_TIME_COUNT;
     }
 
     @Override

coffee-system/src/main/java/com/coffee/bus/websocket/PatientAddHandler.java

@@ -0,0 +1,32 @@
+package com.coffee.bus.websocket;
+
+import com.coffee.common.config.websocket.WebSocketConstant;
+import com.coffee.common.config.websocket.handler.Subscribe;
+import org.springframework.stereotype.Component;
+import org.tio.core.ChannelContext;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName DeviceStateCountHandler.java
+ * @Description 新增病号主题
+ * @createTime 2022年03月25日 14:21:00
+ */
+@Component
+public class PatientAddHandler extends Subscribe {
+
+    @Override
+    public String getId() {
+        return WebSocketConstant.PATIENT_ADD;
+    }
+
+    @Override
+    public void close(ChannelContext channelContext) {
+
+    }
+
+    @Override
+    public boolean needParam() {
+        return false;
+    }
+}

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

@@ -172,13 +172,13 @@ public class DeviceInfoListener {
             //启动新的输注，则撤泵标识取消
             device.setIsUndo(false);
             //分包标识发生了改变，设备开机时间重新计算
-            device.setStartTime(new Date());
+            device.setStartTime(device.getUploadTime());
         }
         boolean first=false;
         if(StrUtil.isNullOrUndefined(usingId)){
             //设备首次运行，记录开机时间
             device.setId(String.valueOf(IdWorker.getId()));
-            device.setStartTime(new Date());
+            device.setStartTime(device.getUploadTime());
             device.setAlias(deviceOperator.getAlias());
             first=true;
         }else {
@@ -259,7 +259,9 @@ public class DeviceInfoListener {
         Set<DeviceTimeSmallInfo> allDevice = patientOperator.getAllDevice();
         //过滤掉已换绑的泵，获取剩余所绑定的泵数据
         if(CollectionUtils.isNotEmpty(allDevice)){
-            Set<DeviceTimeSmallInfo> remainPatientBindDevices = allDevice.stream().filter(bindDevice -> !bindDevice.getDeviceId().equals(deviceId)).collect(Collectors.toSet());
+            Set<DeviceTimeSmallInfo> remainPatientBindDevices = allDevice.stream()
+                    .filter(bindDevice -> !deviceId.equals(bindDevice.getDeviceId()))
+                    .collect(Collectors.toSet());
             if(CollectionUtil.isEmpty(remainPatientBindDevices)){
                 log.error("病号：【{}】临床发生无泵报警",patientCode);
                 patientService.update(new UpdateWrapper<BusPatientEntity>().lambda().eq(BusPatientEntity::getCode,patientCode)
@@ -270,7 +272,7 @@ public class DeviceInfoListener {
                 suppliers.add(()->{
                     //todo 发起无泵报警,处理原先泵的无泵信息
                     patientOperator.setBindDeviceId(null);
-                    patientOperator.setAllDevice(new ArrayList<>());
+                    patientOperator.setAllDevice(new HashSet<>());
                     //发起无泵报警后,将该病人最后一条输注作为显示信息实时传输给前端 //todo
                     return null;
                 });