1710 Rebase - Second Attempt

This commit rebases changes from openecomp-mso/internal-staging-1710 up to and including this codecloud commit:
54483fc6606ddb1591a2e9da61bff8712325f924
Wed Sep 6 18:12:56 2017 -0400

Rebasing was done on a branch on top of this commit in so/master in ONAP:
93fbdfbe46104f8859d4754040f979cb7997c157
Thu Sep 7 16:42:59 2017 +0000

Change-Id: I4ad9abf40da32bf5bdca43e868b8fa2dbcd9dc59
Issue-id: SO-107
Signed-off-by: Arthur Martella <amartell@research.att.com>

1 files changed, 40 insertions, 0 deletions

diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.java
new file mode 100644
index 0000000000..07ed8719b7
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.java
@@ -0,0 +1,40 @@
+package org.openecomp.mso.client.dmaap;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRConsumer;
+
+public class DmaapConsumer {
+
+	private final MRConsumer mrConsumer;
+	public DmaapConsumer() {
+		mrConsumer = null;
+	}
+	public DmaapConsumer (String filepath) throws FileNotFoundException, IOException {
+		
+		mrConsumer = MRClientFactory.createConsumer(filepath);
+	}
+	
+	
+	public MRConsumer getMRConsumer() {
+		return mrConsumer;
+	}
+	public boolean consume(Consumer consumer) throws Exception {
+		boolean accepted = false;
+		while (consumer.continuePolling()) {
+			for (String message : this.getMRConsumer().fetch()) {
+				if (!accepted && consumer.isAccepted(message)) {
+					accepted = true;
+				} 
+				if (accepted) {
+					consumer.processMessage(message);
+				}
+			}
+		}
+		
+		return true;
+	}
+	
+}