添加企业污染因子报警功能

src/main/java/com/cm/tenlion/pollutantdata/PollutantDataApplication.java

@@ -5,6 +5,7 @@ import com.cm.tenlion.pollutantdata.service.alarmlog.IAlarmLogService;
 import com.cm.tenlion.pollutantdata.service.collector.ICollectorService;
 import com.cm.tenlion.pollutantdata.service.datareal.IDataRealService;
 import com.cm.tenlion.pollutantdata.service.enterprise.IEnterpriseService;
+import com.cm.tenlion.pollutantdata.service.enterprisepoll.IEnterprisePollService;
 import com.cm.tenlion.pollutantdata.service.instrument.IInstrumentService;
 import com.cm.tenlion.pollutantdata.service.poll.IPollService;
 import com.cm.tenlion.pollutantdata.utils.net.TCPServer;
@@ -27,6 +28,8 @@ public class PollutantDataApplication {
     @Autowired
     private IEnterpriseService enterpriseService;
     @Autowired
+    private IEnterprisePollService enterprisePollService;
+    @Autowired
     private IInstrumentService instrumentService;
     @Autowired
     private ICollectorService collectorService;
@@ -67,6 +70,7 @@ public class PollutantDataApplication {
                 try {
                     TCPServer tcpServer = new TCPServer(nettyPort);
                     tcpServer.setEnterpriseService(enterpriseService);
+                    tcpServer.setEnterprisePollService(enterprisePollService);
                     tcpServer.setInstrumentService(instrumentService);
                     tcpServer.setCollectorService(collectorService);
                     tcpServer.setDataMinuteService(dataMinuteService);

src/main/java/com/cm/tenlion/pollutantdata/pojo/pos/enterprisepoll/EnterprisePollPO.java

@@ -13,7 +13,9 @@ public class EnterprisePollPO {
     private String enterprisePollId;
     private String enterpriseId;
     private String pollId;
-    private String enterprisePollBeyond;
+    private String pollNo;
+    private String pollNoOld;
+    private Double enterprisePollBeyond;
     private String gmtCreate;
     private String creator;
     private String gmtModified;
@@ -44,11 +46,27 @@ public class EnterprisePollPO {
         this.pollId = pollId;
     }
 
-    public String getEnterprisePollBeyond() {
+    public String getPollNo() {
-        return enterprisePollBeyond == null ? "" : enterprisePollBeyond.trim();
+        return pollNo == null ? "" : pollNo.trim();
     }
 
-    public void setEnterprisePollBeyond(String enterprisePollBeyond) {
+    public void setPollNo(String pollNo) {
+        this.pollNo = pollNo;
+    }
+
+    public String getPollNoOld() {
+        return pollNoOld == null ? "" : pollNoOld.trim();
+    }
+
+    public void setPollNoOld(String pollNoOld) {
+        this.pollNoOld = pollNoOld;
+    }
+
+    public Double getEnterprisePollBeyond() {
+        return enterprisePollBeyond == null ? 0 : enterprisePollBeyond;
+    }
+
+    public void setEnterprisePollBeyond(Double enterprisePollBeyond) {
         this.enterprisePollBeyond = enterprisePollBeyond;
     }
 

src/main/java/com/cm/tenlion/pollutantdata/service/enterprisepoll/IEnterprisePollService.java

@@ -21,6 +21,7 @@ import java.util.Map;
 public interface IEnterprisePollService {
     /**
      * 根据企业ID和污染因子编码获取监控污染因子
+     *
      * @param enterpriseId
      * @param pollNo
      * @return
@@ -29,6 +30,7 @@ public interface IEnterprisePollService {
 
     /**
      * 根据企业ID获取监控污染因子
+     *
      * @param enterpriseId
      * @return
      */
@@ -160,6 +162,15 @@ public interface IEnterprisePollService {
      */
     EnterprisePollPO getPO(String enterprisePollId);
 
+    /**
+     * 企业监控污染因子详情
+     *
+     * @param enterpriseId
+     * @param pollNo 污染因子编码
+     * @return
+     */
+    EnterprisePollPO getPO(String enterpriseId, String pollNo);
+
     /**
      * 企业监控污染因子列表
      *
@@ -184,6 +195,14 @@ public interface IEnterprisePollService {
      */
     List<EnterprisePollPO> listPO(Map<String, Object> params);
 
+    /**
+     * 企业污染因子列表
+     *
+     * @param enterpriseId
+     * @return
+     */
+    List<EnterprisePollPO> listPO(String enterpriseId);
+
     /**
      * 企业监控污染因子分页列表
      *

src/main/java/com/cm/tenlion/pollutantdata/service/enterprisepoll/impl/EnterprisePollServiceImpl.java

@@ -81,7 +81,6 @@ public class EnterprisePollServiceImpl extends AbstractService implements IEnter
     }
 
 
-
     @Override
     public void remove(List<String> ids) {
         remove(null, ids);
@@ -159,6 +158,17 @@ public class EnterprisePollServiceImpl extends AbstractService implements IEnter
         return getPO(params);
     }
 
+    @Override
+    public EnterprisePollPO getPO(String enterpriseId, String pollNo) {
+        if (StringUtils.isBlank(enterpriseId) || StringUtils.isBlank(pollNo)) {
+            return null;
+        }
+        Map<String, Object> params = getHashMap(4);
+        params.put("enterpriseId", enterpriseId);
+        params.put("pollNo", pollNo);
+        return getPO(params);
+    }
+
     @Override
     public List<EnterprisePollDTO> list(Map<String, Object> params) {
         return enterprisePollDao.list(params);
@@ -174,6 +184,16 @@ public class EnterprisePollServiceImpl extends AbstractService implements IEnter
         return enterprisePollDao.listPO(params);
     }
 
+    @Override
+    public List<EnterprisePollPO> listPO(String enterpriseId) {
+        if (StringUtils.isBlank(enterpriseId)) {
+            return new ArrayList<>();
+        }
+        Map<String, Object> params = getHashMap(2);
+        params.put("enterpriseId", enterpriseId);
+        return listPO(params);
+    }
+
     @Override
     public SuccessResultList<List<EnterprisePollDTO>> listPage(ListPage page) {
         PageHelper.startPage(page.getPage(), page.getRows());
@@ -210,5 +230,4 @@ public class EnterprisePollServiceImpl extends AbstractService implements IEnter
     }
 
 
-
 }

src/main/java/com/cm/tenlion/pollutantdata/utils/net/TCPHandler.java

@@ -9,6 +9,7 @@ import com.cm.tenlion.pollutantdata.service.alarmlog.IAlarmLogService;
 import com.cm.tenlion.pollutantdata.service.collector.ICollectorService;
 import com.cm.tenlion.pollutantdata.service.datareal.IDataRealService;
 import com.cm.tenlion.pollutantdata.service.enterprise.IEnterpriseService;
+import com.cm.tenlion.pollutantdata.service.enterprisepoll.IEnterprisePollService;
 import com.cm.tenlion.pollutantdata.service.instrument.IInstrumentService;
 import com.cm.tenlion.pollutantdata.service.poll.IPollService;
 import com.cm.tenlion.pollutantdata.utils.HJ212DataUtil;
@@ -44,6 +45,7 @@ import java.util.List;
 public class TCPHandler extends SimpleChannelInboundHandler<ByteBuf> {
 
     private IEnterpriseService enterpriseService;
+    private IEnterprisePollService enterprisePollService;
     private IInstrumentService instrumentService;
     private ICollectorService collectorService;
     private IDataRealService dataMinuteService;
@@ -51,12 +53,14 @@ public class TCPHandler extends SimpleChannelInboundHandler<ByteBuf> {
     private IAlarmLogService alarmLogService;
 
     public TCPHandler(IEnterpriseService enterpriseService,
+                      IEnterprisePollService enterprisePollService,
                       IInstrumentService instrumentService,
                       ICollectorService collectorService,
                       IDataRealService dataMinuteService,
                       IPollService pollService,
                       IAlarmLogService alarmLogService) {
         this.enterpriseService = enterpriseService;
+        this.enterprisePollService = enterprisePollService;
         this.instrumentService = instrumentService;
         this.collectorService = collectorService;
         this.dataMinuteService = dataMinuteService;
@@ -128,6 +132,7 @@ public class TCPHandler extends SimpleChannelInboundHandler<ByteBuf> {
             realDataHandler.setDataMinuteService(dataMinuteService);
             realDataHandler.setCollectorService(collectorService);
             realDataHandler.setAlarmLogService(alarmLogService);
+            realDataHandler.setEnterprisePollService(enterprisePollService);
             realDataHandler.setInstrumentDTO(instrumentDTO);
             realDataHandler.setEnterpriseDTO(enterpriseDTO);
             realDataHandler.handle(data, needResponse, ctx);

src/main/java/com/cm/tenlion/pollutantdata/utils/net/TCPServer.java

@@ -4,6 +4,7 @@ import com.cm.tenlion.pollutantdata.service.alarmlog.IAlarmLogService;
 import com.cm.tenlion.pollutantdata.service.collector.ICollectorService;
 import com.cm.tenlion.pollutantdata.service.datareal.IDataRealService;
 import com.cm.tenlion.pollutantdata.service.enterprise.IEnterpriseService;
+import com.cm.tenlion.pollutantdata.service.enterprisepoll.IEnterprisePollService;
 import com.cm.tenlion.pollutantdata.service.instrument.IInstrumentService;
 import com.cm.tenlion.pollutantdata.service.poll.IPollService;
 import io.netty.bootstrap.ServerBootstrap;
@@ -32,6 +33,7 @@ import java.util.concurrent.TimeUnit;
 public class TCPServer {
 
     private IEnterpriseService enterpriseService;
+    private IEnterprisePollService enterprisePollService;
     private IInstrumentService instrumentService;
     private ICollectorService collectorService;
     private IDataRealService dataMinuteService;
@@ -84,7 +86,7 @@ public class TCPServer {
                         }
                     });
                     // 这里的对象必须new，否则只有第一次有效，其余无法连接，每次执行的后都需要new
-                    pipeline.addLast(new TCPHandler(enterpriseService, instrumentService, collectorService, dataMinuteService, pollService, alarmLogService));
+                    pipeline.addLast(new TCPHandler(enterpriseService, enterprisePollService, instrumentService, collectorService, dataMinuteService, pollService, alarmLogService));
                 }
             });
 
@@ -107,6 +109,10 @@ public class TCPServer {
         this.enterpriseService = enterpriseService;
     }
 
+    public void setEnterprisePollService(IEnterprisePollService enterprisePollService) {
+        this.enterprisePollService = enterprisePollService;
+    }
+
     public void setInstrumentService(IInstrumentService instrumentService) {
         this.instrumentService = instrumentService;
     }

src/main/java/com/cm/tenlion/pollutantdata/utils/net/handle/RealDataHandler.java

@@ -8,15 +8,16 @@ import com.cm.tenlion.pollutantdata.enums.QnRtdEnum;
 import com.cm.tenlion.pollutantdata.manager.InstrumentManager;
 import com.cm.tenlion.pollutantdata.manager.WebSocketManager;
 import com.cm.tenlion.pollutantdata.pojo.BigDataWebSocketData;
-import com.cm.tenlion.pollutantdata.pojo.dtos.collector.CollectorDTO;
 import com.cm.tenlion.pollutantdata.pojo.dtos.enterprise.EnterpriseDTO;
 import com.cm.tenlion.pollutantdata.pojo.dtos.instrument.InstrumentDTO;
+import com.cm.tenlion.pollutantdata.pojo.pos.enterprisepoll.EnterprisePollPO;
 import com.cm.tenlion.pollutantdata.pojo.pos.poll.PollPO;
 import com.cm.tenlion.pollutantdata.pojo.vos.alarmlog.AlarmLogVO;
 import com.cm.tenlion.pollutantdata.pojo.vos.datareal.DataRealVO;
 import com.cm.tenlion.pollutantdata.service.alarmlog.IAlarmLogService;
 import com.cm.tenlion.pollutantdata.service.collector.ICollectorService;
 import com.cm.tenlion.pollutantdata.service.datareal.IDataRealService;
+import com.cm.tenlion.pollutantdata.service.enterprisepoll.IEnterprisePollService;
 import com.cm.tenlion.pollutantdata.service.poll.IPollService;
 import com.cm.tenlion.pollutantdata.utils.HJ212DataUtil;
 import com.cm.tenlion.pollutantdata.utils.config.IDataHandle;
@@ -44,6 +45,7 @@ public class RealDataHandler implements IDataHandle<HJ212DataUtil.Data> {
     private ICollectorService collectorService;
     private IPollService pollService;
     private IDataRealService dataMinuteService;
+    private IEnterprisePollService enterprisePollService;
     private EnterpriseDTO enterpriseDTO;
     private com.cm.tenlion.pollutantdata.pojo.dtos.instrument.InstrumentDTO instrumentDTO;
     private IAlarmLogService alarmLogService;
@@ -53,6 +55,7 @@ public class RealDataHandler implements IDataHandle<HJ212DataUtil.Data> {
     public void handle(HJ212DataUtil.Data data, boolean isRepo, ChannelHandlerContext ctx) throws Exception {
         List<HJ212DataUtil.RealDataCp> realDataCps = data.getRealDataCps();
         List<PollPO> pollPOs = pollService.listPO(new HashMap<>());
+        List<EnterprisePollPO> enterprisePollPOs = enterprisePollService.listPO(instrumentDTO.getEnterpriseId());
         for (HJ212DataUtil.RealDataCp realDataCp : realDataCps) {
             if (realDataCp.getDataTime() == null) {
                 log.error("数据时间为空");
@@ -102,10 +105,19 @@ public class RealDataHandler implements IDataHandle<HJ212DataUtil.Data> {
             dataRealVO.setIpAddress(ipAddress);
             dataMinuteService.save(dataRealVO);
             // 超过报警值，记录报警日志
+            EnterprisePollPO enterprisePollPO = getEnterprisePoll(realDataCp.getPollId(), enterprisePollPOs);
+            if (enterprisePollPO != null) {
+                // 1. 判断企业是否有污染因子的报警值，如果有以企业为准
+                if (enterprisePollPO.getEnterprisePollBeyond() != null && dataRealVO.getRtd() > enterprisePollPO.getEnterprisePollBeyond()) {
+                    saveAlarmLog(dataRealVO, pollPO.getAlarmValue(), instrumentDTO);
+                }
+            } else {
+                // 2. 如果企业没有设置污染因子，以公共为主
                 if (pollPO.getAlarmValue() != null && dataRealVO.getRtd() > pollPO.getAlarmValue()) {
                     saveAlarmLog(dataRealVO, pollPO.getAlarmValue(), instrumentDTO);
                 }
             }
+        }
         InstrumentManager.getInstance().addInstrument(ctx.channel().id().toString(), instrumentDTO);
         new ResponseDefault(ctx, QnRtdEnum.READY, ExeRtnEnum.SUCCESS, isRepo).response(data);
     }
@@ -135,6 +147,23 @@ public class RealDataHandler implements IDataHandle<HJ212DataUtil.Data> {
         WebSocketManager.getInstance().sendMessage(new BigDataWebSocketData<>(BigDataTypeEnum.ALARM_MESSAGE, message));
     }
 
+    /**
+     * 企业污染因子
+     *
+     * @param pollId
+     * @param enterprisePollPOs
+     * @return
+     */
+    private EnterprisePollPO getEnterprisePoll(String pollId, List<EnterprisePollPO> enterprisePollPOs) {
+        for (EnterprisePollPO enterprisePollPO : enterprisePollPOs) {
+            if (StringUtils.equals(pollId, enterprisePollPO.getPollNo()) ||
+                    StringUtils.equals(pollId, enterprisePollPO.getPollNoOld())) {
+                return enterprisePollPO;
+            }
+        }
+        return null;
+    }
+
     /**
      * 检查污染因子
      *
@@ -180,6 +209,10 @@ public class RealDataHandler implements IDataHandle<HJ212DataUtil.Data> {
         this.dataMinuteService = dataMinuteService;
     }
 
+    public void setEnterprisePollService(IEnterprisePollService enterprisePollService) {
+        this.enterprisePollService = enterprisePollService;
+    }
+
     public void setEnterpriseDTO(EnterpriseDTO enterpriseDTO) {
         this.enterpriseDTO = enterpriseDTO;
     }

src/main/resources/application-test.yml

@@ -1,6 +1,6 @@
 server:
-  port: 8080
+  port: 7004
-  url: http://192.168.0.120:8080/pollutant
+  url: http://192.168.0.103:7004/pollutant
   title: 污染物上报系统
   servlet:
     context-path: /pollutant
@@ -22,11 +22,11 @@ spring:
       max-request-size: 1GB
   datasource:
     druid:
-      url: jdbc:mysql://localhost:3306/db_pollutant_data?useUnicode=true&characterEncoding=utf8&characterSetResults=utf8&autoReconnect=true&failOverReadOnly=false&useSSL=false&serverTimezone=UTC
+      url: jdbc:mysql://192.168.0.151:3306/db_pollutant_data?useUnicode=true&characterEncoding=utf8&characterSetResults=utf8&autoReconnect=true&failOverReadOnly=false&useSSL=false&serverTimezone=UTC
       db-type: mysql
       driver-class-name: com.mysql.jdbc.Driver
       username: root
-      password: 123456
+      password: root
       initial-size: 2
       min-idle: 2
       max-active: 10
@@ -60,8 +60,8 @@ security:
     oauth-server: http://192.168.0.152:7001/usercenter
     oauth-logout: ${security.oauth2.oauth-server}/logout?redirect_uri=${server.url}
     client:
-      client-id: 74e4b55ad48840f9b4de86ce5da58b53
+      client-id: 2a347bba1dc54def83a47493c7578ae2
-      client-secret: VjlWbUFFbkJKMmZ3U29lekROb2x3M3Q1SmEzOGlwV3NzT3ZqSDByQVZoWW1ac2wwZTJHWk5NbXh3L3h3U2c4Rg==
+      client-secret: TEdsOHlnTnc4T2JwTXM3alZldzRSM1ROKzhaQkZaQ24vSWJxREd0TWVLMG1ac2wwZTJHWk5NbXh3L3h3U2c4Rg==
       user-authorization-uri: ${security.oauth2.oauth-server}/oauth_client/authorize
       access-token-uri: ${security.oauth2.oauth-server}/oauth_client/token
       grant-type: authorization_code

src/main/resources/mybatis/mapper/enterprisepoll/enterprise-poll-mapper.xml

@@ -30,6 +30,8 @@
         <result column="enterprise_poll_id" property="enterprisePollId"/>
         <result column="enterprise_id" property="enterpriseId"/>
         <result column="poll_id" property="pollId"/>
+        <result column="poll_no" property="pollNo"/>
+        <result column="poll_no_old" property="pollNoOld"/>
         <result column="enterprise_poll_beyond" property="enterprisePollBeyond"/>
         <result column="gmt_create" property="gmtCreate"/>
         <result column="creator" property="creator"/>
@@ -184,6 +186,8 @@
             t1.enterprise_poll_id,
             t1.enterprise_id,
             t1.poll_id,
+            t1.poll_no,
+            t1.poll_no_old,
             t1.enterprise_poll_beyond,
             t1.gmt_create,
             t1.creator,
@@ -198,6 +202,23 @@
             AND
             t1.enterprise_poll_id = #{enterprisePollId}
             </if>
+        <if test="enterpriseId != null and enterpriseId != ''">
+        AND
+            t1.enterprise_id = #{enterpriseId}
+        </if>
+        <if test="pollId != null and pollId != ''">
+            AND
+            t1.poll_id = #{pollId}
+        </if>
+        <if test="pollNo != null and pollNo != ''">
+        AND (
+            t1.poll_no  = #{pollNo}
+            OR
+            t1.poll_no_old  = #{pollNo}
+            OR
+            t1.poll_id = #{pollNo}
+            )
+        </if>
     </select>
 
     <!-- 企业监控污染因子列表 -->
@@ -279,6 +300,8 @@
             t1.enterprise_poll_id,
             t1.enterprise_id,
             t1.poll_id,
+            t1.poll_no,
+            t1.poll_no_old,
             t1.enterprise_poll_beyond,
             t1.gmt_create,
             t1.creator,
@@ -310,6 +333,10 @@
                 #{enterprisePollIds[${index}]}
             </foreach>
         </if>
+        <if test="enterpriseId != null and enterpriseId != ''">
+        AND
+            t1.enterprise_id = #{enterpriseId}
+        </if>
     </select>
 
     <!-- 企业监控污染因子统计  -->

src/test/java/cn/com/tenlion/pollutantdata/ConnectRunner.java

@@ -85,7 +85,7 @@ public class ConnectRunner implements Runnable {
                     });
                 }
             });
-            ChannelFuture channelFuture = bootstrap.connect("192.168.0.103", 1991).sync();
+            ChannelFuture channelFuture = bootstrap.connect("192.168.0.120", 1991).sync();
             channelFuture.addListener((ChannelFutureListener) future -> {
                 if (!future.isSuccess()) {
                     future.channel().pipeline().fireChannelInactive();