fix ws订阅发布Bug

fix 泵重复查询根据病人+临床区分

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

@@ -0,0 +1,120 @@
+package com.coffee.admin;
+
+import cn.hutool.core.date.DateField;
+import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.util.EnumUtil;
+import cn.hutool.core.util.RandomUtil;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.coffee.bus.entity.BusClinicEntity;
+import com.coffee.bus.entity.BusDeviceEntity;
+import com.coffee.bus.his.strategy.HisStrategyEnum;
+import com.coffee.bus.his.strategy.HisStrategyHandler;
+import com.coffee.bus.his.strategy.HisStrategyManager;
+import com.coffee.bus.his.strategy.HisStrategyManagerRegister;
+import com.coffee.bus.service.LocalBusClinicService;
+import com.coffee.bus.service.LocalBusDeviceService;
+import com.coffee.common.enums.SexEnum;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import static com.coffee.admin.BusPatientTest.getName;
+
+/**
+ * @Author XX
+ * @Date 2022-04-25 19:00:10
+ * @Version 1.0
+ * @Description XXX
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = AdminApplication.class)
+public class HisStrategyTest {
+
+    @Autowired
+    HisStrategyManagerRegister managerRegister;
+
+    @Autowired
+    LocalBusClinicService clinicService;
+
+    String patientCode="120263";
+    String tenantId="1";
+    @Test
+    public void allEquals(){
+        HisStrategyManager<? extends HisStrategyHandler> hisStrategyManager = managerRegister.get(HisStrategyEnum.ALL);
+
+    }
+
+
+    @Test
+    public void allMoreToLess(){
+        HisStrategyManager<? extends HisStrategyHandler> hisStrategyManager = managerRegister.get(HisStrategyEnum.ALL);
+        List<BusClinicEntity> sources = new ArrayList<>();
+        List<BusClinicEntity> targets = clinicService.list(new QueryWrapper<BusClinicEntity>().lambda().eq(BusClinicEntity::getPatientCode, patientCode).eq(BusClinicEntity::getTenantId, tenantId));
+        for (BusClinicEntity target : targets) {
+            BusClinicEntity clinic = new BusClinicEntity();
+            clinic.setStartTime(target.getStartTime()==null?
+                    RandomUtil.randomDate(DateUtil.beginOfMonth(new Date()), DateField.HOUR,-30,30):DateUtil.date(target.getStartTime()).offset(DateField.HOUR_OF_DAY,-1));
+            clinic.setName(BusPatientTest.clinicNames.get(RandomUtil.randomInt(BusPatientTest.clinicNames.size()-1)));
+            clinic.setPatientName(getName());
+            clinic.setPatientGender(EnumUtil.likeValueOf(SexEnum.class,RandomUtil.randomInt(1,2)));
+            clinic.setBedNo(target.getBedNo());
+            clinic.setWard(target.getWard());
+            clinic.setWeight(String.valueOf(RandomUtil.randomInt(90,150)));
+            clinic.setWeight(String.valueOf(RandomUtil.randomInt(150,200)));
+            clinic.setAnaDoctor(getName());
+            clinic.setAnaType(BusPatientTest.anaTypes.get(RandomUtil.randomInt(BusPatientTest.anaTypes.size()-1)));
+            clinic.setAnalType(BusPatientTest.analTypes.get(RandomUtil.randomInt(BusPatientTest.analTypes.size()-1)));
+            clinic.setSurgeryDoctor(getName());
+            clinic.setConfigPerson(getName());
+            clinic.setPatientCode(target.getPatientCode());
+            clinic.setMonitorType(true);
+            clinic.setTenantId("1");
+            sources.add(clinic);
+        }
+        BusClinicEntity clinic = new BusClinicEntity();
+        clinic.setStartTime(DateUtil.beginOfDay(new Date()));
+        clinic.setName(BusPatientTest.clinicNames.get(RandomUtil.randomInt(BusPatientTest.clinicNames.size()-1)));
+        clinic.setPatientName(getName());
+        clinic.setPatientGender(EnumUtil.likeValueOf(SexEnum.class,RandomUtil.randomInt(1,2)));
+        clinic.setBedNo("10");
+        clinic.setWard("10");
+        clinic.setWeight(String.valueOf(RandomUtil.randomInt(90,150)));
+        clinic.setWeight(String.valueOf(RandomUtil.randomInt(150,200)));
+        clinic.setAnaDoctor(getName());
+        clinic.setAnaType(BusPatientTest.anaTypes.get(RandomUtil.randomInt(BusPatientTest.anaTypes.size()-1)));
+        clinic.setAnalType(BusPatientTest.analTypes.get(RandomUtil.randomInt(BusPatientTest.analTypes.size()-1)));
+        clinic.setSurgeryDoctor(getName());
+        clinic.setConfigPerson(getName());
+        clinic.setPatientCode(patientCode);
+        clinic.setMonitorType(true);
+        clinic.setTenantId("1");
+        sources.add(clinic);
+        hisStrategyManager.getHandlers()
+                .stream()
+                .filter(handler-> handler.apply(sources,targets))
+                .forEach(handler->handler.handle(sources,targets));
+    }
+
+
+    @Test
+    public void allLessToMore(){
+        HisStrategyManager<? extends HisStrategyHandler> hisStrategyManager = managerRegister.get(HisStrategyEnum.ALL);
+
+    }
+    @Test
+    public void onlyNew(){
+        HisStrategyManager<? extends HisStrategyHandler> hisStrategyManager = managerRegister.get(HisStrategyEnum.NEW);
+
+    }
+
+    @Test
+    public void part(){
+        HisStrategyManager<? extends HisStrategyHandler> hisStrategyManager = managerRegister.get(HisStrategyEnum.PART);
+    }
+}

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

@@ -1,9 +1,8 @@
 package com.coffee.common.config.websocket;
 
 import cn.hutool.core.collection.CollectionUtil;
-import cn.hutool.core.util.ObjectUtil;
-import cn.hutool.json.JSON;
 import cn.hutool.json.JSONUtil;
+import com.coffee.common.config.websocket.handler.Subscribe;
 import com.coffee.common.config.websocket.handler.TopicWrapper;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -14,6 +13,8 @@ import org.springframework.data.redis.connection.MessageListener;
 import org.tio.core.ChannelContext;
 import org.tio.core.Tio;
 import org.tio.websocket.common.WsResponse;
+
+import java.util.Map;
 import java.util.Set;
 
 /**
@@ -27,12 +28,12 @@ import java.util.Set;
 @Slf4j
 public class DefaultMessageListener implements MessageListener {
     private final String id;
-    private final Set<ChannelContext> channelContexts;
     private final TopicWrapper topicWrapper;
     private final ObjectMapper objectMapper;
 
     @Override
     public void onMessage(Message message, byte[] pattern) {
+        Set<ChannelContext> channelContexts = Subscribe.getSubscribeChannel(topicWrapper.getTopic());
         if (CollectionUtil.isNotEmpty(channelContexts)) {
             channelContexts.parallelStream()
                     .forEach(channel -> {

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

@@ -6,6 +6,7 @@ import cn.hutool.core.util.StrUtil;
 import cn.hutool.json.JSONUtil;
 import com.coffee.common.Constants;
 import com.coffee.common.bo.LoginUser;
+import com.coffee.common.cache.value.Value;
 import com.coffee.common.config.websocket.handler.WsHandler;
 import com.coffee.common.result.R;
 import lombok.AllArgsConstructor;
@@ -93,6 +94,7 @@ public class DefaultWebSocketMsgHandler implements IWsMsgHandler {
                 log.debug("websocket 接收到消息,message:{},token:{},userId:{}",message,channelContext.getToken(),JSONUtil.toJsonStr(channelContext.get(Constants.LOGIN_USER_KEY)));
             }
             MessagingRequest messagingRequest = JSONUtil.toBean(message, MessagingRequest.class);
+
             messagingRequest.validate();
             List<WsHandler> collect = messageHandlers
                     .parallelStream()
@@ -104,7 +106,9 @@ public class DefaultWebSocketMsgHandler implements IWsMsgHandler {
                 channelContext.setClosed(true);
                 return null;
             }
-            collect.forEach(handler->handler.onMessage(messagingRequest,channelContext));
+            synchronized (channelContext){
+                collect.forEach(handler->handler.onMessage(messagingRequest,channelContext));
+            }
         }catch (Exception e){
             e.printStackTrace();
             log.warn("websocket 接收到异常请求,token:{},message:{},userId:{}",channelContext.getToken(),message,JSONUtil.toJsonStr(channelContext.get(Constants.LOGIN_USER_KEY)));

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

@@ -16,6 +16,10 @@ public class WebSocketConstant {
     public static final String MONITOR_STATE_COUNT ="monitor-state-count";
 
     public static final String PATIENT_ADD ="patient-add";
+
+    public static final String DEVICE_REPEAT ="device-repeat";
+
+    public static final String DEVICE_NONE ="device-none";
     /**
      * 病人监控订阅
      */
