X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;ds=sidebyside;f=datarouter-node%2Fsrc%2Fmain%2Fjava%2Forg%2Fonap%2Fdmaap%2Fdatarouter%2Fnode%2FDeliveryQueue.java;h=d447bcc1d3438767a43e33d346f46dda00b1d781;hb=adb2ad2d16e851fbf8dcc71af68949a74463204d;hp=a3df26acee9e449423a02214f00542e6eead39f8;hpb=5ad15107613e7aa41af1a0c1e61c3be2e608e4c4;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 a3df26ac..d447bcc1 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
@@ -26,19 +26,22 @@ 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 java.io.File;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
import org.jetbrains.annotations.Nullable;
/**
* Mechanism for monitoring and controlling delivery of files to a destination.
- *
- * The DeliveryQueue class maintains lists of DeliveryTasks for a single
+ *
+ *
The DeliveryQueue class maintains lists of DeliveryTasks for a single
* destination (a subscription or another data router node) and assigns
* delivery threads to try to deliver them. It also maintains a delivery
* status that causes it to back off on delivery attempts after a failure.
- *
- * If the most recent delivery result was a failure, then no more attempts
+ *
+ *
If the most recent delivery result was a failure, then no more attempts
* will be made for a period of time. Initially, and on the first failure
* following a success, this delay will be DeliveryQueueHelper.getInitFailureTimer() (milliseconds).
* If, after this delay, additional failures occur, each failure will
@@ -50,34 +53,44 @@ import org.jetbrains.annotations.Nullable;
* delivery fails while the delay was active, it will not change the delay
* or change the duration of any subsequent delay.
* If, however, it succeeds, it will cancel the delay.
- *
- * The queue maintains 3 collections of files to deliver: A todo list of
+ * The queue maintains 3 collections of files to deliver: A todoList of
* files that will be attempted, a working set of files that are being
* attempted, and a retry set of files that were attempted and failed.
- * Whenever the todo list is empty and needs to be refilled, a scan of the
+ * Whenever the todoList is empty and needs to be refilled, a scan of the
* spool directory is made and the file names sorted. Any files in the working set are ignored.
* If a DeliveryTask for the file is in the retry set, then that delivery
- * task is placed on the todo list. Otherwise, a new DeliveryTask for the
- * file is created and placed on the todo list.
- * If, when a DeliveryTask is about to be removed from the todo list, its
+ * task is placed on the todoList. Otherwise, a new DeliveryTask for the
+ * file is created and placed on the todoList.
+ * If, when a DeliveryTask is about to be removed from the todoList, its
* age exceeds DeliveryQueueHelper.getExpirationTimer(), then it is instead
* marked as expired.
- *
- * A delivery queue also maintains a skip flag. This flag is true if the
+ *
+ *
A delivery queue also maintains a skip flag. This flag is true if the
* 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 working = new Hashtable<>();
- private Hashtable retry = new Hashtable<>();
+ private HashMap working = new HashMap<>();
+ private HashMap retry = new HashMap<>();
private int todoindex;
private boolean failed;
private long failduration;
private long resumetime;
private File dir;
- private Vector todo = new Vector<>();
+ private List todoList = new ArrayList<>();
+
+ /**
+ * Create a delivery queue for a given destination info.
+ */
+ DeliveryQueue(DeliveryQueueHelper deliveryQueueHelper, DestInfo destinationInfo) {
+ this.deliveryQueueHelper = deliveryQueueHelper;
+ this.destinationInfo = destinationInfo;
+ dir = new File(destinationInfo.getSpool());
+ dir.mkdirs();
+ }
/**
* Try to cancel a delivery task.
@@ -90,8 +103,8 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
DeliveryTask dt = retry.get(pubid);
if (dt == null) {
- for (int i = todoindex; i < todo.size(); i++) {
- DeliveryTask xdt = todo.get(i);
+ for (int i = todoindex; i < todoList.size(); i++) {
+ DeliveryTask xdt = todoList.get(i);
if (xdt.getPublishId().equals(pubid)) {
dt = xdt;
break;
@@ -107,7 +120,8 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
if (dt.isCleaned()) {
return (0);
}
- StatusLog.logExp(dt.getPublishId(), dt.getFeedId(), dt.getSubId(), dt.getURL(), dt.getMethod(), dt.getCType(), dt.getLength(), "diskFull", dt.getAttempts());
+ StatusLog.logExp(dt.getPublishId(), dt.getFeedId(), dt.getSubId(), dt.getURL(),
+ dt.getMethod(), dt.getCType(), dt.getLength(), "diskFull", dt.getAttempts());
dt.clean();
return (dt.getLength());
}
@@ -148,7 +162,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
if (failduration == 0) {
if (destinationInfo.isPrivilegedSubscriber()) {
failduration = deliveryQueueHelper.getWaitForFileProcessFailureTimer();
- } else{
+ } else {
failduration = deliveryQueueHelper.getInitFailureTimer();
}
}
@@ -206,110 +220,108 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
}
while (true) {
- if (todoindex >= todo.size()) {
+ if (todoindex >= todoList.size()) {
todoindex = 0;
- todo = new Vector<>();
+ todoList = new ArrayList<>();
String[] files = dir.list();
- Arrays.sort(files);
- scanForNextTask(files);
- retry = new Hashtable<>();
- }
- DeliveryTask dt = getDeliveryTask(mindate);
- if (dt != null) {
- return dt;
+ if (files != null) {
+ Arrays.sort(files);
+ scanForNextTask(files);
+ }
+ retry = new HashMap<>();
}
- return null;
-
+ return getDeliveryTask(mindate);
}
}
/**
- * Create a delivery queue for a given destination info
- */
- DeliveryQueue(DeliveryQueueHelper deliveryQueueHelper, DestInfo destinationInfo) {
- this.deliveryQueueHelper = deliveryQueueHelper;
- this.destinationInfo = destinationInfo;
- dir = new File(destinationInfo.getSpool());
- dir.mkdirs();
- }
-
- /**
- * Update the destination info for this delivery queue
+ * Update the destination info for this delivery queue.
*/
public void config(DestInfo destinationInfo) {
this.destinationInfo = destinationInfo;
}
/**
- * Get the dest info
+ * Get the dest info.
*/
public DestInfo getDestinationInfo() {
return (destinationInfo);
}
/**
- * Get the config manager
+ * Get the config manager.
*/
public DeliveryQueueHelper getConfig() {
return (deliveryQueueHelper);
}
/**
- * Exceptional condition occurred during delivery
+ * Exceptional condition occurred during delivery.
*/
public void reportDeliveryExtra(DeliveryTask task, long sent) {
StatusLog.logDelExtra(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getLength(), sent);
}
/**
- * Message too old to deliver
+ * Message too old to deliver.
*/
void reportExpiry(DeliveryTask task) {
- StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "retriesExhausted", task.getAttempts());
+ StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
+ task.getCType(), task.getLength(), "retriesExhausted", task.getAttempts());
markExpired(task);
}
/**
- * Completed a delivery attempt
+ * Completed a delivery attempt.
*/
public void reportStatus(DeliveryTask task, int status, String xpubid, String location) {
if (status < 300) {
- StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, xpubid);
+ StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
+ task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, xpubid);
if (destinationInfo.isPrivilegedSubscriber()) {
- task.setResumeTime(System.currentTimeMillis() + deliveryQueueHelper.getWaitForFileProcessFailureTimer());
+ task.setResumeTime(System.currentTimeMillis()
+ + deliveryQueueHelper.getWaitForFileProcessFailureTimer());
markFailWithRetry(task);
} else {
markSuccess(task);
}
} else if (status < 400 && deliveryQueueHelper.isFollowRedirects()) {
- StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
+ StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
+ task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
if (deliveryQueueHelper.handleRedirection(destinationInfo, location, task.getFileId())) {
markRedirect(task);
} else {
- StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
+ StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(),
+ task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
markFailNoRetry(task);
}
- } else if (status < 500 && status != 429) { // Status 429 is the standard response for Too Many Requests and indicates that a file needs to be delivered again at a later time.
- StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
- StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
+ } else if (status < 500 && status != 429) {
+ // Status 429 is the standard response for Too Many Requests and indicates
+ // that a file needs to be delivered again at a later time.
+ StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
+ task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
+ StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
+ task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
markFailNoRetry(task);
} else {
- StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
+ StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
+ task.getCType(), task.getLength(), destinationInfo.getAuthUser(), status, location);
markFailWithRetry(task);
}
}
/**
- * Delivery failed by reason of an exception
+ * Delivery failed by reason of an exception.
*/
public void reportException(DeliveryTask task, Exception exception) {
- StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), destinationInfo.getAuthUser(), -1, exception.toString());
+ StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(),
+ task.getCType(), task.getLength(), destinationInfo.getAuthUser(), -1, exception.toString());
deliveryQueueHelper.handleUnreachable(destinationInfo);
markFailWithRetry(task);
}
/**
- * Get the feed ID for a subscription
+ * Get the feed ID for a subscription.
*
* @param subid The subscription ID
* @return The feed ID
@@ -319,14 +331,14 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
/**
- * Get the URL to deliver a message to given the file ID
+ * Get the URL to deliver a message to given the file ID.
*/
public String getDestURL(String fileid) {
return (deliveryQueueHelper.getDestURL(destinationInfo, fileid));
}
/**
- * Deliver files until there's a failure or there are no more
+ * Deliver files until there's a failure or there are no more.
* files to deliver
*/
public void run() {
@@ -343,21 +355,21 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
/**
- * Is there no work to do for this queue right now?
+ * Is there no work to do for this queue right now?.
*/
synchronized boolean isSkipSet() {
return (peekNext() == null);
}
/**
- * Reset the retry timer
+ * Reset the retry timer.
*/
void resetQueue() {
resumetime = System.currentTimeMillis();
}
/**
- * Get task if in queue and mark as success
+ * Get task if in queue and mark as success.
*/
boolean markTaskSuccess(String pubId) {
DeliveryTask task = working.get(pubId);
@@ -375,6 +387,7 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
return false;
}
+
private void scanForNextTask(String[] files) {
for (String fname : files) {
String pubId = getPubId(fname);
@@ -385,14 +398,14 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
if (dt == null) {
dt = new DeliveryTask(this, pubId);
}
- todo.add(dt);
+ todoList.add(dt);
}
}
@Nullable
private DeliveryTask getDeliveryTask(long mindate) {
- if (todoindex < todo.size()) {
- DeliveryTask dt = todo.get(todoindex);
+ if (todoindex < todoList.size()) {
+ DeliveryTask dt = todoList.get(todoindex);
if (dt.isCleaned()) {
todoindex++;
}
@@ -433,4 +446,4 @@ public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
}
return fname2;
}
-}
+}
\ No newline at end of file