tj

brg-queue-center/app/base/db/consumer.base.js

@@ -43,14 +43,16 @@ class ConsumerBase {
   async execSubDoConsumer(queuedName, actionBody) {
     var execResult = null;
     try {
+      const notifyQueuedName = "NOTIFY-SYTXPUBLIC-MSGQ";
       this.subBeforeConsumer(queuedName, actionBody);
       actionBody.requestId = await this.getBusUid("PUB-");
-      if (queuedName.indexOf("SYTXFAIL-SYTXPUBLIC-MSGQ") < 0) {
+      if ("SYTXFAIL-SYTXPUBLIC-MSGQ" != queuedName) {
         execResult = await this.subDoConsumer(queuedName, actionBody);
-        actionBody.resultInfo = execResult;
+        if (notifyQueuedName != queuedName) {
+          actionBody.resultInfo = execResult;
+        }
         if (execResult.status === 1) {
-          var notifyQueuedName = "NOTIFY-SYTXPUBLIC-MSGQ";
-          if (queuedName.indexOf(notifyQueuedName) < 0) {
+          if (notifyQueuedName != queuedName) {
             if (actionBody.notifyUrl && actionBody.notifyUrl.indexOf("http") >= 0) {
               await this.redisClient.lpushData(notifyQueuedName, actionBody);
             }
@@ -67,10 +69,14 @@ class ConsumerBase {
         return;
       }
       if (actionBody.counter > 4) {
-        await this.pushFailureLogDao.addOpFailureLogs("推送失败", actionBody, actionBody.resultInfo);
+        await this.pushFailureLogDao.addOpFailureLogs("推送失败", actionBody, actionBody.pushAgainResultInfo);
         return;
       }
+      if (actionBody.pushAgainResultInfo) {
+        delete actionBody["pushAgainResultInfo"];
+      }
       execResult = await this.subDoConsumer(queuedName, actionBody);
+      actionBody.pushAgainResultInfo = execResult;
     } catch (error) {
       var stackStr = error.stack ? error.stack : JSON.stringify(error);
       this.errorLogDao.addOpErrorLogs("队列执行execSubDoConsumer存在异常", actionBody, execResult, stackStr, 3);