diff options
23 files changed, 261 insertions, 310 deletions
diff --git a/datafile-app-server/config/application.yaml b/datafile-app-server/config/application.yaml index f2538578..8985b9b8 100644 --- a/datafile-app-server/config/application.yaml +++ b/datafile-app-server/config/application.yaml @@ -15,7 +15,7 @@ logging: org.springframework: ERROR org.springframework.data: ERROR org.springframework.web.reactive.function.client.ExchangeFunctions: ERROR - org.onap.dcaegen2.collectors.datafile: ERROR + org.onap.dcaegen2.collectors.datafile: WARN file: /var/log/ONAP/application.log app: filepath: /opt/app/datafile/config/datafile_endpoints.json diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfig.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfig.java index f89a1012..a30d2826 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfig.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/AppConfig.java @@ -112,10 +112,8 @@ public class AppConfig { setConfiguration(consumerConfiguration, publisherConfiguration, ftpesConfig); } - } catch (IOException e) { - logger.error("Problem with file loading, file: {}", filepath, e); - } catch (JsonSyntaxException e) { - logger.error("Problem with Json deserialization", e); + } catch (JsonSyntaxException | IOException e) { + logger.error("Problem with loading configuration, file: {}", filepath, e); } } diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfig.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfig.java index 52723330..d5c8b3b2 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfig.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/configuration/SchedulerConfig.java @@ -63,7 +63,7 @@ public class SchedulerConfig { private static final Logger logger = LoggerFactory.getLogger(SchedulerConfig.class); private static final Marker ENTRY = MarkerFactory.getMarker("ENTRY"); private static final Marker EXIT = MarkerFactory.getMarker("EXIT"); - private static volatile List<ScheduledFuture<?>> scheduledFutureList = new ArrayList<>(); + private static List<ScheduledFuture<?>> scheduledFutureList = new ArrayList<>(); private Map<String, String> contextMap; private final TaskScheduler taskScheduler; @@ -97,8 +97,7 @@ public class SchedulerConfig { MdcVariables.setMdcContextMap(contextMap); logger.info(EXIT, "Stopped Datafile workflow"); MDC.clear(); - return Mono.defer(() -> Mono - .just(new ResponseEntity<>("Datafile Service has already been stopped!", HttpStatus.CREATED))); + return Mono.just(new ResponseEntity<>("Datafile Service has already been stopped!", HttpStatus.CREATED)); } /** diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DmaapNotFoundException.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DmaapNotFoundException.java deleted file mode 100644 index 36279016..00000000 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/exceptions/DmaapNotFoundException.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * ============LICENSE_START====================================================================== - * Copyright (C) 2018 NOKIA Intellectual Property, 2018-2019 Nordix Foundation. All rights reserved. - * =============================================================================================== - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * ============LICENSE_END======================================================================== - */ - -package org.onap.dcaegen2.collectors.datafile.exceptions; - -/** - * @author <a href="mailto:przemyslaw.wasala@nokia.com">Przemysław Wąsala</a> on 4/13/18 - */ -public class DmaapNotFoundException extends DatafileTaskException { - - private static final long serialVersionUID = 1L; - - public DmaapNotFoundException(String message) { - super(message); - } -} diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/FileData.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/FileData.java index bdb47b2b..d0d1f91a 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/FileData.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/FileData.java @@ -37,10 +37,19 @@ import org.onap.dcaegen2.collectors.datafile.ftp.Scheme; public abstract class FileData { private static final String DATAFILE_TMPDIR = "/tmp/onap_datafile/"; + /** + * @return the file name with no path + */ public abstract String name(); + /** + * @return the URL to use to fetch the file from the PNF + */ public abstract String location(); + /** + * @return the file transfer protocol to use for fetching the file + */ public abstract Scheme scheme(); public abstract String compression(); @@ -49,17 +58,25 @@ public abstract class FileData { public abstract String fileFormatVersion(); + public abstract MessageMetaData messageMetaData(); + + /** + * @return the name of the PNF, must be unique in the network + */ + public String sourceName() { + return messageMetaData().sourceName(); + } + public String remoteFilePath() { return URI.create(location()).getPath(); } public Path getLocalFileName() { - URI uri = URI.create(location()); - return createLocalFileName(uri.getHost(), name()); + return createLocalFileName(messageMetaData().sourceName(), name()); } - public static Path createLocalFileName(String host, String fileName) { - return Paths.get(DATAFILE_TMPDIR, host + "_" + fileName); + public static Path createLocalFileName(String sourceName, String fileName) { + return Paths.get(DATAFILE_TMPDIR, sourceName + "_" + fileName); } public FileServerData fileServerData() { @@ -86,4 +103,4 @@ public abstract class FileData { } return Optional.empty(); } -}
\ No newline at end of file +} diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/FileReadyMessage.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/FileReadyMessage.java index e3293faa..9373a4f2 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/FileReadyMessage.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/model/FileReadyMessage.java @@ -31,9 +31,5 @@ import org.immutables.value.Value; @Value.Immutable @Gson.TypeAdapters public interface FileReadyMessage { - public String pnfName(); - - public MessageMetaData messageMetaData(); - public List<FileData> files(); } diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParser.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParser.java index af45cc99..a3595ecf 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParser.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParser.java @@ -100,9 +100,13 @@ public class JsonMessageParser { Optional<JsonObject> getJsonObjectFromAnArray(JsonElement element) { JsonParser jsonParser = new JsonParser(); - return element.isJsonPrimitive() ? Optional.of(jsonParser.parse(element.getAsString()).getAsJsonObject()) - : element.isJsonObject() ? Optional.of((JsonObject) element) - : Optional.of(jsonParser.parse(element.toString()).getAsJsonObject()); + if (element.isJsonPrimitive()) { + return Optional.of(jsonParser.parse(element.getAsString()).getAsJsonObject()); + } else if (element.isJsonObject()) { + return Optional.of((JsonObject) element); + } else { + return Optional.of(jsonParser.parse(element.toString()).getAsJsonObject()); + } } private Flux<FileReadyMessage> getMessagesFromJsonArray(JsonElement jsonElement) { @@ -133,18 +137,17 @@ public class JsonMessageParser { : logErrorAndReturnEmptyMessageFlux("Incorrect JsonObject - missing header. " + jsonObject)); } + private Mono<FileReadyMessage> transformMessages(JsonObject message) { Optional<MessageMetaData> optionalMessageMetaData = getMessageMetaData(message); if (optionalMessageMetaData.isPresent()) { + MessageMetaData messageMetaData = optionalMessageMetaData.get(); JsonObject notificationFields = message.getAsJsonObject(EVENT).getAsJsonObject(NOTIFICATION_FIELDS); JsonArray arrayOfNamedHashMap = notificationFields.getAsJsonArray(ARRAY_OF_NAMED_HASH_MAP); if (arrayOfNamedHashMap != null) { - List<FileData> allFileDataFromJson = getAllFileDataFromJson(arrayOfNamedHashMap); + List<FileData> allFileDataFromJson = getAllFileDataFromJson(arrayOfNamedHashMap, messageMetaData); if (!allFileDataFromJson.isEmpty()) { - MessageMetaData messageMetaData = optionalMessageMetaData.get(); return Mono.just(ImmutableFileReadyMessage.builder() // - .pnfName(messageMetaData.sourceName()) // - .messageMetaData(messageMetaData) // .files(allFileDataFromJson) // .build()); } else { @@ -152,13 +155,14 @@ public class JsonMessageParser { } } - logger.error("Unable to collect file from xNF. Missing arrayOfNamedHashMap in message. {}", message); + logger.error("VES event parsing. Missing arrayOfNamedHashMap in message. {}", message); return Mono.empty(); } - logger.error("Unable to collect file from xNF. FileReady event has incorrect JsonObject. {}", message); + logger.error("VES event parsing. FileReady event has incorrect JsonObject. {}", message); return Mono.empty(); } + private Optional<MessageMetaData> getMessageMetaData(JsonObject message) { List<String> missingValues = new ArrayList<>(); JsonObject commonEventHeader = message.getAsJsonObject(EVENT).getAsJsonObject(COMMON_EVENT_HEADER); @@ -185,7 +189,7 @@ public class JsonMessageParser { if (missingValues.isEmpty() && isChangeIdentifierCorrect(changeIdentifier) && isChangeTypeCorrect(changeType)) { return Optional.of(messageMetaData); } else { - String errorMessage = "Unable to collect file from xNF."; + String errorMessage = "VES event parsing."; if (!missingValues.isEmpty()) { errorMessage += " Missing data: " + missingValues; } @@ -206,11 +210,11 @@ public class JsonMessageParser { return FILE_READY_CHANGE_IDENTIFIER.equals(changeIdentifier); } - private List<FileData> getAllFileDataFromJson(JsonArray arrayOfAdditionalFields) { + private List<FileData> getAllFileDataFromJson(JsonArray arrayOfAdditionalFields, MessageMetaData messageMetaData) { List<FileData> res = new ArrayList<>(); for (int i = 0; i < arrayOfAdditionalFields.size(); i++) { JsonObject fileInfo = (JsonObject) arrayOfAdditionalFields.get(i); - Optional<FileData> fileData = getFileDataFromJson(fileInfo); + Optional<FileData> fileData = getFileDataFromJson(fileInfo, messageMetaData); if (fileData.isPresent()) { res.add(fileData.get()); @@ -219,7 +223,7 @@ public class JsonMessageParser { return res; } - private Optional<FileData> getFileDataFromJson(JsonObject fileInfo) { + private Optional<FileData> getFileDataFromJson(JsonObject fileInfo, MessageMetaData messageMetaData) { logger.trace("starting to getFileDataFromJson!"); List<String> missingValues = new ArrayList<>(); @@ -230,7 +234,7 @@ public class JsonMessageParser { try { scheme = Scheme.getSchemeFromString(URI.create(location).getScheme()); } catch (Exception e) { - logger.error("Unable to collect file from xNF.", e); + logger.error("VES event parsing.", e); return Optional.empty(); } FileData fileData = ImmutableFileData.builder() // @@ -240,12 +244,12 @@ public class JsonMessageParser { .location(location) // .scheme(scheme) // .compression(getValueFromJson(data, COMPRESSION, missingValues)) // + .messageMetaData(messageMetaData) .build(); if (missingValues.isEmpty()) { return Optional.of(fileData); } - logger.error("Unable to collect file from xNF. File information wrong. Missing data: {} Data: {}", - missingValues, fileInfo); + logger.error("VES event parsing. File information wrong. Missing data: {} Data: {}", missingValues, fileInfo); return Optional.empty(); } diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTask.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTask.java index f6daf733..49e2f01e 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTask.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTask.java @@ -30,8 +30,9 @@ import org.onap.dcaegen2.services.sdk.rest.services.dmaap.client.service.consume import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.web.reactive.function.client.WebClient; -import reactor.core.publisher.Mono; + import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; /** * @author <a href="mailto:henrik.b.andersson@est.tech">Henrik Andersson</a> @@ -39,25 +40,21 @@ import reactor.core.publisher.Flux; public class DMaaPMessageConsumerTask { private static final Logger logger = LoggerFactory.getLogger(DMaaPMessageConsumerTask.class); - private AppConfig datafileAppConfig; - private JsonMessageParser jsonMessageParser; - private DMaaPConsumerReactiveHttpClient dmaaPConsumerReactiveHttpClient; + private final JsonMessageParser jsonMessageParser; + private final DMaaPConsumerReactiveHttpClient dmaaPConsumerReactiveHttpClient; public DMaaPMessageConsumerTask(AppConfig datafileAppConfig) { - this.datafileAppConfig = datafileAppConfig; this.jsonMessageParser = new JsonMessageParser(); + this.dmaaPConsumerReactiveHttpClient = createHttpClient(datafileAppConfig); } - protected DMaaPMessageConsumerTask(AppConfig datafileAppConfig, - DMaaPConsumerReactiveHttpClient dmaaPConsumerReactiveHttpClient, + protected DMaaPMessageConsumerTask(DMaaPConsumerReactiveHttpClient dmaaPConsumerReactiveHttpClient, JsonMessageParser messageParser) { - this.datafileAppConfig = datafileAppConfig; this.dmaaPConsumerReactiveHttpClient = dmaaPConsumerReactiveHttpClient; this.jsonMessageParser = messageParser; } public Flux<FileReadyMessage> execute() { - dmaaPConsumerReactiveHttpClient = resolveClient(); logger.trace("execute called"); return consume((dmaaPConsumerReactiveHttpClient.getDMaaPConsumerResponse())); } @@ -67,15 +64,10 @@ public class DMaaPMessageConsumerTask { return jsonMessageParser.getMessagesFromJson(message); } - protected DmaapConsumerConfiguration resolveConfiguration() { - return datafileAppConfig.getDmaapConsumerConfiguration(); + private static DMaaPConsumerReactiveHttpClient createHttpClient(AppConfig datafileAppConfig) { + DmaapConsumerConfiguration config = datafileAppConfig.getDmaapConsumerConfiguration(); + WebClient client = new DmaapReactiveWebClient().fromConfiguration(config).build(); + return new DMaaPConsumerReactiveHttpClient(config, client); } - protected DMaaPConsumerReactiveHttpClient resolveClient() { - return new DMaaPConsumerReactiveHttpClient(resolveConfiguration(), buildWebClient()); - } - - protected WebClient buildWebClient() { - return new DmaapReactiveWebClient().fromConfiguration(resolveConfiguration()).build(); - } } diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java index 0fef9ab4..4207d1fc 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java @@ -84,12 +84,13 @@ public class DataRouterPublisher { * @param model information about the file to publish * @param numRetries the maximal number of retries if the publishing fails * @param firstBackoff the time to delay the first retry - * @return the HTTP response status as a string + * @param contextMap tracing context variables + * @return the (same) ConsumerDmaapModel */ public Mono<ConsumerDmaapModel> execute(ConsumerDmaapModel model, long numRetries, Duration firstBackoff, Map<String, String> contextMap) { MdcVariables.setMdcContextMap(contextMap); - logger.trace("Method called with arg {}", model); + logger.trace("Publish called with arg {}", model); dmaapProducerReactiveHttpClient = resolveClient(); return Mono.just(model) @@ -117,7 +118,7 @@ public class DataRouterPublisher { logger.trace(response.toString()); return Mono.just(HttpStatus.valueOf(response.getStatusLine().getStatusCode())); } catch (Exception e) { - logger.error("Unable to send file to DataRouter. Data: {}", consumerDmaapModel.getInternalLocation(), e); + logger.warn("Unable to send file to DataRouter. Data: {}", consumerDmaapModel.getInternalLocation(), e); return Mono.error(e); } } diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java index fb27a579..8849b45e 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollector.java @@ -49,31 +49,31 @@ public class FileCollector { this.datafileAppConfig = datafileAppConfig; } - public Mono<ConsumerDmaapModel> execute(FileData fileData, MessageMetaData metaData, long maxNumberOfRetries, - Duration firstBackoffTimeout, Map<String, String> contextMap) { + public Mono<ConsumerDmaapModel> execute(FileData fileData, long maxNumberOfRetries, Duration firstBackoffTimeout, + Map<String, String> contextMap) { MdcVariables.setMdcContextMap(contextMap); logger.trace("Entering execute with {}", fileData); return Mono.just(fileData) // .cache() // - .flatMap(fd -> collectFile(fileData, metaData, contextMap)) // + .flatMap(fd -> collectFile(fileData, contextMap)) // .retryBackoff(maxNumberOfRetries, firstBackoffTimeout); } - private Mono<ConsumerDmaapModel> collectFile(FileData fileData, MessageMetaData metaData, - Map<String, String> contextMap) { + private Mono<ConsumerDmaapModel> collectFile(FileData fileData, Map<String, String> contextMap) { MdcVariables.setMdcContextMap(contextMap); - logger.trace("starting to collectFile"); + logger.trace("starting to collectFile {}", fileData.name()); final String remoteFile = fileData.remoteFilePath(); final Path localFile = fileData.getLocalFileName(); try (FileCollectClient currentClient = createClient(fileData)) { + currentClient.open(); localFile.getParent().toFile().mkdir(); // Create parent directories currentClient.collectFile(remoteFile, localFile); - return Mono.just(getConsumerDmaapModel(fileData, metaData, localFile)); + return Mono.just(getConsumerDmaapModel(fileData, localFile)); } catch (Exception throwable) { - logger.warn("Failed to download file: {}, reason: {}", fileData.name(), throwable); + logger.warn("Failed to download file: {} {}, reason: {}", fileData.sourceName(), fileData.name(), throwable.toString()); return Mono.error(throwable); } } @@ -89,9 +89,9 @@ public class FileCollector { } } - private ConsumerDmaapModel getConsumerDmaapModel(FileData fileData, MessageMetaData metaData, Path localFile) { + private ConsumerDmaapModel getConsumerDmaapModel(FileData fileData, Path localFile) { String location = fileData.location(); - + MessageMetaData metaData = fileData.messageMetaData(); return ImmutableConsumerDmaapModel.builder() // .productName(metaData.productName()) // .vendorName(metaData.vendorName()) // @@ -108,16 +108,13 @@ public class FileCollector { .build(); } - SftpClient createSftpClient(FileData fileData) throws DatafileTaskException { - SftpClient client = new SftpClient(fileData.fileServerData()); - client.open(); - return client; + protected SftpClient createSftpClient(FileData fileData) { + return new SftpClient(fileData.fileServerData()); } - FtpsClient createFtpsClient(FileData fileData) throws DatafileTaskException { + protected FtpsClient createFtpsClient(FileData fileData) { FtpesConfig config = datafileAppConfig.getFtpesConfiguration(); - FtpsClient client = new FtpsClient(fileData.fileServerData()); - client.open(config.keyCert(), config.keyPassword(), Paths.get(config.trustedCA()), config.trustedCAPassword()); - return client; + return new FtpsClient(fileData.fileServerData(), config.keyCert(), config.keyPassword(), + Paths.get(config.trustedCA()), config.trustedCAPassword()); } } diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/PublishedChecker.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/PublishedChecker.java index 0729caa0..41f8e3cd 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/PublishedChecker.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/PublishedChecker.java @@ -54,7 +54,7 @@ public class PublishedChecker { private final Logger logger = LoggerFactory.getLogger(this.getClass()); - private AppConfig appConfig; + private final AppConfig appConfig; /** * Constructor. @@ -88,12 +88,13 @@ public class PublishedChecker { HttpResponse response = producerClient.getDmaapProducerResponseWithCustomTimeout(getRequest, 2000, contextMap); - logger.trace(response.toString()); + logger.trace("{}", response); int status = response.getStatusLine().getStatusCode(); HttpEntity entity = response.getEntity(); - InputStream content = entity.getContent(); - String body = IOUtils.toString(content); - return HttpStatus.SC_OK == status && !"[]".equals(body); + try (InputStream content = entity.getContent()) { + String body = IOUtils.toString(content); + return HttpStatus.SC_OK == status && !"[]".equals(body); + } } catch (Exception e) { logger.warn("Unable to check if file has been published.", e); return false; 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 d41e5c25..b4096c73 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 @@ -20,8 +20,6 @@ import java.nio.file.Files; import java.nio.file.Path; import java.time.Duration; import java.time.Instant; -import java.util.ArrayList; -import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; @@ -29,7 +27,6 @@ import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig; import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel; import org.onap.dcaegen2.collectors.datafile.model.FileData; import org.onap.dcaegen2.collectors.datafile.model.FileReadyMessage; -import org.onap.dcaegen2.collectors.datafile.model.MessageMetaData; import org.onap.dcaegen2.collectors.datafile.model.logging.MdcVariables; import org.onap.dcaegen2.collectors.datafile.service.PublishedFileCache; import org.slf4j.Logger; @@ -44,32 +41,24 @@ import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; /** - * This implements the main flow of the data file collector. Fetch file ready events from the message router, fetch new - * files from the PNF publish these in the data router. + * This implements the main flow of the data file collector. Fetch file ready events from the + * message router, fetch new files from the PNF publish these in the data router. */ @Component public class ScheduledTasks { - private static final int MAX_NUMBER_OF_CONCURRENT_TASKS = 200; - private static final int MAX_ILDLE_THREAD_TIME_TO_LIVE_SECONDS = 10; - - /** Data needed for fetching of one file. */ - private class FileCollectionData { - final FileData fileData; - final MessageMetaData metaData; - - FileCollectionData(FileData fd, MessageMetaData metaData) { - this.fileData = fd; - this.metaData = metaData; - } - } + private static final int NUMBER_OF_WORKER_THREADS = 100; + private static final int MAX_TASKS_FOR_POLLING = 50; + private static final long DATA_ROUTER_MAX_RETRIES = 5; + private static final Duration DATA_ROUTER_INITIAL_RETRY_TIMEOUT = Duration.ofSeconds(2); + private static final long FILE_TRANSFER_MAX_RETRIES = 3; + private static final Duration FILE_TRANSFER_INITIAL_RETRY_TIMEOUT = Duration.ofSeconds(5); private static final Logger logger = LoggerFactory.getLogger(ScheduledTasks.class); private final AppConfig applicationConfiguration; private final AtomicInteger currentNumberOfTasks = new AtomicInteger(); - private final Scheduler scheduler = - Schedulers.newElastic("DataFileCollector", MAX_ILDLE_THREAD_TIME_TO_LIVE_SECONDS); + private final Scheduler scheduler = Schedulers.newParallel("FileCollectorWorker", NUMBER_OF_WORKER_THREADS); PublishedFileCache alreadyPublishedFiles = new PublishedFileCache(); /** @@ -86,21 +75,25 @@ public class ScheduledTasks { * Main function for scheduling for the file collection Workflow. */ public void scheduleMainDatafileEventTask(Map<String, String> contextMap) { - MdcVariables.setMdcContextMap(contextMap); - logger.trace("Execution of tasks was registered"); - applicationConfiguration.loadConfigurationFromFile(); - createMainTask(contextMap).subscribe(model -> onSuccess(model, contextMap), thr -> onError(thr, contextMap), - () -> onComplete(contextMap)); + try { + MdcVariables.setMdcContextMap(contextMap); + logger.trace("Execution of tasks was registered"); + applicationConfiguration.loadConfigurationFromFile(); + createMainTask(contextMap).subscribe(model -> onSuccess(model, contextMap), thr -> onError(thr, contextMap), + () -> onComplete(contextMap)); + } catch (Exception e) { + logger.error("Unexpected exception: ", e); + } } Flux<ConsumerDmaapModel> createMainTask(Map<String, String> contextMap) { return fetchMoreFileReadyMessages() // - .parallel(getParallelism()) // Each FileReadyMessage in a separate thread + .parallel(NUMBER_OF_WORKER_THREADS) // Each FileReadyMessage in a separate thread .runOn(scheduler) // - .flatMap(this::createFileCollectionTask) // + .flatMap(fileReadyMessage -> Flux.fromIterable(fileReadyMessage.files())) // .filter(fileData -> shouldBePublished(fileData, contextMap)) // .doOnNext(fileData -> currentNumberOfTasks.incrementAndGet()) // - .flatMap(fileData -> collectFileFromXnf(fileData, contextMap)) // + .flatMap(fileData -> fetchFile(fileData, contextMap)) // .flatMap(model -> publishToDataRouter(model, contextMap)) // .doOnNext(model -> deleteFile(model.getInternalLocation(), contextMap)) // .doOnNext(model -> currentNumberOfTasks.decrementAndGet()) // @@ -114,63 +107,61 @@ public class ScheduledTasks { alreadyPublishedFiles.purge(now); } - private void onComplete(Map<String, String> contextMap) { - MdcVariables.setMdcContextMap(contextMap); - logger.info("Datafile tasks have been completed"); + protected PublishedChecker createPublishedChecker() { + return new PublishedChecker(applicationConfiguration); } - private void onSuccess(ConsumerDmaapModel model, Map<String, String> contextMap) { - MdcVariables.setMdcContextMap(contextMap); - logger.info("Datafile consumed tasks {}", model.getInternalLocation()); + protected int getCurrentNumberOfTasks() { + return currentNumberOfTasks.get(); } - private void onError(Throwable throwable, Map<String, String> contextMap) { - MdcVariables.setMdcContextMap(contextMap); - logger.error("Chain of tasks have been aborted due to errors in Datafile workflow {}", throwable); + protected DMaaPMessageConsumerTask createConsumerTask() { + return new DMaaPMessageConsumerTask(this.applicationConfiguration); } - private int getParallelism() { - if (MAX_NUMBER_OF_CONCURRENT_TASKS - getCurrentNumberOfTasks() > 0) { - return MAX_NUMBER_OF_CONCURRENT_TASKS - getCurrentNumberOfTasks(); - } else { - return 1; // We need at least one rail/thread - } + protected FileCollector createFileCollector() { + return new FileCollector(applicationConfiguration); + } + + protected DataRouterPublisher createDataRouterPublisher() { + return new DataRouterPublisher(applicationConfiguration); } - private Flux<FileCollectionData> createFileCollectionTask(FileReadyMessage availableFiles) { - List<FileCollectionData> fileCollects = new ArrayList<>(); + private void onComplete(Map<String, String> contextMap) { + MdcVariables.setMdcContextMap(contextMap); + logger.trace("Datafile tasks have been completed"); + } - for (FileData fileData : availableFiles.files()) { - fileCollects.add(new FileCollectionData(fileData, availableFiles.messageMetaData())); - } - return Flux.fromIterable(fileCollects); + private synchronized void onSuccess(ConsumerDmaapModel model, Map<String, String> contextMap) { + MdcVariables.setMdcContextMap(contextMap); + logger.info("Datafile file published {}", model.getInternalLocation()); } - private boolean shouldBePublished(FileCollectionData task, Map<String, String> contextMap) { + private void onError(Throwable throwable, Map<String, String> contextMap) { + MdcVariables.setMdcContextMap(contextMap); + logger.error("Chain of tasks have been aborted due to errors in Datafile workflow {}", throwable.toString()); + } + + private boolean shouldBePublished(FileData fileData, Map<String, String> contextMap) { boolean result = false; - Path localFileName = task.fileData.getLocalFileName(); + Path localFileName = fileData.getLocalFileName(); if (alreadyPublishedFiles.put(localFileName) == null) { result = !createPublishedChecker().execute(localFileName.getFileName().toString(), contextMap); } return result; } - private Mono<ConsumerDmaapModel> collectFileFromXnf(FileCollectionData fileCollect, - Map<String, String> contextMap) { - final long maxNUmberOfRetries = 3; - final Duration initialRetryTimeout = Duration.ofSeconds(5); - + private Mono<ConsumerDmaapModel> fetchFile(FileData fileData, Map<String, String> contextMap) { MdcVariables.setMdcContextMap(contextMap); return createFileCollector() - .execute(fileCollect.fileData, fileCollect.metaData, maxNUmberOfRetries, initialRetryTimeout, - contextMap) - .onErrorResume(exception -> handleCollectFailure(fileCollect.fileData, contextMap)); + .execute(fileData, FILE_TRANSFER_MAX_RETRIES, FILE_TRANSFER_INITIAL_RETRY_TIMEOUT, contextMap) + .onErrorResume(exception -> handleFetchFileFailure(fileData, contextMap)); } - private Mono<ConsumerDmaapModel> handleCollectFailure(FileData fileData, Map<String, String> contextMap) { + private Mono<ConsumerDmaapModel> handleFetchFileFailure(FileData fileData, Map<String, String> contextMap) { MdcVariables.setMdcContextMap(contextMap); Path localFileName = fileData.getLocalFileName(); - logger.error("File fetching failed: {}", localFileName); + logger.error("File fetching failed, fileData {}", fileData); deleteFile(localFileName, contextMap); alreadyPublishedFiles.remove(localFileName); currentNumberOfTasks.decrementAndGet(); @@ -178,19 +169,16 @@ public class ScheduledTasks { } private Mono<ConsumerDmaapModel> publishToDataRouter(ConsumerDmaapModel model, Map<String, String> contextMap) { - final long maxNumberOfRetries = 3; - final Duration initialRetryTimeout = Duration.ofSeconds(5); - - MdcVariables.setMdcContextMap(contextMap); - return createDataRouterPublisher().execute(model, maxNumberOfRetries, initialRetryTimeout, contextMap) - .onErrorResume(exception -> handlePublishFailure(model, exception, contextMap)); + + return createDataRouterPublisher() + .execute(model, DATA_ROUTER_MAX_RETRIES, DATA_ROUTER_INITIAL_RETRY_TIMEOUT, contextMap) + .onErrorResume(exception -> handlePublishFailure(model, contextMap)); } - private Mono<ConsumerDmaapModel> handlePublishFailure(ConsumerDmaapModel model, Throwable exception, - Map<String, String> contextMap) { + private Mono<ConsumerDmaapModel> handlePublishFailure(ConsumerDmaapModel model, Map<String, String> contextMap) { MdcVariables.setMdcContextMap(contextMap); - logger.error("File publishing failed: {}, exception: {}", model.getName(), exception); + logger.error("File publishing failed: {}", model); Path internalFileName = model.getInternalLocation(); deleteFile(internalFileName, contextMap); alreadyPublishedFiles.remove(internalFileName); @@ -202,8 +190,9 @@ public class ScheduledTasks { * Fetch more messages from the message router. This is done in a polling/blocking fashion. */ private Flux<FileReadyMessage> fetchMoreFileReadyMessages() { - logger.trace("Consuming new file ready messages, current number of tasks: {}", getCurrentNumberOfTasks()); - if (getCurrentNumberOfTasks() > MAX_NUMBER_OF_CONCURRENT_TASKS) { + logger.info("Consuming new file ready messages, current number of tasks: {}", getCurrentNumberOfTasks()); + if (getCurrentNumberOfTasks() > MAX_TASKS_FOR_POLLING) { + logger.info("Skipping, current number of tasks: {}", getCurrentNumberOfTasks()); return Flux.empty(); } @@ -215,7 +204,8 @@ public class ScheduledTasks { private Flux<FileReadyMessage> handleConsumeMessageFailure(Throwable exception, Map<String, String> contextMap) { MdcVariables.setMdcContextMap(contextMap); - logger.error("Polling for file ready message failed, exception: {}", exception); + logger.error("Polling for file ready message failed, exception: {}, config: {}", exception.toString(), + this.applicationConfiguration.getDmaapConsumerConfiguration()); return Flux.empty(); } @@ -229,24 +219,4 @@ public class ScheduledTasks { } } - PublishedChecker createPublishedChecker() { - return new PublishedChecker(applicationConfiguration); - } - - int getCurrentNumberOfTasks() { - return currentNumberOfTasks.get(); - } - - DMaaPMessageConsumerTask createConsumerTask() { - return new DMaaPMessageConsumerTask(this.applicationConfiguration); - } - - FileCollector createFileCollector() { - return new FileCollector(applicationConfiguration); - } - - DataRouterPublisher createDataRouterPublisher() { - return new DataRouterPublisher(applicationConfiguration); - } - } diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FileDataTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FileDataTest.java index 1f5827c8..84c5e07b 100644 --- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FileDataTest.java +++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/model/FileDataTest.java @@ -44,6 +44,20 @@ public class FileDataTest { private static final String LOCATION_WITHOUT_USER = FTPES_SCHEME + SERVER_ADDRESS + ":" + PORT_22 + REMOTE_FILE_LOCATION; + + private MessageMetaData messageMetaData() { + return ImmutableMessageMetaData.builder() + .productName("PRODUCT_NAME") + .vendorName("VENDOR_NAME") + .lastEpochMicrosec("LAST_EPOCH_MICROSEC") + .sourceName("SOURCE_NAME") + .startEpochMicrosec("START_EPOCH_MICROSEC") + .timeZoneOffset("TIME_ZONE_OFFSET") + .changeIdentifier("PM_MEAS_CHANGE_IDENTIFIER") + .changeType("FILE_READY_CHANGE_TYPE") + .build(); + } + private FileData properFileDataWithUser() { // @formatter:off return ImmutableFileData.builder() @@ -53,6 +67,7 @@ public class FileDataTest { .fileFormatType("type") .fileFormatVersion("version") .scheme(Scheme.FTPS) + .messageMetaData(messageMetaData()) .build(); // @formatter:on } @@ -66,6 +81,7 @@ public class FileDataTest { .fileFormatType("type") .fileFormatVersion("version") .scheme(Scheme.FTPS) + .messageMetaData(messageMetaData()) .build(); // @formatter:on } diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java index b33180fa..b8aa7da2 100644 --- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java +++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/service/JsonMessageParserTest.java @@ -29,7 +29,6 @@ import java.util.Optional; import org.junit.jupiter.api.Test; import org.mockito.Mockito; -import org.onap.dcaegen2.collectors.datafile.exceptions.DmaapNotFoundException; import org.onap.dcaegen2.collectors.datafile.ftp.Scheme; import org.onap.dcaegen2.collectors.datafile.model.FileData; import org.onap.dcaegen2.collectors.datafile.model.FileReadyMessage; @@ -67,7 +66,7 @@ class JsonMessageParserTest { private static final String NOTIFICATION_FIELDS_VERSION = "1.0"; @Test - void whenPassingCorrectJson_oneFileReadyMessage() throws DmaapNotFoundException { + void whenPassingCorrectJson_oneFileReadyMessage() { AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder() // .name(PM_FILE_NAME) // .location(LOCATION) // @@ -100,12 +99,11 @@ class JsonMessageParserTest { .compression(GZIP_COMPRESSION) // .fileFormatType(FILE_FORMAT_TYPE) // .fileFormatVersion(FILE_FORMAT_VERSION) // + .messageMetaData(messageMetaData) .build(); List<FileData> files = new ArrayList<>(); files.add(expectedFileData); FileReadyMessage expectedMessage = ImmutableFileReadyMessage.builder() - .pnfName(SOURCE_NAME) // - .messageMetaData(messageMetaData) // .files(files) // .build(); @@ -121,7 +119,7 @@ class JsonMessageParserTest { } @Test - void whenPassingCorrectJsonWithTwoEvents_twoMessages() throws DmaapNotFoundException { + void whenPassingCorrectJsonWithTwoEvents_twoMessages() { AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder() // .name(PM_FILE_NAME) // .location(LOCATION) // @@ -154,12 +152,11 @@ class JsonMessageParserTest { .compression(GZIP_COMPRESSION) // .fileFormatType(FILE_FORMAT_TYPE) // .fileFormatVersion(FILE_FORMAT_VERSION) // + .messageMetaData(messageMetaData) .build(); List<FileData> files = new ArrayList<>(); files.add(expectedFileData); FileReadyMessage expectedMessage = ImmutableFileReadyMessage.builder() // - .pnfName(SOURCE_NAME) // - .messageMetaData(messageMetaData) // .files(files) // .build(); @@ -202,7 +199,7 @@ class JsonMessageParserTest { } @Test - void whenPassingCorrectJsonWithTwoEventsFirstNoHeader_oneFileDatan() throws DmaapNotFoundException { + void whenPassingCorrectJsonWithTwoEventsFirstNoHeader_oneFileDatan() { AdditionalField additionalField = new JsonMessage.AdditionalFieldBuilder() // .name(PM_FILE_NAME) // .location(LOCATION) // @@ -235,12 +232,11 @@ class JsonMessageParserTest { .compression(GZIP_COMPRESSION) // .fileFormatType(FILE_FORMAT_TYPE) // .fileFormatVersion(FILE_FORMAT_VERSION) // + .messageMetaData(messageMetaData) .build(); List<FileData> files = new ArrayList<>(); files.add(expectedFileData); FileReadyMessage expectedMessage = ImmutableFileReadyMessage.builder() // - .pnfName(SOURCE_NAME) // - .messageMetaData(messageMetaData) // .files(files) // .build(); @@ -421,12 +417,11 @@ class JsonMessageParserTest { .compression(GZIP_COMPRESSION) // .fileFormatType(FILE_FORMAT_TYPE) // .fileFormatVersion(FILE_FORMAT_VERSION) // + .messageMetaData(messageMetaData) .build(); List<FileData> files = new ArrayList<>(); files.add(expectedFileData); FileReadyMessage expectedMessage = ImmutableFileReadyMessage.builder() // - .pnfName(SOURCE_NAME) // - .messageMetaData(messageMetaData) // .files(files) // .build(); diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTaskImplTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTaskImplTest.java index a695e20d..98c7dc32 100644 --- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTaskImplTest.java +++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DMaaPMessageConsumerTaskImplTest.java @@ -20,7 +20,6 @@ package org.onap.dcaegen2.collectors.datafile.tasks; -import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; @@ -156,13 +155,12 @@ public class DMaaPMessageConsumerTaskImplTest { .compression(GZIP_COMPRESSION) // .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) // .fileFormatVersion(FILE_FORMAT_VERSION) // + .messageMetaData(messageMetaData) .build(); List<FileData> files = new ArrayList<>(); files.add(ftpesFileData); expectedFtpesMessage = ImmutableFileReadyMessage.builder() // - .pnfName(SOURCE_NAME) // - .messageMetaData(messageMetaData) // .files(files) // .build(); @@ -187,6 +185,7 @@ public class DMaaPMessageConsumerTaskImplTest { .compression(GZIP_COMPRESSION) // .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) // .fileFormatVersion(FILE_FORMAT_VERSION) // + .messageMetaData(messageMetaData) .build(); ImmutableConsumerDmaapModel consumerDmaapModel = ImmutableConsumerDmaapModel.builder() // @@ -208,8 +207,6 @@ public class DMaaPMessageConsumerTaskImplTest { files = new ArrayList<>(); files.add(sftpFileData); expectedSftpMessage = ImmutableFileReadyMessage.builder() // - .pnfName(SOURCE_NAME) // - .messageMetaData(messageMetaData) // .files(files) // .build(); } @@ -264,8 +261,6 @@ public class DMaaPMessageConsumerTaskImplTest { .thenReturn(Flux.error(new DatafileTaskException("problemas"))); } - messageConsumerTask = spy(new DMaaPMessageConsumerTask(appConfig, httpClientMock, jsonMessageParserMock)); - when(messageConsumerTask.resolveConfiguration()).thenReturn(dmaapConsumerConfiguration); - doReturn(httpClientMock).when(messageConsumerTask).resolveClient(); + messageConsumerTask = spy(new DMaaPMessageConsumerTask(httpClientMock, jsonMessageParserMock)); } } diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java index ed8b93f1..fe867738 100644 --- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java +++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisherTest.java @@ -86,13 +86,11 @@ class DataRouterPublisherTest { private static final String FEED_ID = "1"; private static final String FILE_CONTENT = "Just a string."; - private static final Map<String, String> CONTEXT_MAP = new HashMap<>(); - private static ConsumerDmaapModel consumerDmaapModel; private static DmaapProducerReactiveHttpClient httpClientMock; private static AppConfig appConfig; private static DmaapPublisherConfiguration publisherConfigurationMock = mock(DmaapPublisherConfiguration.class); - + private final Map<String, String> contextMap = new HashMap<>(); private static DataRouterPublisher publisherTaskUnderTestSpy; /** @@ -125,9 +123,8 @@ class DataRouterPublisherTest { @Test public void whenPassedObjectFits_ReturnsCorrectStatus() throws Exception { prepareMocksForTests(null, Integer.valueOf(HttpStatus.OK.value())); - StepVerifier - .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), CONTEXT_MAP)) + .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), contextMap)) .expectNext(consumerDmaapModel) // .verifyComplete(); @@ -170,7 +167,7 @@ class DataRouterPublisherTest { prepareMocksForTests(new DatafileTaskException("Error"), HttpStatus.OK.value()); StepVerifier - .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 2, Duration.ofSeconds(0), CONTEXT_MAP)) + .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 2, Duration.ofSeconds(0), contextMap)) .expectNext(consumerDmaapModel) // .verifyComplete(); } @@ -181,7 +178,7 @@ class DataRouterPublisherTest { Integer.valueOf(HttpStatus.OK.value())); StepVerifier - .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), CONTEXT_MAP)) + .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), contextMap)) .expectNext(consumerDmaapModel) // .verifyComplete(); @@ -197,7 +194,7 @@ class DataRouterPublisherTest { Integer.valueOf((HttpStatus.BAD_GATEWAY.value()))); StepVerifier - .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), CONTEXT_MAP)) + .create(publisherTaskUnderTestSpy.execute(consumerDmaapModel, 1, Duration.ofSeconds(0), contextMap)) .expectErrorMessage("Retries exhausted: 1/1") // .verify(); diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java index fb49c860..6e17f27b 100644 --- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java +++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/FileCollectorTest.java @@ -67,7 +67,7 @@ public class FileCollectorTest { private static final int PORT_22 = 22; private static final String PM_FILE_NAME = "A20161224.1030-1045.bin.gz"; private static final String REMOTE_FILE_LOCATION = "/ftp/rop/" + PM_FILE_NAME; - private static final Path LOCAL_FILE_LOCATION = FileData.createLocalFileName(SERVER_ADDRESS, PM_FILE_NAME); + private static final Path LOCAL_FILE_LOCATION = FileData.createLocalFileName(SOURCE_NAME, PM_FILE_NAME); private static final String USER = "usr"; private static final String PWD = "pwd"; private static final String FTPES_LOCATION = @@ -93,6 +93,7 @@ public class FileCollectorTest { private FtpsClient ftpsClientMock = mock(FtpsClient.class); private SftpClient sftpClientMock = mock(SftpClient.class); + private final Map<String, String> contextMap = new HashMap<>(); private MessageMetaData createMessageMetaData() { @@ -119,6 +120,7 @@ public class FileCollectorTest { .fileFormatType(MEAS_COLLECT_FILE_FORMAT_TYPE) .fileFormatVersion(FILE_FORMAT_VERSION) .scheme(scheme) + .messageMetaData(createMessageMetaData()) .build(); // @formatter:on } @@ -160,11 +162,11 @@ public class FileCollectorTest { ConsumerDmaapModel expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(FTPES_LOCATION_NO_PORT); - Map<String, String> contextMap = new HashMap<>(); StepVerifier.create( - collectorUndetTest.execute(fileData, createMessageMetaData(), 3, Duration.ofSeconds(0), contextMap)) + collectorUndetTest.execute(fileData, 3, Duration.ofSeconds(0), contextMap)) .expectNext(expectedConsumerDmaapModel).verifyComplete(); + verify(ftpsClientMock, times(1)).open(); verify(ftpsClientMock, times(1)).collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION); verify(ftpsClientMock, times(1)).close(); @@ -176,12 +178,12 @@ public class FileCollectorTest { FileCollector collectorUndetTest = spy(new FileCollector(appConfigMock)); doReturn(sftpClientMock).when(collectorUndetTest).createSftpClient(any()); + FileData fileData = createFileData(SFTP_LOCATION_NO_PORT, Scheme.SFTP); ConsumerDmaapModel expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(SFTP_LOCATION_NO_PORT); - Map<String, String> contextMap = new HashMap<>(); StepVerifier - .create(collectorUndetTest.execute(fileData, createMessageMetaData(), 3, Duration.ofSeconds(0), + .create(collectorUndetTest.execute(fileData, 3, Duration.ofSeconds(0), contextMap)) .expectNext(expectedConsumerDmaapModel) // .verifyComplete(); @@ -191,11 +193,12 @@ public class FileCollectorTest { expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(SFTP_LOCATION); StepVerifier - .create(collectorUndetTest.execute(fileData, createMessageMetaData(), 3, Duration.ofSeconds(0), + .create(collectorUndetTest.execute(fileData, 3, Duration.ofSeconds(0), contextMap)) .expectNext(expectedConsumerDmaapModel) // .verifyComplete(); + verify(sftpClientMock, times(2)).open(); verify(sftpClientMock, times(2)).collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION); verify(sftpClientMock, times(2)).close(); verifyNoMoreInteractions(sftpClientMock); @@ -210,9 +213,8 @@ public class FileCollectorTest { doThrow(new DatafileTaskException("Unable to collect file.")).when(ftpsClientMock) .collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION); - Map<String, String> contextMap = new HashMap<>(); StepVerifier.create( - collectorUndetTest.execute(fileData, createMessageMetaData(), 3, Duration.ofSeconds(0), contextMap)) + collectorUndetTest.execute(fileData, 3, Duration.ofSeconds(0), contextMap)) .expectErrorMessage("Retries exhausted: 3/3").verify(); verify(ftpsClientMock, times(4)).collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION); @@ -228,9 +230,9 @@ public class FileCollectorTest { ConsumerDmaapModel expectedConsumerDmaapModel = createExpectedConsumerDmaapModel(FTPES_LOCATION_NO_PORT); FileData fileData = createFileData(FTPES_LOCATION_NO_PORT, Scheme.FTPS); - Map<String, String> contextMap = new HashMap<>(); + StepVerifier.create( - collectorUndetTest.execute(fileData, createMessageMetaData(), 3, Duration.ofSeconds(0), contextMap)) + collectorUndetTest.execute(fileData, 3, Duration.ofSeconds(0), contextMap)) .expectNext(expectedConsumerDmaapModel).verifyComplete(); verify(ftpsClientMock, times(2)).collectFile(REMOTE_FILE_LOCATION, LOCAL_FILE_LOCATION); diff --git a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java index d781cea3..f6beae02 100644 --- a/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java +++ b/datafile-app-server/src/test/java/org/onap/dcaegen2/collectors/datafile/tasks/ScheduledTasksTest.java @@ -30,8 +30,10 @@ import static org.mockito.Mockito.verifyNoMoreInteractions; import java.nio.file.Paths; import java.time.Duration; +import java.util.HashMap; import java.util.LinkedList; import java.util.List; +import java.util.Map; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -64,6 +66,7 @@ public class ScheduledTasksTest { private PublishedChecker publishedCheckerMock; private FileCollector fileCollectorMock; private DataRouterPublisher dataRouterMock; + private Map<String, String> contextMap = new HashMap<String, String>(); @BeforeEach private void setUp() { @@ -115,6 +118,7 @@ public class ScheduledTasksTest { .location("ftpes://192.168.0.101/ftp/rop/" + PM_FILE_NAME + instanceNumber) // .scheme(Scheme.FTPS) // .compression("") // + .messageMetaData(messageMetaData()) .build(); } @@ -130,9 +134,7 @@ public class ScheduledTasksTest { } private FileReadyMessage createFileReadyMessage(int numberOfFiles, boolean uniqueNames) { - MessageMetaData md = messageMetaData(); - return ImmutableFileReadyMessage.builder().pnfName(md.sourceName()).messageMetaData(md) - .files(files(numberOfFiles, uniqueNames)).build(); + return ImmutableFileReadyMessage.builder().files(files(numberOfFiles, uniqueNames)).build(); } private Flux<FileReadyMessage> fileReadyMessageFlux(int numberOfEvents, int filesPerEvent, boolean uniqueNames) { @@ -185,17 +187,17 @@ public class ScheduledTasksTest { doReturn(false).when(publishedCheckerMock).execute(anyString(), any()); Mono<ConsumerDmaapModel> collectedFile = Mono.just(consumerData()); - doReturn(collectedFile).when(fileCollectorMock).execute(notNull(), notNull(), anyLong(), notNull(), any()); + doReturn(collectedFile).when(fileCollectorMock).execute(notNull(), anyLong(), notNull(), notNull()); doReturn(collectedFile).when(dataRouterMock).execute(notNull(), anyLong(), notNull(), any()); - StepVerifier.create(testedObject.createMainTask(any())).expectSubscription() // + StepVerifier.create(testedObject.createMainTask(contextMap)).expectSubscription() // .expectNextCount(noOfFiles) // .expectComplete() // .verify(); // assertEquals(0, testedObject.getCurrentNumberOfTasks()); verify(consumerMock, times(1)).execute(); - verify(fileCollectorMock, times(noOfFiles)).execute(notNull(), notNull(), anyLong(), notNull(), any()); + verify(fileCollectorMock, times(noOfFiles)).execute(notNull(), anyLong(), notNull(), notNull()); verify(dataRouterMock, times(noOfFiles)).execute(notNull(), anyLong(), notNull(), any()); verifyNoMoreInteractions(dataRouterMock); verifyNoMoreInteractions(fileCollectorMock); @@ -215,19 +217,19 @@ public class ScheduledTasksTest { // First file collect will fail, 3 will succeed doReturn(error, collectedFile, collectedFile, collectedFile) // .when(fileCollectorMock) // - .execute(any(FileData.class), any(MessageMetaData.class), anyLong(), any(Duration.class), any()); + .execute(any(FileData.class), anyLong(), any(Duration.class), notNull()); doReturn(collectedFile).when(dataRouterMock).execute(notNull(), anyLong(), notNull(), any()); doReturn(collectedFile).when(dataRouterMock).execute(notNull(), anyLong(), notNull(), any()); - StepVerifier.create(testedObject.createMainTask(any())).expectSubscription() // + StepVerifier.create(testedObject.createMainTask(contextMap)).expectSubscription() // .expectNextCount(3) // .expectComplete() // .verify(); // assertEquals(0, testedObject.getCurrentNumberOfTasks()); verify(consumerMock, times(1)).execute(); - verify(fileCollectorMock, times(4)).execute(notNull(), notNull(), anyLong(), notNull(), any()); + verify(fileCollectorMock, times(4)).execute(notNull(), anyLong(), notNull(), notNull()); verify(dataRouterMock, times(3)).execute(notNull(), anyLong(), notNull(), any()); verifyNoMoreInteractions(dataRouterMock); verifyNoMoreInteractions(fileCollectorMock); @@ -243,7 +245,7 @@ public class ScheduledTasksTest { doReturn(false).when(publishedCheckerMock).execute(anyString(), any()); Mono<ConsumerDmaapModel> collectedFile = Mono.just(consumerData()); - doReturn(collectedFile).when(fileCollectorMock).execute(notNull(), notNull(), anyLong(), notNull(), any()); + doReturn(collectedFile).when(fileCollectorMock).execute(notNull(), anyLong(), notNull(), notNull()); Mono<Object> error = Mono.error(new Exception("problem")); // One publish will fail, the rest will succeed @@ -251,14 +253,14 @@ public class ScheduledTasksTest { .when(dataRouterMock) // .execute(notNull(), anyLong(), notNull(), any()); - StepVerifier.create(testedObject.createMainTask(any())).expectSubscription() // + StepVerifier.create(testedObject.createMainTask(contextMap)).expectSubscription() // .expectNextCount(3) // 3 completed files .expectComplete() // .verify(); // assertEquals(0, testedObject.getCurrentNumberOfTasks()); verify(consumerMock, times(1)).execute(); - verify(fileCollectorMock, times(4)).execute(notNull(), notNull(), anyLong(), notNull(), any()); + verify(fileCollectorMock, times(4)).execute(notNull(), anyLong(), notNull(), notNull()); verify(dataRouterMock, times(4)).execute(notNull(), anyLong(), notNull(), any()); verifyNoMoreInteractions(dataRouterMock); verifyNoMoreInteractions(fileCollectorMock); @@ -277,17 +279,17 @@ public class ScheduledTasksTest { doReturn(false).when(publishedCheckerMock).execute(anyString(), any()); Mono<ConsumerDmaapModel> collectedFile = Mono.just(consumerData()); - doReturn(collectedFile).when(fileCollectorMock).execute(notNull(), notNull(), anyLong(), notNull(), any()); + doReturn(collectedFile).when(fileCollectorMock).execute(notNull(), anyLong(), notNull(), notNull()); doReturn(collectedFile).when(dataRouterMock).execute(notNull(), anyLong(), notNull(), any()); - StepVerifier.create(testedObject.createMainTask(any())).expectSubscription() // + StepVerifier.create(testedObject.createMainTask(contextMap)).expectSubscription() // .expectNextCount(1) // 99 is skipped .expectComplete() // .verify(); // assertEquals(0, testedObject.getCurrentNumberOfTasks()); verify(consumerMock, times(1)).execute(); - verify(fileCollectorMock, times(1)).execute(notNull(), notNull(), anyLong(), notNull(), any()); + verify(fileCollectorMock, times(1)).execute(notNull(), anyLong(), notNull(), notNull()); verify(dataRouterMock, times(1)).execute(notNull(), anyLong(), notNull(), any()); verifyNoMoreInteractions(dataRouterMock); verifyNoMoreInteractions(fileCollectorMock); diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectClient.java index bca7dfd2..de50f24a 100644 --- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectClient.java +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FileCollectClient.java @@ -25,4 +25,5 @@ import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException; */ public interface FileCollectClient extends AutoCloseable { public void collectFile(String remoteFile, Path localFile) throws DatafileTaskException; + public void open() throws DatafileTaskException; } diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java index 4283debf..492768c2 100644 --- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClient.java @@ -43,7 +43,7 @@ import org.slf4j.LoggerFactory; import org.springframework.core.io.FileSystemResource; /** - * Gets file from xNF with FTPS protocol. + * Gets file from PNF with FTPS protocol. * * @author <a href="mailto:martin.c.yan@est.tech">Martin Yan</a> */ @@ -53,12 +53,22 @@ public class FtpsClient implements FileCollectClient { private final FileServerData fileServerData; private static TrustManager theTrustManager = null; - public FtpsClient(FileServerData fileServerData) { + private final String keyCertPath; + private final String keyCertPassword; + private final Path trustedCAPath; + private final String trustedCAPassword; + + public FtpsClient(FileServerData fileServerData, String keyCertPath, String keyCertPassword, Path trustedCAPath, + String trustedCAPassword) { this.fileServerData = fileServerData; + this.keyCertPath = keyCertPath; + this.keyCertPassword = keyCertPassword; + this.trustedCAPath = trustedCAPath; + this.trustedCAPassword = trustedCAPassword; } - public void open(String keyCertPath, String keyCertPassword, Path trustedCAPath, String trustedCAPassword) - throws DatafileTaskException { + @Override + public void open() throws DatafileTaskException { try { realFtpsClient.setNeedClientAuth(true); realFtpsClient.setKeyManager(createKeyManager(keyCertPath, keyCertPassword)); @@ -135,12 +145,24 @@ public class FtpsClient implements FileCollectClient { logger.trace("setUpConnection successfully!"); } - InputStream createInputStream(Path localFileName) throws IOException { + private TrustManager createTrustManager(Path trustedCAPath, String trustedCAPassword) + throws IOException, KeyStoreException, NoSuchAlgorithmException, CertificateException { + logger.trace("Creating trust manager from file: {}", trustedCAPath); + try (InputStream fis = createInputStream(trustedCAPath)) { + KeyStore keyStore = KeyStore.getInstance("JKS"); + keyStore.load(fis, trustedCAPassword.toCharArray()); + TrustManagerFactory factory = TrustManagerFactory.getInstance("SunX509"); + factory.init(keyStore); + return factory.getTrustManagers()[0]; + } + } + + protected InputStream createInputStream(Path localFileName) throws IOException { FileSystemResource realResource = new FileSystemResource(localFileName); return realResource.getInputStream(); } - OutputStream createOutputStream(Path localFileName) throws IOException { + protected OutputStream createOutputStream(Path localFileName) throws IOException { File localFile = localFileName.toFile(); if (localFile.createNewFile()) { logger.warn("Local file {} already created", localFileName); @@ -150,18 +172,7 @@ public class FtpsClient implements FileCollectClient { return output; } - private TrustManager createTrustManager(Path trustedCAPath, String trustedCAPassword) - throws IOException, KeyStoreException, NoSuchAlgorithmException, CertificateException { - try (InputStream fis = createInputStream(trustedCAPath)) { - KeyStore keyStore = KeyStore.getInstance("JKS"); - keyStore.load(fis, trustedCAPassword.toCharArray()); - TrustManagerFactory factory = TrustManagerFactory.getInstance("SunX509"); - factory.init(keyStore); - return factory.getTrustManagers()[0]; - } - } - - TrustManager getTrustManager(Path trustedCAPath, String trustedCAPassword) + protected TrustManager getTrustManager(Path trustedCAPath, String trustedCAPassword) throws KeyStoreException, NoSuchAlgorithmException, IOException, CertificateException { synchronized (FtpsClient.class) { if (theTrustManager == null) { @@ -171,7 +182,7 @@ public class FtpsClient implements FileCollectClient { } } - KeyManager createKeyManager(String keyCertPath, String keyCertPassword) + protected KeyManager createKeyManager(String keyCertPath, String keyCertPassword) throws IOException, GeneralSecurityException { return KeyManagerUtils.createClientKeyManager(new File(keyCertPath), keyCertPassword); } diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java index dec8af42..4517a755 100644 --- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/ftp/SftpClient.java @@ -47,7 +47,7 @@ public class SftpClient implements FileCollectClient { @Override public void collectFile(String remoteFile, Path localFile) throws DatafileTaskException { - logger.trace("collectFile called"); + logger.trace("collectFile {}", localFile); try { sftpChannel.get(remoteFile, localFile.toString()); @@ -61,7 +61,7 @@ public class SftpClient implements FileCollectClient { @Override public void close() { - logger.trace("close"); + logger.trace("closing sftp session"); if (sftpChannel != null) { sftpChannel.exit(); sftpChannel = null; @@ -72,6 +72,7 @@ public class SftpClient implements FileCollectClient { } } + @Override public void open() throws DatafileTaskException { try { if (session == null) { diff --git a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java index 9bd5d57f..944d3b34 100644 --- a/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java +++ b/datafile-dmaap-client/src/main/java/org/onap/dcaegen2/collectors/datafile/service/producer/DmaapProducerReactiveHttpClient.java @@ -57,11 +57,7 @@ public class DmaapProducerReactiveHttpClient { private final Logger logger = LoggerFactory.getLogger(this.getClass()); - private final String dmaapHostName; - private final Integer dmaapPortNumber; - private final String dmaapProtocol; - private final String user; - private final String pwd; + private final DmaapPublisherConfiguration configuration; /** * Constructor DmaapProducerReactiveHttpClient. @@ -69,11 +65,7 @@ public class DmaapProducerReactiveHttpClient { * @param dmaapPublisherConfiguration - DMaaP producer configuration object */ public DmaapProducerReactiveHttpClient(DmaapPublisherConfiguration dmaapPublisherConfiguration) { - this.dmaapHostName = dmaapPublisherConfiguration.dmaapHostName(); - this.dmaapPortNumber = dmaapPublisherConfiguration.dmaapPortNumber(); - this.dmaapProtocol = dmaapPublisherConfiguration.dmaapProtocol(); - this.user = dmaapPublisherConfiguration.dmaapUserName(); - this.pwd = dmaapPublisherConfiguration.dmaapUserPassword(); + this.configuration = dmaapPublisherConfiguration; } public HttpResponse getDmaapProducerResponseWithRedirect(HttpUriRequest request, Map<String, String> contextMap) @@ -85,7 +77,7 @@ public class DmaapProducerReactiveHttpClient { logger.trace(INVOKE, "Starting to produce to DR {}", request); Future<HttpResponse> future = webClient.execute(request, null); HttpResponse response = future.get(); - logger.trace(INVOKE_RETURN, "Response from DR {}", response.toString()); + logger.trace(INVOKE_RETURN, "Response from DR {}", response); return response; } catch (Exception e) { throw new DatafileTaskException("Unable to create web client.", e); @@ -101,7 +93,7 @@ public class DmaapProducerReactiveHttpClient { logger.trace(INVOKE, "Starting to produce to DR {}", request); Future<HttpResponse> future = webClient.execute(request, null); HttpResponse response = future.get(); - logger.trace(INVOKE_RETURN, "Response from DR {}", response.toString()); + logger.trace(INVOKE_RETURN, "Response from DR {}", response); return response; } catch (Exception e) { throw new DatafileTaskException("Unable to create web client.", e); @@ -109,7 +101,7 @@ public class DmaapProducerReactiveHttpClient { } public void addUserCredentialsToHead(HttpUriRequest request) { - String plainCreds = user + ":" + pwd; + String plainCreds = configuration.dmaapUserName() + ":" + configuration.dmaapUserPassword(); byte[] plainCredsBytes = plainCreds.getBytes(StandardCharsets.ISO_8859_1); byte[] base64CredsBytes = Base64.encodeBase64(plainCredsBytes); String base64Creds = new String(base64CredsBytes); @@ -119,9 +111,9 @@ public class DmaapProducerReactiveHttpClient { public UriBuilder getBaseUri() { return new DefaultUriBuilderFactory().builder() // - .scheme(dmaapProtocol) // - .host(dmaapHostName) // - .port(dmaapPortNumber); + .scheme(configuration.dmaapProtocol()) // + .host(configuration.dmaapHostName()) // + .port(configuration.dmaapPortNumber()); } private CloseableHttpAsyncClient createWebClient(boolean expectRedirect, int requestTimeout) diff --git a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java index e2882606..9e6c29f8 100644 --- a/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java +++ b/datafile-dmaap-client/src/test/java/org/onap/dcaegen2/collectors/datafile/ftp/FtpsClientTest.java @@ -73,7 +73,8 @@ public class FtpsClientTest { @BeforeEach protected void setUp() throws Exception { - clientUnderTestSpy = spy(new FtpsClient(createFileServerData())); + clientUnderTestSpy = spy(new FtpsClient(createFileServerData(), FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, + TRUSTED_CA_PASSWORD)); clientUnderTestSpy.realFtpsClient = ftpsClientMock; } @@ -104,7 +105,7 @@ public class FtpsClientTest { doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD); doReturn(HttpStatus.OK.value()).when(ftpsClientMock).getReplyCode(); - clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD); + clientUnderTestSpy.open(); doReturn(true).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock); clientUnderTestSpy.collectFile(REMOTE_FILE_PATH, LOCAL_FILE_PATH); @@ -124,9 +125,8 @@ public class FtpsClientTest { public void collectFileFaultyOwnKey_shouldFail() throws Exception { doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH); - assertThatThrownBy(() -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, - TRUSTED_CA_PASSWORD)).hasMessageContaining( - "Could not open connection: java.io.FileNotFoundException:"); + assertThatThrownBy(() -> clientUnderTestSpy.open()) + .hasMessageContaining("Could not open connection: java.io.FileNotFoundException:"); verify(ftpsClientMock).setNeedClientAuth(true); @@ -142,9 +142,8 @@ public class FtpsClientTest { doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD); doThrow(new IOException("problem")).when(clientUnderTestSpy).createInputStream(TRUSTED_CA_PATH); - assertThatThrownBy( - () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD)) - .hasMessage("Could not open connection: java.io.IOException: problem"); + assertThatThrownBy(() -> clientUnderTestSpy.open()) + .hasMessage("Could not open connection: java.io.IOException: problem"); } @Test @@ -153,9 +152,8 @@ public class FtpsClientTest { doReturn(keyManagerMock).when(clientUnderTestSpy).createKeyManager(FTP_KEY_PATH, FTP_KEY_PASSWORD); doReturn(inputStreamMock).when(clientUnderTestSpy).createInputStream(TRUSTED_CA_PATH); - assertThatThrownBy( - () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD)) - .hasMessage("Could not open connection: java.io.EOFException"); + assertThatThrownBy(() -> clientUnderTestSpy.open()) + .hasMessage("Could not open connection: java.io.EOFException"); } @Test @@ -166,9 +164,7 @@ public class FtpsClientTest { doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH); doReturn(false).when(ftpsClientMock).login(USERNAME, PASSWORD); - assertThatThrownBy( - () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD)) - .hasMessage("Unable to log in to xNF. 127.0.0.1"); + assertThatThrownBy(() -> clientUnderTestSpy.open()).hasMessage("Unable to log in to xNF. 127.0.0.1"); verify(ftpsClientMock).setNeedClientAuth(true); verify(ftpsClientMock).setKeyManager(keyManagerMock); @@ -185,9 +181,8 @@ public class FtpsClientTest { doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD); doReturn(503).when(ftpsClientMock).getReplyCode(); - assertThatThrownBy( - () -> clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD)) - .hasMessage("Unable to connect to xNF. 127.0.0.1 xNF reply code: 503"); + assertThatThrownBy(() -> clientUnderTestSpy.open()) + .hasMessage("Unable to connect to xNF. 127.0.0.1 xNF reply code: 503"); verify(ftpsClientMock).setNeedClientAuth(true); verify(ftpsClientMock).setKeyManager(keyManagerMock); @@ -205,7 +200,7 @@ public class FtpsClientTest { doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH); doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD); doReturn(HttpStatus.OK.value()).when(ftpsClientMock).getReplyCode(); - clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD); + clientUnderTestSpy.open(); doReturn(false).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock); @@ -224,7 +219,7 @@ public class FtpsClientTest { doReturn(outputStreamMock).when(clientUnderTestSpy).createOutputStream(LOCAL_FILE_PATH); doReturn(true).when(ftpsClientMock).login(USERNAME, PASSWORD); doReturn(HttpStatus.OK.value()).when(ftpsClientMock).getReplyCode(); - clientUnderTestSpy.open(FTP_KEY_PATH, FTP_KEY_PASSWORD, TRUSTED_CA_PATH, TRUSTED_CA_PASSWORD); + clientUnderTestSpy.open(); when(ftpsClientMock.isConnected()).thenReturn(false); doThrow(new IOException("problem")).when(ftpsClientMock).retrieveFile(REMOTE_FILE_PATH, outputStreamMock); |