fix websocket 订阅发布模式

add 泵状态枚举

coffee-admin/src/main/java/com/coffee/admin/AdminApplication.java

@@ -5,6 +5,7 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
 import org.springframework.context.annotation.Import;
+import org.springframework.scheduling.annotation.EnableScheduling;
 import org.tio.websocket.starter.EnableTioWebSocketServer;
 import springfox.documentation.swagger2.annotations.EnableSwagger2;
 
@@ -17,6 +18,7 @@ import springfox.documentation.swagger2.annotations.EnableSwagger2;
 @Import(cn.hutool.extra.spring.SpringUtil.class)
 @EnableSwagger2
 @EnableTioWebSocketServer
+@EnableScheduling
 public class AdminApplication {
 
     public static void main(String[] args) {

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

@@ -35,7 +35,7 @@ public class DefaultMessageListener implements MessageListener {
 //                    .filter(channelContext -> !channelContext.isClosed)
                     .forEach(channel -> Tio.send(channel,
                             WsResponse.fromText(JSONUtil.toJsonStr(R.success(
-                                    JSONUtil.parseArray(new String(message.getBody())).get(1)))
+                                    new String(message.getBody())))
                                     ,"utf-8")));
         }
     }

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

@@ -56,13 +56,11 @@ public abstract class Subscribe implements WsHandler {
             redisTemplate.execute(new RedisCallback<Object>() {
                 @Override
                 public Object doInRedis(RedisConnection connection) throws DataAccessException {
-                    connection.subscribe(new DefaultMessageListener(channelContexts),topic.getBytes());
+                    connection.pSubscribe(new DefaultMessageListener(channelContexts),topic.getBytes());
                     redisConnectionMap.put(topic,connection);
                     return null;
                 }
             });
-            new Thread(()->singleThreadEventLoop.scheduleAtFixedRate(()->redisTemplate.convertAndSend(topic, MessageResponse.of(getId(),"result","主题为["+topic+"]的回复消息")),1,3, TimeUnit.SECONDS))
-                    .start();
         }
         channelContexts.add(channelContext);
         subscribeTopics.put(topic,channelContexts);

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