@@ -64,4 +68,27 @@ public class WebSocketConstant {
     public static TopicWrapper getPatientAdd(String tenantId){
         return getTopic(PATIENT_ADD,null,null,tenantId);
     }
+
+
+    /**
+     * 描述: 获取医院临床设备重复数量统计
+     * @author lifang
+     * @date 2022/5/13 9:50
+     * @param tenantId
+     * @return TopicWrapper
+     */
+    public static TopicWrapper getDeviceRepeat(String tenantId){
+        return getTopic(DEVICE_REPEAT,null,null,tenantId);
+    }
+
+    /**
+     * 描述: 获取医院临床设备无绑定数量统计
+     * @author lifang
+     * @date 2022/5/13 9:50
+     * @param tenantId
+     * @return TopicWrapper
+     */
+    public static TopicWrapper getDeviceNone(String tenantId){
+        return getTopic(DEVICE_NONE,null,null,tenantId);
+    }
 }

coffee-common/src/main/java/com/coffee/common/config/websocket/handler/Subscribe.java

@@ -7,6 +7,7 @@ import com.coffee.common.config.websocket.DefaultMessageListener;
 import com.coffee.common.config.websocket.MessagingRequest;
 import com.coffee.common.config.websocket.WebSocketConstant;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.redis.connection.RedisConnection;
