modified sample SDNO code for SO

Issue-Id: SO-309
Change-Id: I998d5573d60d99987bb99133967ca012ef8dba6f
Signed-off-by: Benjamin, Max (mb388a) <mb388a@us.att.com>
modified RestClient to accept rest properties

Issue-Id: SO-309
Change-Id: I68bd88f9f942573dc470dda2eb60dc747e782768
Signed-off-by: Benjamin, Max (mb388a) <mb388a@us.att.com>

moved SDNO beans to their own package

Issue-Id: SO-309
Change-Id: I0b35edfe0ac3ad336320d8bc62d2efd98fff1b0b
Signed-off-by: Benjamin, Max (mb388a) <mb388a@us.att.com>

added enum for ServiceType in policy

Issue-Id: SO-309
Change-Id: Icdc95b237a427bbacac4cf5d815fc71cebb91675
Signed-off-by: Benjamin, Max (mb388a) <mb388a@us.att.com>

dmaap rest client implementation

Issue-Id: SO-309
Change-Id: I1bf8b85820524adca462d7d96c4c0beef5892864
Signed-off-by: Benjamin, Max (mb388a) <mb388a@us.att.com>

dmaap client interface for SO component

Issue-Id: SO-309
Change-Id: Ieccfb22678c52af7ffcf8750a38462ba32486182
Signed-off-by: Benjamin, Max (mb388a) <mb388a@us.att.com>
added required guava dependency to pom.xml

Issue-Id: SO-309
Change-Id: I3096bf54888e4960e60bb73b13d8cd3223217271
Signed-off-by: Benjamin, Max (mb388a) <mb388a@us.att.com>

14 files changed, 693 insertions, 64 deletions

diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/Consumer.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/Consumer.java
index f1bafde997..0e00ae5da8 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/Consumer.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/Consumer.java
@@ -22,33 +22,5 @@ package org.openecomp.mso.client.dmaap;
 
 public interface Consumer {
 
-	/**
-	 * Should this consumer continue to consume messages from the topic?
-	 * @return
-	 */
-	public boolean continuePolling();
-	/**
-	 * Process a message from a DMaaP topic
-	 * 
-	 * @param message
-	 * @throws Exception
-	 */
-	public void processMessage(String message) throws Exception;
-	/**
-	 * Has the request been accepted by the receiving system?
-	 * Should the consumer move to processing messages?
-	 * 
-	 * @param message
-	 * @return
-	 */
-	public boolean isAccepted(String message);
-	/**
-	 * The request id to filter messages on
-	 * @return
-	 */
-	public String getRequestId();
-	/**
-	 * Logic that defines when the consumer should stop processing messages
-	 */
-	public void stopProcessingMessages();
+	public Iterable<String> fetch();
 }
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DefaultDmaapPropertiesImpl.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DefaultDmaapPropertiesImpl.java
new file mode 100644
index 0000000000..9af1fd3f7e
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DefaultDmaapPropertiesImpl.java
@@ -0,0 +1,38 @@
+/*-
+ * ============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.dmaap;
+
+import java.util.Map;
+
+import org.openecomp.mso.bpmn.core.PropertyConfiguration;
+
+public class DefaultDmaapPropertiesImpl implements DmaapProperties {
+
+	private final Map<String, String> properties;
+	public DefaultDmaapPropertiesImpl() {
+		this.properties = PropertyConfiguration.getInstance().getProperties("mso.bpmn.urn.properties");
+	}
+	@Override
+	public Map<String, String> getProperties() {
+		return this.properties;
+	} 
+
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapClient.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapClient.java
new file mode 100644
index 0000000000..7862c9d41d
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapClient.java
@@ -0,0 +1,68 @@
+/*-
+ * ============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.dmaap;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.Base64;
+import java.util.Map;
+import java.util.Properties;
+
+import org.openecomp.mso.bpmn.core.PropertyConfiguration;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+
+import com.att.eelf.configuration.EELFLogger;
+import com.att.eelf.configuration.EELFManager;
+
+public abstract class DmaapClient {
+	
+	protected final EELFLogger auditLogger = EELFManager.getInstance().getAuditLogger();
+	protected final Map<String, String> msoProperties;
+	protected final Properties properties;
+	public DmaapClient(String filepath) throws FileNotFoundException, IOException {
+		Resource resource = new ClassPathResource(filepath);
+		DmaapProperties dmaapProperties = DmaapPropertiesLoader.getInstance().getImpl();
+		if (dmaapProperties == null) {
+			dmaapProperties = new DefaultDmaapPropertiesImpl();
+		}
+		this.msoProperties = dmaapProperties.getProperties();
+		this.properties = new Properties();
+		this.properties.load(resource.getInputStream());
+		this.properties.put("password", this.deobfuscatePassword(this.getPassword()));
+		this.properties.put("username", this.getUserName());
+		this.properties.put("topic", this.getTopic());
+	}
+	protected String deobfuscatePassword(String password) {
+		
+		try {
+			return new String(Base64.getDecoder().decode(password.getBytes()));
+		} catch(IllegalArgumentException iae) {
+			
+			return password;
+		}
+	}
+	
+	
+	public abstract String getUserName();
+	public abstract String getPassword();
+	public abstract String getTopic();
+}
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
index 2a618763ff..033951612d 100644
--- 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
@@ -22,39 +22,110 @@ package org.openecomp.mso.client.dmaap;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.util.concurrent.TimeUnit;
 
-import com.att.nsa.mr.client.MRClientFactory;
-import com.att.nsa.mr.client.MRConsumer;
+import org.openecomp.mso.client.dmaap.exceptions.DMaaPConsumerFailure;
+import org.openecomp.mso.client.dmaap.exceptions.ExceededMaximumPollingTime;
+import org.openecomp.mso.client.dmaap.rest.RestConsumer;
 
-public class DmaapConsumer {
+import com.google.common.base.Stopwatch;
 
-	private final MRConsumer mrConsumer;
-	public DmaapConsumer() {
-		mrConsumer = null;
-	}
-	public DmaapConsumer (String filepath) throws FileNotFoundException, IOException {
-		
-		mrConsumer = MRClientFactory.createConsumer(filepath);
+public abstract class DmaapConsumer extends DmaapClient {
+
+	public DmaapConsumer() throws FileNotFoundException, IOException {
+		super("dmaap/default-consumer.properties");
 	}
 	
-	
-	public MRConsumer getMRConsumer() {
-		return mrConsumer;
+	public Consumer getConsumer() throws FileNotFoundException, IOException {
+		return new RestConsumer(this.properties);
 	}
-	public boolean consume(Consumer consumer) throws Exception {
+	public boolean consume() throws Exception {
+		
+		Consumer mrConsumer = this.getConsumer();
+		int iterations = 0;
 		boolean accepted = false;
-		while (consumer.continuePolling()) {
-			for (String message : this.getMRConsumer().fetch()) {
-				if (!accepted && consumer.isAccepted(message)) {
-					accepted = true;
-				} 
-				if (accepted) {
-					consumer.processMessage(message);
+		Stopwatch stopwatch = Stopwatch.createUnstarted();
+		try {
+			while (this.continuePolling()) {
+				if (stopwatch.elapsed(TimeUnit.MILLISECONDS) >= this.getMaximumElapsedTime()) {
+					final String message = "exceeded maximum retries on " + this.getRequestId() + " on " + this.getTopic();
+					auditLogger.error(message);
+					throw new ExceededMaximumPollingTime(message);
 				}
+				stopwatch.start();
+				Iterable<String> itr = mrConsumer.fetch();
+				stopwatch.stop();
+				for (String message : itr) {
+					if (!accepted && this.isAccepted(message)) {
+						auditLogger.info("accepted message found for " + this.getRequestId() + " on " + this.getTopic());
+						accepted = true;
+					} 
+					if (accepted) {
+						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();
 			}
 		}
-		
-		return true;
 	}
 	
+	/**
+	 * 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
+	 */
+	public abstract void processMessage(String 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
+	 */
+	public abstract boolean isFailure(String message);
+	/**
+	 * The request id to filter messages on
+	 * @return
+	 */
+	public abstract String getRequestId();
+	/**
+	 * 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/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapProperties.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapProperties.java
new file mode 100644
index 0000000000..7bdd7dfe40
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapProperties.java
@@ -0,0 +1,32 @@
+/*-
+ * ============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.dmaap;
+
+import java.util.Map;
+
+public interface DmaapProperties {
+
+	/**
+	 * A map of strings which contains the properties for a dmaap client
+	 * @return
+	 */
+	public Map<String, String> getProperties();
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapPropertiesLoader.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapPropertiesLoader.java
new file mode 100644
index 0000000000..a21dbe8477
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapPropertiesLoader.java
@@ -0,0 +1,49 @@
+/*-
+ * ============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.dmaap;
+
+import java.util.Iterator;
+import java.util.ServiceLoader;
+
+public class DmaapPropertiesLoader {
+
+	private final ServiceLoader<DmaapProperties> services;
+	private DmaapPropertiesLoader() {
+		services = ServiceLoader.load(DmaapProperties.class);
+	}
+	
+	private static class Helper {
+		private static final DmaapPropertiesLoader INSTANCE = new DmaapPropertiesLoader();
+	}
+	
+	public static DmaapPropertiesLoader getInstance() {
+		return Helper.INSTANCE;
+	}
+	
+	public DmaapProperties getImpl() {
+		Iterator<DmaapProperties> propertyImpls = services.iterator();
+		while (propertyImpls.hasNext()) {
+			return propertyImpls.next();
+		}
+		
+		return null;
+	}
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapPublisher.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapPublisher.java
index 4d70a16b73..d2752c531b 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapPublisher.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/DmaapPublisher.java
@@ -22,29 +22,30 @@ package org.openecomp.mso.client.dmaap;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
-import java.util.concurrent.TimeUnit;
 
-import com.att.nsa.mr.client.MRBatchingPublisher;
-import com.att.nsa.mr.client.MRClientFactory;
+import org.openecomp.mso.client.dmaap.rest.RestPublisher;
 
-public class DmaapPublisher {
+public abstract class DmaapPublisher extends DmaapClient {
 	
-	private final long seconds;
-	private final MRBatchingPublisher publisher;
-	
-	public DmaapPublisher(String filepath) throws FileNotFoundException, IOException {
+	private long seconds;
+	private final Publisher publisher;
+	public DmaapPublisher() throws FileNotFoundException, IOException {
+		super("dmaap/default-consumer.properties");
+		this.publisher = new RestPublisher(properties);
 		this.seconds = 20;
-		this.publisher = MRClientFactory.createBatchingPublisher(filepath);
+		
 	}
 	
-	public DmaapPublisher(String filepath, long seconds) throws FileNotFoundException, IOException {
+	public DmaapPublisher(long seconds) throws FileNotFoundException, IOException {
+		this();
 		this.seconds = seconds;
-		this.publisher = MRClientFactory.createBatchingPublisher(filepath);
 	}
 	
 	public void send(String json) throws IOException, InterruptedException {
+		auditLogger.info("publishing message to dmaap topic " + this.getTopic() + ": " + json);
 		publisher.send(json);
-		publisher.close(seconds, TimeUnit.SECONDS);
+		//publisher.close(seconds, TimeUnit.SECONDS);
 	}
 
+
 }
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/Publisher.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/Publisher.java
new file mode 100644
index 0000000000..d89ee6e5c6
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/Publisher.java
@@ -0,0 +1,26 @@
+/*-
+ * ============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.dmaap;
+
+public interface Publisher {
+
+	public void send(String json);
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/exceptions/DMaaPConsumerFailure.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/exceptions/DMaaPConsumerFailure.java
new file mode 100644
index 0000000000..29472b2180
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/exceptions/DMaaPConsumerFailure.java
@@ -0,0 +1,34 @@
+/*-
+ * ============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.dmaap.exceptions;
+
+public class DMaaPConsumerFailure extends Exception {
+
+	private static final long serialVersionUID = 2499229901897110362L;
+
+	public DMaaPConsumerFailure() {
+		super();
+	}
+	
+	public DMaaPConsumerFailure(String message) {
+		super(message);
+	}
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/exceptions/ExceededMaximumPollingTime.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/exceptions/ExceededMaximumPollingTime.java
new file mode 100644
index 0000000000..c9d675067e
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/exceptions/ExceededMaximumPollingTime.java
@@ -0,0 +1,34 @@
+/*-
+ * ============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.dmaap.exceptions;
+
+public class ExceededMaximumPollingTime extends RuntimeException  {
+
+	private static final long serialVersionUID = 2331207691092906423L;
+
+	public ExceededMaximumPollingTime() {
+		super();
+	}
+	
+	public ExceededMaximumPollingTime(String message) {
+		super(message);
+	}
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/DMaaPRestClient.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/DMaaPRestClient.java
new file mode 100644
index 0000000000..124e2c3a28
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/DMaaPRestClient.java
@@ -0,0 +1,57 @@
+/*-
+ * ============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.dmaap.rest;
+
+import java.net.URL;
+import java.util.Base64;
+import java.util.Map;
+import java.util.Optional;
+
+import javax.ws.rs.client.ClientResponseFilter;
+
+import org.openecomp.mso.client.ResponseExceptionMapperImpl;
+import org.openecomp.mso.client.policy.RestClient;
+
+public class DMaaPRestClient  extends RestClient {
+
+	private final String username;
+	private final String password;
+	public DMaaPRestClient(URL url, String contentType, String username, String password) {
+		super(url, contentType);
+		this.username = username;
+		this.password = password;
+	}
+
+	@Override
+	protected void initializeHeaderMap(Map<String, String> headerMap) {
+		headerMap.put("Authorization", "Basic " + Base64.getEncoder().encodeToString(new String(username + ":" + password).getBytes()));
+	}
+
+	@Override
+	protected Optional<ClientResponseFilter> addResponseFilter() {
+		return Optional.of(new ResponseExceptionMapperImpl());
+	}
+
+	@Override
+	public RestClient addRequestId(String requestId) {
+		return null;
+	}
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/PropertiesBean.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/PropertiesBean.java
new file mode 100644
index 0000000000..fb914a0c13
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/PropertiesBean.java
@@ -0,0 +1,131 @@
+/*-
+ * ============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.dmaap.rest;
+
+import java.util.Properties;
+
+public class PropertiesBean {
+
+	private String username;
+	private String password;
+	private String environment;
+	private String partition;
+	private String contentType;
+	private String host;
+	private String topic;
+	private String timeout;
+	
+	
+	public PropertiesBean(Properties properties) {
+		this.withUsername(properties.getProperty("username"))
+		.withPassword(properties.getProperty("password"))
+		.withTopic(properties.getProperty("topic"))
+		.withEnvironment(properties.getProperty("environment"))
+		.withHost(properties.getProperty("host"))
+		.withTimeout(properties.getProperty("timeout", "20000"))
+		.withPartition(properties.getProperty("partition"))
+		.withContentType(properties.getProperty("contentType", "application/json"));
+	}
+	public String getUsername() {
+		return username;
+	}
+	public void setUsername(String username) {
+		this.username = username;
+	}
+	public PropertiesBean withUsername(String username) {
+		this.username = username;
+		return this;
+	}
+	public String getPassword() {
+		return password;
+	}
+	public void setPassword(String password) {
+		this.password = password;
+	}
+	public PropertiesBean withPassword(String password) {
+		this.password = password;
+		return this;
+	}
+	public String getEnvironment() {
+		return environment;
+	}
+	public void setEnvironment(String environment) {
+		this.environment = environment;
+	}
+	public PropertiesBean withEnvironment(String environment) {
+		this.environment = environment;
+		return this;
+	}
+	public String getPartition() {
+		return partition;
+	}
+	public void setPartition(String partition) {
+		this.partition = partition;
+	}
+	public PropertiesBean withPartition(String partition) {
+		this.partition = partition;
+		return this;
+	}
+	public String getContentType() {
+		return contentType;
+	}
+	public void setContentType(String contentType) {
+		this.contentType = contentType;
+	}
+	public PropertiesBean withContentType(String contentType) {
+		this.contentType = contentType;
+		return this;
+	}
+	public String getHost() {
+		return host;
+	}
+	public void setHost(String host) {
+		this.host = host;
+	}
+	public PropertiesBean withHost(String host) {
+		this.host = host;
+		return this;
+	}
+	public String getTopic() {
+		return topic;
+	}
+	public void setTopic(String topic) {
+		this.topic = topic;
+	}
+	public PropertiesBean withTopic(String topic) {
+		this.topic = topic;
+		return this;
+	}
+	public String getTimeout() {
+		return timeout;
+	}
+	public void setTimeout(String timeout) {
+		this.timeout = timeout;
+	}
+	public PropertiesBean withTimeout(String timeout) {
+		this.timeout = timeout;
+		return this;
+	}
+	
+	
+	
+	
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/RestConsumer.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/RestConsumer.java
new file mode 100644
index 0000000000..ff199e2373
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/RestConsumer.java
@@ -0,0 +1,60 @@
+/*-
+ * ============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.dmaap.rest;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.List;
+import java.util.Properties;
+
+import javax.ws.rs.core.GenericType;
+import javax.ws.rs.core.UriBuilder;
+
+import org.openecomp.mso.client.dmaap.Consumer;
+import org.openecomp.mso.client.policy.RestClient;
+
+public class RestConsumer implements Consumer {
+
+	private final RestClient client;
+	public RestConsumer(Properties properties) {
+		PropertiesBean bean = new PropertiesBean(properties);
+		client = new DMaaPRestClient(this.createURL(bean), bean.getContentType(), bean.getUsername(), bean.getPassword());
+	}
+	
+	private URL createURL(PropertiesBean properties) {
+		try {
+			return UriBuilder.fromUri("http://" + properties.getHost())
+					.path("events").path(properties.getTopic())
+					.path(properties.getPartition())
+					.path("consumer1")
+					.queryParam("timeout",  properties.getTimeout()).build().toURL();
+		} catch (MalformedURLException e) {
+			throw new RuntimeException(e);
+		}
+	}
+	
+	@Override
+	public Iterable<String> fetch() {
+		
+		return client.get(new GenericType<List<String>>() {});
+	}
+
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/RestPublisher.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/RestPublisher.java
new file mode 100644
index 0000000000..e8e685932a
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/dmaap/rest/RestPublisher.java
@@ -0,0 +1,56 @@
+/*-
+ * ============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.dmaap.rest;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Properties;
+
+import javax.ws.rs.core.UriBuilder;
+import javax.ws.rs.core.UriBuilderException;
+
+import org.openecomp.mso.client.dmaap.Publisher;
+import org.openecomp.mso.client.policy.RestClient;
+
+public class RestPublisher implements Publisher {
+
+	private final RestClient client;
+
+	public RestPublisher(Properties properties) {
+		PropertiesBean bean = new PropertiesBean(properties);
+		client = new DMaaPRestClient(this.createURL(bean), bean.getContentType(), bean.getUsername(), bean.getPassword());
+	}
+	
+	private URL createURL(PropertiesBean properties) {
+		try {
+			return UriBuilder.fromUri("http://" + properties.getHost())
+					.path("events").path(properties.getTopic())
+					.queryParam("timeout",  properties.getTimeout()).build().toURL();
+		} catch (MalformedURLException e) {
+			throw new RuntimeException(e);
+		}
+	}
+	
+	@Override
+	public void send(String json) {
+		client.post(json);
+	}
+}