aboutsummaryrefslogtreecommitdiffstats
path: root/common/src/main/java
diff options
context:
space:
mode:
authorLukasz Muszkieta <lukasz.muszkieta@nokia.com>2018-03-16 14:42:35 +0100
committerRob Daugherty <rd472p@att.com>2018-03-27 22:02:47 +0000
commitce984167cd55716c8a40324c7f079ed39f5fbfd7 (patch)
tree51763e03460de2ac828382beeb288c082ff19210 /common/src/main/java
parentb4473da1a753e63e66f5a9a3b1984fd4736273c2 (diff)
pnf ready event consumer
Change-Id: I63802ea60d318626ae32e734167d2bce602d72e4 Issue-ID: SO-466 Signed-off-by: Lukasz Muszkieta <lukasz.muszkieta@nokia.com>
Diffstat (limited to 'common/src/main/java')
-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
3 files changed, 111 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;
}