AT&T 1712 and 1802 release code

This is code from AT&T's 1712 and 1802 releases.

Change-Id: Ie1e85851e94bc66c4d9514a0226c221939531a04
Issue-ID: SO-425
Signed-off-by: Rob Daugherty <rd472p@att.com>

13 files changed, 774 insertions, 0 deletions

diff --git a/common/src/main/java/org/openecomp/mso/client/dmaap/Consumer.java b/common/src/main/java/org/openecomp/mso/client/dmaap/Consumer.java
new file mode 100644
index 0000000000..0e00ae5da8
--- /dev/null
+++ b/common/src/main/java/org/openecomp/mso/client/dmaap/Consumer.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 Consumer {
+
+	public Iterable<String> fetch();
+}
diff --git a/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapClient.java b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapClient.java
new file mode 100644
index 0000000000..aabeb8aabf
--- /dev/null
+++ b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapClient.java
@@ -0,0 +1,73 @@
+/*-
+ * ============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.IOException;
+import java.util.Base64;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Properties;
+
+import org.openecomp.mso.client.defaultproperties.DefaultDmaapPropertiesImpl;
+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 IOException {
+		Resource resource = new ClassPathResource(filepath);
+		DmaapProperties dmaapProperties = DmaapPropertiesLoader.getInstance().getNewImpl();
+		if (dmaapProperties == null) {
+			auditLogger.error("No RestProperty implementation found on classpath, loading default");
+			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());
+		if (this.getHost().isPresent()) {
+			this.properties.put("host", this.getHost().get());
+		}
+	}
+	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();
+	public abstract Optional<String> getHost();
+}
diff --git a/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.java b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.java
new file mode 100644
index 0000000000..033951612d
--- /dev/null
+++ b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapConsumer.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;
+
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+
+import org.openecomp.mso.client.dmaap.exceptions.DMaaPConsumerFailure;
+import org.openecomp.mso.client.dmaap.exceptions.ExceededMaximumPollingTime;
+import org.openecomp.mso.client.dmaap.rest.RestConsumer;
+
+import com.google.common.base.Stopwatch;
+
+public abstract class DmaapConsumer extends DmaapClient {
+
+	public DmaapConsumer() throws FileNotFoundException, IOException {
+		super("dmaap/default-consumer.properties");
+	}
+	
+	public Consumer getConsumer() throws FileNotFoundException, IOException {
+		return new RestConsumer(this.properties);
+	}
+	public boolean consume() throws Exception {
+		
+		Consumer mrConsumer = this.getConsumer();
+		int iterations = 0;
+		boolean accepted = false;
+		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();
+			}
+		}
+	}
+	
+	/**
+	 * 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/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapProperties.java b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapProperties.java
new file mode 100644
index 0000000000..7bdd7dfe40
--- /dev/null
+++ b/common/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/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapPropertiesLoader.java b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapPropertiesLoader.java
new file mode 100644
index 0000000000..3c861991f6
--- /dev/null
+++ b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapPropertiesLoader.java
@@ -0,0 +1,62 @@
+/*-
+ * ============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.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() {
+		return this.getImpl(false);
+	}
+	public DmaapProperties getNewImpl() {
+		return this.getImpl(true);
+	}
+	private DmaapProperties getImpl(boolean forceNewInstance) {
+		for (DmaapProperties service : services) {
+			if (forceNewInstance) {
+				try {
+					return service.getClass().newInstance();
+				} catch (InstantiationException | IllegalAccessException e) {
+					/* all spi implementations must provide a public
+					 * no argument constructor
+					 */
+				}
+			} else {
+				return service;
+			}
+		}
+		
+		return null;
+	}
+}
diff --git a/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapPublisher.java b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapPublisher.java
new file mode 100644
index 0000000000..d2752c531b
--- /dev/null
+++ b/common/src/main/java/org/openecomp/mso/client/dmaap/DmaapPublisher.java
@@ -0,0 +1,51 @@
+/*-
+ * ============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 org.openecomp.mso.client.dmaap.rest.RestPublisher;
+
+public abstract class DmaapPublisher extends DmaapClient {
+	
+	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;
+		
+	}
+	
+	public DmaapPublisher(long seconds) throws FileNotFoundException, IOException {
+		this();
+		this.seconds = seconds;
+	}
+	
+	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);
+	}
+
+
+}
diff --git a/common/src/main/java/org/openecomp/mso/client/dmaap/Publisher.java b/common/src/main/java/org/openecomp/mso/client/dmaap/Publisher.java
new file mode 100644
index 0000000000..ed1b3f3796
--- /dev/null
+++ b/common/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/common/src/main/java/org/openecomp/mso/client/dmaap/exceptions/DMaaPConsumerFailure.java b/common/src/main/java/org/openecomp/mso/client/dmaap/exceptions/DMaaPConsumerFailure.java
new file mode 100644
index 0000000000..b1ae2b7e89
--- /dev/null
+++ b/common/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/common/src/main/java/org/openecomp/mso/client/dmaap/exceptions/ExceededMaximumPollingTime.java b/common/src/main/java/org/openecomp/mso/client/dmaap/exceptions/ExceededMaximumPollingTime.java
new file mode 100644
index 0000000000..c0cda6b0ed
--- /dev/null
+++ b/common/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/common/src/main/java/org/openecomp/mso/client/dmaap/rest/DMaaPRestClient.java b/common/src/main/java/org/openecomp/mso/client/dmaap/rest/DMaaPRestClient.java
new file mode 100644
index 0000000000..ad744ad31b
--- /dev/null
+++ b/common/src/main/java/org/openecomp/mso/client/dmaap/rest/DMaaPRestClient.java
@@ -0,0 +1,58 @@
+/*-
+ * ============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 java.util.UUID;
+
+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, UUID.randomUUID(), contentType);
+		this.username = username;
+		this.password = password;
+	}
+
+	@Override
+	protected void initializeHeaderMap(Map<String, String> headerMap) {
+		headerMap.put("Authorization", "Basic " + Base64.getEncoder().encodeToString((username + ":" + password).getBytes()));
+	}
+
+	@Override
+	protected Optional<ClientResponseFilter> addResponseFilter() {
+		return Optional.of(new ResponseExceptionMapperImpl());
+	}
+
+	@Override
+	public RestClient addRequestId(UUID requestId) {
+		return this;
+	}
+}
diff --git a/common/src/main/java/org/openecomp/mso/client/dmaap/rest/PropertiesBean.java b/common/src/main/java/org/openecomp/mso/client/dmaap/rest/PropertiesBean.java
new file mode 100644
index 0000000000..e661dd87bd
--- /dev/null
+++ b/common/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/common/src/main/java/org/openecomp/mso/client/dmaap/rest/RestConsumer.java b/common/src/main/java/org/openecomp/mso/client/dmaap/rest/RestConsumer.java
new file mode 100644
index 0000000000..c2fb6f676f
--- /dev/null
+++ b/common/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(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/common/src/main/java/org/openecomp/mso/client/dmaap/rest/RestPublisher.java b/common/src/main/java/org/openecomp/mso/client/dmaap/rest/RestPublisher.java
new file mode 100644
index 0000000000..3dab12f559
--- /dev/null
+++ b/common/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(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);
+	}
+}