From ce984167cd55716c8a40324c7f079ed39f5fbfd7 Mon Sep 17 00:00:00 2001 From: Lukasz Muszkieta Date: Fri, 16 Mar 2018 14:42:35 +0100 Subject: pnf ready event consumer Change-Id: I63802ea60d318626ae32e734167d2bce602d72e4 Issue-ID: SO-466 Signed-off-by: Lukasz Muszkieta --- .../scripts/PnfReadyEventHandler.java | 28 +++++++ common/pom.xml | 32 ++------ .../openecomp/mso/client/dmaap/DmaapConsumer.java | 37 ++++----- .../client/sdno/dmaap/PnfReadyEventConsumer.java | 93 ++++++++++++++++++++++ .../sdno/dmaap/SDNOHealthCheckDmaapConsumer.java | 6 +- .../client/dmaap/PnfReadyEventConsumerTest.java | 85 ++++++++++++++++++++ 6 files changed, 230 insertions(+), 51 deletions(-) create mode 100644 bpmn/MSOInfrastructureBPMN/src/main/groovy/org/openecomp/mso/bpmn/infrastructure/scripts/PnfReadyEventHandler.java create mode 100644 common/src/main/java/org/openecomp/mso/client/sdno/dmaap/PnfReadyEventConsumer.java create mode 100644 common/src/test/java/org/openecomp/mso/client/dmaap/PnfReadyEventConsumerTest.java diff --git a/bpmn/MSOInfrastructureBPMN/src/main/groovy/org/openecomp/mso/bpmn/infrastructure/scripts/PnfReadyEventHandler.java b/bpmn/MSOInfrastructureBPMN/src/main/groovy/org/openecomp/mso/bpmn/infrastructure/scripts/PnfReadyEventHandler.java new file mode 100644 index 0000000000..f89b6a7a58 --- /dev/null +++ b/bpmn/MSOInfrastructureBPMN/src/main/groovy/org/openecomp/mso/bpmn/infrastructure/scripts/PnfReadyEventHandler.java @@ -0,0 +1,28 @@ +package org.openecomp.mso.bpmn.infrastructure.scripts; + +import org.camunda.bpm.engine.delegate.DelegateExecution; +import org.openecomp.mso.bpmn.common.scripts.ExceptionUtil; +import org.openecomp.mso.client.dmaap.DmaapConsumer; +import org.openecomp.mso.client.sdno.dmaap.PnfReadyEventConsumer; + +public class PnfReadyEventHandler { + + private ExceptionUtil exceptionUtil; + + private static final String TOPIC_NAME = "VES event"; + + public PnfReadyEventHandler() { + exceptionUtil = new ExceptionUtil(); + } + + public void getPnfReadyEventFromDmaap (DelegateExecution execution) throws Exception { + Object correlationIdVar = execution.getVariable("correlationId"); + if (!(correlationIdVar instanceof String)) { + exceptionUtil.buildAndThrowWorkflowException(execution, 500, "correlationId variable is not String type"); + } + String correlationId = (String) correlationIdVar; + DmaapConsumer dmaapConsumer = new PnfReadyEventConsumer(correlationId); + dmaapConsumer.consume(); + // TODO inform camunda process that event has been received + } +} diff --git a/common/pom.xml b/common/pom.xml index 23692832e4..313131a7c2 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -160,32 +160,12 @@ commons-lang3 3.4 - - + + org.assertj + assertj-core + 3.9.0 + test + diff --git a/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.java b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.java index 033951612d..6a01fb61ba 100644 --- a/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.java +++ b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.java @@ -20,29 +20,25 @@ package org.openecomp.mso.client.dmaap; -import java.io.FileNotFoundException; +import com.google.common.base.Stopwatch; import java.io.IOException; import java.util.concurrent.TimeUnit; - import org.openecomp.mso.client.dmaap.exceptions.DMaaPConsumerFailure; import org.openecomp.mso.client.dmaap.exceptions.ExceededMaximumPollingTime; import org.openecomp.mso.client.dmaap.rest.RestConsumer; -import com.google.common.base.Stopwatch; - public abstract class DmaapConsumer extends DmaapClient { - public DmaapConsumer() throws FileNotFoundException, IOException { + public DmaapConsumer() throws IOException { super("dmaap/default-consumer.properties"); } - - public Consumer getConsumer() throws FileNotFoundException, IOException { + + public Consumer getConsumer() { return new RestConsumer(this.properties); } + public boolean consume() throws Exception { - - Consumer mrConsumer = this.getConsumer(); - int iterations = 0; + Consumer mrConsumer = this.getConsumer(); boolean accepted = false; Stopwatch stopwatch = Stopwatch.createUnstarted(); try { @@ -59,32 +55,28 @@ public abstract class DmaapConsumer extends DmaapClient { if (!accepted && this.isAccepted(message)) { auditLogger.info("accepted message found for " + this.getRequestId() + " on " + this.getTopic()); accepted = true; - } + } if (accepted) { + auditLogger.info("received dmaap message: " + message); if (this.isFailure(message)) { this.stopProcessingMessages(); - auditLogger.info("received dmaap message: " + message); final String errorMsg = "failure received from dmaap topic " + this.getTopic(); auditLogger.error(errorMsg); throw new DMaaPConsumerFailure(errorMsg); } else { - auditLogger.info("received dmaap message: " + message); this.processMessage(message); } } } - iterations++; } return true; - } catch (Exception e ) { - throw e; } finally { if (stopwatch.isRunning()) { stopwatch.stop(); } } } - + /** * Should this consumer continue to consume messages from the topic? * @return @@ -92,7 +84,7 @@ public abstract class DmaapConsumer extends DmaapClient { public abstract boolean continuePolling(); /** * Process a message from a DMaaP topic - * + * * @param message * @throws Exception */ @@ -100,14 +92,14 @@ public abstract class DmaapConsumer extends DmaapClient { /** * Has the request been accepted by the receiving system? * Should the consumer move to processing messages? - * + * * @param message * @return */ public abstract boolean isAccepted(String message); /** * has the request failed? - * + * * @param message * @return */ @@ -121,11 +113,14 @@ public abstract class DmaapConsumer extends DmaapClient { * Logic that defines when the consumer should stop processing messages */ public abstract void stopProcessingMessages(); - + /** * time in milliseconds */ public int getMaximumElapsedTime() { return 180000; } + + + } diff --git a/common/src/main/java/org/openecomp/mso/client/sdno/dmaap/PnfReadyEventConsumer.java b/common/src/main/java/org/openecomp/mso/client/sdno/dmaap/PnfReadyEventConsumer.java new file mode 100644 index 0000000000..08e35f62f8 --- /dev/null +++ b/common/src/main/java/org/openecomp/mso/client/sdno/dmaap/PnfReadyEventConsumer.java @@ -0,0 +1,93 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP - SO + * ================================================================================ + * Copyright (C) 2017 AT&T Intellectual Property. 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.openecomp.mso.client.sdno.dmaap; + +import java.io.IOException; +import java.util.Optional; +import javax.ws.rs.NotSupportedException; +import org.openecomp.mso.client.dmaap.DmaapConsumer; +import org.openecomp.mso.jsonpath.JsonPathUtil; + +public class PnfReadyEventConsumer extends DmaapConsumer { + + private static final String JSON_PATH_CORRELATION_ID = "$.pnfRegistrationFields.correlationId"; + + private boolean continuePolling = true; + private String correlationId; + + public PnfReadyEventConsumer(String correlationId) throws IOException { + this.correlationId = correlationId; + } + + @Override + public boolean continuePolling() { + return continuePolling; + } + + @Override + public void processMessage(String message) { + } + + @Override + public boolean isAccepted(String message) { + Optional correlationIdOpt = JsonPathUtil.getInstance().locateResult(message, JSON_PATH_CORRELATION_ID); + if (correlationIdOpt.isPresent()) { + continuePolling = false; + return correlationIdOpt.get().equals(correlationId); + } + return false; + } + + @Override + public boolean isFailure(String message) { + throw new NotSupportedException(); + } + + @Override + public void stopProcessingMessages() { + continuePolling = false; + } + + @Override + public String getRequestId() { + throw new NotSupportedException(); + } + + @Override + public String getUserName() { + throw new NotSupportedException(); + } + + @Override + public String getPassword() { + throw new NotSupportedException(); + } + + @Override + public String getTopic() { + throw new NotSupportedException(); + } + + @Override + public Optional getHost() { + throw new NotSupportedException(); + } +} diff --git a/common/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapConsumer.java b/common/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapConsumer.java index 59adeb2026..ca5888caca 100644 --- a/common/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapConsumer.java +++ b/common/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapConsumer.java @@ -20,10 +20,8 @@ package org.openecomp.mso.client.sdno.dmaap; -import java.io.FileNotFoundException; import java.io.IOException; import java.util.Optional; - import org.openecomp.mso.client.dmaap.DmaapConsumer; import org.openecomp.mso.client.exceptions.SDNOException; import org.openecomp.mso.jsonpath.JsonPathUtil; @@ -34,11 +32,11 @@ public class SDNOHealthCheckDmaapConsumer extends DmaapConsumer { private boolean continuePolling = true; private final static String healthDiagnosticPath = "body.output.*"; - public SDNOHealthCheckDmaapConsumer() throws FileNotFoundException, IOException { + public SDNOHealthCheckDmaapConsumer() throws IOException { this("none"); } - public SDNOHealthCheckDmaapConsumer(String uuid) throws FileNotFoundException, IOException { + public SDNOHealthCheckDmaapConsumer(String uuid) throws IOException { super(); this.uuid = uuid; } diff --git a/common/src/test/java/org/openecomp/mso/client/dmaap/PnfReadyEventConsumerTest.java b/common/src/test/java/org/openecomp/mso/client/dmaap/PnfReadyEventConsumerTest.java new file mode 100644 index 0000000000..1561f75140 --- /dev/null +++ b/common/src/test/java/org/openecomp/mso/client/dmaap/PnfReadyEventConsumerTest.java @@ -0,0 +1,85 @@ +/*- + * ============LICENSE_START======================================================= + * ONAP - SO + * ================================================================================ + * Copyright (C) 2018 Huawei Intellectual Property. 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.openecomp.mso.client.dmaap; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Optional; +import org.junit.Test; +import org.openecomp.mso.client.sdno.dmaap.PnfReadyEventConsumer; + +public class PnfReadyEventConsumerTest { + + private static final String CORRELATION_ID = "correlation_id_test"; + + private static final String JSON_WITH_CORRELATION_ID = " {\"pnfRegistrationFields\": {\n" + + " \"correlationId\": \"correlation_id_test\"\n" + + " }}"; + + @Test + public void eventIsFoundForGivenCorrelationId2() throws Exception { + PnfReadyEventConsumerForTesting testedObjectSpy = spy(new PnfReadyEventConsumerForTesting(CORRELATION_ID)); + Consumer consumerMock = mock(Consumer.class); + when(testedObjectSpy.getConsumer()).thenReturn(consumerMock); + when(consumerMock.fetch()).thenReturn(Arrays.asList(JSON_WITH_CORRELATION_ID)); + testedObjectSpy.consume(); + assertThat(testedObjectSpy.continuePolling()).isFalse(); + } + + // TODO this is temporary class, when methods are defined, it will be deleted + private class PnfReadyEventConsumerForTesting extends PnfReadyEventConsumer { + + public PnfReadyEventConsumerForTesting(String correlationId) throws IOException { + super(correlationId); + } + + @Override + public String getUserName(){ + return "userNameTest"; + } + @Override + public String getPassword(){ + return "passTest"; + } + @Override + public String getTopic(){ + return "topicTest"; + } + @Override + public Optional getHost(){ + return Optional.of("http://localhost"); + } + @Override + public boolean isFailure(String message) { + return false; + } + @Override + public String getRequestId() { + return "requestTest"; + } + } + +} -- cgit 1.2.3-korg