@@ -2,6 +2,9 @@ package com.coffee.common.crud;
 
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.util.ObjectUtil;
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.db.sql.Direction;
+import cn.hutool.db.sql.Order;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.core.enums.SqlMethod;
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
@@ -199,6 +202,10 @@ public abstract class BaseService<M extends BaseMapper<E>, E,PK extends Serializ
     public Page<E> list(QueryParamEntity<E> param) {
         //是否为分页查询
         if(ObjectUtil.isNotNull(param.getPage())){
+            param.getPage().setSearchCount(false);
+            QueryWrapper<E> wrapper = build(param);
+            Page<E> page = this.page(param.getPage(),wrapper);
+            page.setTotal(this.count(wrapper));
             return this.page(param.getPage(),build(param));
         }else {
             try {
@@ -225,12 +232,28 @@ public abstract class BaseService<M extends BaseMapper<E>, E,PK extends Serializ
         QueryWrapper<E> queryWrapper = new QueryWrapper<>();
         Set<String> includes = param.getIncludes();
         Set<Term> wheres = param.getWheres();
+        Set<Order> orders = param.getOrders();
         if(CollectionUtil.isNotEmpty(includes)){
             queryWrapper.select(includes.toArray(new String[includes.size()+1]));
         }
         if (CollectionUtil.isNotEmpty(wheres)) {
             wheres.forEach(term -> term.getType().build(queryWrapper,term));
         }
+        if(CollectionUtil.isNotEmpty(orders)){
+            orders.forEach(order -> {
+                String field = order.getField();
+                String column = StrUtil.toUnderlineCase(field).toLowerCase();
+                Direction direction = order.getDirection();
+                if(Direction.ASC==direction){
+                    queryWrapper.orderByAsc(column);
+                }else {
+                    queryWrapper.orderByDesc(column);
+                }
+            });
+        }else {
+            queryWrapper.orderByDesc("id");
+        }
+
         return queryWrapper;
     }
 
@@ -271,7 +294,7 @@ public abstract class BaseService<M extends BaseMapper<E>, E,PK extends Serializ
      * @Param [entity]
      * @return void
      **/
-   public abstract void validateBeforeUpdate(E entity) ;
+    public abstract void validateBeforeUpdate(E entity) ;
 
 
 

coffee-common/src/main/java/com/coffee/common/entity/QueryParamEntity.java

@@ -2,11 +2,14 @@ package com.coffee.common.entity;
 
 import cn.hutool.core.collection.CollectionUtil;
 import cn.hutool.core.util.ObjectUtil;
+import cn.hutool.db.sql.Order;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.coffee.common.entity.param.Term;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.Data;
+import org.springframework.data.domain.Sort;
+
 import java.util.Set;
 
 /**
@@ -31,4 +34,7 @@ public class QueryParamEntity<T> {
     @Schema(description = "分页查询")
     private Page<T> page;
 
+    @Schema(description = "排序规则,field传入字段名，Direction 传输ASC、DESC")
+    private Set<Order> orders;
+
 }

coffee-common/src/main/java/com/coffee/common/redis/RedisConfig.java

@@ -1,5 +1,6 @@
 package com.coffee.common.redis;
 
+import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import com.fasterxml.jackson.annotation.PropertyAccessor;
@@ -32,14 +33,16 @@ public class RedisConfig extends CachingConfigurerSupport {
         // 使用Jackson2JsonRedisSerialize 替换默认序列化
         Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
 
+
         ObjectMapper objectMapper = new ObjectMapper();
         objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
         objectMapper.activateDefaultTyping(LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.WRAPPER_ARRAY);
 
         jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
 
+        FastJsonRedisSerializer<Object> fastJsonRedisSerializer = new FastJsonRedisSerializer<>(Object.class);
         // 设置value的序列化规则和 key的序列化规则
-        redisTemplate.setValueSerializer(jackson2JsonRedisSerializer);
+        redisTemplate.setValueSerializer(fastJsonRedisSerializer);
         redisTemplate.setKeySerializer(new StringRedisSerializer());
 
         redisTemplate.setHashKeySerializer(jackson2JsonRedisSerializer);

coffee-framework/src/main/java/com/coffee/framework/config/mybatisplus/MybatisPlusConfig.java

@@ -59,7 +59,7 @@ public class MybatisPlusConfig {
             ignoreTableName =
                     tableClass
                             .stream()
-                            .filter(aClass -> aClass.getSuperclass() == TenantGenericEntity.class)
+                            .filter(aClass -> aClass.getSuperclass() != TenantGenericEntity.class)
                             .map(aClass -> {
                                 TableName tableName = aClass.getAnnotation(TableName.class);
                                 return tableName.value();
@@ -145,13 +145,13 @@ public class MybatisPlusConfig {
                 if(url.toString().endsWith("/login")){
                     return true;
                 }
-                return true;
+//                return true;
                 //判断当前有用户是否为系统级用户,若是,则忽略逻辑隔离
-//                LoginUser loginUser = (LoginUser) StpUtil.getTokenSession().get(Constants.LOGIN_USER_KEY);
-//                if(1==loginUser.getIsSys()){
-//                    return true;
-//                }
-//                return CollectionUtil.isEmpty(ignoreTableName)||ignoreTableName.contains(tableName);
+                LoginUser loginUser = (LoginUser) StpUtil.getTokenSession().get(Constants.LOGIN_USER_KEY);
+                if(1==loginUser.getIsSys()){
+                    return true;
+                }
+                return CollectionUtil.isEmpty(ignoreTableName)||ignoreTableName.contains(tableName);
             }
 
             @Override

coffee-framework/src/main/java/com/coffee/framework/config/mybatisplus/TenantIdManager.java

@@ -28,9 +28,9 @@ public class TenantIdManager {
         if(url.toString().endsWith("/login")){
             return "";
         }
-        return "";
-//        LoginUser loginUser = (LoginUser) StpUtil.getTokenSession().get(Constants.LOGIN_USER_KEY);
-//        return Optional.ofNullable(loginUser.getSysUser().getTenantId()).orElse("");
+//        return "";
+        LoginUser loginUser = (LoginUser) StpUtil.getTokenSession().get(Constants.LOGIN_USER_KEY);
+        return Optional.ofNullable(loginUser.getSysUser().getTenantId()).orElse("");
     }
 
 }

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

@@ -27,9 +27,9 @@ import java.util.*;
 @ApiModel(value="评价模板", description="评价模板")
 public class BusEvaluationTmpEntity extends TenantGenericEntity<String,String> {
 
-    @ApiModelProperty(value = "不需要评价的字段,如calm、leftArm等")
+    @ApiModelProperty(value = "需要评价的字段,如calm、leftArm等")
     @TableField(typeHandler = FastjsonTypeHandler.class)
-    private List<String> excludeConfig;
+    private List<String> includeConfig;
 
 
     @TableField(fill = FieldFill.INSERT)

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

@@ -1,27 +1,18 @@
 package com.coffee.bus.entity;
 
-import com.alibaba.fastjson.annotation.JSONField;
 import com.baomidou.mybatisplus.annotation.TableField;
-import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.coffee.bus.enums.NetPumpAlarmEnum;
 import com.coffee.bus.enums.NetPumpStatusEnum;
 import com.coffee.common.config.mybatis.DateToBigIntHandler;
 import com.coffee.common.entity.TenantGenericEntity;
-import com.fasterxml.jackson.annotation.JsonEnumDefaultValue;
-import com.fasterxml.jackson.annotation.JsonSetter;
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
-import com.fasterxml.jackson.databind.ser.std.EnumSerializer;
 import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.experimental.Accessors;
 import org.apache.ibatis.type.EnumOrdinalTypeHandler;
-
-import java.io.Serializable;
 import java.math.BigDecimal;
-import java.time.LocalDateTime;
 import java.util.Date;
 
 /**
@@ -139,9 +130,6 @@ public class BusNetPumpEntity extends TenantGenericEntity<String,String> {
     @ApiModelProperty(value = "是否已撤泵")
     private Boolean isRemove;
 
-    @ApiModelProperty(value = "状态标记")
-    private Integer stateFlag;
-
     @ApiModelProperty(value = "显示状态")
     private String viewState;
 

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

@@ -14,6 +14,7 @@ import java.time.Clock;
  * @version 1.0.0
  * @ClassName DeviceInfoEvent.java
  * @Description 所有通过第三方传过来的数据，请先解析成netpump格式类型，然后分发到系统中
+ *
  * @see  BusNetPumpEntity
  * @createTime 2022年03月21日 16:21:00
  */

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

@@ -2,7 +2,6 @@ package com.coffee.bus.websocket;
 
 import com.coffee.common.config.websocket.MessagingRequest;
 import com.coffee.common.config.websocket.handler.Subscribe;
-import com.coffee.common.config.websocket.handler.WsHandler;
 import org.springframework.stereotype.Component;
 import org.tio.core.ChannelContext;
 

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

@@ -25,7 +25,11 @@ public class WebSocketConstant {
      */
     public static String getTopic(String id,String productName,String param,String tenantId){
         productName=StrUtil.isEmptyIfStr(productName)?"default":productName;
-//        tenantId=StrUtil.isNullOrUndefined(tenantId)?"*":tenantId;
-        return id+":"+productName+":"+param+":"+tenantId;
+        tenantId=StrUtil.isNullOrUndefined(tenantId)?"*":tenantId;
+        return id+"-"+productName+"-"+param+"-"+tenantId;
+    }
+
+    public static String getDeviceInfoDetailTopic(String productName,String param,String tenantId){
+        return getTopic(DEVICE_INFO_DETAIL,productName,param,tenantId);
     }
 }

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

@@ -0,0 +1,59 @@
+package com.coffee.bus.websocket.listener;
+
+import cn.hutool.core.util.RandomUtil;
+import cn.hutool.extra.spring.SpringUtil;
+import com.coffee.bus.entity.BusNetPumpEntity;
+import com.coffee.bus.listener.event.bean.DeviceInfoEvent;
+import com.coffee.bus.service.LocalBusNetPumpService;
+import com.coffee.bus.websocket.WebSocketConstant;
+import lombok.AllArgsConstructor;
+import org.springframework.context.event.EventListener;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.math.BigDecimal;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * @author lifang
+ * @version 1.0.0
+ * @ClassName DeviceInfoListener.java
+ * @Description TODO
+ * @createTime 2022年03月27日 11:21:00
+ */
+@Component
+@AllArgsConstructor
+public class DeviceInfoListener {
+
+    private final RedisTemplate redisTemplate;
+
+    private final LocalBusNetPumpService netPumpService;
+    /**
+     * 监听上传的数据信息，
+     * 若设备详情发生变化，则及时通知相应的ws通道
+     * @param infoEvent
+     */
+    @EventListener
+    public void deviceInfoDetail(DeviceInfoEvent infoEvent){
+        BusNetPumpEntity pump = infoEvent.getContent();
+        String topic = WebSocketConstant.getDeviceInfoDetailTopic(null, pump.getId(), pump.getTenantId());
+        redisTemplate.convertAndSend(topic, pump);
+    }
+
+
+    @Scheduled(cron = "0/3 * * * * ?")
+    public void send(){
+        List<BusNetPumpEntity> list = netPumpService.list();
+        list.forEach(pump->{
+            pump.setRemainQuantity(BigDecimal.valueOf(RandomUtil.randomInt(100)));
+            pump.setContinueQuantity(BigDecimal.valueOf(RandomUtil.randomInt(100)));
+            pump.setSelfControlLockTime(BigDecimal.valueOf(RandomUtil.randomInt(100)));
+            pump.setValidTime(RandomUtil.randomInt(100));
+            pump.setInvalidTime(RandomUtil.randomInt(100));
+            pump.setTotalCount(RandomUtil.randomInt(100));
+            SpringUtil.publishEvent(new DeviceInfoEvent(this,pump,pump.getId()));
+        });
+    }
+}

coffee-system/src/main/java/com/coffee/system/common/dto/SysUserAddDTO.java

@@ -137,7 +137,6 @@ public class SysUserAddDTO implements Serializable {
      */
     private List<Long> roleIds;
 
-    @NotNull(message = "用户标识不能为空")
     private Integer isSys;
 
     private Long tenantId;

coffee-system/src/main/java/com/coffee/system/common/dto/SysUserEditDTO.java

@@ -123,7 +123,6 @@ public class SysUserEditDTO implements Serializable {
     /**
      * 状态 0否；1是
      */
-    @NotNull(message = "是否为系统级别用户")
     private Integer isSys;
 
     /**

coffee-system/src/main/java/com/coffee/system/service/impl/SysUserServiceImpl.java

@@ -125,6 +125,9 @@ public class SysUserServiceImpl extends ServiceImpl<SysUserMapper, SysUser> impl
         if (!this.checkUniqueStaffNumber(req.getStaffNumber(), null)) {
             throw new CustomException("工号已存在");
         }
+        if(req.getIsSys()==null){
+            req.setIsSys(0);
+        }
         SysUser entity = BeanUtil.copyProperties(req, SysUser.class);
         // 密码加密
         entity.setPassword(SecurityUtil.encryptPassword(entity.getPassword()));
@@ -173,15 +176,16 @@ public class SysUserServiceImpl extends ServiceImpl<SysUserMapper, SysUser> impl
         if (!this.checkUniqueStaffNumber(req.getStaffNumber(), req.getId())) {
             throw new CustomException("工号已存在");
         }
-        if(Boolean.TRUE.equals(req.getIsSys())){
+        if(req.getIsSys()!=null&&1==req.getIsSys()){
             try {
-                LoginUser loginUser = (LoginUser) StpUtil.getSession().get(Constants.LOGIN_USER_KEY);
-                if(!Boolean.TRUE.equals(loginUser.getIsSys())){
+                LoginUser loginUser = (LoginUser) StpUtil.getTokenSession().get(Constants.LOGIN_USER_KEY);
+                if(req.getIsSys()==null||1!=loginUser.getIsSys()){
                     throw new CustomException("当前用户没有操作系统级别用户的权限");
                 }
             }catch (Exception e){
-                StpUtil.logout();
-                throw NotLoginException.newInstance("", NotLoginException.DEFAULT_MESSAGE);
+//                StpUtil.logout();
+                throw new CustomException("当前用户没有操作系统级别用户的权限");
+//                throw NotLoginException.newInstance("", NotLoginException.DEFAULT_MESSAGE);
             }
         }
         // 当前用户不是超级管理员，不允许修改