diff options
Diffstat (limited to 'datafile-app-server/src')
5 files changed, 19 insertions, 34 deletions
diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTask.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTask.java index 0b81df5b..8508cd12 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTask.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTask.java @@ -18,9 +18,7 @@ package org.onap.dcaegen2.collectors.datafile.tasks; import org.onap.dcaegen2.collectors.datafile.config.DmaapPublisherConfiguration; import org.onap.dcaegen2.collectors.datafile.model.ConsumerDmaapModel; -import org.onap.dcaegen2.collectors.datafile.service.DmaapReactiveWebClient; import org.onap.dcaegen2.collectors.datafile.service.producer.DmaapProducerReactiveHttpClient; -import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Flux; @@ -30,15 +28,9 @@ import reactor.core.publisher.Flux; */ abstract class DmaapPublisherTask { - abstract Flux<String> publish(ConsumerDmaapModel consumerDmaapModel); - - abstract DmaapProducerReactiveHttpClient resolveClient(); - protected abstract DmaapPublisherConfiguration resolveConfiguration(); - protected abstract Flux<String> execute(ConsumerDmaapModel consumerDmaapModel); + protected abstract DmaapProducerReactiveHttpClient resolveClient(); - WebClient buildWebClient() { - return new DmaapReactiveWebClient().fromConfiguration(resolveConfiguration()).build(); - } + protected abstract Flux<String> execute(ConsumerDmaapModel consumerDmaapModel); } diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTaskImpl.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTaskImpl.java index b4ee3a9d..201b33d1 100644 --- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTaskImpl.java +++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DmaapPublisherTaskImpl.java @@ -37,7 +37,6 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask { private static final Logger logger = LoggerFactory.getLogger(DmaapPublisherTaskImpl.class); private final Config datafileAppConfig; - private DmaapProducerReactiveHttpClient dmaapProducerReactiveHttpClient; @Autowired public DmaapPublisherTaskImpl(AppConfig datafileAppConfig) { @@ -45,16 +44,10 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask { } @Override - public Flux<String> publish(ConsumerDmaapModel consumerDmaapModel) { - logger.trace("Publishing on DMaaP DataRouter {}", consumerDmaapModel); - return dmaapProducerReactiveHttpClient.getDmaapProducerResponse(consumerDmaapModel); - } - - @Override public Flux<String> execute(ConsumerDmaapModel consumerDmaapModel) { - dmaapProducerReactiveHttpClient = resolveClient(); logger.trace("Method called with arg {}", consumerDmaapModel); - return publish(consumerDmaapModel); + DmaapProducerReactiveHttpClient dmaapProducerReactiveHttpClient = resolveClient(); + return dmaapProducerReactiveHttpClient.getDmaapProducerResponse(consumerDmaapModel); } @Override @@ -63,8 +56,8 @@ public class DmaapPublisherTaskImpl extends DmaapPublisherTask { } @Override - DmaapProducerReactiveHttpClient resolveClient() { - return new DmaapProducerReactiveHttpClient(resolveConfiguration()).createDmaapWebClient(buildWebClient()); + protected DmaapProducerReactiveHttpClient resolveClient() { + return new DmaapProducerReactiveHttpClient(resolveConfiguration()); } } 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 c263c95c..171dd024 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 @@ -61,7 +61,7 @@ public class ScheduledTasks { logger.trace("Execution of tasks was registered"); consumeFromDmaapMessage() - .doOnError(DmaapEmptyResponseException.class, error -> logger.error("Nothing to consume from DMaaP")) + .doOnError(DmaapEmptyResponseException.class, error -> logger.info("Nothing to consume from DMaaP")) .flatMap(this::collectFilesFromXnf).flatMap(this::publishToDmaapConfiguration) .subscribe(this::onSuccess, this::onError, this::onComplete); } diff --git a/datafile-app-server/src/main/resources/datafile_endpoints.json b/datafile-app-server/src/main/resources/datafile_endpoints.json index 188129e8..dff77d2f 100644 --- a/datafile-app-server/src/main/resources/datafile_endpoints.json +++ b/datafile-app-server/src/main/resources/datafile_endpoints.json @@ -3,10 +3,10 @@ "dmaap": { "dmaapConsumerConfiguration": { "consumerGroup": "notification", - "consumerId": "1", + "consumerId": "C12", "dmaapContentType": "application/json", "dmaapHostName": "localhost", - "dmaapPortNumber": 3904, + "dmaapPortNumber": 2222, "dmaapProtocol": "http", "dmaapTopicName": "/events/unauthenticated.VES_NOTIFICATION_OUTPUT", "dmaapUserName": "admin", @@ -17,11 +17,11 @@ "dmaapProducerConfiguration": { "dmaapContentType": "application/octet-stream", "dmaapHostName": "localhost", - "dmaapPortNumber": 3905, - "dmaapProtocol": "http", + "dmaapPortNumber": 3907, + "dmaapProtocol": "https", "dmaapTopicName": "publish", - "dmaapUserName": "admin", - "dmaapUserPassword": "admin" + "dmaapUserName": "dradmin", + "dmaapUserPassword": "dradmin" } } } diff --git a/datafile-app-server/src/test/resources/datafile_endpoints.json b/datafile-app-server/src/test/resources/datafile_endpoints.json index 188129e8..dff77d2f 100644 --- a/datafile-app-server/src/test/resources/datafile_endpoints.json +++ b/datafile-app-server/src/test/resources/datafile_endpoints.json @@ -3,10 +3,10 @@ "dmaap": { "dmaapConsumerConfiguration": { "consumerGroup": "notification", - "consumerId": "1", + "consumerId": "C12", "dmaapContentType": "application/json", "dmaapHostName": "localhost", - "dmaapPortNumber": 3904, + "dmaapPortNumber": 2222, "dmaapProtocol": "http", "dmaapTopicName": "/events/unauthenticated.VES_NOTIFICATION_OUTPUT", "dmaapUserName": "admin", @@ -17,11 +17,11 @@ "dmaapProducerConfiguration": { "dmaapContentType": "application/octet-stream", "dmaapHostName": "localhost", - "dmaapPortNumber": 3905, - "dmaapProtocol": "http", + "dmaapPortNumber": 3907, + "dmaapProtocol": "https", "dmaapTopicName": "publish", - "dmaapUserName": "admin", - "dmaapUserPassword": "admin" + "dmaapUserName": "dradmin", + "dmaapUserPassword": "dradmin" } } } |