指令同步日志

nursing-unit-services/nu-services-biz/src/main/java/com/nu/mq/directive/listener/DirectiveMQListener.java

@@ -2,9 +2,11 @@ package com.nu.mq.directive.listener;
 
 import cn.hutool.core.bean.BeanUtil;
 import cn.hutool.core.collection.ListUtil;
+import com.baomidou.dynamic.datasource.annotation.DSTransactional;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.nu.dto.*;
+import com.nu.enums.MQStatus;
 import com.nu.modules.directivetag.body.entity.DirectiveBodyTag;
 import com.nu.modules.directivetag.body.entity.DirectiveBodyTagRelation;
 import com.nu.modules.directivetag.body.service.IDirectiveBodyTagService;
@@ -24,6 +26,7 @@ import com.nu.modules.servicetype.service.IConfigServiceTypeService;
 import com.nu.modules.sysconfig.entity.SysConfig;
 import com.nu.modules.sysconfig.service.ISysConfigService;
 import com.nu.utils.FileDownloader;
+import com.nu.utils.RabbitMQUtil;
 import com.nu.utils.SafetyUtil;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.compress.utils.Lists;
@@ -36,6 +39,7 @@ import org.springframework.amqp.rabbit.annotation.RabbitListener;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
 
 import java.io.File;
 import java.net.URLEncoder;
@@ -70,23 +74,38 @@ public class DirectiveMQListener {
     private IConfigServiceCategoryService serviceCategoryService;
     @Autowired
     private IConfigServiceTypeService serviceTypeService;
-//    @Autowired
+    @Autowired
-//    private RabbitMQUtil rabbitMQUtil;
+    private RabbitMQUtil rabbitMQUtil;
 
-    @RabbitListener(
+    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "#{directiveAsyncDQNP.getSyncDirectiveQueueName()}"), exchange = @Exchange(name = "hldy.directive", type = ExchangeTypes.DIRECT), key = "#{directiveAsyncDQNP.getSyncDirectiveKeyName()}"), errorHandler = "directiveMQErrorHandler")
-            bindings = @QueueBinding(
+    @Transactional(rollbackFor = {Exception.class})
-                    value = @Queue(name = "#{directiveAsyncDQNP.getSyncDirectiveQueueName()}"),
-                    exchange = @Exchange(name = "hldy.directive", type = ExchangeTypes.DIRECT),
-                    key = "#{directiveAsyncDQNP.getSyncDirectiveKeyName()}"
-            ),
-            errorHandler = "directiveMQErrorHandler"
-    )
     public void handleAuditResult(DirectiveMQDto dto) {
         dto.setIzInc(true);
         dto.setIdStr(dto.getDirectiveList().stream().map(d -> d.getId()).collect(Collectors.joining(",")));
-        //增量处理
+        try {
+            //增量处理数据
             handleIncremental(dto);
+            //数据处理出错的话就不能继续处理指令资源了 所以方法放到这个位置
             handleCreateMedia(dto);
+        } catch (Exception e) {
+            e.printStackTrace();
+            //返回错误日志
+            StatusMQDto statusMQDto = new StatusMQDto();
+            statusMQDto.setStatus(MQStatus.PROCESS_FAILED.getCode());
+            statusMQDto.setMessage(e.getMessage());
+            statusMQDto.setAsyncId(dto.getAsyncId());
+            statusMQDto.setCode("data");
+            rabbitMQUtil.sendToExchange("hldy.fwzl", "fwzl.async.result", statusMQDto);
+            throw new RuntimeException("数据同步失败");
+        }
+        //在这里返回数据成功和日志
+        StatusMQDto statusMQDto = new StatusMQDto();
+        statusMQDto.setStatus(MQStatus.SUCCESS.getCode());
+        statusMQDto.setMessage("数据同步成功！");
+        statusMQDto.setAsyncId(dto.getAsyncId());
+        statusMQDto.setCode("data");
+        rabbitMQUtil.sendToExchange("hldy.fwzl", "fwzl.async.result", statusMQDto);
+        //资源日志在handleCreateMedia中有处理
     }
 
     /**
@@ -94,15 +113,9 @@ public class DirectiveMQListener {
      *
      * @param mqDto
      */
