aboutsummaryrefslogtreecommitdiffstats
path: root/common/src
diff options
context:
space:
mode:
authorRob Daugherty <rd472p@att.com>2018-03-28 17:59:09 +0000
committerGerrit Code Review <gerrit@onap.org>2018-03-28 17:59:09 +0000
commit8f394febaf7423e6ea66007565f8aeebe23991e6 (patch)
treee9dd30e26a2f7d4908790ec778570cffcfd09b0e /common/src
parent35a1423d200962910d508a8f9a0daba9d6c668d2 (diff)
parentce984167cd55716c8a40324c7f079ed39f5fbfd7 (diff)
Merge "pnf ready event consumer"
Diffstat (limited to 'common/src')
-rw-r--r--common/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.java37
-rw-r--r--common/src/main/java/org/openecomp/mso/client/sdno/dmaap/PnfReadyEventConsumer.java93
-rw-r--r--common/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapConsumer.java6
-rw-r--r--common/src/test/java/org/openecomp/mso/client/dmaap/PnfReadyEventConsumerTest.java85
4 files changed, 196 insertions, 25 deletions
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<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();
+ }
+}
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<String> getHost(){
+ return Optional.of("http://localhost");
+ }
+ @Override
+ public boolean isFailure(String message) {
+ return false;
+ }
+ @Override
+ public String getRequestId() {
+ return "requestTest";
+ }
+ }
+
+}