事件推送功能

iot-access/access-boot/src/main/java/com/njcn/access/runner/AccessApplicationRunner.java

@@ -1,47 +1,47 @@
-package com.njcn.access.runner;
-
-import com.njcn.access.service.ICsEquipmentDeliveryService;
-import com.njcn.access.service.ICsTopicService;
-import com.njcn.access.service.impl.CsDeviceServiceImpl;
-import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.boot.ApplicationArguments;
-import org.springframework.boot.ApplicationRunner;
-import org.springframework.stereotype.Component;
-
-import javax.annotation.Resource;
-import java.util.List;
-import java.util.Objects;
-
-/**
- * 类的介绍：用来重新发起设备的接入，存在程序意外停止了，缓存失效导致无法更新装置的状态，所以需要在程序启动时发起设备的接入
- *
- * @author xuyang
- * @version 1.0.0
- * @createTime 2023/8/28 13:57
- */
-@Component
-@Slf4j
-public class AccessApplicationRunner implements ApplicationRunner {
-
-    @Resource
-    private CsDeviceServiceImpl csDeviceService;
-
-    @Resource
-    private ICsTopicService csTopicService;
-
-    @Resource
-    private ICsEquipmentDeliveryService csEquipmentDeliveryService;
-
-    @Override
-    public void run(ApplicationArguments args){
-        List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getAll();
-        list.forEach(item->{
-            String version = csTopicService.getVersion(item.getNdid());
-            if (!Objects.isNull(version)){
-                csDeviceService.devAccess(item.getNdid(),version);
-            }
-        });
-    }
-
-}
+//package com.njcn.access.runner;
+//
+//import com.njcn.access.service.ICsEquipmentDeliveryService;
+//import com.njcn.access.service.ICsTopicService;
+//import com.njcn.access.service.impl.CsDeviceServiceImpl;
+//import com.njcn.csdevice.pojo.po.CsEquipmentDeliveryPO;
+//import lombok.extern.slf4j.Slf4j;
+//import org.springframework.boot.ApplicationArguments;
+//import org.springframework.boot.ApplicationRunner;
+//import org.springframework.stereotype.Component;
+//
+//import javax.annotation.Resource;
+//import java.util.List;
+//import java.util.Objects;
+//
+///**
+// * 类的介绍：用来重新发起设备的接入，存在程序意外停止了，缓存失效导致无法更新装置的状态，所以需要在程序启动时发起设备的接入
+// *
+// * @author xuyang
+// * @version 1.0.0
+// * @createTime 2023/8/28 13:57
+// */
+//@Component
+//@Slf4j
+//public class AccessApplicationRunner implements ApplicationRunner {
+//
+//    @Resource
+//    private CsDeviceServiceImpl csDeviceService;
+//
+//    @Resource
+//    private ICsTopicService csTopicService;
+//
+//    @Resource
+//    private ICsEquipmentDeliveryService csEquipmentDeliveryService;
+//
+//    @Override
+//    public void run(ApplicationArguments args){
+//        List<CsEquipmentDeliveryPO> list = csEquipmentDeliveryService.getAll();
+//        list.forEach(item->{
+//            String version = csTopicService.getVersion(item.getNdid());
+//            if (!Objects.isNull(version)){
+//                csDeviceService.devAccess(item.getNdid(),version);
+//            }
+//        });
+//    }
+//
+//}

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/EventServiceImpl.java

@@ -7,7 +7,6 @@ import com.njcn.common.pojo.exception.BusinessException;
 import com.njcn.csdevice.api.CsDeviceUserFeignClient;
 import com.njcn.csdevice.api.CsLineFeignClient;
 import com.njcn.csdevice.api.EquipmentFeignClient;
-import com.njcn.csdevice.pojo.po.CsDeviceUserPO;
 import com.njcn.csdevice.pojo.po.CsLinePO;
 import com.njcn.csharmonic.pojo.po.CsEventPO;
 import com.njcn.csharmonic.pojo.po.CsEventUserPO;
@@ -22,6 +21,8 @@ import com.njcn.system.api.EpdFeignClient;
 import com.njcn.system.enums.DicDataEnum;
 import com.njcn.system.pojo.dto.EpdDTO;
 import com.njcn.system.pojo.po.DictData;
+import com.njcn.user.api.AppInfoSetFeignClient;
+import com.njcn.user.pojo.po.app.AppInfoSet;
 import com.njcn.zlevent.service.ICsEventService;
 import com.njcn.zlevent.service.ICsEventUserService;
 import com.njcn.zlevent.service.IEventService;
@@ -32,12 +33,14 @@ import org.influxdb.dto.BatchPoints;
 import org.influxdb.dto.Point;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.CollectionUtils;
 
 import java.text.SimpleDateFormat;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 /**
  * 类的介绍：
@@ -69,6 +72,8 @@ public class EventServiceImpl implements IEventService {
 
     private final CsDeviceUserFeignClient csDeviceUserFeignClient;
 
+    private final AppInfoSetFeignClient appInfoSetFeignClient;
+
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void analysis(AppEventMessage appEventMessage) {
@@ -139,6 +144,12 @@ public class EventServiceImpl implements IEventService {
         if (CollectionUtil.isNotEmpty(records)) {
             influxDbUtils.batchInsert(influxDbUtils.getDbName(), "", InfluxDB.ConsistencyLevel.ALL, TimeUnit.MILLISECONDS, records);
         }
+        //todo 通知用户有新的事件发生
+        List<String> eventUser = getEventUser(deviceId);
+
+
+
+
     }
 
     /**
@@ -210,4 +221,20 @@ public class EventServiceImpl implements IEventService {
         return result;
     }
 
+    /**
+     * 获取需要通知暂态事件的用户
+     */
+    public List<String> getEventUser(String devId) {
+        List<String> result = new ArrayList<>();
+        //获取设备下主用户和子用户集合
+        List<String> list = csDeviceUserFeignClient.findUserById(devId).getData();
+        //查询哪些用户打开了事件提示
+        if (CollectionUtil.isNotEmpty(list)){
+            List<AppInfoSet> appInfoSet = appInfoSetFeignClient.getListById(list).getData();
+            result = appInfoSet.stream()
+                    .filter(person -> person.getEventInfo() == 1)
+                    .map(AppInfoSet::getUserId).collect(Collectors.toList());
+        }
+        return result;
+    }
 }