aboutsummaryrefslogtreecommitdiffstats
path: root/src/main/java
diff options
context:
space:
mode:
Diffstat (limited to 'src/main/java')
-rw-r--r--src/main/java/org/onap/dcae/common/ConfigProcessors.java2
-rw-r--r--src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java28
2 files changed, 18 insertions, 12 deletions
diff --git a/src/main/java/org/onap/dcae/common/ConfigProcessors.java b/src/main/java/org/onap/dcae/common/ConfigProcessors.java
index 040a3e6e..d53bf23b 100644
--- a/src/main/java/org/onap/dcae/common/ConfigProcessors.java
+++ b/src/main/java/org/onap/dcae/common/ConfigProcessors.java
@@ -202,7 +202,7 @@ public class ConfigProcessors {
}
}
- private String performOperation(String operation, String value) {
+ String performOperation(String operation, String value) {
log.info("performOperation");
if ("convertMBtoKB".equals(operation)) {
float kbValue = Float.parseFloat(value) * 1024;
diff --git a/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java b/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java
index 08e16e0c..a7c0a0eb 100644
--- a/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java
+++ b/src/main/java/org/onap/dcae/common/publishing/DMaaPEventPublisher.java
@@ -2,7 +2,7 @@
* ============LICENSE_START=======================================================
* org.onap.dcaegen2.collectors.ves
* ================================================================================
- * Copyright (C) 2017,2020 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017,2020,2023 AT&T Intellectual Property. All rights reserved.
* Copyright (C) 2018-2021 Nokia. All rights reserved.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
@@ -33,6 +33,7 @@ import java.util.List;
import java.util.Objects;
import static org.onap.dcae.common.publishing.MessageRouterHttpStatusMapper.getHttpStatus;
+
/**
* @author Pawel Szalapski (pawel.szalapski@nokia.com)
*/
@@ -48,9 +49,10 @@ public class DMaaPEventPublisher {
/**
* Reload Dmaap configuration
+ *
* @param dmaapConfiguration Dmaap configuration
*/
- public void reload(Map<String, PublisherConfig> dmaapConfiguration){
+ public void reload(Map<String, PublisherConfig> dmaapConfiguration) {
dMaaPConfig = dmaapConfiguration;
log.info("reload dmaap configuration");
}
@@ -58,22 +60,26 @@ public class DMaaPEventPublisher {
public HttpStatus sendEvent(List<VesEvent> vesEvents, String dmaapId) {
clearVesUniqueIdFromEvent(vesEvents);
io.vavr.collection.List<String> events = mapListOfEventsToVavrList(vesEvents);
- Flux<MessageRouterPublishResponse> messageRouterPublishFlux = dmaapPublisher.publishEvents(events, dMaaPConfig.get(dmaapId));
+ HttpStatus rc = messageRouterPublishResponse(events, dmaapId);
+ return rc;
+ }
+
+ HttpStatus messageRouterPublishResponse(io.vavr.collection.List<String> events, String dmaapId) {
+ Flux<MessageRouterPublishResponse> messageRouterPublishFlux =
+ dmaapPublisher.publishEvents(events, dMaaPConfig.get(dmaapId));
MessageRouterPublishResponse messageRouterPublishResponse = messageRouterPublishFlux.blockFirst();
return getHttpStatus(Objects.requireNonNull(messageRouterPublishResponse));
+
}
private io.vavr.collection.List<String> mapListOfEventsToVavrList(List<VesEvent> vesEvents) {
- return io.vavr.collection.List.ofAll(vesEvents)
- .map(event -> event.asJsonObject().toString());
+ return io.vavr.collection.List.ofAll(vesEvents).map(event -> event.asJsonObject().toString());
}
private void clearVesUniqueIdFromEvent(List<VesEvent> events) {
- events.stream()
- .filter(event -> event.hasType(VesEvent.VES_UNIQUE_ID))
- .forEach(event -> {
- log.debug("Removing VESuniqueid object from event");
- event.removeElement(VesEvent.VES_UNIQUE_ID);
- });
+ events.stream().filter(event -> event.hasType(VesEvent.VES_UNIQUE_ID)).forEach(event -> {
+ log.debug("Removing VESuniqueid object from event");
+ event.removeElement(VesEvent.VES_UNIQUE_ID);
+ });
}
}