X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=datarouter-node%2Fsrc%2Fmain%2Fjava%2Forg%2Fonap%2Fdmaap%2Fdatarouter%2Fnode%2FDeliveryQueue.java;h=a3df26acee9e449423a02214f00542e6eead39f8;hb=5ad15107613e7aa41af1a0c1e61c3be2e608e4c4;hp=bef8dab22d72f9e835fc66bdfdbf97f79bee44e1;hpb=875daad0a737115702458d1850ddee87ac4cea30;p=dmaap%2Fdatarouter.git

diff --git a/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java b/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java
index bef8dab2..a3df26ac 100644
--- a/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java
+++ b/datarouter-node/src/main/java/org/onap/dmaap/datarouter/node/DeliveryQueue.java
@@ -24,8 +24,11 @@
 
 package org.onap.dmaap.datarouter.node;
 
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
 import java.io.*;
 import java.util.*;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * Mechanism for monitoring and controlling delivery of files to a destination.
@@ -64,6 +67,7 @@ import java.util.*;
  * failure timer is active or if no files are found in a directory scan.
  */
 public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
+    private static EELFLogger logger = EELFManager.getInstance().getLogger(DeliveryQueue.class);
     private DeliveryQueueHelper deliveryQueueHelper;
     private DestInfo destinationInfo;
     private Hashtable<String, DeliveryTask> working = new Hashtable<>();
@@ -113,6 +117,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
      */
     private synchronized void markSuccess(DeliveryTask task) {
         working.remove(task.getPublishId());
+        logger.info(task.getPublishId() + " marked as success.");
         task.clean();
         failed = false;
         failduration = 0;
@@ -122,6 +127,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
      * Mark that a delivery task has expired.
      */
     private synchronized void markExpired(DeliveryTask task) {
+        logger.info(task.getPublishId() + " marked as expired.");
         task.clean();
     }
 
@@ -130,6 +136,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
      */
     private synchronized void markFailNoRetry(DeliveryTask task) {
         working.remove(task.getPublishId());
+        logger.info(task.getPublishId() + " marked as failed permanently");
         task.clean();
         failed = false;
         failduration = 0;
@@ -159,6 +166,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
      */
     private synchronized void markRedirect(DeliveryTask task) {
         working.remove(task.getPublishId());
+        logger.info(task.getPublishId() + " marked as redirected.");
         retry.put(task.getPublishId(), task);
     }
 
@@ -167,6 +175,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
      */
     private synchronized void markFailWithRetry(DeliveryTask task) {
         working.remove(task.getPublishId());
+        logger.info(task.getPublishId() + " marked as temporarily failed.");
         retry.put(task.getPublishId(), task);
         fdupdate();
     }
@@ -202,53 +211,15 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
                 todo = new Vector<>();
                 String[] files = dir.list();
                 Arrays.sort(files);
-                for (String fname : files) {
-                    if (!fname.endsWith(".M")) {
-                        continue;
-                    }
-                    String fname2 = fname.substring(0, fname.length() - 2);
-                    long pidtime = 0;
-                    int dot = fname2.indexOf('.');
-                    if (dot < 1) {
-                        continue;
-                    }
-                    try {
-                        pidtime = Long.parseLong(fname2.substring(0, dot));
-                    } catch (Exception e) {
-                    }
-                    if (pidtime < 1000000000000L) {
-                        continue;
-                    }
-                    if (working.get(fname2) != null) {
-                        continue;
-                    }
-                    DeliveryTask dt = retry.get(fname2);
-                    if (dt == null) {
-                        dt = new DeliveryTask(this, fname2);
-                    }
-                    todo.add(dt);
-                }
+                scanForNextTask(files);
                 retry = new Hashtable<>();
             }
-            if (todoindex < todo.size()) {
-                DeliveryTask dt = todo.get(todoindex);
-                if (dt.isCleaned()) {
-                    todoindex++;
-                    continue;
-                }
-                if (destinationInfo.isPrivilegedSubscriber() && dt.getResumeTime() > System.currentTimeMillis()) {
-                    retry.put(dt.getPublishId(), dt);
-                    todoindex++;
-                    continue;
-                }
-                if (dt.getDate() >= mindate) {
-                    return (dt);
-                }
-                todoindex++;
-                reportExpiry(dt);
-                continue;
+            DeliveryTask dt = getDeliveryTask(mindate);
+            if (dt != null) {
+                return dt;
             }
-            return (null);
+            return null;
+
         }
     }
 
@@ -359,11 +330,12 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
      * files to deliver
      */
     public void run() {
-        DeliveryTask t;
+        DeliveryTask task;
         long endtime = System.currentTimeMillis() + deliveryQueueHelper.getFairTimeLimit();
         int filestogo = deliveryQueueHelper.getFairFileLimit();
-        while ((t = getNext()) != null) {
-            t.run();
+        while ((task = getNext()) != null) {
+            logger.info("Processing file: " + task.getPublishId());
+            task.run();
             if (--filestogo <= 0 || System.currentTimeMillis() > endtime) {
                 break;
             }
@@ -403,4 +375,62 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
         }
         return false;
     }
+    private void scanForNextTask(String[] files) {
+        for (String fname : files) {
+            String pubId = getPubId(fname);
+            if (pubId == null) {
+                continue;
+            }
+            DeliveryTask dt = retry.get(pubId);
+            if (dt == null) {
+                dt = new DeliveryTask(this, pubId);
+            }
+            todo.add(dt);
+        }
+    }
+
+    @Nullable
+    private DeliveryTask getDeliveryTask(long mindate) {
+        if (todoindex < todo.size()) {
+            DeliveryTask dt = todo.get(todoindex);
+            if (dt.isCleaned()) {
+                todoindex++;
+            }
+            if (destinationInfo.isPrivilegedSubscriber() && dt.getResumeTime() > System.currentTimeMillis()) {
+                retry.put(dt.getPublishId(), dt);
+                todoindex++;
+            }
+            if (dt.getDate() >= mindate) {
+                return (dt);
+            }
+            todoindex++;
+            reportExpiry(dt);
+        }
+        return null;
+    }
+
+    @Nullable
+    private String getPubId(String fname) {
+        if (!fname.endsWith(".M")) {
+            return null;
+        }
+        String fname2 = fname.substring(0, fname.length() - 2);
+        long pidtime = 0;
+        int dot = fname2.indexOf('.');
+        if (dot < 1) {
+            return null;
+        }
+        try {
+            pidtime = Long.parseLong(fname2.substring(0, dot));
+        } catch (Exception e) {
+            logger.error("Exception", e);
+        }
+        if (pidtime < 1000000000000L) {
+            return null;
+        }
+        if (working.get(fname2) != null) {
+            return null;
+        }
+        return fname2;
+    }
 }