--- /dev/null
+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
+ }
+}
<artifactId>commons-lang3</artifactId>
<version>3.4</version>
</dependency>
- <!--for yang decoder-->
- <!--<dependency>
- <groupId>org.opendaylight.yangtools</groupId>
- <artifactId>yang-data-codec-gson</artifactId>
- <version>1.1.1-Carbon</version>
- </dependency>
- <dependency>
- <groupId>org.opendaylight.mdsal</groupId>
- <artifactId>mdsal-binding-dom-adapter</artifactId>
- <version>2.2.1-Carbon</version>
- </dependency>
- <dependency>
- <groupId>org.opendaylight.mdsal</groupId>
- <artifactId>mdsal-dom-broker</artifactId>
- <version>2.2.1-Carbon</version>
- </dependency>
- <dependency>
- <groupId>org.opendaylight.netconf</groupId>
- <artifactId>sal-rest-connector</artifactId>
- <version>1.5.1-Carbon</version>
- </dependency>
- <dependency>
- <groupId>org.dom4j</groupId>
- <artifactId>dom4j</artifactId>
- <version>2.0.0</version>
- </dependency>-->
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <version>3.9.0</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
<resources>
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 {
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
public abstract boolean continuePolling();
/**
* Process a message from a DMaaP topic
- *
+ *
* @param message
* @throws Exception
*/
/**
* 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
*/
* Logic that defines when the consumer should stop processing messages
*/
public abstract void stopProcessingMessages();
-
+
/**
* time in milliseconds
*/
public int getMaximumElapsedTime() {
return 180000;
}
+
+
+
}
--- /dev/null
+/*-
+ * ============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<String> 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<String> getHost() {
+ throw new NotSupportedException();
+ }
+}
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;
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;
}
--- /dev/null
+/*-
+ * ============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<String> getHost(){
+ return Optional.of("http://localhost");
+ }
+ @Override
+ public boolean isFailure(String message) {
+ return false;
+ }
+ @Override
+ public String getRequestId() {
+ return "requestTest";
+ }
+ }
+
+}
Code Review / so.git / commitdiff