Merge "Bugfixes, Generalizing Data File Collection to handle any type of file"

1 files changed, 7 insertions, 9 deletions

diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java
index bac52659..0f220fde 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java
@@ -117,10 +117,10 @@ public class ScheduledTasks {
                 .runOn(scheduler) //
                 .doOnNext(fileReadyMessage -> threadPoolQueueSize.decrementAndGet()) //
                 .flatMap(fileReadyMessage -> Flux.fromIterable(fileReadyMessage.files())) //
-                .doOnNext(fileData -> currentNumberOfTasks.incrementAndGet()) //
                 .flatMap(fileData -> createMdcContext(fileData, context)) //
                 .filter(this::isFeedConfigured) //
                 .filter(this::shouldBePublished) //
+                .doOnNext(fileData -> currentNumberOfTasks.incrementAndGet()) //
                 .flatMap(this::fetchFile, false, 1, 1) //
                 .flatMap(this::publishToDataRouter, false, 1, 1) //
                 .doOnNext(publishInfo -> deleteFile(publishInfo.getInternalLocation(), publishInfo.getContext())) //
@@ -210,21 +210,19 @@ public class ScheduledTasks {
     }
 
     private boolean shouldBePublished(FileDataWithContext fileData) {
-        boolean result = false;
         Path localFilePath = fileData.fileData.getLocalFilePath();
         if (publishedFilesCache.put(localFilePath) == null) {
             try {
-                result = !createPublishedChecker().isFilePublished(fileData.fileData.name(),
+                boolean result = !createPublishedChecker().isFilePublished(fileData.fileData.name(),
                         fileData.fileData.messageMetaData().changeIdentifier(), fileData.context);
+                return result;
             } catch (DatafileTaskException e) {
                 logger.error("Cannot check if a file {} is published", fileData.fileData.name(), e);
+                return true; // Publish it then
             }
+        } else {
+            return false;
         }
-        if (!result) {
-            currentNumberOfTasks.decrementAndGet();
-        }
-
-        return result;
     }
 
     private Mono<FilePublishInformation> fetchFile(FileDataWithContext fileData) {
@@ -269,7 +267,7 @@ public class ScheduledTasks {
     private Flux<FileReadyMessage> fetchMoreFileReadyMessages() {
         logger.info(
                 "Consuming new file ready messages, current number of tasks: {}, published files: {}, "
-                + "number of subscriptions: {}",
+                        + "number of subscriptions: {}",
                 getCurrentNumberOfTasks(), publishedFilesCache.size(), this.currentNumberOfSubscriptions.get());
 
         Map<String, String> context = MDC.getCopyOfContextMap();