summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorHenrik Andersson <henrik.b.andersson@est.tech>2019-06-04 06:58:01 +0000
committerGerrit Code Review <gerrit@onap.org>2019-06-04 06:58:01 +0000
commit2c4f790714012f3da3c8bdb86572d8e6b30e86db (patch)
tree58c260aa0faf233c7a4e4f350ef132d66b943a5c
parentc82fc8635874821324b15f9516d5ea04cf4a8704 (diff)
parentf26ab6b4e6ac57759037d2c854b7390c7afb6456 (diff)
Merge "Bugfixes, Generalizing Data File Collection to handle any type of file"
-rw-r--r--datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasks.java16
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();