add

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

@@ -0,0 +1,44 @@
+package com.coffee.common.config.websocket;
+
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.json.JSONUtil;
+import com.coffee.common.result.R;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.util.SerializationUtils;
+import org.tio.core.ChannelContext;
+import org.tio.core.Tio;
+import org.tio.websocket.common.WsResponse;
+
+import java.io.UnsupportedEncodingException;
+import java.util.Set;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName DefaultMessageListener.java
+ * @Description TODO
+ * @createTime 2022年03月25日 14:42:00
+ */
+@Data
+@Slf4j
+public class DefaultMessageListener implements MessageListener {
+    private final String id;
+    private final Set<ChannelContext> channelContexts;
+
+
+    @Override
+    public void onMessage(Message message, byte[] pattern) {
+        if (CollectionUtil.isNotEmpty(channelContexts)) {
+            channelContexts.parallelStream()
+                    .filter(channelContext -> !channelContext.isClosed)
+                    .forEach(channel -> Tio.send(channel,
+                            WsResponse.fromText(JSONUtil.toJsonStr(R.success(
+                                    MessageResponse.of(id,"result",new String(message.getBody()))))
+                                    ,"utf-8")));
+        }
+    }
+}

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

@@ -0,0 +1,134 @@
+package com.coffee.common.config.websocket.handler;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.coffee.common.Constants;
+import com.coffee.common.bo.LoginUser;
+import com.coffee.common.config.websocket.DefaultMessageListener;
+import com.coffee.common.config.websocket.MessagingRequest;
+import com.coffee.common.config.websocket.WebSocketConstant;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.util.ConcurrentReferenceHashMap;
+import org.tio.core.ChannelContext;
+
+import javax.annotation.Resource;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName Subscribe.java
+ * @Description TODO
+ * @createTime 2022年03月25日 14:18:00
+ */
+public abstract class Subscribe implements WsHandler {
+    @Resource
+    private RedisTemplate<String,Object> redisTemplate;
+
+    public static final String SUBSCRIBE_TOPIC="subscribe-topic";
+    /**
+     * 存储主题与ws通道关联
+     */
+    private Map<String,Set<ChannelContext>> subscribeTopics=new ConcurrentHashMap<>();
+
+
+    /**
+     * 存储主题与redis通道关联
+     */
+    private Map<String,RedisConnection> redisConnectionMap=new ConcurrentReferenceHashMap<>();
+
+
+    public String getTopic(String productName,String param,String tenantId){
+        return  WebSocketConstant.getTopic(this.getId(),productName, param, tenantId);
+
+    };
+
+
+    @Override
+    public void onMessage(MessagingRequest message, ChannelContext channelContext) {
+        LoginUser loginUser = (LoginUser) channelContext.get(Constants.LOGIN_USER_KEY);
+        if(loginUser==null){
+            channelContext.setClosed(true);
+            return;
+        }
+        //获取所有设备id
+        List<String> params = message.getParams();
+        if(CollectionUtil.isEmpty(params)){
+            return;
+        }
+        //需要处理的主题
+        List<String> subScribeTopic =
+                params.stream().map(deviceId -> getTopic(message.getProductName(), deviceId, loginUser.getTenantId()))
+                        .collect(Collectors.toList());
+        MessagingRequest.Type type = message.getType();
+        if(MessagingRequest.Type.sub==type){
+            //订阅主题
+            subScribeTopic.forEach(topic->this.subscribe(channelContext,topic));
+        }else {
+            //取消订阅主题
+            subScribeTopic.forEach(topic->this.unsubscribe(channelContext,topic));
+        }
+    }
+
+    /**
+     * ws 订阅主题
+     * @param channelContext
+     * @param topic
+     */
+    public void subscribe(ChannelContext channelContext, String topic){
+        //同一主题只订阅一次
+        Set<ChannelContext> channelContexts = Optional.ofNullable(subscribeTopics.get(topic)).orElse(new HashSet<>());
+        if(!subscribeTopics.containsKey(topic)){
+            channelContexts.add(channelContext);
+            redisTemplate.execute(new RedisCallback<Object>() {
+                @Override
+                public Object doInRedis(RedisConnection connection) throws DataAccessException {
+                    connection.pSubscribe(new DefaultMessageListener(getId(),channelContexts),topic.getBytes());
+                    redisConnectionMap.put(topic,connection);
+                    return null;
+                }
+            });
+        }
+        subscribeTopics.put(topic,channelContexts);
+        //将主题与ws通道绑定
+        Object result = Optional.ofNullable(channelContext.get(SUBSCRIBE_TOPIC)).orElse(new HashSet<>());
+        Set<String> subscribeTopicSet= (Set<String>) result;
+        subscribeTopicSet.add(topic);
+        channelContext.set(SUBSCRIBE_TOPIC,subscribeTopicSet);
+    };
+
+    /**
+     * ws取消订阅主题
+     * @param channelContext
+     * @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);
+                redisTemplate.execute(new RedisCallback<Object>() {
+                    @Override
+                    public Object doInRedis(RedisConnection connection) throws DataAccessException {
+                        RedisConnection redisConnection = redisConnectionMap.get(topic);
+                        if (redisConnection!=null) {
+                            redisConnection.getSubscription().unsubscribe(topic.getBytes());
+                        }
+                        return null;
+                    }
+                });
+                redisConnectionMap.remove(topic);
+            }
+        }
+    };
+
+
+}

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

@@ -8,7 +8,6 @@ import com.coffee.bus.controller.vo.NoPumpConfig;
 import com.coffee.bus.controller.vo.UndoConfig;
 import com.coffee.bus.entity.BusDeviceRunningEntity;
 import com.coffee.bus.enums.DeviceStatusEnum;
-import com.coffee.bus.enums.NetPumpWarnEnum;
 import com.coffee.bus.service.LocalBusDeviceRunningService;
 import com.coffee.common.crud.BaseService;
 import com.coffee.common.crud.controller.BaseQueryController;

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

@@ -3,7 +3,7 @@ package com.coffee.bus.entity;
 import com.baomidou.mybatisplus.annotation.TableField;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.coffee.bus.enums.DeviceStatusEnum;
-import com.coffee.bus.enums.NetPumpWarnEnum;
+import com.coffee.bus.enums.DeviceWarnEnum;
 import com.coffee.common.config.mybatis.DateToBigIntHandler;
 import com.coffee.common.entity.TenantGenericEntity;
 import com.coffee.common.enums.SexEnum;
@@ -98,7 +98,7 @@ public class BusDeviceHistoryEntity extends TenantGenericEntity<String,String> {
 
     @ApiModelProperty(value = "提醒信息")
     @TableField(typeHandler = EnumOrdinalTypeHandler.class)
-    private NetPumpWarnEnum warn;
+    private DeviceWarnEnum warn;
 
     @ApiModelProperty(value = "开始时间")
     @TableField(typeHandler = DateToBigIntHandler.class)

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

@@ -0,0 +1,55 @@
+package com.coffee.bus.enums;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+import io.swagger.annotations.ApiModelProperty;
+import io.swagger.annotations.ApiOperation;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName AlarmEnum.java
+ * @Description 泵状态类型
+ * @createTime 2022年03月27日 09:49:00
+ */
+@AllArgsConstructor
+@JsonFormat(shape = JsonFormat.Shape.OBJECT)
+public enum DeviceStatusEnum {
+    /**
+     * 请在添加或修改的过程中不要改变枚举的顺序！！！！！！！！！！！！！！！
+     * 如若新增,请在最后尾部新增
+     * 切记不可删除！！！！！！！！！！！！
+     */
+    Running(0,"正在运行",false),
+    Waiting(1,"待机中",false),
+    Shutdown(2,"关机",false),
+
+    //    以下为报警信息状态
+    Finished(3,"输液结束",true),
+    LowBattery(4,"低电量报警",true),
+    Jam(5,"堵塞报警",true),
+    Bubble(6,"气泡报警",true),
+    NoSignal(7,"不在服务区",true),
+    Machine(8,"机械故障",true),
+    NotFastened(9,"未扣合股涨",true);
+
+
+    /**
+     * 与枚举ordinal保持一致
+     */
+    @Getter
+    @ApiModelProperty("运行状态编码")
+    private Integer code;
+    @Getter
+    @ApiModelProperty("运行状态内容")
+    private String text;
+    /**
+     * 是否为报警状态
+     */
+    @ApiModelProperty("是否为报警状态")
+    @Getter
+    private Boolean alarm;
+
+}

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

@@ -0,0 +1,40 @@
+package com.coffee.bus.enums;
+
+import com.alibaba.fastjson.annotation.JSONType;
+import com.fasterxml.jackson.annotation.*;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName AlarmEnum.java
+ * @Description 泵报警类型
+ * @createTime 2022年03月27日 09:49:00
+ */
+@AllArgsConstructor
+@JsonFormat(shape = JsonFormat.Shape.OBJECT)
+public enum DeviceWarnEnum {
+    /**
+     * 请在添加或修改的过程中不要改变枚举的顺序！！！！！！！！！！！！！！！
+     * 如若新增,请在最后尾部新增
+     * 切记不可删除！！！！！！！！！！！！
+     */
+    UpLimit(0,"加档受限"),
+    ComingEnd(1,"即将输液完毕"),
+    LowInfusion(2,"低输注状态"),
+    LowBattery(3,"电量不足"),
+    FlowRateLimit(4,"流速已达上限"),
+    InsufficientAnalgesia(5,"镇痛不足");
+
+
+    /**
+     * 与枚举ordinal保持一致
+     */
+    @Getter
+    private Integer code;
+    @Getter
+    private String text;
+
+}