@@ -17,7 +18,9 @@ import org.tio.core.ChannelContext;
 
 import javax.annotation.Resource;
 import java.util.*;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
 /**
@@ -27,6 +30,7 @@ import java.util.stream.Collectors;
  * @Description TODO
  * @createTime 2022年03月25日 14:18:00
  */
+@Slf4j
 public abstract class Subscribe implements WsHandler {
     @Autowired
     private RedisTemplate<String,Object> redisTemplate;
@@ -38,13 +42,13 @@ public abstract class Subscribe implements WsHandler {
     /**
      * 存储主题与ws通道关联
      */
-    private Map<String,Set<ChannelContext>> subscribeTopics=new ConcurrentHashMap<>();
+    private static Map<String,Set<ChannelContext>> subscribeTopics=new ConcurrentHashMap<>();
 
 
     /**
      * 存储主题与redis通道关联
      */
-    private Map<String,RedisConnection> redisConnectionMap=new ConcurrentReferenceHashMap<>();
+    private static Map<String,RedisConnection> redisConnectionMap=new ConcurrentReferenceHashMap<>();
 
 
     public TopicWrapper getTopic(String productName,String param,String tenantId){
@@ -52,6 +56,10 @@ public abstract class Subscribe implements WsHandler {
 
     };
 
+    public static Set<ChannelContext> getSubscribeChannel(String topic){
+        return subscribeTopics.get(topic);
+    }
+
 
     @Override
     public void onMessage(MessagingRequest message, ChannelContext channelContext) {
@@ -78,11 +86,16 @@ public abstract class Subscribe implements WsHandler {
         MessagingRequest.Type type = message.getType();
         if(MessagingRequest.Type.sub==type){
             //订阅主题
-            subScribeTopic.forEach(topicWrapper->this.subscribe(channelContext,topicWrapper));
+            for (TopicWrapper topicWrapper : subScribeTopic) {
+                this.subscribe(channelContext,topicWrapper);
+            }
+//            subScribeTopic.forEach(topicWrapper->this.subscribe(channelContext,topicWrapper));
         }else {
             //取消订阅主题
             subScribeTopic.forEach(topicWrapper->this.unsubscribe(channelContext,topicWrapper.getTopic()));
         }
+        log.error("订阅成功{}",subScribeTopic.stream().map(TopicWrapper::getTopic).collect(Collectors.toList()));
+
     }
 
     /**
@@ -91,6 +104,7 @@ public abstract class Subscribe implements WsHandler {
      * @param topicWrapper
      */
     public void subscribe(ChannelContext channelContext, TopicWrapper topicWrapper){
+
         //将主题与ws通道绑定
         Object result = Optional.ofNullable(channelContext.get(SUBSCRIBE_TOPIC)).orElse(new HashSet<>());
         Set<String> subscribeTopicSet= (Set<String>) result;
@@ -100,19 +114,20 @@ public abstract class Subscribe implements WsHandler {
         subscribeTopicSet.add(topicWrapper.getTopic());
         channelContext.set(SUBSCRIBE_TOPIC,subscribeTopicSet);
         //同一主题只订阅一次
-        Set<ChannelContext> channelContexts = Optional.ofNullable(subscribeTopics.get(topicWrapper.getTopic())).orElse(new HashSet<>());
+        Set<ChannelContext> channelContexts = subscribeTopics.computeIfAbsent(topicWrapper.getTopic(), k -> new HashSet<>());
         channelContexts.add(channelContext);
-        if(!subscribeTopics.containsKey(topicWrapper.getTopic())){
+        if(!redisConnectionMap.containsKey(topicWrapper.getTopic())){
             redisTemplate.execute(new RedisCallback<Object>() {
                 @Override
                 public Object doInRedis(RedisConnection connection) throws DataAccessException {
-                    connection.pSubscribe(new DefaultMessageListener(getId(),channelContexts,topicWrapper,objectMapper),topicWrapper.getTopic().getBytes());
+                    CompletableFuture.runAsync(()->
+                            connection.subscribe(new DefaultMessageListener(getId(),topicWrapper,objectMapper),topicWrapper.getTopic().getBytes()));
                     redisConnectionMap.put(topicWrapper.getTopic(),connection);
                     return null;
                 }
             });
+
         }
-        subscribeTopics.put(topicWrapper.getTopic(),channelContexts);
     };
 
     /**
@@ -121,14 +136,12 @@ public abstract class Subscribe implements WsHandler {
      * @param topic
      */
     public void unsubscribe(ChannelContext channelContext, String topic){
-        if(subscribeTopics.containsKey(topic)){
-            Set<ChannelContext> channelContexts = subscribeTopics.get(topic);
-            if(CollectionUtil.isNotEmpty(channelContexts)){
-                channelContexts.remove(channelContext);
-            }
-            //重新获取集合,避免多线程发生冲突，再次判断此时是否为空
-            if(CollectionUtil.isEmpty(subscribeTopics.get(topic))){
-                subscribeTopics.remove(topic);
+        Object result = Optional.ofNullable(channelContext.get(SUBSCRIBE_TOPIC)).orElse(new HashSet<>());
+        Set<String> subscribeTopicSet= (Set<String>) result;
+        subscribeTopicSet.remove(topic);
+        subscribeTopics.computeIfPresent(topic,(k,v)->{
+            v.remove(channelContext);
+            if(CollectionUtil.isEmpty(v)){
                 redisTemplate.execute(new RedisCallback<Object>() {
                     @Override
                     public Object doInRedis(RedisConnection connection) throws DataAccessException {
@@ -141,7 +154,8 @@ public abstract class Subscribe implements WsHandler {
                 });
                 redisConnectionMap.remove(topic);
             }
-        }
+            return v;
+        });
     };
 
 

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

@@ -1,10 +1,13 @@
 package com.coffee.bus.controller;
 
+import cn.dev33.satoken.SaManager;
 import cn.dev33.satoken.annotation.SaCheckPermission;
+import cn.dev33.satoken.stp.StpLogic;
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
+import com.baomidou.mybatisplus.core.mapper.Mapper;
 import com.coffee.bus.controller.vo.MonitorFinishedVo;
 import com.coffee.bus.entity.*;
 import com.coffee.bus.enums.DeviceStatusEnum;
@@ -15,6 +18,8 @@ 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.service.dto.*;
+import com.coffee.common.crud.BaseService;
+import com.coffee.common.crud.controller.BaseQueryController;
 import com.coffee.common.exception.CustomException;
 import com.coffee.common.result.R;
 import com.coffee.common.result.ResultCode;
@@ -36,8 +41,8 @@ import java.util.stream.Collectors;
 @RestController
 @AllArgsConstructor
 @RequestMapping("/bus/patient")
-@Api(tags = "医院病人管理",description = "关于病人的相关操作")
-public class BusPatientController  {
+@Api(tags = "医院病人管理",description = "关于病人的相关操作,查询操作权限【bus:patient:query】")
+public class BusPatientController  implements BaseQueryController<BusPatientEntity,String> {
     private final LocalBusPatientService patientService;
 
     private final LocalBusDeviceRunningService deviceRunningService;
@@ -284,4 +289,18 @@ public class BusPatientController  {
     }
 
 
+    @Override
+    public BaseService<? extends Mapper<BusPatientEntity>, BusPatientEntity, String> getService() {
+        return patientService;
+    }
+
+    @Override
+    public String getPermissionPrefix() {
+        return "bus:patient:";
+    }
+
+    @Override
+    public StpLogic getStpLogin() {
+        return SaManager.getStpLogic("");
+    }
 }

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

@@ -47,5 +47,7 @@ public class PatientDeviceRepeatDomain {
     private Date infusionStartTime;
     @ApiModelProperty("是否为主泵")
     private Boolean master;
+    @ApiModelProperty("手术id")
+    private String clinicId;
 
 }

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

@@ -8,7 +8,6 @@ import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
-import org.springframework.util.Assert;
 
 import java.util.Comparator;
 import java.util.List;
@@ -41,7 +40,7 @@ public class DefaultHisNewStrategyHandler implements HisNewStrategyHandler {
         if(CollectionUtil.isEmpty(source)){
             return;
         }
-        log.info("\"拉取his只获取最新的一跳病人数据,对其进行处理\",拉取手术信息{},数据库数据{}", JSONUtil.toJsonStr(source),JSONUtil.toJsonStr(target));
+        log.info("拉取his只获取最新的一跳病人数据,对其进行处理,拉取手术信息{},数据库数据{}", JSONUtil.toJsonStr(source),JSONUtil.toJsonStr(target));
         //是否插入新的数据
         boolean insert=true;
         //是否为设置为当前的临床数据

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

@@ -1,11 +1,13 @@
 package com.coffee.bus.service;
 
+import cn.hutool.core.collection.CollectionUtil;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import cn.hutool.extra.spring.SpringUtil;
 import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
 import com.coffee.aliyun.utils.Constants;
 import com.coffee.aliyun.utils.Items;
 import com.coffee.aliyun.utils.PumpParams;
+import com.coffee.bus.entity.BusClinicEntity;
 import com.coffee.bus.entity.BusDeviceEntity;
 import com.coffee.bus.entity.BusInfusionHistoryEntity;
 import com.coffee.bus.entity.BusPatientEntity;
@@ -17,8 +19,11 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.Assert;
 
+import java.util.Comparator;
 import java.util.Date;
+import java.util.List;
 import java.util.Objects;
 
 /**
@@ -38,6 +43,10 @@ public class LocalBusInfusionHistoryService extends BaseService<BusInfusionHisto
     @Lazy
     private LocalBusPatientService patientService;
 
+    @Autowired
+    @Lazy
+    private LocalBusClinicService clinicService;
+
     @Override
     public void validateBeforeSave(BusInfusionHistoryEntity entity) {
 
@@ -141,12 +150,24 @@ public class LocalBusInfusionHistoryService extends BaseService<BusInfusionHisto
      */
     @Transactional(rollbackFor = Exception.class)
     public void adjustInfusionByClinic(String clinicId, String patientCode, String tenantId, Date startTime){
-        this.update(new UpdateWrapper<BusInfusionHistoryEntity>()
+        Assert.hasText(clinicId,"手术id不可为空");
+        Assert.hasText(patientCode,"病号不可为空");
+        Assert.hasText(tenantId,"医院id不可为空");
+        Assert.notNull(startTime,"手术开始时间不可为空");
+
+        List<BusInfusionHistoryEntity> infusionHistories = this.list(new QueryWrapper<BusInfusionHistoryEntity>()
                 .lambda()
-                .set(BusInfusionHistoryEntity::getClinicId,clinicId)
-                .eq(BusInfusionHistoryEntity::getTenantId,tenantId)
-                .eq(BusInfusionHistoryEntity::getPatientCode,patientCode)
-                .lt(BusInfusionHistoryEntity::getStartTime,startTime)
-                .gt(BusInfusionHistoryEntity::getClinicStartTime,startTime));
+                .eq(BusInfusionHistoryEntity::getTenantId, tenantId)
+                .eq(BusInfusionHistoryEntity::getPatientCode, patientCode)
+                .lt(BusInfusionHistoryEntity::getStartTime, startTime)
+                .gt(BusInfusionHistoryEntity::getClinicStartTime, startTime));
+        if(CollectionUtil.isNotEmpty(infusionHistories)){
+            infusionHistories.sort(Comparator.comparing(BusInfusionHistoryEntity::getStartTime));
+            infusionHistories.forEach(infusion->infusion.setClinicId(clinicId));
+            clinicService.update(new UpdateWrapper<BusClinicEntity>().lambda().eq(BusClinicEntity::getId,clinicId)
+            .set(BusClinicEntity::getMonitorStartTime,infusionHistories.get(0).getStartTime()));
+
+        }
+
     }
 }

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

@@ -56,6 +56,7 @@ public class LocalBusPatientService extends BaseService<BusPatientMapper, BusPat
     private PatientRegistry patientRegistry;
 
     @Autowired
+    @Lazy
     private WsPublishUtils wsPublishUtils;
 
     private ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
@@ -114,7 +115,7 @@ public class LocalBusPatientService extends BaseService<BusPatientMapper, BusPat
         List<PatientDeviceRepeatDomain> patientDeviceRepeats = this.baseMapper.selectRepeatDevice();
         Map<String, PatientDeviceRepeatResult> resultMap = new HashMap<>();
         patientDeviceRepeats.forEach(deviceRepeat->{
-            PatientDeviceRepeatResult repeatResult = resultMap.computeIfAbsent(deviceRepeat.getCode(),k->
+            PatientDeviceRepeatResult repeatResult = resultMap.computeIfAbsent(deviceRepeat.getCode()+deviceRepeat.getClinicId(),k->
                     PatientDeviceRepeatResult.of(
                             deviceRepeat.getName(),
                             deviceRepeat.getGender(),
@@ -123,6 +124,7 @@ public class LocalBusPatientService extends BaseService<BusPatientMapper, BusPat
                             deviceRepeat.getWard(),
                             deviceRepeat.getBedNo(),
                             deviceRepeat.getClinicName(),
+                            deviceRepeat.getClinicId(),
                             new ArrayList<>())
             );
             List<PatientDeviceRepeatResult.DeviceRunningSmallInfo> deviceRunningSmallInfos = Optional.ofNullable(repeatResult.getDevices()).orElse(new ArrayList<>());

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

@@ -5,6 +5,7 @@ import com.coffee.bus.enums.DeviceTypeEnum;
 import com.coffee.common.enums.SexEnum;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
+import io.swagger.annotations.ApiOperation;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 
@@ -42,6 +43,9 @@ public class PatientDeviceRepeatResult {
     @ApiModelProperty("手术名称")
     private String clinicName;
 
+    @ApiModelProperty("手术id")
+    private String clinicId;
+
     @ApiModelProperty("设备集合")
     private List<DeviceRunningSmallInfo> devices;
 

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

@@ -1,10 +1,18 @@
 package com.coffee.bus.utils;
 
 
+import cn.hutool.json.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.coffee.bus.entity.BusPatientEntity;
+import com.coffee.bus.enums.PatientAlarmEnum;
+import com.coffee.bus.service.LocalBusPatientService;
+import com.coffee.common.config.websocket.WebSocketConstant;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
+import org.springframework.util.Assert;
 
 /**
  * @author lifang
@@ -18,8 +26,82 @@ import org.springframework.stereotype.Component;
 @Slf4j
 public class WsPublishUtils {
     private final RedisTemplate redisTemplate;
-
+    private final LocalBusPatientService patientService;
+    @Async
     public void publish(String topic,Object msg){
         redisTemplate.convertAndSend(topic,msg);
     }
+
+    /**
+     * 描述: 推送设备输注消息
+     * @author lifang
+     * @date 2022/5/13 9:39
+     * @param patientCode
+     * @param tenantId
+     * @return void
+     */
+    @Async
+    public void publishPatientMonitor(String patientCode,String tenantId){
+        Assert.hasText(tenantId,"医院id不能为空");
+        this.publish(WebSocketConstant.getPatientMonitor(null, patientCode, tenantId).getTopic(),
+                patientService.lookMonitorByPatientCode(patientCode,tenantId));
+    }
+
+    /**
+     * 描述: 推送医院设备状态统计数量
+     * @author lifang
+     * @date 2022/5/13 9:39
+     * @param
+     * @return void
+     */
+    @Async
+    public void publishMonitorStateCount(String tenantId){
+        Assert.hasText(tenantId,"医院id不能为空");
+        this.publish(WebSocketConstant.getMonitorStateCount(tenantId).getTopic(),patientService.statusStats(tenantId));
+    }
+
+    /**
+     * 描述: 推送医院临床设备重复数量统计
+     * @author lifang
+     * @date 2022/5/13 9:39
+     * @param
+     * @return void
+     */
+    @Async
+    public void publishPatientAdd(String patientCode,String tenantId){
+        Assert.hasText(tenantId,"医院id不能为空");
+        this.publish(WebSocketConstant.getPatientAdd(tenantId).getTopic(),
+                new JSONObject().putOpt("patientCode",patientCode));
+    }
+    /**
+     * 描述: 推送医院临床设备重复数量统计
+     * @author lifang
+     * @date 2022/5/13 9:39
+     * @param
+     * @return void
+     */
+    @Async
+    public void publishDeviceRepeat(String tenantId){
+        Assert.hasText(tenantId,"医院id不能为空");
+        this.publish(WebSocketConstant.getDeviceRepeat(tenantId).getTopic(),
+                new JSONObject().putOpt("count",   patientService.count(new QueryWrapper<BusPatientEntity>().lambda().eq(BusPatientEntity::getTenantId,tenantId)
+                        .eq(BusPatientEntity::getAlarm, PatientAlarmEnum.DEVICE_REPEAT)))
+        );
+    }
+
+    /**
+     * 描述: 推送临床设备无绑定数量统计
+     * @author lifang
+     * @date 2022/5/13 9:39
+     * @param
+     * @return void
+     */
+    @Async
+    public void publishDeviceNone(String tenantId){
+        Assert.hasText(tenantId,"医院id不能为空");
+        this.publish(WebSocketConstant.getDeviceNone(tenantId).getTopic(),
+                new JSONObject().putOpt("count",  patientService.count(new QueryWrapper<BusPatientEntity>().lambda().eq(BusPatientEntity::getTenantId,tenantId)
+                        .eq(BusPatientEntity::getAlarm, PatientAlarmEnum.DEVICE_NONE)))
+        );
+    }
 }

coffee-system/src/main/java/com/coffee/bus/websocket/DeviceNoneHandler.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 DeviceRepeatHandler.java
+ * @Description 临床无绑定设备
+ * @createTime 2022年03月25日 14:21:00
+ */
+@Component
+public class DeviceNoneHandler extends Subscribe {
+
+    @Override
+    public String getId() {
+        return WebSocketConstant.DEVICE_NONE;
+    }
+
+    @Override
+    public void close(ChannelContext channelContext) {
+
+    }
+
+    @Override
+    public boolean needParam() {
+        return false;
+    }
+}

coffee-system/src/main/java/com/coffee/bus/websocket/DeviceRepeatHandler.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 DeviceRepeatHandler.java
+ * @Description 临床绑定重复设备
+ * @createTime 2022年03月25日 14:21:00
+ */
+@Component
+public class DeviceRepeatHandler extends Subscribe {
+
+    @Override
+    public String getId() {
+        return WebSocketConstant.DEVICE_REPEAT;
+    }
+
+    @Override
+    public void close(ChannelContext channelContext) {
+
+    }
+
+    @Override
+    public boolean needParam() {
+        return false;
+    }
+}

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

@@ -141,13 +141,8 @@ public class DeviceInfoListener {
 
             //异步推送
             cacheOperation.add(()->{
-                CompletableFuture.runAsync(()->{
-                    //推送设备上报输注消息
-                    wsPublishUtils.publish(WebSocketConstant.getPatientMonitor(null, device.getPatientCode(), device.getTenantId()).getTopic(),
-                            patientService.lookMonitorByPatientCode(device.getPatientCode(),device.getTenantId()));
-                    //推送设备状态
-                    wsPublishUtils.publish(WebSocketConstant.getMonitorStateCount(device.getTenantId()).getTopic(),patientService.statusStats(device.getTenantId()));
-                });
+                wsPublishUtils.publishPatientMonitor(device.getPatientCode(),device.getTenantId());
+                wsPublishUtils.publishMonitorStateCount(device.getTenantId());
                 return null;
             });
 
@@ -280,7 +275,9 @@ public class DeviceInfoListener {
                     //todo 发起无泵报警,处理原先泵的无泵信息
                     patientOperator.setBindDeviceId(null);
                     patientOperator.setAllDevice(new HashSet<>());
-                    //发起无泵报警后,将该病人最后一条输注作为显示信息实时传输给前端 //todo
+                    wsPublishUtils.publishDeviceNone(hospitalId);
+                    wsPublishUtils.publishPatientMonitor(patientCode,hospitalId);
+                    //发起无泵报警后,将该病人最后一条输注作为显示信息实时传输给前端
                     return null;
                 });
             }else {
@@ -305,13 +302,8 @@ public class DeviceInfoListener {
                         String newMasterId = master.get().getDeviceId();
                         patientOperator.setBindDeviceId(newMasterId);
                         patientOperator.setAllDevice(remainPatientBindDevices);
-                        //更换输注信息后，将最新的输注信息传输实时传输给前端 //todo
-
-                        CompletableFuture.runAsync(()->{
-                            wsPublishUtils.publish(WebSocketConstant.getPatientMonitor(null,patientCode, hospitalId).getTopic(),
-                                    patientService.lookMonitorByPatientCode(patientCode,hospitalId));
-                        });
-
+                        //更换输注信息后，将最新的输注信息传输实时传输给前端
+                        wsPublishUtils.publishPatientMonitor(patientCode,hospitalId);
                         return null;
                     });
                     if(remainPatientBindDevices.size()==1){
@@ -355,7 +347,12 @@ public class DeviceInfoListener {
             patientService.update(new UpdateWrapper<BusPatientEntity>().lambda()
                     .eq(BusPatientEntity::getCode,device.getPatientCode())
                     .eq(BusPatientEntity::getTenantId,device.getTenantId())
+                    .set(Boolean.TRUE.equals(device.getMaster()),BusPatientEntity::getInfusionId,device.getInfusionId())
                     .set(BusPatientEntity::getAlarm,PatientAlarmEnum.DEVICE_REPEAT));
+            suppliers.add(()->{
+                wsPublishUtils.publishDeviceRepeat(device.getPatientCode());
+                return null;
+            });
         }
     }
 

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

@@ -14,6 +14,7 @@
         <result column="device_id" property="deviceId"/>
         <result column="device_alias" property="deviceAlias"/>
         <result column="device_run_state" property="deviceRunState"/>
+        <result column="clinic_id" property="clinicId"/>
         <result column="device_alarm" property="deviceAlarm" />
         <result column="infusion_start_time" property="infusionStartTime"/>
         <result column="master" property="master"/>
@@ -99,6 +100,7 @@
          c.patient_age as age,
          c.ward,
          c.bed_no,
+         c.id as clinic_id,
          c.`name` as clinic_name,
          d.id as device_running_id,
          d.device_id as device_id,