Merge "modified sample SDNO code for SO"

34 files changed, 1630 insertions, 421 deletions

diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/PreconditionFailedException.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/PreconditionFailedException.java
new file mode 100644
index 0000000000..df28baac74
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/PreconditionFailedException.java
@@ -0,0 +1,35 @@
+/*-
+ * ============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;
+
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Response;
+
+public class PreconditionFailedException extends WebApplicationException {
+	/**
+	 * 
+	 */
+	private static final long serialVersionUID = 1L;
+
+	public PreconditionFailedException(String message) {
+        super(message, Response.Status.PRECONDITION_FAILED);
+    }
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/ResponseExceptionMapper.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/ResponseExceptionMapper.java
new file mode 100644
index 0000000000..e0e3e936be
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/ResponseExceptionMapper.java
@@ -0,0 +1,98 @@
+/*-
+ * ============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;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Optional;
+
+import javax.annotation.Priority;
+import javax.ws.rs.BadRequestException;
+import javax.ws.rs.ForbiddenException;
+import javax.ws.rs.InternalServerErrorException;
+import javax.ws.rs.NotAcceptableException;
+import javax.ws.rs.NotAllowedException;
+import javax.ws.rs.NotAuthorizedException;
+import javax.ws.rs.NotFoundException;
+import javax.ws.rs.NotSupportedException;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.client.ClientRequestContext;
+import javax.ws.rs.client.ClientResponseContext;
+import javax.ws.rs.client.ClientResponseFilter;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.ext.Provider;
+
+@Provider
+@Priority(value = 1)
+public abstract class ResponseExceptionMapper implements ClientResponseFilter {
+
+	@Override
+	public void filter(ClientRequestContext requestContext, ClientResponseContext responseContext) throws IOException {
+		if (responseContext.getStatus() >= 300) {
+			String message = "empty message";
+			if (responseContext.hasEntity()) {
+				Optional<String> result = this.extractMessage(responseContext.getEntityStream());
+				if (result.isPresent()) {
+					message = result.get();
+				}
+			}
+			Response.Status status = Response.Status.fromStatusCode(responseContext.getStatus());
+			WebApplicationException webAppException;
+			switch (status) {
+			case BAD_REQUEST:
+				webAppException = new BadRequestException(message);
+				break;
+			case UNAUTHORIZED:
+				webAppException = new NotAuthorizedException(message);
+				break;
+			case FORBIDDEN:
+				webAppException = new ForbiddenException(message);
+				break;
+			case NOT_FOUND:
+				webAppException = new NotFoundException(message);
+				break;
+			case METHOD_NOT_ALLOWED:
+				webAppException = new NotAllowedException(message);
+				break;
+			case NOT_ACCEPTABLE:
+				webAppException = new NotAcceptableException(message);
+				break;
+			case PRECONDITION_FAILED:
+				webAppException = new PreconditionFailedException(message);
+				break;
+			case UNSUPPORTED_MEDIA_TYPE:
+				webAppException = new NotSupportedException(message);
+				break;
+			case INTERNAL_SERVER_ERROR:
+				webAppException = new InternalServerErrorException(message);
+				break;
+			case SERVICE_UNAVAILABLE:
+				webAppException = new WebApplicationException(message);
+				break;
+			default:
+				webAppException = new WebApplicationException(message);
+			}
+			throw webAppException;
+		}
+	}
+	
+	public abstract Optional<String> extractMessage(InputStream stream) throws IOException;
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/ResponseExceptionMapperImpl.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/ResponseExceptionMapperImpl.java
new file mode 100644
index 0000000000..0845a2fbcd
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/ResponseExceptionMapperImpl.java
@@ -0,0 +1,39 @@
+/*-
+ * ============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;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Optional;
+
+import org.apache.commons.io.IOUtils;
+
+public class ResponseExceptionMapperImpl extends ResponseExceptionMapper {
+
+	@Override
+	public Optional<String> extractMessage(InputStream stream) throws IOException {
+		final String input = IOUtils.toString(stream, "UTF-8");
+		IOUtils.closeQuietly(stream);
+		return Optional.of(input);
+	}
+	
+
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/RestProperties.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/RestProperties.java
new file mode 100644
index 0000000000..ae8862de5e
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/RestProperties.java
@@ -0,0 +1,30 @@
+/*-
+ * ============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;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+
+public interface RestProperties {
+
+	public URL getEndpoint() throws MalformedURLException;
+	public String getSystemName();
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/RestPropertiesLoader.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/RestPropertiesLoader.java
new file mode 100644
index 0000000000..6d49d9800f
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/RestPropertiesLoader.java
@@ -0,0 +1,55 @@
+/*-
+ * ============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;
+
+import java.util.Iterator;
+import java.util.ServiceLoader;
+
+public class RestPropertiesLoader {
+
+	private final ServiceLoader<RestProperties> services;
+	private RestPropertiesLoader() {
+		services = ServiceLoader.load(RestProperties.class);
+	}
+	
+	private static class Helper {
+		private static final RestPropertiesLoader INSTANCE = new RestPropertiesLoader();
+	}
+	
+	public static RestPropertiesLoader getInstance() {
+		return Helper.INSTANCE;
+	}
+	
+	public <T> T getImpl(Class<? extends RestProperties> clazz) {
+		T result = null;
+		Iterator<RestProperties> propertyImpls = services.iterator();
+		RestProperties item;
+		while (propertyImpls.hasNext()) {
+			item = propertyImpls.next();
+			if (clazz.isAssignableFrom(item.getClass())) {
+				result = (T)item;
+				break;
+			}
+		}
+		
+		return result;
+	}
+}
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);
+	}
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/LoggingFilter.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/LoggingFilter.java
index 665b9052eb..e02941944a 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/LoggingFilter.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/LoggingFilter.java
@@ -1,130 +1,138 @@
-/*-

- * ============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.policy;

-

-import java.io.BufferedInputStream;

-

-import java.io.ByteArrayOutputStream;

-import java.io.FilterOutputStream;

-import java.io.IOException;

-import java.io.InputStream;

-import java.io.OutputStream;

-import java.nio.charset.Charset;

-import java.nio.charset.StandardCharsets;

-import java.util.logging.Logger;

-import javax.annotation.Priority;

-import javax.ws.rs.WebApplicationException;

-import javax.ws.rs.client.ClientRequestContext;

-import javax.ws.rs.client.ClientRequestFilter;

-import javax.ws.rs.client.ClientResponseContext;

-import javax.ws.rs.client.ClientResponseFilter;

-import javax.ws.rs.ext.WriterInterceptor;

-import javax.ws.rs.ext.WriterInterceptorContext;

-

-@Priority(Integer.MIN_VALUE)

-public class LoggingFilter implements ClientRequestFilter, ClientResponseFilter, WriterInterceptor {

-

-	private static final Logger logger = Logger.getLogger(LoggingFilter.class.getName());

-	private static final String ENTITY_STREAM_PROPERTY = "LoggingFilter.entityStream";

-	private static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;

-	private final int maxEntitySize = 1024 * 8;

-

-	private void log(StringBuilder sb) {

-		logger.info(sb.toString());

-	}

-

-	private InputStream logInboundEntity(final StringBuilder b, InputStream stream, final Charset charset)

-			throws IOException {

-		InputStream inputStream = stream;

-		if (!inputStream.markSupported()) {

-			inputStream = new BufferedInputStream(inputStream);

-		}

-		inputStream.mark(maxEntitySize + 1);

-		final byte[] entity = new byte[maxEntitySize + 1];

-		final int entitySize = inputStream.read(entity);

-		b.append(new String(entity, 0, Math.min(entitySize, maxEntitySize), charset));

-		if (entitySize > maxEntitySize) {

-			b.append("...more...");

-		}

-		b.append('\n');

-		inputStream.reset();

-		return inputStream;

-	}

-

-	@Override

-	public void filter(ClientRequestContext requestContext) throws IOException {

-		if (requestContext.hasEntity()) {

-			final OutputStream stream = new LoggingStream(requestContext.getEntityStream());

-			requestContext.setEntityStream(stream);

-			requestContext.setProperty(ENTITY_STREAM_PROPERTY, stream);

-		}

-	}

-

-	@Override

-	public void filter(ClientRequestContext requestContext, ClientResponseContext responseContext) throws IOException {

-		final StringBuilder sb = new StringBuilder();

-		if (responseContext.hasEntity()) {

-			responseContext.setEntityStream(logInboundEntity(sb, responseContext.getEntityStream(), DEFAULT_CHARSET));

-			log(sb);

-		}

-

-	}

-

-	@Override

-	public void aroundWriteTo(WriterInterceptorContext context) throws IOException, WebApplicationException {

-		final LoggingStream stream = (LoggingStream) context.getProperty(ENTITY_STREAM_PROPERTY);

-		context.proceed();

-		if (stream != null) {

-			log(stream.getStringBuilder(DEFAULT_CHARSET));

-		}

-	}

-

-	private class LoggingStream extends FilterOutputStream {

-

-		private final StringBuilder sb = new StringBuilder();

-		private final ByteArrayOutputStream baos = new ByteArrayOutputStream();

-

-		LoggingStream(OutputStream out) {

-			super(out);

-		}

-

-		StringBuilder getStringBuilder(Charset charset) {

-			// write entity to the builder

-			final byte[] entity = baos.toByteArray();

-

-			sb.append(new String(entity, 0, entity.length, charset));

-			if (entity.length > maxEntitySize) {

-				sb.append("...more...");

-			}

-			sb.append('\n');

-

-			return sb;

-		}

-

-		@Override

-		public void write(final int i) throws IOException {

-			if (baos.size() <= maxEntitySize) {

-				baos.write(i);

-			}

-			out.write(i);

-		}

-	}

-}

+/*-
+ * ============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.policy;
+
+import java.io.BufferedInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.FilterOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.logging.Logger;
+
+import javax.annotation.Priority;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.client.ClientRequestContext;
+import javax.ws.rs.client.ClientRequestFilter;
+import javax.ws.rs.client.ClientResponseContext;
+import javax.ws.rs.client.ClientResponseFilter;
+import javax.ws.rs.ext.WriterInterceptor;
+import javax.ws.rs.ext.WriterInterceptorContext;
+
+@Priority(Integer.MIN_VALUE)
+public class LoggingFilter implements ClientRequestFilter, ClientResponseFilter, WriterInterceptor {
+
+	private static final Logger logger = Logger.getLogger(LoggingFilter.class.getName());
+	private static final String ENTITY_STREAM_PROPERTY = "LoggingFilter.entityStream";
+	private static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;
+	private final int maxEntitySize;
+
+	public LoggingFilter() {
+		maxEntitySize = 1024 * 1024;
+	}
+
+	public LoggingFilter(int maxPayloadSize) {
+		this.maxEntitySize = Integer.min(maxPayloadSize, 1024 * 1024);
+	}
+
+	private void log(StringBuilder sb) {
+		logger.info(sb.toString());
+	}
+
+	private InputStream logInboundEntity(final StringBuilder b, InputStream stream, final Charset charset)
+			throws IOException {
+		if (!stream.markSupported()) {
+			stream = new BufferedInputStream(stream);
+		}
+		stream.mark(maxEntitySize + 1);
+		final byte[] entity = new byte[maxEntitySize + 1];
+		final int entitySize = stream.read(entity);
+		if (entitySize != -1) {
+			b.append(new String(entity, 0, Math.min(entitySize, maxEntitySize), charset));
+		}
+		if (entitySize > maxEntitySize) {
+			b.append("...more...");
+		}
+		b.append('\n');
+		stream.reset();
+		return stream;
+	}
+
+	@Override
+	public void filter(ClientRequestContext requestContext) throws IOException {
+		if (requestContext.hasEntity()) {
+			final OutputStream stream = new LoggingStream(requestContext.getEntityStream());
+			requestContext.setEntityStream(stream);
+			requestContext.setProperty(ENTITY_STREAM_PROPERTY, stream);
+		}
+	}
+
+	@Override
+	public void filter(ClientRequestContext requestContext, ClientResponseContext responseContext) throws IOException {
+		final StringBuilder sb = new StringBuilder();
+		if (responseContext.hasEntity()) {
+			responseContext.setEntityStream(logInboundEntity(sb, responseContext.getEntityStream(), DEFAULT_CHARSET));
+			log(sb);
+		}
+	}
+
+	@Override
+	public void aroundWriteTo(WriterInterceptorContext context) throws IOException, WebApplicationException {
+		final LoggingStream stream = (LoggingStream) context.getProperty(ENTITY_STREAM_PROPERTY);
+		context.proceed();
+		if (stream != null) {
+			log(stream.getStringBuilder(DEFAULT_CHARSET));
+		}
+	}
+
+	private class LoggingStream extends FilterOutputStream {
+
+		private final StringBuilder sb = new StringBuilder();
+		private final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+		LoggingStream(OutputStream out) {
+			super(out);
+		}
+
+		StringBuilder getStringBuilder(Charset charset) {
+			// write entity to the builder
+			final byte[] entity = baos.toByteArray();
+
+			sb.append(new String(entity, 0, entity.length, charset));
+			if (entity.length > maxEntitySize) {
+				sb.append("...more...");
+			}
+			sb.append('\n');
+
+			return sb;
+		}
+
+		@Override
+		public void write(final int i) throws IOException {
+			if (baos.size() <= maxEntitySize) {
+				baos.write(i);
+			}
+			out.write(i);
+		}
+	}
+}
\ No newline at end of file
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/PolicyRestClient.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/PolicyRestClient.java
index 71ae56f55b..4ed2a887ef 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/PolicyRestClient.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/PolicyRestClient.java
@@ -1,6 +1,6 @@
 /*-
  * ============LICENSE_START=======================================================
- * ONAP - SO
+ * OPENECOMP - MSO
  * ================================================================================
  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
  * ================================================================================
@@ -20,50 +20,47 @@
 
 package org.openecomp.mso.client.policy;
 
+import java.net.MalformedURLException;
 import java.util.Map;
+import java.util.Optional;
 import java.util.UUID;
 
-import javax.ws.rs.client.Entity;
-import javax.ws.rs.core.MediaType;
+import javax.ws.rs.client.ClientResponseFilter;
+import javax.ws.rs.core.UriBuilder;
+import javax.ws.rs.core.UriBuilderException;
 
+import org.openecomp.mso.client.ResponseExceptionMapperImpl;
+import org.openecomp.mso.client.RestProperties;
+import org.openecomp.mso.client.policy.entities.PolicyServiceType;
 import org.springframework.stereotype.Service;
 
 @Service
 public class PolicyRestClient extends RestClient {
 
-	private static final String ENDPOINT_KEY = "policy.endpoint";
 	private static final String X_ECOMP_REQUESTID = String.valueOf(UUID.randomUUID());
-
-	public PolicyRestClient() {
-		super(ENDPOINT_KEY);
+	private final PolicyRestProperties properties;
+	public PolicyRestClient(PolicyRestProperties props, PolicyServiceType serviceType) {
+		super(props, Optional.of(UriBuilder.fromPath(serviceType.toString()).build()));
+		this.properties = props;
+		this.getClient();
 	}
 
 	@Override
 	protected void initializeHeaderMap(Map<String, String> headerMap) {
-		headerMap.put("ClientAuth", properties.get("policy.client.auth"));
-		headerMap.put("Authorization", properties.get("policy.auth"));
-		headerMap.put("Environment", properties.get("policy.environment"));
-		headerMap.put("X-ECOMP-RequestID", X_ECOMP_REQUESTID);
+		headerMap.put("ClientAuth", properties.getClientAuth());
+		headerMap.put("Authorization", properties.getAuth());
+		headerMap.put("Environment", properties.getEnvironment());
+		this.addRequestId(X_ECOMP_REQUESTID);
 	}
 
-	public PolicyDecision getDecision(String serviceType, String vnfType, String bbID, String workStep,
-			String errorCode) {
-		DecisionAttributes decisionAttributes = new DecisionAttributes();
-		decisionAttributes.setServiceType(serviceType);
-		decisionAttributes.setVNFType(vnfType);
-		decisionAttributes.setBBID(bbID);
-		decisionAttributes.setWorkStep(workStep);
-		decisionAttributes.setErrorCode(errorCode);
-
-		return this.getDecision(decisionAttributes);
+	@Override
+	protected Optional<ClientResponseFilter> addResponseFilter() {
+		return Optional.of(new ResponseExceptionMapperImpl());
 	}
 
-	private PolicyDecision getDecision(DecisionAttributes decisionAttributes) {
-		PolicyDecisionRequest decisionRequest = new PolicyDecisionRequest();
-		decisionRequest.setDecisionAttributes(decisionAttributes);
-		decisionRequest.setEcompcomponentName(ECOMP_COMPONENT_NAME);
-
-		return this.getBuilder().accept(MediaType.APPLICATION_JSON_TYPE)
-				.post(Entity.entity(decisionRequest, MediaType.APPLICATION_JSON)).readEntity(PolicyDecision.class);
+	@Override
+	public RestClient addRequestId(String requestId) {
+		this.headerMap.put("X-ECOMP-RequestID", requestId);
+		return this;
 	}
 }
\ No newline at end of file
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/PolicyRestProperties.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/PolicyRestProperties.java
new file mode 100644
index 0000000000..d9336768fc
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/PolicyRestProperties.java
@@ -0,0 +1,64 @@
+/*-
+ * ============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.policy;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.Map;
+
+import org.openecomp.mso.bpmn.core.PropertyConfiguration;
+import org.openecomp.mso.client.RestProperties;
+
+public class PolicyRestProperties implements RestProperties {
+
+	
+	final Map<String, String> props;
+	public PolicyRestProperties() {
+		this.props = PropertyConfiguration.getInstance().getProperties("mso.bpmn.urn.properties");
+
+	}
+	@Override
+	public URL getEndpoint() {
+		try {
+			return new URL(props.getOrDefault("policy.endpoint", ""));
+		} catch (MalformedURLException e) {
+			return null;
+		}
+	}
+
+	@Override
+	public String getSystemName() {
+		return "MSO";
+	}
+	
+	public String getClientAuth() {
+		return props.get("policy.client.auth");
+	}
+	
+	public String getAuth() {
+		return props.get("policy.auth");
+	}
+	
+	public String getEnvironment() {
+		return props.get("policy.environment");
+	}
+
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/RestClient.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/RestClient.java
index 81c072ff33..74b1c3f802 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/RestClient.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/RestClient.java
@@ -20,51 +20,104 @@
 
 package org.openecomp.mso.client.policy;
 
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URL;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.logging.Logger;
+import java.util.Optional;
 
+import javax.ws.rs.client.Client;
 import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.ClientResponseFilter;
+import javax.ws.rs.client.Entity;
 import javax.ws.rs.client.Invocation.Builder;
 import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.GenericType;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.UriBuilder;
+import javax.ws.rs.ext.ContextResolver;
 
-
-import org.openecomp.mso.bpmn.core.PropertyConfiguration;
+import org.apache.log4j.Logger;
+import org.openecomp.mso.client.RestProperties;
 import org.openecomp.mso.logger.MsoLogger;
 import org.springframework.stereotype.Service;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 @Service
 public abstract class RestClient {
 	protected static final String ECOMP_COMPONENT_NAME = "MSO";
-
+	
+	private static final int MAX_PAYLOAD_SIZE = 1024 * 1024;
 	private WebTarget webTarget;
 
 	protected final Map<String, String> headerMap;
 	protected final MsoLogger msoLogger;
-	protected Map<String, String> properties;
-	protected String host;
+	protected URL host;
+	protected Optional<URI> path;
+	protected Logger logger;
+	protected String accept;
+	protected String contentType;
 
-	protected RestClient(String endpointKey) {
-		Logger logger = Logger.getLogger(getClass().getName());
+	protected RestClient(RestProperties props, Optional<URI> path) {
+		logger = Logger.getLogger(getClass().getName());
 		msoLogger = MsoLogger.getMsoLogger(MsoLogger.Catalog.GENERAL);
-		
-		properties = PropertyConfiguration.getInstance().getProperties("mso.bpmn.urn.properties");
+
 		headerMap = new HashMap<>();
-		initializeHeaderMap(headerMap);
+		try {
+			host = props.getEndpoint();
+		} catch (MalformedURLException e) {
+			logger.error("url not valid", e);
+			throw new RuntimeException(e);
+		}
+		
+		this.path = path;
+		initializeClient(getClient());
+	}
 
-		host = this.getHost(endpointKey);
+	protected RestClient(RestProperties props, Optional<URI> path, String accept, String contentType) {
+		this(props, path);
+		this.accept = accept;
+		this.contentType = contentType;
 
-		webTarget = ClientBuilder.newClient().register(logger).register(new LoggingFilter())
-				.register(new CommonObjectMapperProvider()).target(host);
 	}
 
-	private String getHost(String key) {
-		return properties.get(key);
+	protected RestClient(URL host, String contentType) {
+		headerMap = new HashMap<>();
+		logger = Logger.getLogger(getClass().getName());
+		msoLogger = MsoLogger.getMsoLogger(MsoLogger.Catalog.GENERAL);
+		this.path = Optional.empty();
+		this.host = host;
+		this.contentType = contentType;
+		initializeClient(getClient());
+	}
+
+	/**
+	 * Override method to return false to disable logging.
+	 * 
+	 * @return true - to enable logging, false otherwise
+	 */
+	protected boolean enableLogging() {
+		return true;
+	}
+	
+	/**
+	 * Override method to return custom value for max payload size.
+	 * 
+	 * @return Default value for MAX_PAYLOAD_SIZE = 1024 * 1024
+	 */
+	protected int getMaxPayloadSize()
+	{
+		return MAX_PAYLOAD_SIZE;
 	}
 
 	protected Builder getBuilder() {
+
 		Builder builder = webTarget.request();
+		initializeHeaderMap(headerMap);
 
 		for (Entry<String, String> entry : headerMap.entrySet()) {
 			builder.header(entry.getKey(), entry.getValue());
@@ -73,4 +126,100 @@ public abstract class RestClient {
 	}
 
 	protected abstract void initializeHeaderMap(Map<String, String> headerMap);
+
+	protected abstract Optional<ClientResponseFilter> addResponseFilter();
+
+	public abstract RestClient addRequestId(String requestId);
+
+	protected ContextResolver<ObjectMapper> getMapper() {
+		return new CommonObjectMapperProvider();
+	}
+
+	protected String getAccept() {
+		return accept;
+	}
+
+	protected String getContentType() {
+		return contentType;
+	}
+
+	protected String getMergeContentType() {
+		return "application/merge-patch+json";
+	}
+
+	protected Client getClient() {
+		return ClientBuilder.newBuilder().build();
+	}
+
+	protected void initializeClient(Client client) {
+		if (this.enableLogging()) {
+			client.register(logger).register(new LoggingFilter(this.getMaxPayloadSize()));
+		}
+		client.register(this.getMapper());
+		Optional<ClientResponseFilter> responseFilter = this.addResponseFilter();
+		if (responseFilter.isPresent()) {
+			client.register(responseFilter.get());
+		}
+		if (!path.isPresent()) {
+			webTarget = client.target(host.toString());
+		} else {
+			webTarget = client.target(UriBuilder.fromUri(host + path.get().toString()));
+		}
+		this.accept = MediaType.APPLICATION_JSON;
+		this.contentType = MediaType.APPLICATION_JSON;
+	}
+
+	public Response get() {
+		return this.getBuilder().accept(this.getAccept()).get();
+	}
+
+	public Response post(Object obj) {
+		return this.getBuilder().accept(this.getAccept()).post(Entity.entity(obj, this.getContentType()));
+	}
+
+	public Response patch(Object obj) {
+		return this.getBuilder().header("X-HTTP-Method-Override", "PATCH").accept(this.getAccept())
+				.post(Entity.entity(obj, this.getMergeContentType()));
+	}
+
+	public Response put(Object obj) {
+		return this.getBuilder().accept(this.getAccept()).put(Entity.entity(obj, this.getContentType()));
+	}
+
+	public Response delete() {
+		return this.getBuilder().accept(this.getAccept()).delete();
+	}
+
+	public Response delete(Object obj) {
+		return this.getBuilder().header("X-HTTP-Method-Override", "DELETE").accept(this.getAccept())
+				.put(Entity.entity(obj, this.getContentType()));
+	}
+
+	public <T> T get(Class<T> resultClass) {
+		return this.get().readEntity(resultClass);
+	}
+
+	public <T> T get(GenericType<T> resultClass) {
+		return this.get().readEntity(resultClass);
+	}
+
+	public <T> T post(Object obj, Class<T> resultClass) {
+		return this.post(obj).readEntity(resultClass);
+	}
+
+	public <T> T patch(Object obj, Class<T> resultClass) {
+		return this.patch(obj).readEntity(resultClass);
+	}
+
+	public <T> T put(Object obj, Class<T> resultClass) {
+		return this.put(obj).readEntity(resultClass);
+	}
+
+	public <T> T delete(Class<T> resultClass) {
+		return this.delete().readEntity(resultClass);
+	}
+	
+	public <T> T delete(Object obj, Class<T> resultClass) {
+		return this.delete(obj).readEntity(resultClass);
+	}
 }
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/entities/PolicyServiceType.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/entities/PolicyServiceType.java
new file mode 100644
index 0000000000..b5ab63ce35
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/policy/entities/PolicyServiceType.java
@@ -0,0 +1,48 @@
+/*-
+ * ============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.policy.entities;
+
+public enum PolicyServiceType {
+	GET_CONFIG("getConfig"),
+	SEND_EVENT("sendEvent"),
+	PUSH_POLICY("pushPolicy"),
+	CREATE_POLICY("createPolicy"),
+	UPDATE_POLICY("updatePolicy"),
+	GET_DECISION("getDecision"),
+	GET_METRICS("getMetrics"),
+	DELETE_POLICY("deletePolicy"),
+	LIST_CONFIG("listConfig"),
+	CREATE_DICTIONARY_ITEM("createDictionaryItem"),
+	UPDATE_DICTIONARY_ITEM("updateDictionaryItem"),
+	GET_DICTIONARY_ITEMS("getDictionaryItems");
+	
+	private final String name;
+	
+	private PolicyServiceType(String name) {
+		this.name = name;
+	}
+	
+	@Override
+	public String toString() {
+		return name;
+	}
+	
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/SDNOValidator.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/SDNOValidator.java
index eb18e10bc3..c746c0d448 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/SDNOValidator.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/SDNOValidator.java
@@ -32,6 +32,6 @@ public interface SDNOValidator {
 	 * @throws IOException
 	 * @throws Exception
 	 */
-	public void healthDiagnostic(String vnfName, String uuid) throws IOException, Exception;
+	public void healthDiagnostic(String vnfId, String requestingUserId) throws IOException, Exception;
 
 }
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/SDNOValidatorImpl.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/SDNOValidatorImpl.java
index 92a00eee65..bdb4aa94ee 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/SDNOValidatorImpl.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/SDNOValidatorImpl.java
@@ -30,7 +30,7 @@ import org.openecomp.mso.client.dmaap.DmaapPublisher;
 import org.openecomp.mso.client.exceptions.SDNOException;
 import org.openecomp.mso.jsonpath.JsonPathUtil;
 
