Merge "Update docker image"
diff --git a/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java b/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java
index 72d1509..e083466 100644
--- a/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java
+++ b/src/main/java/org/onap/dcaegen2/services/pmmapper/App.java
@@ -24,6 +24,7 @@
import io.undertow.Undertow;
import io.undertow.util.StatusCodes;
+import lombok.NonNull;
import org.onap.dcaegen2.services.pmmapper.config.ConfigHandler;
import org.onap.dcaegen2.services.pmmapper.datarouter.DataRouterSubscriber;
import org.onap.dcaegen2.services.pmmapper.exceptions.CBSConfigException;
@@ -32,12 +33,16 @@
import org.onap.dcaegen2.services.pmmapper.exceptions.MapperConfigException;
import org.onap.dcaegen2.services.pmmapper.exceptions.TooManyTriesException;
import org.onap.dcaegen2.services.pmmapper.mapping.Mapper;
+import org.onap.dcaegen2.services.pmmapper.model.Event;
import org.onap.dcaegen2.services.pmmapper.model.MapperConfig;
import org.onap.dcaegen2.services.pmmapper.healthcheck.HealthCheckHandler;
import org.onap.dcaegen2.services.pmmapper.utils.XMLValidator;
import org.onap.logging.ref.slf4j.ONAPLogAdapter;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxSink;
+import reactor.core.scheduler.Schedulers;
import java.nio.file.Path;
import java.nio.file.Paths;
@@ -46,29 +51,59 @@
private static final ONAPLogAdapter logger = new ONAPLogAdapter(LoggerFactory.getLogger(App.class));
private static Path mappingTemplate = Paths.get("/opt/app/pm-mapper/etc/mapping.ftl");
private static Path xmlSchema = Paths.get("/opt/app/pm-mapper/etc/measCollec_plusString.xsd");
+ private static FluxSink<Event> fluxSink;
public static void main(String[] args) throws InterruptedException, TooManyTriesException, CBSConfigException, EnvironmentConfigException, CBSServerError, MapperConfigException {
+ Flux<Event> flux = Flux.create(eventFluxSink -> fluxSink = eventFluxSink);
HealthCheckHandler healthCheckHandler = new HealthCheckHandler();
Mapper mapper = new Mapper(mappingTemplate);
XMLValidator validator = new XMLValidator(xmlSchema);
- DataRouterSubscriber dataRouterSubscriber = new DataRouterSubscriber(event -> {
- event.getHttpServerExchange().unDispatch();
- event.getHttpServerExchange().getResponseSender().send(StatusCodes.OK_STRING);
- MDC.setContextMap(event.getMdc());
- if(!validator.validate(event)){
- logger.unwrap().info("Event failed validation against schema.");
- } else {
- String ves = mapper.map(event);
- logger.unwrap().info("Mapped Event: {}", ves);
- }
- });
+ flux.onBackpressureDrop(App::handleBackPressure)
+ .doOnNext(App::receiveRequest)
+ .limitRate(1)
+ .parallel()
+ .runOn(Schedulers.newParallel(""), 1)
+ .doOnNext(event -> MDC.setContextMap(event.getMdc()))
+ .filter(validator::validate)
+ .map(mapper::map)
+ .subscribe(event -> logger.unwrap().info("Event Processed"));
+
+ DataRouterSubscriber dataRouterSubscriber = new DataRouterSubscriber(fluxSink::next);
MapperConfig mapperConfig = new ConfigHandler().getMapperConfig();
dataRouterSubscriber.start(mapperConfig);
Undertow.builder()
.addHttpListener(8081, "0.0.0.0")
- .setHandler(Handlers.routing().add("put", "/sub", dataRouterSubscriber)
- .add("get", "/healthcheck", healthCheckHandler))
+ .setHandler(Handlers.routing().add("put", "/delivery", dataRouterSubscriber)
+ .add("get", "/healthcheck", healthCheckHandler))
.build().start();
}
+
+ /**
+ * Takes the exchange from an event, responds with a 429 and un-dispatches the exchange.
+ * @param event to be ignored.
+ */
+ public static void handleBackPressure(@NonNull Event event) {
+ logger.unwrap().debug("Event will not be processed, responding with 429");
+ event.getHttpServerExchange()
+ .setStatusCode(StatusCodes.TOO_MANY_REQUESTS)
+ .getResponseSender()
+ .send(StatusCodes.TOO_MANY_REQUESTS_STRING);
+ event.getHttpServerExchange()
+ .unDispatch();
+ }
+
+ /**
+ * Takes the exchange from an event, responds with a 200 and un-dispatches the exchange.
+ * @param event to be received.
+ */
+ public static void receiveRequest(@NonNull Event event) {
+ logger.unwrap().debug("Event will be processed, responding with 200");
+ event.getHttpServerExchange()
+ .setStatusCode(StatusCodes.OK)
+ .getResponseSender()
+ .send(StatusCodes.OK_STRING);
+ event.getHttpServerExchange()
+ .unDispatch();
+ }
}
diff --git a/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java b/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
new file mode 100644
index 0000000..82650f3
--- /dev/null
+++ b/src/test/java/org/onap/dcaegen2/services/pmmapper/AppTest.java
@@ -0,0 +1,66 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 Nordix Foundation.
+ * ================================================================================
+ * 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.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.dcaegen2.services.pmmapper;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import io.undertow.util.StatusCodes;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.onap.dcaegen2.services.pmmapper.model.Event;
+import org.onap.dcaegen2.services.pmmapper.model.EventMetadata;
+
+
+@ExtendWith(MockitoExtension.class)
+class AppTest {
+
+ @Test
+ void testHandleBackPressureNullValue() {
+ assertThrows(NullPointerException.class, () -> App.handleBackPressure(null));
+ }
+
+ @Test
+ void testHandleBackPressure() {
+ Event event = utils.EventUtils.makeMockEvent("", mock(EventMetadata.class));
+ App.handleBackPressure(event);
+ verify(event.getHttpServerExchange(), times(1)).setStatusCode(StatusCodes.TOO_MANY_REQUESTS);
+ verify(event.getHttpServerExchange(), times(1)).unDispatch();
+ }
+
+ @Test
+ void testReceiveRequestNullValue() {
+ assertThrows(NullPointerException.class, () -> App.receiveRequest(null));
+ }
+
+ @Test
+ void testReceiveRequest() {
+ Event event = utils.EventUtils.makeMockEvent("", mock(EventMetadata.class));
+ App.receiveRequest(event);
+ verify(event.getHttpServerExchange(), times(1)).setStatusCode(StatusCodes.OK);
+ verify(event.getHttpServerExchange(), times(1)).unDispatch();
+ }
+
+
+}