-    @RabbitListener(
+    @RabbitListener(bindings = @QueueBinding(value = @Queue(name = "#{directiveAsyncDQNP.getCreateMediaQueueName()}"), exchange = @Exchange(name = "hldy.directive", type = ExchangeTypes.DIRECT), key = "#{directiveAsyncDQNP.getCreateMediaKeyName()}"), errorHandler = "directiveMQErrorHandler")
-            bindings = @QueueBinding(
-                    value = @Queue(name = "#{directiveAsyncDQNP.getCreateMediaQueueName()}"),
-                    exchange = @Exchange(name = "hldy.directive", type = ExchangeTypes.DIRECT),
-                    key = "#{directiveAsyncDQNP.getCreateMediaKeyName()}"
-            ),
-            errorHandler = "directiveMQErrorHandler"
-    )
     public void handleCreateMedia(DirectiveMQDto mqDto) {
+        try {
             mqDto.getDirectiveList().stream().forEach(dto -> {
                 //查询现有服务指令
                 QueryWrapper<ConfigServiceDirective> qw = new QueryWrapper<>();
@@ -141,6 +154,7 @@ public class DirectiveMQListener {
                                 MediaAsyncErrorLog mediaAsyncErrorLog = new MediaAsyncErrorLog();
                                 mediaAsyncErrorLog.setMediaid(previewFile);
                                 mediaAsyncErrorLogService.save(mediaAsyncErrorLog);
+                                throw new RuntimeException("服务指令图片文件拉取错误,指令id" + currentDirective.getId());
                             }
                         }
                     }
@@ -166,6 +180,7 @@ public class DirectiveMQListener {
                                 mediaAsyncErrorLog.setMediaid(immediateFile);
                                 mediaAsyncErrorLogService.save(mediaAsyncErrorLog);
                                 e.printStackTrace();
+                                throw new RuntimeException("即时指令图标文件拉取错误,指令id" + currentDirective.getId());
                             }
                         }
                     }
@@ -191,6 +206,7 @@ public class DirectiveMQListener {
                                 mediaAsyncErrorLog.setMediaid(mp3File);
                                 mediaAsyncErrorLogService.save(mediaAsyncErrorLog);
                                 e.printStackTrace();
+                                throw new RuntimeException("指令音频文件拉取错误,指令id" + currentDirective.getId());
                             }
                         }
                     }
@@ -216,12 +232,31 @@ public class DirectiveMQListener {
                                 mediaAsyncErrorLog.setMediaid(mp4File);
                                 mediaAsyncErrorLogService.save(mediaAsyncErrorLog);
                                 e.printStackTrace();
+                                throw new RuntimeException("指令视频文件拉取错误,指令id" + currentDirective.getId());
                             }
                         }
                     }
                     directiveService.updateById(configServiceDirective);
                 }
             });
+            //在这里返回数据成功和日志
+            StatusMQDto statusMQDto = new StatusMQDto();
+            statusMQDto.setStatus(MQStatus.SUCCESS.getCode());
+            statusMQDto.setMessage("文件同步成功！");
+            statusMQDto.setAsyncId(mqDto.getAsyncId());
+            statusMQDto.setCode("file");
+            rabbitMQUtil.sendToExchange("hldy.fwzl", "fwzl.async.result", statusMQDto);
+        } catch (Exception e) {
+            e.printStackTrace();
+            //返回错误日志
+            StatusMQDto statusMQDto = new StatusMQDto();
+            statusMQDto.setStatus(MQStatus.PROCESS_FAILED.getCode());
+            statusMQDto.setMessage(e.getMessage());
+            statusMQDto.setAsyncId(mqDto.getAsyncId());
+            statusMQDto.setCode("file");
+            rabbitMQUtil.sendToExchange("hldy.fwzl", "fwzl.async.result", statusMQDto);
+        }
+
     }
 //    @RabbitListener(
 //            bindings = @QueueBinding(
@@ -471,17 +506,12 @@ public class DirectiveMQListener {
         QueryWrapper<ConfigServiceDirective> dtw = new QueryWrapper<>();
         dtw.select("id");
         List<ConfigServiceDirective> tempList = directiveService.list(dtw);
-        Set<String> existingIds = tempList.stream()
+        Set<String> existingIds = tempList.stream().map(ConfigServiceDirective::getId).map(String::valueOf).collect(Collectors.toSet());
-                .map(ConfigServiceDirective::getId)
-                .map(String::valueOf)
-                .collect(Collectors.toSet());
 
         String idStr = dto.getIdStr();
         List<String> inputIds = Arrays.asList(idStr.split(","));
 
-        List<String> uniqueIds = inputIds.stream()
+        List<String> uniqueIds = inputIds.stream().filter(id -> !existingIds.contains(id)).collect(Collectors.toList());
-                .filter(id -> !existingIds.contains(id))
-                .collect(Collectors.toList());
 
         dto.setIdList(uniqueIds);