-public class SDNOValidatorImpl implements SDNOValidator, Consumer {
+public class SDNOValidatorImpl implements SDNOValidator {
 
 	private final static String aafUserName = "something";
 	private final static String clientName = "MSO";
@@ -51,103 +51,4 @@ public class SDNOValidatorImpl implements SDNOValidator, Consumer {
 		//block and continue to poll waiting for response
 	}
 
-	protected SDNO buildRequestDiagnostic(String vnfName, String uuid, String oamIp) {
-		
-		Input input = new Input();
-		SDNO parentRequest = new SDNO();
-		Body body = new Body();
-		parentRequest.setBody(body);
-		parentRequest.setNodeType("vPE");
-		parentRequest.setOperation("health-diagnostic");
-		
-		body.setInput(input);
-		
-		RequestHealthDiagnostic request = new RequestHealthDiagnostic();
-		request.setRequestClientName(clientName);
-		request.setRequestNodeName(vnfName);
-		request.setRequestNodeIp(oamIp); //generic-vnf oam ip
-		request.setRequestUserId(aafUserName); //mech id?
-		request.setRequestId(uuid); //something to identify this request by for polling
-		
-		input.setRequestHealthDiagnostic(request);
-		
-		return parentRequest;
-	}
-	protected void submitRequest(String json) throws FileNotFoundException, IOException, InterruptedException {
-		DmaapPublisher publisher = new DmaapPublisher(this.producerFilePath);
-		publisher.send(json);
-	}
-	protected boolean pollForResponse(DmaapConsumer consumer, String uuid) throws Exception {
-		this.uuid = uuid;
-		return consumer.consume(this);
-	}
-	
-	@Override
-	public boolean continuePolling() {
-		return continuePolling;
-	}
-	
-	@Override
-	public void stopProcessingMessages() {
-		continuePolling = false;
-	}
-	@Override
-	public void processMessage(String message) throws Exception {
-		if (isHealthDiagnostic(message, this.getRequestId())) {
-			if (!healthDiagnosticSuccessful(message)) {
-				Optional<String> statusMessage = this.getStatusMessage(message);
-				if (statusMessage.isPresent()) {
-					throw new SDNOException(statusMessage.get());
-				} else {
-					throw new SDNOException();
-				}
-			} else {
-				stopProcessingMessages();
-			}
-		}
-	}
-	
-	@Override
-	public boolean isAccepted(String message) {
-		if (isResultInfo(message)) {
-			Optional<String> code = isAccepted(message, this.getRequestId());
-			if (code.isPresent()) {
-				if ("202".equals(code.get())) {
-					return true;
-				} else {
-					//TODO check other statuses 400 and 500
-				}
-			} else {
-				//TODO throw error
-			}
-		}
-		
-		return false;
-	}
-	
-	@Override
-	public String getRequestId() {
-		return uuid;
-	}
-	
-	protected Optional<String> isAccepted(String json, String uuid) {
-		return JsonPathUtil.getInstance().locateResult(json, String.format("$.result-info[?(@.status=='ACCEPTED' && @.request-id=='%s')].code", uuid));
-	}
-	
-	protected boolean isResultInfo(String json) {
-		return JsonPathUtil.getInstance().pathExists(json, "$[?(@.result-info)]");
-	}
-	
-	protected boolean isHealthDiagnostic(String json, String uuid) {
-		return JsonPathUtil.getInstance().pathExists(json, String.format("$[?(@.result-info.request-id=='%s')].%s", uuid, healthDiagnosticPath));
-	}
-	
-	protected boolean healthDiagnosticSuccessful(String json) {
-		return JsonPathUtil.getInstance().pathExists(json, "$." + healthDiagnosticPath + "[?(@.response-status=='Success')]");
-	}
-	
-	protected Optional<String> getStatusMessage(String json) {
-		return JsonPathUtil.getInstance().locateResult(json, "$." + healthDiagnosticPath + ".response-details-json");
-	}
-	
 }
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/Body.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/beans/Body.java
index 2db4dea069..cc8ce81dec 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/Body.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/beans/Body.java
@@ -18,7 +18,7 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.openecomp.mso.client.sdno;
+package org.openecomp.mso.client.sdno.beans;
 
 import java.io.Serializable;
 import java.util.HashMap;
@@ -41,7 +41,7 @@ public class Body implements Serializable
     @JsonProperty("input")
     private Input input;
     @JsonIgnore
-    private Map<String, Object> additionalProperties = new HashMap<>();
+    private Map<String, Object> additionalProperties = new HashMap<String, Object>();
     private final static long serialVersionUID = 9101706044452851559L;
 
     @JsonProperty("input")
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/Input.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/beans/Input.java
index 25eb2ed270..7b9bf6f2c7 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/Input.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/beans/Input.java
@@ -18,7 +18,7 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.openecomp.mso.client.sdno;
+package org.openecomp.mso.client.sdno.beans;
 
 import java.io.Serializable;
 import java.util.HashMap;
@@ -33,15 +33,17 @@ import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @JsonPropertyOrder({
-    "request-healthdiagnostic"
+    "request-healthdiagnostic",
+    "request-hd-custom"
 })
 public class Input implements Serializable
 {
 
     @JsonProperty("request-healthdiagnostic")
     private RequestHealthDiagnostic RequestHealthDiagnostic;
+    
     @JsonIgnore
-    private Map<String, Object> additionalProperties = new HashMap<>();
+    private Map<String, Object> additionalProperties = new HashMap<String, Object>();
     private final static long serialVersionUID = 7155546785389227528L;
 
     @JsonProperty("request-healthdiagnostic")
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/RequestHealthDiagnostic.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/beans/RequestHealthDiagnostic.java
index b1ed77b513..104dbb7ee6 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/RequestHealthDiagnostic.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/beans/RequestHealthDiagnostic.java
@@ -18,7 +18,7 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.openecomp.mso.client.sdno;
+package org.openecomp.mso.client.sdno.beans;
 
 import java.io.Serializable;
 import java.util.HashMap;
@@ -59,7 +59,7 @@ public class RequestHealthDiagnostic implements Serializable
     @JsonProperty("health-diagnostic-code")
     private String healthDiagnosticCode;
     @JsonIgnore
-    private Map<String, Object> additionalProperties = new HashMap<>();
+    private Map<String, Object> additionalProperties = new HashMap<String, Object>();
     private final static long serialVersionUID = 1166788526178388021L;
 
     @JsonProperty("request-client-name")
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/ResultInfo.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/beans/ResultInfo.java
index 0997b2df90..4d029cb8cb 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/ResultInfo.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/beans/ResultInfo.java
@@ -18,7 +18,7 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.openecomp.mso.client.sdno;
+package org.openecomp.mso.client.sdno.beans;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -51,7 +51,7 @@ private String requestId;
 @JsonProperty("status")
 private String status;
 @JsonIgnore
-private Map<String, Object> additionalProperties = new HashMap<>();
+private Map<String, Object> additionalProperties = new HashMap<String, Object>();
 
 @JsonProperty("client-name")
 public String getClientName() {
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/SDNO.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/beans/SDNO.java
index caa3ff58e9..4edd54ca09 100644
--- a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/SDNO.java
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/beans/SDNO.java
@@ -18,7 +18,7 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.openecomp.mso.client.sdno;
+package org.openecomp.mso.client.sdno.beans;
 
 import java.io.Serializable;
 import java.util.HashMap;
@@ -34,6 +34,7 @@ import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @JsonPropertyOrder({
     "operation",
+    "nodeLoc",
     "nodeType",
     "body"
 })
@@ -42,12 +43,14 @@ public class SDNO implements Serializable
 
     @JsonProperty("operation")
     private String operation;
+    @JsonProperty("nodeLoc")
+    private String nodeLoc;
     @JsonProperty("nodeType")
     private String nodeType;
     @JsonProperty("body")
     private Body body;
     @JsonIgnore
-    private Map<String, Object> additionalProperties = new HashMap<>();
+    private Map<String, Object> additionalProperties = new HashMap<String, Object>();
     private final static long serialVersionUID = -5303297382564282650L;
 
     @JsonProperty("operation")
@@ -59,8 +62,23 @@ public class SDNO implements Serializable
     public void setOperation(String operation) {
         this.operation = operation;
     }
+    
+    @JsonProperty("nodeLoc")
+    public String getNodeLoc() {
+		return nodeLoc;
+	}
+    
+    @JsonProperty("nodeLoc")
+	public void setNodeLoc(String nodeLoc) {
+		this.nodeLoc = nodeLoc;
+	}
+    
+    public SDNO withNodeLoc(String nodeLoc) {
+    	this.nodeLoc = nodeLoc;
+    	return this;
+    }
 
-    public SDNO withOperation(String operation) {
+	public SDNO withOperation(String operation) {
         this.operation = operation;
         return this;
     }
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapConsumer.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapConsumer.java
new file mode 100644
index 0000000000..f23d882b53
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapConsumer.java
@@ -0,0 +1,156 @@
+/*-
+ * ============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.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;
+
+public class SDNOHealthCheckDmaapConsumer extends DmaapConsumer {
+
+	private final String uuid;
+	private boolean continuePolling = true;
+	private final static String healthDiagnosticPath = "body.output.*";
+
+	public SDNOHealthCheckDmaapConsumer() throws FileNotFoundException, IOException {
+		this("none");
+	}
+	
+	public SDNOHealthCheckDmaapConsumer(String uuid) throws FileNotFoundException, IOException {
+		super();
+		this.uuid = uuid;
+	}
+	
+	@Override
+	public String getUserName() {
+		return msoProperties.get("sdno.health-check.dmaap.username");
+	}
+
+	@Override
+	public String getPassword() {
+		return msoProperties.get("sdno.health-check.dmaap.password");
+	}
+
+	@Override
+	public String getTopic() {
+		return msoProperties.get("sdno.health-check.dmaap.subscriber.topic");
+	}
+
+	@Override
+	public boolean continuePolling() {
+		return continuePolling;
+	}
+	
+	@Override
+	public void stopProcessingMessages() {
+		continuePolling = false;
+	}
+	@Override
+	public void processMessage(String message) throws Exception {
+		if (isHealthDiagnostic(message, this.getRequestId())) {
+			if (!healthDiagnosticSuccessful(message)) {
+				Optional<String> statusMessage = this.getStatusMessage(message);
+				if (statusMessage.isPresent()) {
+					throw new SDNOException("failed with message " + statusMessage.get());
+				} else {
+					throw new SDNOException("failed with no status message");
+				}
+			} else {
+				auditLogger.info("successful health diagnostic found for request: " + this.getRequestId());
+				stopProcessingMessages();
+			}
+		}
+	}
+	
+	@Override
+	public boolean isAccepted(String message) {
+		if (isResultInfo(message)) {
+			Optional<String> code = isAccepted(message, this.getRequestId());
+			if (code.isPresent()) {
+				if ("202".equals(code.get())) {
+					return true;
+				} else {
+					//TODO check other statuses 400 and 500
+				}
+			} else {
+				//TODO throw error
+			}
+		}
+		
+		return false;
+	}
+	
+	@Override
+	public boolean isFailure(String message) {
+		if (isResultInfo(message)) {
+			Optional<String> code = isFailure(message, this.getRequestId());
+			if (code.isPresent()) {
+				if ("500".equals(code.get())) {
+					return true;
+				} else {
+					//TODO check other statuses 400 and 500
+				}
+			} else {
+				//TODO throw error
+			}
+		}
+		
+		return false;
+	}
+	
+	@Override
+	public String getRequestId() {
+		return uuid;
+	}
+	
+	protected Optional<String> isAccepted(String json, String uuid) {
+		return JsonPathUtil.getInstance().locateResult(json, String.format("$.result-info[?(@.status=='ACCEPTED' && @.request-id=='%s')].code", uuid));
+	}
+	
+	protected Optional<String> isFailure(String json, String uuid) {
+		return JsonPathUtil.getInstance().locateResult(json, String.format("$.result-info[?(@.status=='FAILURE' && @.request-id=='%s')].code", uuid));
+	}
+	
+	protected boolean isResultInfo(String json) {
+		return JsonPathUtil.getInstance().pathExists(json, "$[?(@.result-info)]");
+	}
+	
+	protected boolean isHealthDiagnostic(String json, String uuid) {
+		return JsonPathUtil.getInstance().pathExists(json, String.format("$[?(@.result-info.request-id=='%s')].%s", uuid, healthDiagnosticPath));
+	}
+	
+	protected boolean healthDiagnosticSuccessful(String json) {
+		return JsonPathUtil.getInstance().pathExists(json, "$." + healthDiagnosticPath + "[?(@.response-status=='Success')]");
+	}
+	
+	protected Optional<String> getStatusMessage(String json) {
+		return JsonPathUtil.getInstance().locateResult(json, "$." + healthDiagnosticPath + ".error-message");
+	}
+	
+	@Override
+	public int getMaximumElapsedTime() {
+		return 300000;
+	}
+}
diff --git a/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapPublisher.java b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapPublisher.java
new file mode 100644
index 0000000000..73f06b8e58
--- /dev/null
+++ b/bpmn/MSOCommonBPMN/src/main/java/org/openecomp/mso/client/sdno/dmaap/SDNOHealthCheckDmaapPublisher.java
@@ -0,0 +1,50 @@
+/*-
+ * ============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.FileNotFoundException;
+import java.io.IOException;
+
+import org.openecomp.mso.client.dmaap.DmaapPublisher;
+
+public class SDNOHealthCheckDmaapPublisher extends DmaapPublisher {
+	
+	public SDNOHealthCheckDmaapPublisher() throws FileNotFoundException, IOException {
+		super();
+	}
+	
+	@Override
+	public String getUserName() {
+		return msoProperties.get("sdno.health-check.dmaap.username");
+	}
+
+	@Override
+	public String getPassword() {
+		return msoProperties.get("sdno.health-check.dmaap.password");
+	}
+
+	@Override
+	public String getTopic() {
+		return msoProperties.get("sdno.health-check.dmaap.publisher.topic");
+	}
+	
+
+}
diff --git a/bpmn/MSOCommonBPMN/src/test/java/org/openecomp/mso/client/policy/PolicyDecisionTest.java b/bpmn/MSOCommonBPMN/src/test/java/org/openecomp/mso/client/policy/PolicyDecisionTest.java
deleted file mode 100644
index 8370f74851..0000000000
--- a/bpmn/MSOCommonBPMN/src/test/java/org/openecomp/mso/client/policy/PolicyDecisionTest.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*-
- * ============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.policy;
-
-import static org.junit.Assert.assertEquals;
-
-import java.io.IOException;
-import java.util.Map;
-
-import org.junit.BeforeClass;
-import org.junit.Ignore;
-import org.junit.Test;
-
-public class PolicyDecisionTest {
-	
-	@BeforeClass
-	public static void setUp() {
-		System.setProperty("mso.config.path", "src/test/resources");
-	}
-	@Test
-	public void successReadProperties() throws IOException, Exception {
-		PolicyRestClient client = new PolicyRestClient();
-		Map<String, String> headerMap = client.headerMap;
-		
-		assertEquals("Found expected Client Auth", headerMap.get("ClientAuth"), "Basic bTAzNzQzOnBvbGljeVIwY2sk");
-		assertEquals("Found expected Authorization", headerMap.get("Authorization"), "Basic dGVzdHBkcDphbHBoYTEyMw==");
-		assertEquals("Found expected Environment", headerMap.get("Environment"), "TEST");
-		assertEquals("Has X-ECOMP-RequestID", headerMap.containsKey("X-ECOMP-RequestID"), true);
-	}
-	
-	@Test
-	@Ignore // IGNORED FOR 1710 MERGE TO ONAP
-	public void restEndpointCall() throws IOException, Exception {
-		PolicyRestClient client = new PolicyRestClient();
-		PolicyDecision decision = client.getDecision("X", "Y", "BB1", "1", "123");
-		assertEquals("Decision is correct", decision.getDecision(), "PERMIT");
-		assertEquals("Decision details is correct", decision.getDetails(), "Abort");
-	}
-/*
-	@Test
-	public void failure() throws IOException, Exception {
-		when(dmaapConsumer.getMRConsumer()).thenReturn(mrConsumer);
-		when(mrConsumer.fetch())
-				.thenReturn(Arrays.asList(new String[] { getJson("response.json"), getJson("output-failure.json") }));
-
-		SDNOValidatorImpl validator = new SDNOValidatorImpl();
-		thrown.expect(SDNOException.class);
-		thrown.expectMessage("my message");
-		boolean result = validator.pollForResponse(dmaapConsumer, "xyz123");
-
-	}*/
-}