1 /*******************************************************************************
\r
2 * ============LICENSE_START==================================================
\r
4 * * ===========================================================================
\r
5 * * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
\r
6 * * ===========================================================================
\r
7 * * Licensed under the Apache License, Version 2.0 (the "License");
\r
8 * * you may not use this file except in compliance with the License.
\r
9 * * You may obtain a copy of the License at
\r
11 * * http://www.apache.org/licenses/LICENSE-2.0
\r
13 * * Unless required by applicable law or agreed to in writing, software
\r
14 * * distributed under the License is distributed on an "AS IS" BASIS,
\r
15 * * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
\r
16 * * See the License for the specific language governing permissions and
\r
17 * * limitations under the License.
\r
18 * * ============LICENSE_END====================================================
\r
20 * * ECOMP is a trademark and service mark of AT&T Intellectual Property.
\r
22 ******************************************************************************/
\r
25 package org.onap.dmaap.datarouter.node;
\r
31 * Mechanism for monitoring and controlling delivery of files to a destination.
\r
33 * The DeliveryQueue class maintains lists of DeliveryTasks for a single
\r
34 * destination (a subscription or another data router node) and assigns
\r
35 * delivery threads to try to deliver them. It also maintains a delivery
\r
36 * status that causes it to back off on delivery attempts after a failure.
\r
38 * If the most recent delivery result was a failure, then no more attempts
\r
39 * will be made for a period of time. Initially, and on the first failure
\r
40 * following a success, this delay will be DeliveryQueueHelper.getInitFailureTimer() (milliseconds).
\r
41 * If, after this delay, additional failures occur, each failure will
\r
42 * multiply the delay by DeliveryQueueHelper.getFailureBackoff() up to a
\r
43 * maximum delay specified by DeliveryQueueHelper.getMaxFailureTimer().
\r
44 * Note that this behavior applies to the delivery queue as a whole and not
\r
45 * to individual files in the queue. If multiple files are being
\r
46 * delivered and one fails, the delay will be started. If a second
\r
47 * delivery fails while the delay was active, it will not change the delay
\r
48 * or change the duration of any subsequent delay.
\r
49 * If, however, it succeeds, it will cancel the delay.
\r
51 * The queue maintains 3 collections of files to deliver: A todo list of
\r
52 * files that will be attempted, a working set of files that are being
\r
53 * attempted, and a retry set of files that were attempted and failed.
\r
54 * Whenever the todo list is empty and needs to be refilled, a scan of the
\r
55 * spool directory is made and the file names sorted. Any files in the working set are ignored.
\r
56 * If a DeliveryTask for the file is in the retry set, then that delivery
\r
57 * task is placed on the todo list. Otherwise, a new DeliveryTask for the
\r
58 * file is created and placed on the todo list.
\r
59 * If, when a DeliveryTask is about to be removed from the todo list, its
\r
60 * age exceeds DeliveryQueueHelper.getExpirationTimer(), then it is instead
\r
61 * marked as expired.
\r
63 * A delivery queue also maintains a skip flag. This flag is true if the
\r
64 * failure timer is active or if no files are found in a directory scan.
\r
66 public class DeliveryQueue implements Runnable, DeliveryTaskHelper {
\r
67 private DeliveryQueueHelper dqh;
\r
68 private DestInfo di;
\r
69 private Hashtable<String, DeliveryTask> working = new Hashtable<String, DeliveryTask>();
\r
70 private Hashtable<String, DeliveryTask> retry = new Hashtable<String, DeliveryTask>();
\r
71 private int todoindex;
\r
72 private boolean failed;
\r
73 private long failduration;
\r
74 private long resumetime;
\r
76 private Vector<DeliveryTask> todo = new Vector<DeliveryTask>();
\r
78 * Try to cancel a delivery task.
\r
79 * @return The length of the task in bytes or 0 if the task cannot be cancelled.
\r
81 public synchronized long cancelTask(String pubid) {
\r
82 if (working.get(pubid) != null) {
\r
85 DeliveryTask dt = retry.get(pubid);
\r
87 for (int i = todoindex; i < todo.size(); i++) {
\r
88 DeliveryTask xdt = todo.get(i);
\r
89 if (xdt.getPublishId().equals(pubid)) {
\r
96 dt = new DeliveryTask(this, pubid);
\r
97 if (dt.getFileId() == null) {
\r
101 if (dt.isCleaned()) {
\r
104 StatusLog.logExp(dt.getPublishId(), dt.getFeedId(), dt.getSubId(), dt.getURL(), dt.getMethod(), dt.getCType(), dt.getLength(), "diskFull", dt.getAttempts());
\r
106 return(dt.getLength());
\r
109 * Mark that a delivery task has succeeded.
\r
111 public synchronized void markSuccess(DeliveryTask task) {
\r
112 working.remove(task.getPublishId());
\r
118 * Mark that a delivery task has expired.
\r
120 public synchronized void markExpired(DeliveryTask task) {
\r
124 * Mark that a delivery task has failed permanently.
\r
126 public synchronized void markFailNoRetry(DeliveryTask task) {
\r
127 working.remove(task.getPublishId());
\r
132 private void fdupdate() {
\r
135 if (failduration == 0) {
\r
136 failduration = dqh.getInitFailureTimer();
\r
138 resumetime = System.currentTimeMillis() + failduration;
\r
139 long maxdur = dqh.getMaxFailureTimer();
\r
140 failduration = (long)(failduration * dqh.getFailureBackoff());
\r
141 if (failduration > maxdur) {
\r
142 failduration = maxdur;
\r
147 * Mark that a delivery task has been redirected.
\r
149 public synchronized void markRedirect(DeliveryTask task) {
\r
150 working.remove(task.getPublishId());
\r
151 retry.put(task.getPublishId(), task);
\r
154 * Mark that a delivery task has temporarily failed.
\r
156 public synchronized void markFailWithRetry(DeliveryTask task) {
\r
157 working.remove(task.getPublishId());
\r
158 retry.put(task.getPublishId(), task);
\r
162 * Get the next task.
\r
164 public synchronized DeliveryTask getNext() {
\r
165 DeliveryTask ret = peekNext();
\r
168 working.put(ret.getPublishId(), ret);
\r
173 * Peek at the next task.
\r
175 public synchronized DeliveryTask peekNext() {
\r
176 long now = System.currentTimeMillis();
\r
177 long mindate = now - dqh.getExpirationTimer();
\r
179 if (now > resumetime) {
\r
186 if (todoindex >= todo.size()) {
\r
188 todo = new Vector<DeliveryTask>();
\r
189 String[] files = dir.list();
\r
190 Arrays.sort(files);
\r
191 for (String fname: files) {
\r
192 if (!fname.endsWith(".M")) {
\r
195 String fname2 = fname.substring(0, fname.length() - 2);
\r
197 int dot = fname2.indexOf('.');
\r
202 pidtime = Long.parseLong(fname2.substring(0, dot));
\r
203 } catch (Exception e) {
\r
205 if (pidtime < 1000000000000L) {
\r
208 if (working.get(fname2) != null) {
\r
211 DeliveryTask dt = retry.get(fname2);
\r
213 dt = new DeliveryTask(this, fname2);
\r
217 retry = new Hashtable<String, DeliveryTask>();
\r
219 if (todoindex < todo.size()) {
\r
220 DeliveryTask dt = todo.get(todoindex);
\r
221 if (dt.isCleaned()) {
\r
225 if (dt.getDate() >= mindate) {
\r
236 * Create a delivery queue for a given destination info
\r
238 public DeliveryQueue(DeliveryQueueHelper dqh, DestInfo di) {
\r
241 dir = new File(di.getSpool());
\r
245 * Update the destination info for this delivery queue
\r
247 public void config(DestInfo di) {
\r
251 * Get the dest info
\r
253 public DestInfo getDestInfo() {
\r
257 * Get the config manager
\r
259 public DeliveryQueueHelper getConfig() {
\r
263 * Exceptional condition occurred during delivery
\r
265 public void reportDeliveryExtra(DeliveryTask task, long sent) {
\r
266 StatusLog.logDelExtra(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getLength(), sent);
\r
269 * Message too old to deliver
\r
271 public void reportExpiry(DeliveryTask task) {
\r
272 StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "retriesExhausted", task.getAttempts());
\r
276 * Completed a delivery attempt
\r
278 public void reportStatus(DeliveryTask task, int status, String xpubid, String location) {
\r
279 if (status < 300) {
\r
280 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), di.getAuthUser(), status, xpubid);
\r
282 } else if (status < 400 && dqh.isFollowRedirects()) {
\r
283 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), di.getAuthUser(), status, location);
\r
284 if (dqh.handleRedirection(di, location, task.getFileId())) {
\r
285 markRedirect(task);
\r
287 StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
\r
288 markFailNoRetry(task);
\r
290 } else if (status < 500) {
\r
291 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), di.getAuthUser(), status, location);
\r
292 StatusLog.logExp(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), "notRetryable", task.getAttempts());
\r
293 markFailNoRetry(task);
\r
295 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), di.getAuthUser(), status, location);
\r
296 markFailWithRetry(task);
\r
300 * Delivery failed by reason of an exception
\r
302 public void reportException(DeliveryTask task, Exception exception) {
\r
303 StatusLog.logDel(task.getPublishId(), task.getFeedId(), task.getSubId(), task.getURL(), task.getMethod(), task.getCType(), task.getLength(), di.getAuthUser(), -1, exception.toString());
\r
304 dqh.handleUnreachable(di);
\r
305 markFailWithRetry(task);
\r
308 * Get the feed ID for a subscription
\r
309 * @param subid The subscription ID
\r
310 * @return The feed ID
\r
312 public String getFeedId(String subid) {
\r
313 return(dqh.getFeedId(subid));
\r
316 * Get the URL to deliver a message to given the file ID
\r
318 public String getDestURL(String fileid) {
\r
319 return(dqh.getDestURL(di, fileid));
\r
322 * Deliver files until there's a failure or there are no more
\r
325 public void run() {
\r
327 long endtime = System.currentTimeMillis() + dqh.getFairTimeLimit();
\r
328 int filestogo = dqh.getFairFileLimit();
\r
329 while ((t = getNext()) != null) {
\r
331 if (--filestogo <= 0 || System.currentTimeMillis() > endtime) {
\r
337 * Is there no work to do for this queue right now?
\r
339 public synchronized boolean isSkipSet() {
\r
340 return(peekNext() == null);
\r
343 * Reset the retry timer
\r
345 public void resetQueue() {
\r
346 resumetime = System.currentTimeMillis();
\r