Revert package name changes

Reverted package name changes to avoid any potential issues. Renamed maven
group id only.

Issue-id: DMAAP-74
Change-Id: I36c2aef063050c265640b79e6dc0e8ab7add8d22
Signed-off-by: Varun Gudisena <vg411h@att.com>

9 files changed, 3132 insertions, 0 deletions

diff --git a/src/main/java/com/att/nsa/mr/client/impl/Clock.java b/src/main/java/com/att/nsa/mr/client/impl/Clock.java
new file mode 100644
index 0000000..ace791e
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/Clock.java
@@ -0,0 +1,63 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 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=========================================================
+ *
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package com.att.nsa.mr.client.impl;
+
+public class Clock
+{
+	public synchronized static Clock getIt ()
+	{
+		if ( sfClock == null )
+		{
+			sfClock = new Clock ();
+		}
+		return sfClock;
+	}
+
+	/**
+	 * Get the system's current time in milliseconds.
+	 * @return the current time
+	 */
+	public static long now ()
+	{
+		return getIt().nowImpl ();
+	}
+
+	/**
+	 * Get current time in milliseconds
+	 * @return current time in ms
+	 */
+	protected long nowImpl ()
+	{
+		return System.currentTimeMillis ();
+	}
+
+	protected Clock ()
+	{
+	}
+
+	private static Clock sfClock = null;
+
+	protected synchronized static void register ( Clock testClock )
+	{
+		sfClock = testClock;
+	}
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRBaseClient.java b/src/main/java/com/att/nsa/mr/client/impl/MRBaseClient.java
new file mode 100644
index 0000000..a9fd9bd
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRBaseClient.java
@@ -0,0 +1,393 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 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=========================================================
+ *
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package com.att.nsa.mr.client.impl;
+
+import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.concurrent.TimeUnit;
+
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.Response;
+
+import org.apache.http.HttpException;
+import org.glassfish.jersey.client.authentication.HttpAuthenticationFeature;
+import org.glassfish.jersey.internal.util.Base64;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.json.JSONTokener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.nsa.apiClient.http.CacheUse;
+import com.att.nsa.apiClient.http.HttpClient;
+import com.att.nsa.mr.client.MRClient;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+//import com.fasterxml.jackson.core.JsonProcessingException;
+
+public class MRBaseClient extends HttpClient implements MRClient 
+{
+	
+	private static final String MR_AUTH_CONSTANT = "X-CambriaAuth";
+	private static final String MR_DATE_CONSTANT = "X-CambriaDate";
+	
+	protected MRBaseClient ( Collection<String> hosts ) throws MalformedURLException
+	{
+		super ( ConnectionType.HTTP, hosts, MRConstants.kStdMRServicePort );
+	}
+
+	protected MRBaseClient ( Collection<String> hosts, int stdSvcPort ) throws MalformedURLException {
+		super ( ConnectionType.HTTP,hosts, stdSvcPort);
+
+		fLog = LoggerFactory.getLogger ( this.getClass().getName () );
+	}
+
+	protected MRBaseClient ( Collection<String> hosts, String clientSignature ) throws MalformedURLException
+	{
+		super(ConnectionType.HTTP, hosts, MRConstants.kStdMRServicePort, clientSignature, CacheUse.NONE, 1, 1L, TimeUnit.MILLISECONDS, 32, 32, 600000);
+
+		fLog = LoggerFactory.getLogger ( this.getClass().getName () );
+	}
+
+
+	@Override
+	public void close ()
+	{
+	}
+
+	protected Set<String> jsonArrayToSet ( JSONArray a )
+	{
+		if ( a == null ) return null;
+
+		final TreeSet<String> set = new TreeSet<String> ();
+		for ( int i=0; i<a.length (); i++ )
+		{
+			set.add ( a.getString ( i ));
+		}
+		return set;
+	}
+
+	public void logTo ( Logger log )
+	{
+		fLog = log;
+		replaceLogger ( log );
+	}
+
+	protected Logger getLog ()
+	{
+		return fLog;
+	}
+
+	private Logger fLog;
+	
+	public JSONObject post(final String path, final byte[] data, final String contentType, final String username, final String password, final String protocolFlag) throws HttpException, JSONException{
+		if ((null != username && null != password)) {
+			WebTarget target = null;
+
+			Response response = null;
+			
+			target = getTarget(path, username, password);
+			String encoding = Base64.encodeAsString(username+":"+password);
+			
+			
+			response = target.request().header("Authorization", "Basic " + encoding).post(Entity.entity(data, contentType));
+			
+			return getResponseDataInJson(response);
+		} else {
+			throw new HttpException("Authentication Failed: Username/password/AuthKey/AuthDate parameter(s) cannot be null or empty.");
+		}
+	}
+	public String postWithResponse(final String path, final byte[] data, final String contentType, final String username, final String password, final String protocolFlag) throws HttpException, JSONException{
+		String responseData = null;
+		if ((null != username && null != password)) {
+			WebTarget target = null;
+
+			Response response = null;
+			
+			target = getTarget(path, username, password);
+			String encoding = Base64.encodeAsString(username+":"+password);
+			
+			
+			response = target.request().header("Authorization", "Basic " + encoding).post(Entity.entity(data, contentType));
+			
+			responseData = response.readEntity(String.class);
+			return responseData;
+		} else {
+			throw new HttpException("Authentication Failed: Username/password/AuthKey/AuthDate parameter(s) cannot be null or empty.");
+		}
+	}
+	public JSONObject postAuth(final String path, final byte[] data, final String contentType, final String authKey,final String authDate,final String username, final String password, final String protocolFlag) throws HttpException, JSONException{
+		if ((null != username && null != password)) {
+			WebTarget target = null;
+
+			Response response = null;
+				target= getTarget(path,username, password);
+				response = target.request()
+						.header(MR_AUTH_CONSTANT, authKey)
+						.header(MR_DATE_CONSTANT, authDate)
+						.post(Entity.entity(data, contentType));
+				
+			return getResponseDataInJson(response);
+		} else {
+			throw new HttpException("Authentication Failed: Username/password/AuthKey/AuthDate parameter(s) cannot be null or empty.");
+		}
+	}
+	public String postAuthwithResponse(final String path, final byte[] data, final String contentType, final String authKey,final String authDate,final String username, final String password, final String protocolFlag) throws HttpException, JSONException{
+		String responseData = null;
+		if ((null != username && null != password)) {
+			WebTarget target = null;
+
+			Response response = null;
+				target= getTarget(path,username, password);
+				response = target.request()
+						.header(MR_AUTH_CONSTANT, authKey)
+						.header(MR_DATE_CONSTANT, authDate)
+						.post(Entity.entity(data, contentType));
+				responseData = response.readEntity(String.class);
+				return responseData;
+			
+		} else {
+			throw new HttpException("Authentication Failed: Username/password/AuthKey/AuthDate parameter(s) cannot be null or empty.");
+		}
+	}
+
+
+	public JSONObject get(final String path, final String username, final String password, final String protocolFlag) throws HttpException, JSONException {
+		if (null != username && null != password) {
+			
+			WebTarget target = null;
+
+			Response response = null;
+			if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
+				target=getTarget(path);
+				response = target.request()
+						.header(MR_AUTH_CONSTANT, username)
+						.header(MR_DATE_CONSTANT, password)
+						.get();
+			} else {
+				target = getTarget(path, username, password);
+				String encoding = Base64.encodeAsString(username+":"+password);
+				
+				response = target.request().header("Authorization", "Basic " + encoding).get();	
+						
+			}
+			return getResponseDataInJson(response);
+		} else {
+			throw new HttpException("Authentication Failed: Username/password/AuthKey/Authdate parameter(s) cannot be null or empty.");
+		}
+	}
+	
+	
+	public String getResponse(final String path, final String username, final String password, final String protocolFlag) throws HttpException, JSONException {
+		String responseData = null;
+		if (null != username && null != password) {
+			
+			WebTarget target = null;
+
+			Response response = null;
+			if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
+				target=getTarget(path);
+				response = target.request()
+						.header(MR_AUTH_CONSTANT, username)
+						.header(MR_DATE_CONSTANT, password)
+						.get();
+			} else {
+				target = getTarget(path, username, password);
+				String encoding = Base64.encodeAsString(username+":"+password);				
+				response = target.request().header("Authorization", "Basic " + encoding).get();							
+			}
+			MRClientFactory.HTTPHeadersMap=response.getHeaders();
+		
+			String transactionid=response.getHeaderString("transactionid");
+				if (transactionid!=null && !transactionid.equalsIgnoreCase("")) {
+					fLog.info("TransactionId : " + transactionid);
+			}
+			
+			responseData = response.readEntity(String.class);
+			return responseData;
+		} else {
+			throw new HttpException("Authentication Failed: Username/password/AuthKey/Authdate parameter(s) cannot be null or empty.");
+		}
+	}
+	
+	public JSONObject getAuth(final String path, final String authKey, final String authDate,final String username, final String password, final String protocolFlag) throws HttpException, JSONException {
+		if (null != username && null != password) {
+			
+			WebTarget target = null;
+
+			Response response = null;
+				target=getTarget(path, username, password);
+				response = target.request()
+						.header(MR_AUTH_CONSTANT, authKey)
+						.header(MR_DATE_CONSTANT, authDate)
+						.get();
+						
+			return getResponseDataInJson(response);
+		} else {
+			throw new HttpException("Authentication Failed: Username/password/AuthKey/Authdate parameter(s) cannot be null or empty.");
+		}
+	}
+	
+	public String getAuthResponse(final String path, final String authKey, final String authDate,final String username, final String password, final String protocolFlag) throws HttpException, JSONException {
+		String responseData = null;
+		if (null != username && null != password) {
+			
+			WebTarget target = null;
+
+			Response response = null;
+				target=getTarget(path, username, password);
+				response = target.request()
+						.header(MR_AUTH_CONSTANT, authKey)
+						.header(MR_DATE_CONSTANT, authDate)
+						.get();
+				
+				MRClientFactory.HTTPHeadersMap=response.getHeaders();
+				
+				String transactionid=response.getHeaderString("transactionid");
+					if (transactionid!=null && !transactionid.equalsIgnoreCase("")) {
+						fLog.info("TransactionId : " + transactionid);
+				}
+						
+				responseData = response.readEntity(String.class);
+				return responseData;
+		} else {
+			throw new HttpException("Authentication Failed: Username/password/AuthKey/Authdate parameter(s) cannot be null or empty.");
+		}
+	}
+
+	private WebTarget getTarget(final String path, final String username, final String password) {
+
+		Client client = ClientBuilder.newClient();
+
+		
+			// Using UNIVERSAL as it supports both BASIC and DIGEST authentication types.
+			HttpAuthenticationFeature feature = HttpAuthenticationFeature.universal(username, password);
+			client.register(feature);
+		
+		return client.target(path);
+	}
+
+
+	private WebTarget getTarget(final String path) {
+
+		Client client = ClientBuilder.newClient();
+		return client.target(path);
+	}
+	private JSONObject getResponseDataInJson(Response response) throws JSONException {
+		try {
+			MRClientFactory.HTTPHeadersMap=response.getHeaders();
+		//	fLog.info("DMAAP response status: " + response.getStatus());
+			
+						
+			//MultivaluedMap<String, Object> headersMap = response.getHeaders();
+			//for(String key : headersMap.keySet()) {
+			String transactionid=response.getHeaderString("transactionid");
+			if (transactionid!=null && !transactionid.equalsIgnoreCase("")) {
+				fLog.info("TransactionId : " + transactionid);
+			}
+
+			/*final String responseData = response.readEntity(String.class);
+			JSONTokener jsonTokener = new JSONTokener(responseData);
+			JSONObject jsonObject = null;
+			final char firstChar = jsonTokener.next();
+			jsonTokener.back();
+			if ('[' == firstChar) {
+				JSONArray jsonArray = new JSONArray(jsonTokener);
+				jsonObject = new JSONObject();
+				jsonObject.put("result", jsonArray);
+			} else {
+				jsonObject = new JSONObject(jsonTokener);
+			}
+
+			return jsonObject;*/
+			
+
+			if(response.getStatus()==403) {
+				JSONObject jsonObject = null;
+				jsonObject = new JSONObject();
+				JSONArray jsonArray = new JSONArray();
+				jsonArray.put(response.getEntity());
+				jsonObject.put("result", jsonArray);
+				jsonObject.put("status", response.getStatus());
+				return jsonObject;
+			}
+			String responseData = response.readEntity(String.class);
+				
+			JSONTokener jsonTokener = new JSONTokener(responseData);
+			JSONObject jsonObject = null;
+			final char firstChar = jsonTokener.next();
+	    	jsonTokener.back();
+			if ('[' == firstChar) {
+				JSONArray jsonArray = new JSONArray(jsonTokener);
+				jsonObject = new JSONObject();
+				jsonObject.put("result", jsonArray);
+				jsonObject.put("status", response.getStatus());
+			} else {
+				jsonObject = new JSONObject(jsonTokener);
+				jsonObject.put("status", response.getStatus());
+			}
+
+			return jsonObject;
+		} catch (JSONException excp) {
+			fLog.error("DMAAP - Error reading response data.", excp);
+			return null;
+		}
+
+	}
+	
+	public String getHTTPErrorResponseMessage(String responseString){
+		
+		String response = null;
+		int beginIndex = 0;
+		int endIndex = 0;
+		if(responseString.contains("<body>")){
+			
+			beginIndex = responseString.indexOf("body>")+5;
+			endIndex = responseString.indexOf("</body");
+			response = responseString.substring(beginIndex,endIndex);
+		}
+		
+		return response;
+		
+	}
+	
+	public String getHTTPErrorResponseCode(String responseString){
+		
+		String response = null;
+		int beginIndex = 0;
+		int endIndex = 0;
+		if(responseString.contains("<title>")){
+			beginIndex = responseString.indexOf("title>")+6;
+			endIndex = responseString.indexOf("</title");
+			response = responseString.substring(beginIndex,endIndex);
+		}
+				
+		return response;		
+	}
+	
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRBatchPublisher.java b/src/main/java/com/att/nsa/mr/client/impl/MRBatchPublisher.java
new file mode 100644
index 0000000..597e4da
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRBatchPublisher.java
@@ -0,0 +1,485 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 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=========================================================
+ *
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package com.att.nsa.mr.client.impl;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;
+import java.util.zip.GZIPOutputStream;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.nsa.apiClient.http.HttpClient;
+import com.att.nsa.apiClient.http.HttpException;
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.response.MRPublisherResponse;
+
+/**
+ * This is a batching publisher class that allows the client to publish messages
+ * in batches that are limited in terms of size and/or hold time.
+ * 
+ * @author author
+ * @deprecated This class's tricky locking doesn't quite work
+ *
+ */
+@Deprecated
+public class MRBatchPublisher implements MRBatchingPublisher
+{
+	public static final long kMinMaxAgeMs = 1;
+
+	/**
+	 * Create a batch publisher.
+	 * 
+	 * @param baseUrls the base URLs, like "localhost:8080". This class adds the correct application path.
+	 * @param topic the topic to publish to
+	 * @param maxBatchSize the maximum size of a batch
+	 * @param maxAgeMs the maximum age of a batch
+	 */
+	public MRBatchPublisher ( Collection<String> baseUrls, String topic, int maxBatchSize, long maxAgeMs, boolean compress )
+	{
+		if ( maxAgeMs < kMinMaxAgeMs )
+		{
+			fLog.warn ( "Max age in ms is less than the minimum. Overriding to " + kMinMaxAgeMs );
+			maxAgeMs = kMinMaxAgeMs;
+		}
+
+		try {
+			fSender = new Sender ( baseUrls, topic, maxBatchSize, maxAgeMs, compress );
+		} catch (MalformedURLException e) {
+			throw new RuntimeException(e);
+		}
+
+		// FIXME: this strategy needs an overhaul -- why not just run a thread that knows how to wait for
+		// the oldest msg to hit max age? (locking is complicated, but should be do-able)
+		fExec = new ScheduledThreadPoolExecutor ( 1 );
+		fExec.scheduleAtFixedRate ( fSender, 100, 50, TimeUnit.MILLISECONDS );
+	}
+
+	@Override
+	public void setApiCredentials ( String apiKey, String apiSecret )
+	{
+		fSender.setApiCredentials ( apiKey, apiSecret );
+	}
+
+	@Override
+	public void clearApiCredentials ()
+	{
+		fSender.clearApiCredentials ();
+	}
+
+	/**
+	 * Send the given message with the given partition
+	 * @param partition
+	 * @param msg
+	 * @throws IOException
+	 */
+	@Override
+	public int send ( String partition, String msg ) throws IOException
+	{
+		return send ( new message ( partition, msg ) );
+	}
+	@Override
+	public int send ( String msg ) throws IOException
+	{
+		return send ( new message ( "",msg ) );
+	}
+	/**
+	 * Send the given message
+	 * @param userMsg a message
+	 * @throws IOException
+	 */
+	@Override
+	public int send ( message userMsg ) throws IOException
+	{
+		final LinkedList<message> list = new LinkedList<message> ();
+		list.add ( userMsg );
+		return send ( list );
+	}
+
+	/**
+	 * Send the given set of messages
+	 * @param msgs the set of messages, sent in order of iteration
+	 * @return the number of messages in the pending queue (this could actually be less than the size of the given collection, depending on thread timing)
+	 * @throws IOException
+	 */
+	@Override
+	public int send ( Collection<message> msgs ) throws IOException
+	{
+		if ( msgs.size() > 0 )
+		{
+			fSender.queue ( msgs );
+		}
+		return fSender.size ();
+	}
+
+	@Override
+	public int getPendingMessageCount ()
+	{
+		return fSender.size ();
+	}
+
+	/**
+	 * Send any pending messages and close this publisher.
+	 * @throws IOException
+	 * @throws InterruptedException 
+	 */
+	@Override
+	public void close ()
+	{
+		try
+		{
+			final List<message> remains = close ( Long.MAX_VALUE, TimeUnit.MILLISECONDS );
+			if ( remains.size() > 0 )
+			{
+				fLog.warn ( "Closing publisher with " + remains.size() + " messages unsent. "
+					+ "(Consider using the alternate close method to capture unsent messages in this case.)" );
+			}
+		}
+		catch ( InterruptedException e )
+		{
+			fLog.warn ( "Possible message loss. " + e.getMessage(), e );
+		}
+		catch ( IOException e )
+		{
+			fLog.warn ( "Possible message loss. " + e.getMessage(), e );
+		}
+	}
+
+	public List<message> close ( long time, TimeUnit unit ) throws InterruptedException, IOException
+	{
+		fExec.setContinueExistingPeriodicTasksAfterShutdownPolicy ( false );
+		fExec.setExecuteExistingDelayedTasksAfterShutdownPolicy ( false );
+		fExec.shutdown ();
+
+		final long waitInMs = TimeUnit.MILLISECONDS.convert ( time, unit );
+		final long timeoutAtMs = System.currentTimeMillis () + waitInMs;
+		while ( System.currentTimeMillis () < timeoutAtMs && getPendingMessageCount() > 0 )
+		{
+			fSender.checkSend ( true );
+			Thread.sleep ( 250 );
+		}
+
+		final LinkedList<message> result = new LinkedList<message> ();
+		fSender.drainTo ( result );
+		return result;
+	}
+
+	private final ScheduledThreadPoolExecutor fExec;
+	private final Sender fSender;
+	
+	private static class TimestampedMessage extends message
+	{
+		public TimestampedMessage ( message m )
+		{
+			super ( m );
+			timestamp = System.currentTimeMillis ();
+		}
+		public final long timestamp;
+	}
+
+	private Logger fLog = LoggerFactory.getLogger ( MRBatchPublisher.class );
+
+	private class Sender extends MRBaseClient implements Runnable
+	{
+		public Sender ( Collection<String> baseUrls, String topic, int maxBatch, long maxAgeMs, boolean compress ) throws MalformedURLException
+		{
+			super ( baseUrls );
+
+			fNextBatch = new LinkedList<TimestampedMessage> ();
+			fSendingBatch = null;
+			fTopic = topic;
+			fMaxBatchSize = maxBatch;
+			fMaxAgeMs = maxAgeMs;
+			fCompress = compress;
+			fLock = new ReentrantReadWriteLock ();
+			fWriteLock = fLock.writeLock ();
+			fReadLock = fLock.readLock ();
+			fDontSendUntilMs = 0;
+		}
+
+		public void drainTo ( LinkedList<message> list )
+		{
+			fWriteLock.lock ();
+			try
+			{
+				if ( fSendingBatch != null )
+				{
+					list.addAll ( fSendingBatch );
+				}
+				list.addAll ( fNextBatch );
+
+				fSendingBatch = null;
+				fNextBatch.clear ();
+			}
+			finally
+			{
+				fWriteLock.unlock ();
+			}
+		}
+
+		/**
+		 * Called periodically by the background executor.
+		 */
+		@Override
+		public void run ()
+		{
+			try
+			{
+				checkSend ( false );
+			}
+			catch ( IOException e )
+			{
+				fLog.warn ( "MR background send: " + e.getMessage () );
+			}
+		}
+
+		public int size ()
+		{
+			fReadLock.lock ();
+			try
+			{
+				return fNextBatch.size () + ( fSendingBatch == null ? 0 : fSendingBatch.size () );
+			}
+			finally
+			{
+				fReadLock.unlock ();
+			}
+		}
+		
+		/**
+		 * Called to queue a message.
+		 * @param m
+		 * @throws IOException 
+		 */
+		public void queue ( Collection<message> msgs ) throws IOException
+		{
+			fWriteLock.lock ();
+			try
+			{
+				for ( message userMsg : msgs )
+				{
+					if ( userMsg != null )
+					{
+						fNextBatch.add ( new TimestampedMessage ( userMsg ) );
+					}
+					else
+					{
+						fLog.warn ( "MRBatchPublisher::Sender::queue received a null message." );
+					}
+				}
+			}
+			finally
+			{
+				fWriteLock.unlock();
+			}
+			checkSend ( false );
+		}
+
+		/**
+		 * Send a batch if the queue is long enough, or the first pending message is old enough.
+		 * @param force
+		 * @throws IOException
+		 */
+		public void checkSend ( boolean force ) throws IOException
+		{
+			// hold a read lock just long enough to evaluate whether a batch
+			// should be sent
+			boolean shouldSend = false;
+			fReadLock.lock ();
+			try
+			{
+				if ( fNextBatch.size () > 0 )
+				{
+					final long nowMs = System.currentTimeMillis ();
+					shouldSend = ( force || fNextBatch.size() >= fMaxBatchSize );
+					if ( !shouldSend )
+					{
+						final long sendAtMs = fNextBatch.getFirst ().timestamp + fMaxAgeMs;
+						shouldSend = sendAtMs <= nowMs;
+					}
+					
+					// however, unless forced, wait after an error
+					shouldSend = force || ( shouldSend && nowMs >= fDontSendUntilMs ); 
+				}
+				// else: even in 'force', there's nothing to send, so shouldSend=false is fine
+			}
+			finally
+			{
+				fReadLock.unlock ();
+			}
+
+			// if a send is required, acquire a write lock, swap out the next batch,
+			// swap in a fresh batch, and release the lock for the caller to start
+			// filling a batch again. After releasing the lock, send the current
+			// batch. (There could be more messages added between read unlock and
+			// write lock, but that's fine.)
+			if ( shouldSend )
+			{
+				fSendingBatch = null;
+
+				fWriteLock.lock ();
+				try
+				{
+					fSendingBatch = fNextBatch;
+					fNextBatch = new LinkedList<TimestampedMessage> ();
+				}
+				finally
+				{
+					fWriteLock.unlock ();
+				}
+
+				if ( !doSend ( fSendingBatch, this, fTopic, fCompress, fLog ) )
+				{
+					fLog.warn ( "Send failed, rebuilding send queue." );
+
+					// note the time for back-off
+					fDontSendUntilMs = sfWaitAfterError + System.currentTimeMillis ();
+
+					// the send failed. reconstruct the pending queue
+					fWriteLock.lock ();
+					try
+					{
+						final LinkedList<TimestampedMessage> nextGroup = fNextBatch;
+						fNextBatch = fSendingBatch;
+						fNextBatch.addAll ( nextGroup );
+						fSendingBatch = null;
+						fLog.info ( "Send queue rebuilt; " + fNextBatch.size () + " messages to send." );
+					}
+					finally
+					{
+						fWriteLock.unlock ();
+					}
+				}
+				else
+				{
+					fWriteLock.lock ();
+					try
+					{
+						fSendingBatch = null;
+					}
+					finally
+					{
+						fWriteLock.unlock ();
+					}
+				}
+			}
+		}
+
+		private LinkedList<TimestampedMessage> fNextBatch;
+		private LinkedList<TimestampedMessage> fSendingBatch;
+		private final String fTopic;
+		private final int fMaxBatchSize;
+		private final long fMaxAgeMs;
+		private final boolean fCompress;
+		private final ReentrantReadWriteLock fLock;
+		private final WriteLock fWriteLock;
+		private final ReadLock fReadLock;
+		private long fDontSendUntilMs;
+		private static final long sfWaitAfterError = 1000;
+	}
+
+	// this is static so that it's clearly not using any mutable member data outside of a lock
+	private static boolean doSend ( LinkedList<TimestampedMessage> toSend, HttpClient client, String topic, boolean compress, Logger log )
+	{
+		// it's possible for this call to be made with an empty list. in this case, just return.
+		if ( toSend.size() < 1 )
+		{
+			return true;
+		}
+
+		final long nowMs = System.currentTimeMillis ();
+		final String url = MRConstants.makeUrl ( topic );
+
+		log.info ( "sending " + toSend.size() + " msgs to " + url + ". Oldest: " + ( nowMs - toSend.getFirst().timestamp ) + " ms"  );
+
+		final ByteArrayOutputStream baseStream = new ByteArrayOutputStream ();
+		try
+		{
+			OutputStream os = baseStream;
+			if ( compress )
+			{
+				os = new GZIPOutputStream ( baseStream );
+			}
+			for ( TimestampedMessage m : toSend )
+			{
+				os.write ( ( "" + m.fPartition.length () ).getBytes() );
+				os.write ( '.' );
+				os.write ( ( "" + m.fMsg.length () ).getBytes() );
+				os.write ( '.' );
+				os.write ( m.fPartition.getBytes() );
+				os.write ( m.fMsg.getBytes() );
+				os.write ( '\n' );
+			}
+			os.close ();
+		}
+		catch ( IOException e )
+		{
+			log.warn ( "Problem writing stream to post: " + e.getMessage () );
+			return false;
+		}
+
+		boolean result = false;
+		final long startMs = System.currentTimeMillis ();
+		try
+		{
+			client.post ( url, compress ?
+				MRFormat.CAMBRIA_ZIP.toString () :
+				MRFormat.CAMBRIA.toString (),
+				baseStream.toByteArray(), false );
+			result = true;
+		}
+		catch ( HttpException e )
+		{
+			log.warn ( "Problem posting to MR: " + e.getMessage() );
+		}
+		catch ( IOException e )
+		{
+			log.warn ( "Problem posting to MR: " + e.getMessage() );
+		}
+
+		log.info ( "MR response (" + (System.currentTimeMillis ()-startMs) + " ms): OK" );
+		return result;
+	}
+
+	@Override
+	public void logTo ( Logger log )
+	{
+		fLog = log;
+	}
+
+	@Override
+	public MRPublisherResponse sendBatchWithResponse() {
+		// TODO Auto-generated method stub
+		return null;
+	}
+	
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRClientVersionInfo.java b/src/main/java/com/att/nsa/mr/client/impl/MRClientVersionInfo.java
new file mode 100644
index 0000000..fa9207d
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRClientVersionInfo.java
@@ -0,0 +1,54 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 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=========================================================
+ *
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package com.att.nsa.mr.client.impl;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Properties;
+
+public class MRClientVersionInfo
+{
+	public static String getVersion ()
+	{
+		return version;
+	}
+
+	private static final Properties props = new Properties();
+	private static final String version;
+	static
+	{
+		String use = null;
+		try
+		{
+			final InputStream is = MRClientVersionInfo.class.getResourceAsStream ( "/MRClientVersion.properties" );
+			if ( is != null )
+			{
+				props.load ( is );
+				use = props.getProperty ( "MRClientVersion", null );
+			}
+		}
+		catch ( IOException e )
+		{
+		}
+		version = use;
+	}
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRConstants.java b/src/main/java/com/att/nsa/mr/client/impl/MRConstants.java
new file mode 100644
index 0000000..803670c
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRConstants.java
@@ -0,0 +1,180 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 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=========================================================
+ *
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package com.att.nsa.mr.client.impl;
+
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import org.apache.http.HttpHost;
+
+class MRConstants
+{
+	private static final String PROTOCOL = "http";
+	public static final String context = "/";
+	public static final String kBasePath = "events/";
+	//public static final int kStdMRServicePort = 3904;
+	public static final int kStdMRServicePort = 8080;
+
+	public static String escape ( String s )
+	{
+		try
+		{
+			return URLEncoder.encode ( s, "UTF-8");
+		}
+		catch ( UnsupportedEncodingException e )
+		{
+			throw new RuntimeException ( e );
+		}
+	}
+
+	public static String makeUrl ( String rawTopic )
+	{
+		final String cleanTopic = escape ( rawTopic );
+		
+		final StringBuffer url = new StringBuffer().
+			append ( MRConstants.context ).
+			append ( MRConstants.kBasePath ).
+			append ( cleanTopic );
+		return url.toString ();
+	}
+	
+	public static String makeUrl ( final String host, final String rawTopic )
+	{
+		final String cleanTopic = escape ( rawTopic );
+
+		final StringBuffer url = new StringBuffer();
+		
+		if (!host.startsWith("http") || !host.startsWith("https") ) {
+			url.append( PROTOCOL + "://" );
+		}
+		url.append(host);
+		url.append ( MRConstants.context );
+		url.append ( MRConstants.kBasePath );
+		url.append ( cleanTopic );
+		return url.toString ();
+	}
+
+	public static String makeUrl ( final String host, final String rawTopic, final String transferprotocol,final String parttion )
+	{
+		final String cleanTopic = escape ( rawTopic );
+
+		final StringBuffer url = new StringBuffer();
+		
+		if (transferprotocol !=null && !transferprotocol.equals("")) {
+			url.append( transferprotocol + "://" );
+		}else{
+			url.append( PROTOCOL + "://" );
+		}
+		url.append(host);
+		url.append ( MRConstants.context );
+		url.append ( MRConstants.kBasePath );
+		url.append ( cleanTopic );
+		if(parttion!=null && !parttion.equalsIgnoreCase(""))
+			url.append("?partitionKey=").append(parttion);
+		return url.toString ();
+	}
+	public static String makeConsumerUrl ( String topic, String rawConsumerGroup, String rawConsumerId )
+	{
+		final String cleanConsumerGroup = escape ( rawConsumerGroup );
+		final String cleanConsumerId = escape ( rawConsumerId );
+		return MRConstants.context + MRConstants.kBasePath + topic + "/" + cleanConsumerGroup + "/" + cleanConsumerId;
+	}
+
+	/**
+	 * Create a list of HttpHosts from an input list of strings. Input strings have
+	 * host[:port] as format. If the port section is not provided, the default port is used.
+	 * 
+	 * @param hosts
+	 * @return a list of hosts
+	 */
+	public static List<HttpHost> createHostsList(Collection<String> hosts)
+	{
+		final ArrayList<HttpHost> convertedHosts = new ArrayList<HttpHost> ();
+		for ( String host : hosts )
+		{
+			if ( host.length () == 0 ) continue;
+			convertedHosts.add ( hostForString ( host ) );
+		}
+		return convertedHosts;
+	}
+
+	/**
+	 * Return an HttpHost from an input string. Input string has
+	 * host[:port] as format. If the port section is not provided, the default port is used.
+	 * 
+	 * @param hosts
+	 * @return a list of hosts
+	 */
+	public static HttpHost hostForString ( String host )
+	{
+		if ( host.length() < 1 ) throw new IllegalArgumentException ( "An empty host entry is invalid." );
+		
+		String hostPart = host;
+		int port = kStdMRServicePort;
+
+		final int colon = host.indexOf ( ':' );
+		if ( colon == 0 ) throw new IllegalArgumentException ( "Host entry '" + host + "' is invalid." );
+		if ( colon > 0 )
+		{
+			hostPart = host.substring ( 0, colon ).trim();
+
+			final String portPart = host.substring ( colon + 1 ).trim();
+			if ( portPart.length () > 0 )
+			{
+				try
+				{
+					port = Integer.parseInt ( portPart );
+				}
+				catch ( NumberFormatException x )
+				{
+					throw new IllegalArgumentException ( "Host entry '" + host + "' is invalid.", x );
+				}
+			}
+			// else: use default port on "foo:"
+		}
+
+		return new HttpHost ( hostPart, port );
+	}
+
+	public static String makeConsumerUrl(String host, String fTopic, String fGroup, String fId,final String transferprotocol) {
+		final String cleanConsumerGroup = escape ( fGroup );
+		final String cleanConsumerId = escape ( fId );
+		
+		StringBuffer url = new StringBuffer();
+		
+		if (transferprotocol !=null && !transferprotocol.equals("")) {
+			url.append( transferprotocol + "://" );
+		}else{
+			url.append( PROTOCOL + "://" );
+		}
+		
+		url.append(host);
+		url.append(context);
+		url.append(kBasePath);
+		url.append(fTopic + "/" + cleanConsumerGroup + "/" + cleanConsumerId);
+		
+		return url.toString();
+	}
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java b/src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java
new file mode 100644
index 0000000..fdd1b89
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRConsumerImpl.java
@@ -0,0 +1,709 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 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=========================================================
+ *
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package com.att.nsa.mr.client.impl;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URLEncoder;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.http.HttpException;
+import org.apache.http.HttpStatus;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+import org.json.JSONTokener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.att.aft.dme2.api.DME2Client;
+import com.att.aft.dme2.api.DME2Exception;
+import com.att.nsa.mr.client.MRClientFactory;
+import com.att.nsa.mr.client.MRConsumer;
+import com.att.nsa.mr.client.response.MRConsumerResponse;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+public class MRConsumerImpl extends MRBaseClient implements MRConsumer
+{
+	
+	private static final String SUCCESS_MESSAGE = "Success";
+	
+	
+	private Logger fLog = LoggerFactory.getLogger ( this.getClass().getName () );
+	public static List<String> stringToList ( String str )
+	{
+		final LinkedList<String> set = new LinkedList<String> ();
+		if ( str != null )
+		{
+			final String[] parts = str.trim ().split ( "," );
+			for ( String part : parts )
+			{
+				final String trimmed = part.trim();
+				if ( trimmed.length () > 0 )
+				{
+					set.add ( trimmed );
+				}
+			}
+		}
+		return set;
+	}
+	
+	public MRConsumerImpl ( Collection<String> hostPart, final String topic, final String consumerGroup,
+			final String consumerId, int timeoutMs, int limit, String filter, String apiKey_username, String apiSecret_password ) throws MalformedURLException
+		{
+			this( hostPart, topic, consumerGroup, consumerId, timeoutMs, limit, filter, apiKey_username, apiSecret_password, false );
+		}
+	
+	public MRConsumerImpl ( Collection<String> hostPart, final String topic, final String consumerGroup,
+		final String consumerId, int timeoutMs, int limit, String filter, String apiKey, String apiSecret, boolean allowSelfSignedCerts ) throws MalformedURLException
+	{
+		super ( hostPart, topic + "::" + consumerGroup + "::" + consumerId );
+
+		fTopic = topic;
+		fGroup = consumerGroup;
+		fId = consumerId;
+		fTimeoutMs = timeoutMs;
+		fLimit = limit;
+		fFilter = filter;
+
+		//setApiCredentials ( apiKey, apiSecret );
+	}
+
+	@Override
+	public Iterable<String> fetch () throws IOException,Exception
+	{
+		// fetch with the timeout and limit set in constructor
+		return fetch ( fTimeoutMs, fLimit );
+	}
+
+	@Override
+	public Iterable<String> fetch ( int timeoutMs, int limit ) throws IOException,Exception
+	{
+		final LinkedList<String> msgs = new LinkedList<String> ();
+
+// FIXME: the timeout on the socket needs to be at least as long as the long poll
+//		// sanity check for long poll timeout vs. socket read timeout
+//		final int maxReasonableTimeoutMs = CambriaSingletonHttpClient.sfSoTimeoutMs * 9/10;
+//		if ( timeoutMs > maxReasonableTimeoutMs )
+//		{
+//			log.warn ( "Long poll time (" + timeoutMs + ") is too high w.r.t. socket read timeout (" +
+//				CambriaSingletonHttpClient.sfSoTimeoutMs + "). Reducing long poll timeout to " + maxReasonableTimeoutMs + "." );
+//			timeoutMs = maxReasonableTimeoutMs;
+//		}
+
+	//	final String urlPath = createUrlPath ( timeoutMs, limit );
+
+		//getLog().info ( "UEB GET " + urlPath );
+		try
+		{
+			if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) {
+				DMEConfigure(timeoutMs, limit);
+			try 
+			{
+				//getLog().info ( "Receiving msgs from: " + url+subContextPath );
+				String reply = sender.sendAndWait(timeoutMs+10000L);				
+			//	System.out.println("Message received = "+reply);
+				final JSONObject o =getResponseDataInJson(reply);
+				//msgs.add(reply);
+				if ( o != null )
+				{
+					final JSONArray a = o.getJSONArray ( "result" );
+				//	final int b = o.getInt("status" );
+					//if ( a != null && a.length()>0 )
+					if ( a != null)
+					{
+						for ( int i=0; i<a.length (); i++ )
+						{
+							//msgs.add("DMAAP response status: "+Integer.toString(b));
+							if (a.get(i) instanceof String)
+								msgs.add ( a.getString(i) );
+							else
+						    	msgs.add ( a.getJSONObject(i).toString() );
+							
+							
+						}
+					}
+//					else if(a != null && a.length()<1){
+//						msgs.add ("[]");		
+//						}
+				}
+			}	
+			catch ( JSONException e )
+				{
+					// unexpected response
+					reportProblemWithResponse ();
+				}
+				catch ( HttpException e )
+				{
+					throw new IOException ( e );
+				}	
+			}
+			
+			if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) {
+				final String urlPath = createUrlPath (MRConstants.makeConsumerUrl ( host, fTopic, fGroup, fId,props.getProperty("Protocol")), timeoutMs, limit );
+			
+				
+				try
+				{
+					final JSONObject o = get ( urlPath, username, password, protocolFlag );
+
+					if ( o != null )
+					{
+						final JSONArray a = o.getJSONArray ( "result" );
+						final int b = o.getInt("status" );
+						//if ( a != null && a.length()>0 )
+						if ( a != null)
+						{
+							for ( int i=0; i<a.length (); i++ )
+							{
+								msgs.add("DMAAP response status: "+Integer.toString(b));
+								if (a.get(i) instanceof String)
+									msgs.add ( a.getString(i) );
+								else
+									msgs.add ( a.getJSONObject(i).toString() );
+								
+							}
+						}
+//						else if(a != null && a.length()<1)
+//							{
+//								msgs.add ("[]");		
+//							}
+					}
+				}
+				catch ( JSONException e )
+				{
+					// unexpected response
+					reportProblemWithResponse ();
+				}
+				catch ( HttpException e )
+				{
+					throw new IOException ( e );
+				}
+			} 
+			
+			if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
+				final String urlPath = createUrlPath (MRConstants.makeConsumerUrl ( host, fTopic, fGroup, fId ,props.getProperty("Protocol")), timeoutMs, limit );
+				
+
+			try 
+			{
+				final JSONObject o = getAuth(urlPath, authKey, authDate, username, password, protocolFlag );
+				if ( o != null )
+				{
+					final JSONArray a = o.getJSONArray ( "result" );
+					final int b = o.getInt("status" );
+					//if ( a != null && a.length()>0)
+					if ( a != null)
+					{
+						for ( int i=0; i<a.length (); i++ )
+						{
+							msgs.add("DMAAP response status: "+Integer.toString(b));
+							if (a.get(i) instanceof String)
+								msgs.add ( a.getString(i) );
+							else
+						    	msgs.add ( a.getJSONObject(i).toString() );
+							
+						}
+					}
+//					else if(a != null && a.length()<1){
+//						msgs.add ("[]");		
+//						}
+				}
+			}
+			catch ( JSONException e )
+			{
+				// unexpected response
+				reportProblemWithResponse ();
+			}
+			catch ( HttpException e )
+			{
+				throw new IOException ( e );
+			}
+				
+			}
+			
+		} catch ( JSONException e ) {
+			// unexpected response
+			reportProblemWithResponse ();
+		} catch (HttpException e) {
+			throw new IOException(e);
+		} catch (Exception e ) {
+			throw e;
+		}
+
+
+		return msgs;
+	}
+
+	private JSONObject getResponseDataInJson(String response) {
+	try {
+		
+		
+		//fLog.info("DMAAP response status: " + response.getStatus());
+
+		//	final String responseData = response.readEntity(String.class);
+		JSONTokener jsonTokener = new JSONTokener(response);
+		JSONObject jsonObject = null;
+		final char firstChar = jsonTokener.next();
+    	jsonTokener.back();
+		if ('[' == firstChar) {
+			JSONArray jsonArray = new JSONArray(jsonTokener);
+			jsonObject = new JSONObject();
+			jsonObject.put("result", jsonArray);
+		} else {
+			jsonObject = new JSONObject(jsonTokener);
+		}
+
+		return jsonObject;
+	} catch (JSONException excp) {
+	//	fLog.error("DMAAP - Error reading response data.", excp);
+		return null;
+	}
+	
+	
+	
+}
+	
+	private JSONObject getResponseDataInJsonWithResponseReturned(String response) {
+			JSONTokener jsonTokener = new JSONTokener(response);
+			JSONObject jsonObject = null;
+			final char firstChar = jsonTokener.next();
+	    	jsonTokener.back();
+	    	if(null != response && response.length()==0){
+	    		return null;
+	    	}
+	    	
+			if ('[' == firstChar) {
+				JSONArray jsonArray = new JSONArray(jsonTokener);
+				jsonObject = new JSONObject();
+				jsonObject.put("result", jsonArray);
+			} else if('{' == firstChar){
+				return null;
+			} else if('<' == firstChar){
+				return null;
+			}else{
+				jsonObject = new JSONObject(jsonTokener);
+			}
+
+			return jsonObject;
+		
+	}
+	
+	private final String fTopic;
+	private final String fGroup;
+	private final String fId;
+	private final int fTimeoutMs;
+	private final int fLimit;
+	private String fFilter;
+	private String username;
+	private String password;
+	private String host;
+	 private  String latitude;
+		private  String longitude;
+		private  String version;
+		private  String serviceName;
+		private  String env;
+		private  String partner;
+		private String routeOffer;
+		private  String subContextPath;
+		private  String protocol;
+		private  String methodType;
+		private  String url;
+		private  String dmeuser;
+		private  String dmepassword;
+		private  String contenttype;
+	    private DME2Client sender;
+		public String protocolFlag = ProtocolTypeConstants.DME2.getValue();
+		public String consumerFilePath;
+		private String authKey;
+		private String authDate;
+        private Properties props;
+    	private HashMap<String, String> DMETimeOuts;
+    	private String handlers;
+    	public static String routerFilePath;
+    	public static String getRouterFilePath() {
+    		return routerFilePath;
+    	}
+
+    	public static void setRouterFilePath(String routerFilePath) {
+    		MRSimplerBatchPublisher.routerFilePath = routerFilePath;
+    	}
+		public String getConsumerFilePath() {
+			return consumerFilePath;
+		}
+
+		public void setConsumerFilePath(String consumerFilePath) {
+			this.consumerFilePath = consumerFilePath;
+		}
+
+		public String getProtocolFlag() {
+			return protocolFlag;
+		}
+
+		public void setProtocolFlag(String protocolFlag) {
+			this.protocolFlag = protocolFlag;
+		}
+		
+		private void DMEConfigure(int timeoutMs, int limit)throws IOException,DME2Exception, URISyntaxException{ 
+			latitude = props.getProperty("Latitude");
+			longitude = props.getProperty("Longitude");
+			version = props.getProperty("Version");
+			serviceName = props.getProperty("ServiceName");
+			env = props.getProperty("Environment");
+			partner = props.getProperty("Partner");
+			routeOffer = props.getProperty("routeOffer");
+			
+			subContextPath=props.getProperty("SubContextPath")+fTopic+"/"+fGroup+"/"+fId;
+		//	subContextPath=createUrlPath (subContextPath, timeoutMs, limit);
+			//if (timeoutMs != -1) subContextPath=createUrlPath (subContextPath, timeoutMs);
+			
+			protocol = props.getProperty("Protocol"); 
+			methodType = props.getProperty("MethodType");
+			dmeuser = props.getProperty("username");
+			dmepassword = props.getProperty("password");
+			contenttype = props.getProperty("contenttype");
+			handlers = props.getProperty("sessionstickinessrequired");
+			//url =protocol+"://DME2SEARCH/"+ "service="+serviceName+"/"+"version="+version+"/"+"envContext="+env+"/"+"partner="+partner;
+		//	url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&routeOffer="+partner;
+		
+			/**
+			 * Changes to DME2Client url to use Partner for auto failover between data centers
+			 * When Partner value is not provided use the routeOffer value for auto failover within a cluster 
+			 */
+			
+			String preferredRouteKey = readRoute("preferredRouteKey");
+						
+			if (partner != null && !partner.isEmpty() && preferredRouteKey != null && !preferredRouteKey.isEmpty()) 
+			{ 
+				url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&partner="+partner+"&routeoffer="+preferredRouteKey; 
+			}else  if (partner != null && !partner.isEmpty()) 
+			{ 
+				url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&partner="+partner; 
+			}
+			else if (routeOffer!=null && !routeOffer.isEmpty()) 
+			{ 
+				url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&routeoffer="+routeOffer;
+			}
+			
+			//fLog.info("url :"+url);
+						
+			if(timeoutMs != -1 )url=url+"&timeout="+timeoutMs;
+			if(limit != -1 )url=url+"&limit="+limit;
+
+			DMETimeOuts = new HashMap<String, String>();
+			DMETimeOuts.put("AFT_DME2_EP_READ_TIMEOUT_MS", props.getProperty("AFT_DME2_EP_READ_TIMEOUT_MS"));
+			DMETimeOuts.put("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", props.getProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS"));
+			DMETimeOuts.put("AFT_DME2_EP_CONN_TIMEOUT", props.getProperty("AFT_DME2_EP_CONN_TIMEOUT"));
+			DMETimeOuts.put("Content-Type", contenttype);
+			System.setProperty("AFT_LATITUDE", latitude);
+			System.setProperty("AFT_LONGITUDE", longitude);
+			System.setProperty("AFT_ENVIRONMENT",props.getProperty("AFT_ENVIRONMENT"));
+		//	System.setProperty("DME2.DEBUG", "true");
+			
+			//SSL changes
+			System.setProperty("AFT_DME2_CLIENT_SSL_INCLUDE_PROTOCOLS",
+					"SSLv3,TLSv1,TLSv1.1");
+			System.setProperty("AFT_DME2_CLIENT_IGNORE_SSL_CONFIG", "false");
+			System.setProperty("AFT_DME2_CLIENT_KEYSTORE_PASSWORD", "changeit");
+			//SSL changes
+            
+			sender = new DME2Client(new URI(url), timeoutMs+10000L);
+			sender.setAllowAllHttpReturnCodes(true);
+			sender.setMethod(methodType);
+			sender.setSubContext(subContextPath);	
+			if(dmeuser != null && dmepassword != null){
+			sender.setCredentials(dmeuser, dmepassword);
+			//System.out.println(dmepassword);
+			}
+			sender.setHeaders(DMETimeOuts);
+			sender.setPayload("");               
+			
+			if(handlers.equalsIgnoreCase("yes")){
+				sender.addHeader("AFT_DME2_EXCHANGE_REQUEST_HANDLERS", props.getProperty("AFT_DME2_EXCHANGE_REQUEST_HANDLERS"));
+				sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", props.getProperty("AFT_DME2_EXCHANGE_REPLY_HANDLERS"));
+				sender.addHeader("AFT_DME2_REQ_TRACE_ON", props.getProperty("AFT_DME2_REQ_TRACE_ON"));
+				}else{
+					sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", "com.att.nsa.mr.dme.client.HeaderReplyHandler");
+				}
+		/*	HeaderReplyHandler headerhandler= new HeaderReplyHandler(); 
+			sender.setReplyHandler(headerhandler);*/
+//			} catch (DME2Exception x) {
+//				getLog().warn(x.getMessage(), x);
+//				System.out.println("XXXXXXXXXXXX"+x);
+//			} catch (URISyntaxException x) {
+//				System.out.println(x);
+//				getLog().warn(x.getMessage(), x);
+//			} catch (Exception x) {
+//				System.out.println("XXXXXXXXXXXX"+x);
+//				getLog().warn(x.getMessage(), x);
+//			}
+		}
+	public Properties getProps() {
+			return props;
+		}
+
+		public void setProps(Properties props) {
+			this.props = props;
+		}
+
+	protected String createUrlPath (String url, int timeoutMs , int limit ) throws IOException
+	{
+		final StringBuffer contexturl= new StringBuffer(url);
+	//	final StringBuffer url = new StringBuffer ( CambriaConstants.makeConsumerUrl ( host, fTopic, fGroup, fId ) );
+		final StringBuffer adds = new StringBuffer ();
+		if ( timeoutMs > -1 ) adds.append ( "timeout=" ).append ( timeoutMs ); 
+		if ( limit > -1 )
+		{
+			if ( adds.length () > 0 )
+			{
+				adds.append ( "&" );
+			}
+			adds.append ( "limit=" ).append ( limit );
+		}
+		if ( fFilter != null && fFilter.length () > 0 )
+		{
+			try {
+				if ( adds.length () > 0 )
+				{
+					adds.append ( "&" );
+				}
+				adds.append("filter=").append(URLEncoder.encode(fFilter, "UTF-8"));
+			} catch (UnsupportedEncodingException e) {
+				throw new RuntimeException(e.getMessage() + "....say whaaaat?!");
+			}
+		}
+		if ( adds.length () > 0 )
+		{
+			contexturl.append ( "?" ).append ( adds.toString () );
+		}
+		
+		//sender.setSubContext(url.toString());
+		return contexturl.toString ();
+	}
+
+	public String getUsername() {
+		return username;
+	}
+
+	public void setUsername(String username) {
+		this.username = username;
+	}
+
+	public String getPassword() {
+		return password;
+	}
+
+	public void setPassword(String password) {
+		this.password = password;
+	}
+
+	public String getHost() {
+		return host;
+	}
+
+	public void setHost(String host) {
+		this.host = host;
+	}
+
+	public String getAuthKey() {
+		return authKey;
+	}
+
+	public void setAuthKey(String authKey) {
+		this.authKey = authKey;
+	}
+
+	public String getAuthDate() {
+		return authDate;
+	}
+
+	public void setAuthDate(String authDate) {
+		this.authDate = authDate;
+	}
+
+	public String getfFilter() {
+		return fFilter;
+	}
+
+	public void setfFilter(String fFilter) {
+		this.fFilter = fFilter;
+	}
+	
+	private String readRoute(String routeKey) {
+
+		try {
+			
+			MRClientFactory.prop.load(new FileReader(new File (MRClientFactory.routeFilePath)));
+
+		} catch (Exception ex) {
+			fLog.error("Reply Router Error " + ex.toString() );
+		}
+		String routeOffer = MRClientFactory.prop.getProperty(routeKey);		
+		return routeOffer;
+	}
+	
+	@Override
+	public MRConsumerResponse fetchWithReturnConsumerResponse() {
+
+		// fetch with the timeout and limit set in constructor
+		return fetchWithReturnConsumerResponse(fTimeoutMs, fLimit);
+	}
+
+	@Override
+	public MRConsumerResponse fetchWithReturnConsumerResponse(int timeoutMs,
+			int limit) {
+		final LinkedList<String> msgs = new LinkedList<String>();
+		MRConsumerResponse mrConsumerResponse = new MRConsumerResponse();
+		try {
+			if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(
+					protocolFlag)) {
+				DMEConfigure(timeoutMs, limit);
+
+				String reply = sender.sendAndWait(timeoutMs + 10000L);
+
+				final JSONObject o = getResponseDataInJsonWithResponseReturned(reply);
+
+				if (o != null) {
+					final JSONArray a = o.getJSONArray("result");
+
+					if (a != null) {
+						for (int i = 0; i < a.length(); i++) {							
+							if (a.get(i) instanceof String)
+								msgs.add(a.getString(i));
+							else
+								msgs.add(a.getJSONObject(i).toString());
+
+						}
+					}
+
+				}
+				createMRConsumerResponse(reply, mrConsumerResponse);
+			}
+
+			if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(
+					protocolFlag)) {
+				final String urlPath = createUrlPath(
+						MRConstants.makeConsumerUrl(host, fTopic, fGroup, fId,
+								props.getProperty("Protocol")), timeoutMs,
+						limit);
+
+				String response = getResponse(urlPath, username, password,
+						protocolFlag);
+
+				final JSONObject o = getResponseDataInJsonWithResponseReturned(response);
+
+				if (o != null) {
+					final JSONArray a = o.getJSONArray("result");
+
+					if (a != null) {
+						for (int i = 0; i < a.length(); i++) {							
+							if (a.get(i) instanceof String)
+								msgs.add(a.getString(i));
+							else
+								msgs.add(a.getJSONObject(i).toString());
+
+						}
+					}
+
+				}
+				createMRConsumerResponse(response, mrConsumerResponse);
+			}
+
+			if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(
+					protocolFlag)) {
+				final String urlPath = createUrlPath(
+						MRConstants.makeConsumerUrl(host, fTopic, fGroup, fId,
+								props.getProperty("Protocol")), timeoutMs,
+						limit);
+
+				String response  = getAuthResponse(urlPath, authKey, authDate,
+						username, password, protocolFlag);
+				final JSONObject o = getResponseDataInJsonWithResponseReturned(response);			
+				if (o != null) {
+					final JSONArray a = o.getJSONArray("result");
+
+					if (a != null) {
+						for (int i = 0; i < a.length(); i++) {
+							if (a.get(i) instanceof String)
+								msgs.add(a.getString(i));
+							else
+								msgs.add(a.getJSONObject(i).toString());
+
+						}
+					}
+
+				}
+				createMRConsumerResponse(response, mrConsumerResponse);
+			}
+			
+			
+			
+		} catch (JSONException e) {	
+			mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+			mrConsumerResponse.setResponseMessage(e.getMessage());
+		} catch (HttpException e) {			
+			mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+			mrConsumerResponse.setResponseMessage(e.getMessage());
+		}catch(DME2Exception e){			
+			mrConsumerResponse.setResponseCode(e.getErrorCode());
+			mrConsumerResponse.setResponseMessage(e.getErrorMessage());
+		}catch (Exception e) {			
+			mrConsumerResponse.setResponseMessage(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+			mrConsumerResponse.setResponseMessage(e.getMessage());
+		}
+		mrConsumerResponse.setActualMessages(msgs);
+		return mrConsumerResponse;
+	}
+
+	private void createMRConsumerResponse(String reply, MRConsumerResponse mrConsumerResponse) {
+		
+		if(reply.startsWith("{")){
+			JSONObject jObject = new JSONObject(reply);
+			String message = jObject.getString("message");
+			int status = jObject.getInt("status");
+		
+			mrConsumerResponse.setResponseCode(Integer.toString(status));
+			
+			if(null != message){
+				mrConsumerResponse.setResponseMessage(message);	
+			}	
+		}else if (reply.startsWith("<")){
+			mrConsumerResponse.setResponseCode(getHTTPErrorResponseCode(reply));
+			mrConsumerResponse.setResponseMessage(getHTTPErrorResponseMessage(reply));			
+		}else{
+			mrConsumerResponse.setResponseCode(String.valueOf(HttpStatus.SC_OK));
+			mrConsumerResponse.setResponseMessage(SUCCESS_MESSAGE);	
+		}
+		
+	}
+
+ 
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRFormat.java b/src/main/java/com/att/nsa/mr/client/impl/MRFormat.java
new file mode 100644
index 0000000..09a317b
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRFormat.java
@@ -0,0 +1,49 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 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=========================================================
+ *
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package com.att.nsa.mr.client.impl;
+
+enum MRFormat
+{
+	/**
+	 * Messages are sent using MR's message format.
+	 */
+	CAMBRIA
+	{
+		public String toString() { return "application/cambria"; }
+	},
+
+	/**
+	 * Messages are sent using MR's message format with compression.
+	 */
+	CAMBRIA_ZIP
+	{
+		public String toString() { return "application/cambria-zip"; }
+	},
+
+	/**
+	 * messages are sent as simple JSON objects.
+	 */
+	JSON
+	{
+		public String toString() { return "application/json"; }
+	}
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRMetaClient.java b/src/main/java/com/att/nsa/mr/client/impl/MRMetaClient.java
new file mode 100644
index 0000000..d32a7ef
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRMetaClient.java
@@ -0,0 +1,260 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 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=========================================================
+ *
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package com.att.nsa.mr.client.impl;
+
+import java.io.IOException;
+import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.Set;
+import java.util.TreeSet;
+
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+
+import com.att.nsa.apiClient.credentials.ApiCredential;
+import com.att.nsa.apiClient.http.HttpException;
+import com.att.nsa.apiClient.http.HttpObjectNotFoundException;
+import com.att.nsa.mr.client.MRIdentityManager;
+import com.att.nsa.mr.client.MRTopicManager;
+
+public class MRMetaClient extends MRBaseClient implements MRTopicManager, MRIdentityManager
+{
+	public MRMetaClient ( Collection<String> baseUrls ) throws MalformedURLException
+	{
+		super ( baseUrls );
+	}
+
+	@Override
+	public Set<String> getTopics () throws IOException
+	{
+		final TreeSet<String> set = new TreeSet<String> ();
+		try
+		{
+			final JSONObject topicSet = get ( "/topics" );
+			final JSONArray a = topicSet.getJSONArray ( "topics" );
+			for ( int i=0; i<a.length (); i++ )
+			{
+				set.add ( a.getString ( i ) );
+			}
+		}
+		catch ( HttpObjectNotFoundException e )
+		{
+			getLog().warn ( "No /topics endpoint on service." );
+		}
+		catch ( JSONException e )
+		{
+			getLog().warn ( "Bad /topics result from service." );
+		}
+		catch ( HttpException e )
+		{
+			throw new IOException ( e );
+		}
+		return set;
+	}
+
+	@Override
+	public TopicInfo getTopicMetadata ( String topic ) throws HttpObjectNotFoundException, IOException
+	{
+		try
+		{
+			final JSONObject topicData = get ( "/topics/" + MRConstants.escape ( topic ) );
+			return new TopicInfo ()
+			{
+				@Override
+				public String getOwner ()
+				{
+					return topicData.optString ( "owner", null );
+				}
+
+				@Override
+				public String getDescription ()
+				{
+					return topicData.optString ( "description", null );
+				}
+
+				@Override
+				public Set<String> getAllowedProducers ()
+				{
+					final JSONObject acl = topicData.optJSONObject ( "writerAcl" );
+					if ( acl != null && acl.optBoolean ( "enabled", true ) )
+					{
+						return jsonArrayToSet ( acl.optJSONArray ( "users" ) );
+					}
+					return null;
+				}
+
+				@Override
+				public Set<String> getAllowedConsumers ()
+				{
+					final JSONObject acl = topicData.optJSONObject ( "readerAcl" );
+					if ( acl != null && acl.optBoolean ( "enabled", true ) )
+					{
+						return jsonArrayToSet ( acl.optJSONArray ( "users" ) );
+					}
+					return null;
+				}
+			};
+		}
+		catch ( JSONException e )
+		{
+			throw new IOException ( e );
+		}
+		catch ( HttpException e )
+		{
+			throw new IOException ( e );
+		}
+	}
+
+	@Override
+	public void createTopic ( String topicName, String topicDescription, int partitionCount, int replicationCount ) throws HttpException, IOException
+	{
+		final JSONObject o = new JSONObject ();
+		o.put ( "topicName", topicName );
+		o.put ( "topicDescription", topicDescription );
+		o.put ( "partitionCount", partitionCount );
+		o.put ( "replicationCount", replicationCount );
+		post ( "/topics/create", o, false );
+	}
+
+	@Override
+	public void deleteTopic ( String topic ) throws HttpException, IOException
+	{
+		delete ( "/topics/" + MRConstants.escape ( topic ) );
+	}
+
+	@Override
+	public boolean isOpenForProducing ( String topic ) throws HttpObjectNotFoundException, IOException
+	{
+		return null == getAllowedProducers ( topic );
+	}
+
+	@Override
+	public Set<String> getAllowedProducers ( String topic ) throws HttpObjectNotFoundException, IOException
+	{
+		return getTopicMetadata ( topic ).getAllowedProducers ();
+	}
+
+	@Override
+	public void allowProducer ( String topic, String apiKey ) throws HttpObjectNotFoundException, HttpException, IOException
+	{
+		put ( "/topics/" + MRConstants.escape ( topic ) + "/producers/" + MRConstants.escape ( apiKey ), new JSONObject() );
+	}
+
+	@Override
+	public void revokeProducer ( String topic, String apiKey ) throws HttpException, IOException
+	{
+		delete ( "/topics/" + MRConstants.escape ( topic ) + "/producers/" + MRConstants.escape ( apiKey ) );
+	}
+
+	@Override
+	public boolean isOpenForConsuming ( String topic ) throws HttpObjectNotFoundException, IOException
+	{
+		return null == getAllowedConsumers ( topic );
+	}
+
+	@Override
+	public Set<String> getAllowedConsumers ( String topic ) throws HttpObjectNotFoundException, IOException
+	{
+		return getTopicMetadata ( topic ).getAllowedConsumers ();
+	}
+
+	@Override
+	public void allowConsumer ( String topic, String apiKey ) throws HttpObjectNotFoundException, HttpException, IOException
+	{
+		put ( "/topics/" + MRConstants.escape ( topic ) + "/consumers/" + MRConstants.escape ( apiKey ), new JSONObject() );
+	}
+
+	@Override
+	public void revokeConsumer ( String topic, String apiKey ) throws HttpException, IOException
+	{
+		delete ( "/topics/" + MRConstants.escape ( topic ) + "/consumers/" + MRConstants.escape ( apiKey ) );
+	}
+
+	@Override
+	public ApiCredential createApiKey ( String email, String description ) throws HttpException, MRApiException, IOException
+	{
+		try
+		{
+			final JSONObject o = new JSONObject ();
+			o.put ( "email", email );
+			o.put ( "description", description );
+			final JSONObject reply = post ( "/apiKeys/create", o, true );
+			return new ApiCredential ( reply.getString ( "key" ), reply.getString ( "secret" ) );
+		}
+		catch ( JSONException e )
+		{
+			// the response doesn't meet our expectation
+			throw new MRApiException ( "The API key response is incomplete.", e );
+		}
+	}
+
+	@Override
+	public ApiKey getApiKey ( String apiKey ) throws HttpObjectNotFoundException, HttpException, IOException
+	{
+		final JSONObject keyEntry = get ( "/apiKeys/" + MRConstants.escape ( apiKey ) );
+		if ( keyEntry == null )
+		{
+			return null;
+		}
+
+		return new ApiKey ()
+		{
+			@Override
+			public String getEmail ()
+			{
+				final JSONObject aux = keyEntry.optJSONObject ( "aux" );
+				if ( aux != null )
+				{
+					return aux.optString ( "email" );
+				}
+				return null;
+			}
+
+			@Override
+			public String getDescription ()
+			{
+				final JSONObject aux = keyEntry.optJSONObject ( "aux" );
+				if ( aux != null )
+				{
+					return aux.optString ( "description" );
+				}
+				return null;
+			}
+		};
+	}
+
+	@Override
+	public void updateCurrentApiKey ( String email, String description ) throws HttpObjectNotFoundException, HttpException, IOException
+	{
+		final JSONObject o = new JSONObject ();
+		if ( email != null ) o.put ( "email", email );
+		if ( description != null ) o.put ( "description", description );
+		patch ( "/apiKeys/" + MRConstants.escape ( getCurrentApiKey() ), o );
+	}
+
+	@Override
+	public void deleteCurrentApiKey () throws HttpException, IOException
+	{
+		delete ( "/apiKeys/" + MRConstants.escape ( getCurrentApiKey() ) );
+	}
+}
diff --git a/src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java b/src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java
new file mode 100644
index 0000000..ad925b1
--- /dev/null
+++ b/src/main/java/com/att/nsa/mr/client/impl/MRSimplerBatchPublisher.java
@@ -0,0 +1,939 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 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=========================================================
+ *
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package com.att.nsa.mr.client.impl;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.zip.GZIPOutputStream;
+
+import javax.ws.rs.core.MultivaluedMap;
+
+import org.apache.http.HttpException;
+import org.apache.http.HttpStatus;
+import org.json.JSONArray;
+import org.json.JSONObject;
+
+import com.att.aft.dme2.api.DME2Client;
+import com.att.aft.dme2.api.DME2Exception;
+import com.att.nsa.mr.client.HostSelector;
+import com.att.nsa.mr.client.MRBatchingPublisher;
+import com.att.nsa.mr.client.response.MRPublisherResponse;
+import com.att.nsa.mr.test.clients.ProtocolTypeConstants;
+
+public class MRSimplerBatchPublisher extends MRBaseClient implements MRBatchingPublisher
+{
+	public static class Builder 
+	{
+		public Builder ()
+		{
+		}
+
+		public Builder againstUrls ( Collection<String> baseUrls )
+		{
+			fUrls = baseUrls;
+			return this;
+		}
+
+		public Builder onTopic ( String topic )
+		{
+			fTopic = topic;
+			return this;
+		}
+
+		public Builder batchTo ( int maxBatchSize, long maxBatchAgeMs )
+		{
+			fMaxBatchSize = maxBatchSize;
+			fMaxBatchAgeMs = maxBatchAgeMs;
+			return this;
+		}
+
+		public Builder compress ( boolean compress )
+		{
+			fCompress = compress;
+			return this;
+		}
+		
+		public Builder httpThreadTime ( int threadOccuranceTime )
+		{
+			this.threadOccuranceTime = threadOccuranceTime;
+			return this;
+		}
+		
+		public Builder allowSelfSignedCertificates( boolean allowSelfSignedCerts )
+		{
+			fAllowSelfSignedCerts = allowSelfSignedCerts;
+			return this;
+		}
+		
+		public Builder withResponse ( boolean withResponse)
+		{
+			fWithResponse = withResponse;
+			return this;
+		}
+		public MRSimplerBatchPublisher build ()
+		{
+			if(!fWithResponse) 
+			{
+				try {
+					return new MRSimplerBatchPublisher ( fUrls, fTopic, fMaxBatchSize, fMaxBatchAgeMs, fCompress, fAllowSelfSignedCerts,threadOccuranceTime);
+				} catch (MalformedURLException e) {
+					throw new RuntimeException(e);
+				}
+			} else 
+			{
+				try {
+					return new MRSimplerBatchPublisher ( fUrls, fTopic, fMaxBatchSize, fMaxBatchAgeMs, fCompress, fAllowSelfSignedCerts, fMaxBatchSize);
+				} catch (MalformedURLException e) {
+					throw new RuntimeException(e);
+				}
+			}
+				
+		}
+
+		private Collection<String> fUrls;
+		private String fTopic;
+		private int fMaxBatchSize = 100;
+		private long fMaxBatchAgeMs = 1000;
+		private boolean fCompress = false;
+		private int threadOccuranceTime = 50;
+		private boolean fAllowSelfSignedCerts = false;
+		private boolean fWithResponse = false;
+		
+	};
+
+	@Override
+	public int send ( String partition, String msg )
+	{
+		return send ( new message ( partition, msg ) );
+	}
+	@Override
+	public int send ( String msg )
+	{
+		return send ( new message ( null, msg ) );
+	}
+
+
+	@Override
+	public int send ( message msg )
+	{
+		final LinkedList<message> list = new LinkedList<message> ();
+		list.add ( msg );
+		return send ( list );
+	}
+	
+	
+
+	@Override
+	public synchronized int send ( Collection<message> msgs )
+	{
+		if ( fClosed )
+		{
+			throw new IllegalStateException ( "The publisher was closed." );
+		}
+		
+		for ( message userMsg : msgs )
+		{
+			fPending.add ( new TimestampedMessage ( userMsg ) );
+		}
+		return getPendingMessageCount ();
+	}
+
+	@Override
+	public synchronized int getPendingMessageCount ()
+	{
+		return fPending.size ();
+	}
+
+	@Override
+	public void close ()
+	{
+		try
+		{
+			final List<message> remains = close ( Long.MAX_VALUE, TimeUnit.MILLISECONDS );
+			if ( remains.size() > 0 )
+			{
+				getLog().warn ( "Closing publisher with " + remains.size() + " messages unsent. "
+					+ "Consider using MRBatchingPublisher.close( long timeout, TimeUnit timeoutUnits ) to recapture unsent messages on close." );
+			}
+		}
+		catch ( InterruptedException e )
+		{
+			getLog().warn ( "Possible message loss. " + e.getMessage(), e );
+		}
+		catch ( IOException e )
+		{
+			getLog().warn ( "Possible message loss. " + e.getMessage(), e );
+		}
+	}
+
+	@Override
+	public List<message> close ( long time, TimeUnit unit ) throws IOException, InterruptedException
+	{
+		synchronized ( this )
+		{
+			fClosed = true;
+
+			// stop the background sender
+			fExec.setContinueExistingPeriodicTasksAfterShutdownPolicy ( false );
+			fExec.setExecuteExistingDelayedTasksAfterShutdownPolicy ( false );
+			fExec.shutdown ();
+		}
+
+		final long now = Clock.now ();
+		final long waitInMs = TimeUnit.MILLISECONDS.convert ( time, unit );
+		final long timeoutAtMs = now + waitInMs;
+
+		while ( Clock.now() < timeoutAtMs && getPendingMessageCount() > 0 )
+		{
+			send ( true );
+			Thread.sleep ( 250 );
+		}
+
+		synchronized ( this )
+		{
+			final LinkedList<message> result = new LinkedList<message> ();
+			fPending.drainTo ( result );
+			return result;
+		}
+	}
+
+	/**
+	 * Possibly send a batch to the MR server. This is called by the background thread
+	 * and the close() method
+	 * 
+	 * @param force
+	 */
+	private synchronized void send ( boolean force )
+	{
+		if ( force || shouldSendNow () )
+		{
+			if ( !sendBatch () )
+			{
+				getLog().warn ( "Send failed, " + fPending.size() + " message to send." );
+
+				// note the time for back-off
+				fDontSendUntilMs = sfWaitAfterError + Clock.now ();
+			}
+		}
+	}
+
+	private synchronized boolean shouldSendNow ()
+	{
+		boolean shouldSend = false;
+		if ( fPending.size () > 0 )
+		{
+			final long nowMs = Clock.now ();
+
+			shouldSend = ( fPending.size() >= fMaxBatchSize );
+			if ( !shouldSend )
+			{
+				final long sendAtMs = fPending.peek().timestamp + fMaxBatchAgeMs;
+				shouldSend = sendAtMs <= nowMs;
+			}
+
+			// however, wait after an error
+			shouldSend = shouldSend && nowMs >= fDontSendUntilMs; 
+		}
+		return shouldSend;
+	}
+
+	private synchronized boolean sendBatch ()
+	{
+		// it's possible for this call to be made with an empty list. in this case, just return.
+		if ( fPending.size() < 1 )
+		{
+			return true;
+		}
+
+		final long nowMs = Clock.now ();
+		
+		host = this.fHostSelector.selectBaseHost();
+		
+		final String httpurl = MRConstants.makeUrl ( host, fTopic,props.getProperty("Protocol"),props.getProperty("partition") );
+		
+
+		try
+		{
+			/*final String contentType =
+				fCompress ?
+					MRFormat.CAMBRIA_ZIP.toString () :
+					MRFormat.CAMBRIA.toString () 
+			;*/
+            
+			final ByteArrayOutputStream baseStream = new ByteArrayOutputStream ();
+			OutputStream os = baseStream;
+			final String contentType = props.getProperty("contenttype");
+			if(contentType.equalsIgnoreCase("application/json")){
+				JSONArray jsonArray = new JSONArray();
+				for ( TimestampedMessage m : fPending )
+				{
+					JSONObject jsonObject = new JSONObject(m.fMsg);
+								
+						jsonArray.put(jsonObject);
+				
+				}
+				os.write (jsonArray.toString().getBytes() );	
+				os.close();
+
+				}else if (contentType.equalsIgnoreCase("text/plain")){
+					for ( TimestampedMessage m : fPending )
+					{										
+						os.write ( m.fMsg.getBytes() );
+						os.write ( '\n' );
+					}
+					os.close ();
+				} else if (contentType.equalsIgnoreCase("application/cambria") ||  (contentType.equalsIgnoreCase("application/cambria-zip"))){
+					if ( contentType.equalsIgnoreCase("application/cambria-zip") )
+					{
+						os = new GZIPOutputStream ( baseStream );
+					}
+					for ( TimestampedMessage m : fPending )
+					{
+						
+						os.write ( ( "" + m.fPartition.length () ).getBytes() );
+						os.write ( '.' );
+						os.write ( ( "" + m.fMsg.length () ).getBytes() );
+						os.write ( '.' );
+						os.write ( m.fPartition.getBytes() );
+						os.write ( m.fMsg.getBytes() );
+						os.write ( '\n' );
+					}
+					os.close ();
+				}else{
+					for ( TimestampedMessage m : fPending )
+					{										
+						os.write ( m.fMsg.getBytes() );
+					
+					}
+					os.close ();
+				}
+             
+		
+
+			final long startMs = Clock.now ();
+			if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) {
+				
+			
+				DME2Configue();
+				
+				Thread.sleep(5);
+				getLog().info ( "sending " + fPending.size() + " msgs to " + url+subContextPath + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms"  );
+				sender.setPayload(os.toString());               
+				String dmeResponse = sender.sendAndWait(5000L);
+				
+				final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):"
+						+ dmeResponse.toString();
+				getLog().info(logLine);
+				fPending.clear();
+				return true;
+			} 
+			
+			if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
+				getLog().info ( "sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms"  );
+				final JSONObject result = postAuth(httpurl, baseStream.toByteArray(), contentType, authKey, authDate, username, password,protocolFlag);
+				//System.out.println(result.getInt("status"));
+				//Here we are checking for error response. If HTTP status
+				//code is not within the http success response code
+				//then we consider this as error and return false
+				if(result.getInt("status") < 200 || result.getInt("status") > 299) {
+					return false;
+				}
+				final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + result.toString();
+				getLog().info(logLine);
+				fPending.clear();
+				return true;
+			} 
+			
+			if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) {
+				getLog().info ( "sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms"  );
+				final JSONObject result = post(httpurl, baseStream.toByteArray(), contentType, username, password, protocolFlag);
+				
+				
+				//System.out.println(result.getInt("status"));
+				//Here we are checking for error response. If HTTP status
+				//code is not within the http success response code
+				//then we consider this as error and return false
+				if(result.getInt("status") < 200 || result.getInt("status") > 299) {
+					return false;
+				}
+				final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + result.toString();
+				getLog().info(logLine);
+				fPending.clear();
+				return true;
+			}
+		}
+		catch ( IllegalArgumentException x ) {
+			getLog().warn ( x.getMessage(), x );
+		} catch ( IOException x ) {
+			getLog().warn ( x.getMessage(), x );
+		} catch (HttpException x) {
+			getLog().warn ( x.getMessage(), x );
+		} catch (Exception x) {
+			getLog().warn(x.getMessage(), x);
+		}
+		return false;
+	}
+
+	public synchronized MRPublisherResponse sendBatchWithResponse () 
+	{
+		// it's possible for this call to be made with an empty list. in this case, just return.
+		if ( fPending.size() < 1 )
+		{
+			pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
+			pubResponse.setResponseMessage("No Messages to send");
+			return pubResponse;
+		}
+
+		final long nowMs = Clock.now ();
+		
+		host = this.fHostSelector.selectBaseHost();
+		
+		final String httpurl = MRConstants.makeUrl ( host, fTopic,props.getProperty("Protocol"),props.getProperty("partition") );
+		OutputStream os=null;
+		try
+		{
+			
+			final ByteArrayOutputStream baseStream = new ByteArrayOutputStream ();
+			 os = baseStream;
+			final String contentType = props.getProperty("contenttype");
+			if(contentType.equalsIgnoreCase("application/json")){
+				JSONArray jsonArray = new JSONArray();
+				for ( TimestampedMessage m : fPending )
+				{
+					JSONObject jsonObject = new JSONObject(m.fMsg);
+								
+						jsonArray.put(jsonObject);
+				
+				}
+				os.write (jsonArray.toString().getBytes() );	
+				}else if (contentType.equalsIgnoreCase("text/plain")){
+					for ( TimestampedMessage m : fPending )
+					{										
+						os.write ( m.fMsg.getBytes() );
+						os.write ( '\n' );
+					}
+				} else if (contentType.equalsIgnoreCase("application/cambria") ||  (contentType.equalsIgnoreCase("application/cambria-zip"))){
+					if ( contentType.equalsIgnoreCase("application/cambria-zip") )
+					{
+						os = new GZIPOutputStream ( baseStream );
+					}
+					for ( TimestampedMessage m : fPending )
+					{
+						
+						os.write ( ( "" + m.fPartition.length () ).getBytes() );
+						os.write ( '.' );
+						os.write ( ( "" + m.fMsg.length () ).getBytes() );
+						os.write ( '.' );
+						os.write ( m.fPartition.getBytes() );
+						os.write ( m.fMsg.getBytes() );
+						os.write ( '\n' );
+					}
+					os.close ();
+				}else{
+					for ( TimestampedMessage m : fPending )
+					{										
+						os.write ( m.fMsg.getBytes() );
+					
+					}
+				}
+             
+		
+
+			final long startMs = Clock.now ();
+			if (ProtocolTypeConstants.DME2.getValue().equalsIgnoreCase(protocolFlag)) {
+				
+			
+				try {
+				DME2Configue();
+				
+				Thread.sleep(5);
+				getLog().info ( "sending " + fPending.size() + " msgs to " + url+subContextPath + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms"  );
+				sender.setPayload(os.toString());               
+						
+				
+				String dmeResponse = sender.sendAndWait(5000L);
+				System.out.println("dmeres->"+dmeResponse);		
+				
+				
+				pubResponse = createMRPublisherResponse(dmeResponse,pubResponse);
+				
+				if(Integer.valueOf(pubResponse.getResponseCode()) < 200 || Integer.valueOf(pubResponse.getResponseCode()) > 299) {
+					
+					return pubResponse;
+				}
+				final String logLine = String.valueOf((Clock.now() - startMs))
+						+ dmeResponse.toString();
+				getLog().info(logLine);
+				fPending.clear();
+				
+				}
+				catch (DME2Exception x) {
+					getLog().warn(x.getMessage(), x);
+					pubResponse.setResponseCode(x.getErrorCode());
+					pubResponse.setResponseMessage(x.getErrorMessage());
+				} catch (URISyntaxException x) {
+					
+					getLog().warn(x.getMessage(), x);
+					pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
+					pubResponse.setResponseMessage(x.getMessage());
+				} catch (Exception x) {
+
+					pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+					pubResponse.setResponseMessage(x.getMessage());
+					
+				}
+				
+				return pubResponse;
+			} 
+			
+			if (ProtocolTypeConstants.AUTH_KEY.getValue().equalsIgnoreCase(protocolFlag)) {
+				getLog().info ( "sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms"  );
+				final String result = postAuthwithResponse(httpurl, baseStream.toByteArray(), contentType, authKey, authDate, username, password,protocolFlag);
+				//System.out.println(result.getInt("status"));
+				//Here we are checking for error response. If HTTP status
+				//code is not within the http success response code
+				//then we consider this as error and return false
+				
+				
+				pubResponse = createMRPublisherResponse(result,pubResponse);
+				
+				if(Integer.valueOf(pubResponse.getResponseCode()) < 200 || Integer.valueOf(pubResponse.getResponseCode()) > 299) {
+					
+					return pubResponse;
+				}
+				
+				final String logLine = "MR reply ok (" + (Clock.now() - startMs) + " ms):" + result.toString();
+				getLog().info(logLine);
+				fPending.clear();
+				return pubResponse;
+			} 
+			
+			if (ProtocolTypeConstants.AAF_AUTH.getValue().equalsIgnoreCase(protocolFlag)) {
+				getLog().info ( "sending " + fPending.size() + " msgs to " + httpurl + ". Oldest: " + ( nowMs - fPending.peek().timestamp ) + " ms"  );
+				final String result = postWithResponse(httpurl, baseStream.toByteArray(), contentType, username, password, protocolFlag);
+				
+				//System.out.println(result.getInt("status"));
+				//Here we are checking for error response. If HTTP status
+				//code is not within the http success response code
+				//then we consider this as error and return false
+				pubResponse = createMRPublisherResponse(result,pubResponse);
+				
+				if(Integer.valueOf(pubResponse.getResponseCode()) < 200 || Integer.valueOf(pubResponse.getResponseCode()) > 299) {
+					
+					return pubResponse;
+				}
+				
+				final String logLine = String.valueOf((Clock.now() - startMs));
+				getLog().info(logLine);
+				fPending.clear();
+				return pubResponse;
+			}
+		}
+		catch ( IllegalArgumentException x ) {
+			getLog().warn ( x.getMessage(), x );
+			pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
+			pubResponse.setResponseMessage(x.getMessage());
+			
+		} catch ( IOException x ) {
+			getLog().warn ( x.getMessage(), x );
+			pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+			pubResponse.setResponseMessage(x.getMessage());
+			
+		} catch (HttpException x) {
+			getLog().warn ( x.getMessage(), x );
+			pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
+			pubResponse.setResponseMessage(x.getMessage());
+			
+		} catch (Exception x) {
+			getLog().warn(x.getMessage(), x);
+			
+			pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+			pubResponse.setResponseMessage(x.getMessage());
+			
+		}
+		
+		finally {
+			if (fPending.size()>0) {
+				getLog().warn ( "Send failed, " + fPending.size() + " message to send." );
+				pubResponse.setPendingMsgs(fPending.size());
+			}
+			if (os != null) {
+				try {
+				os.close();
+				} catch (Exception x) {
+					getLog().warn(x.getMessage(), x);
+					pubResponse.setResponseCode(String.valueOf(HttpStatus.SC_INTERNAL_SERVER_ERROR));
+					pubResponse.setResponseMessage("Error in closing Output Stream");
+				}
+				}
+		}
+		
+		return pubResponse;
+	}
+	
+private MRPublisherResponse createMRPublisherResponse(String reply, MRPublisherResponse mrPubResponse) {
+		
+	 if (reply.isEmpty()) 
+	 {
+		 
+		 mrPubResponse.setResponseCode(String.valueOf(HttpStatus.SC_BAD_REQUEST));
+		 mrPubResponse.setResponseMessage("Please verify the Producer properties");
+	 }
+	 else if(reply.startsWith("{"))
+	 {
+			JSONObject jObject = new JSONObject(reply);
+			if(jObject.has("message") && jObject.has("status"))
+			{
+				String message = jObject.getString("message");
+				if(null != message)
+				{
+					mrPubResponse.setResponseMessage(message);	
+				}
+				mrPubResponse.setResponseCode(Integer.toString(jObject.getInt("status")));
+			}
+			else
+			 {
+					mrPubResponse.setResponseCode(String.valueOf(HttpStatus.SC_OK));
+					mrPubResponse.setResponseMessage(reply);	
+			 }
+     }
+	 else if (reply.startsWith("<"))
+	 {
+		 String responseCode = getHTTPErrorResponseCode(reply);
+		 if( responseCode.contains("403"))
+			{
+			 responseCode = "403";
+			}	
+		 mrPubResponse.setResponseCode(responseCode);
+			mrPubResponse.setResponseMessage(getHTTPErrorResponseMessage(reply));	
+	 }
+	 
+		return mrPubResponse;
+	}
+
+	private final String fTopic;
+	private final int fMaxBatchSize;
+	private final long fMaxBatchAgeMs;
+	private final boolean fCompress;
+	private int threadOccuranceTime;
+	private boolean fClosed;
+	private String username;
+	private String password;
+	private String host;
+	
+	//host selector
+	private HostSelector fHostSelector = null;
+	
+	private final LinkedBlockingQueue<TimestampedMessage> fPending;
+	private long fDontSendUntilMs;
+	private final ScheduledThreadPoolExecutor fExec;
+
+	private String latitude;
+	private String longitude;
+	private String version;
+	private String serviceName;
+	private String env;
+	private String partner;
+	private String routeOffer;
+	private String subContextPath;
+	private String protocol;
+	private String methodType;
+	private String url;
+	private String dmeuser;
+	private String dmepassword;
+	private String contentType;
+	private static final long sfWaitAfterError = 10000;
+	private HashMap<String, String> DMETimeOuts;
+	private DME2Client sender;
+	public String protocolFlag = ProtocolTypeConstants.DME2.getValue();
+	public String producerFilePath;
+	private String authKey;
+	private String authDate;
+	private String handlers;
+	private Properties props;
+	public static String routerFilePath;
+	public static Map<String, String> headers=new HashMap<String, String>();
+	public static MultivaluedMap<String, Object> headersMap;
+	
+	
+	private MRPublisherResponse pubResponse;
+	
+	public MRPublisherResponse getPubResponse() {
+		return pubResponse;
+	}
+	public void setPubResponse(MRPublisherResponse pubResponse) {
+		this.pubResponse = pubResponse;
+	}
+	
+	public static String getRouterFilePath() {
+		return routerFilePath;
+	}
+
+	public static void setRouterFilePath(String routerFilePath) {
+		MRSimplerBatchPublisher.routerFilePath = routerFilePath;
+	}
+
+	public Properties getProps() {
+		return props;
+	}
+
+	public void setProps(Properties props) {
+		this.props = props;
+	}
+
+	public String getProducerFilePath() {
+		return producerFilePath;
+	}
+
+	public void setProducerFilePath(String producerFilePath) {
+		this.producerFilePath = producerFilePath;
+	}
+
+	public String getProtocolFlag() {
+		return protocolFlag;
+	}
+
+	public void setProtocolFlag(String protocolFlag) {
+		this.protocolFlag = protocolFlag;
+	}
+	
+	
+	private void DME2Configue() throws Exception {
+		try {
+			
+		/*	FileReader reader = new FileReader(new File (producerFilePath));
+			Properties props = new Properties();		
+			props.load(reader);*/
+			latitude = props.getProperty("Latitude");
+			longitude = props.getProperty("Longitude");
+			version = props.getProperty("Version");
+			serviceName = props.getProperty("ServiceName");
+			env = props.getProperty("Environment");
+			partner = props.getProperty("Partner");
+			routeOffer = props.getProperty("routeOffer");
+			subContextPath = props.getProperty("SubContextPath")+fTopic;
+			/*if(props.getProperty("partition")!=null && !props.getProperty("partition").equalsIgnoreCase("")){
+				subContextPath=subContextPath+"?partitionKey="+props.getProperty("partition");
+			}*/			
+			protocol = props.getProperty("Protocol");
+			methodType = props.getProperty("MethodType");
+			dmeuser = props.getProperty("username");
+			dmepassword = props.getProperty("password");
+			contentType = props.getProperty("contenttype");
+			handlers = props.getProperty("sessionstickinessrequired");
+			routerFilePath= props.getProperty("DME2preferredRouterFilePath");
+			
+			/**
+			 * Changes to DME2Client url to use Partner for auto failover between data centers
+			 * When Partner value is not provided use the routeOffer value for auto failover within a cluster 
+			 */
+			
+
+			String partitionKey = props.getProperty("partition");
+			
+			if (partner != null && !partner.isEmpty() ) 
+			{ 
+				url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&partner="+partner; 
+                if(partitionKey!=null && !partitionKey.equalsIgnoreCase("")){
+                    url = url + "&partitionKey=" + partitionKey;
+                }
+			}
+			else if (routeOffer!=null && !routeOffer.isEmpty()) 
+			{ 
+				url = protocol + "://"+serviceName+"?version="+version+"&envContext="+env+"&routeoffer="+routeOffer;
+                if(partitionKey!=null && !partitionKey.equalsIgnoreCase("")){
+                    url = url + "&partitionKey=" + partitionKey;
+                }
+			}
+			 
+			DMETimeOuts = new HashMap<String, String>();
+			DMETimeOuts.put("AFT_DME2_EP_READ_TIMEOUT_MS", props.getProperty("AFT_DME2_EP_READ_TIMEOUT_MS"));
+			DMETimeOuts.put("AFT_DME2_ROUNDTRIP_TIMEOUT_MS", props.getProperty("AFT_DME2_ROUNDTRIP_TIMEOUT_MS"));
+			DMETimeOuts.put("AFT_DME2_EP_CONN_TIMEOUT", props.getProperty("AFT_DME2_EP_CONN_TIMEOUT"));
+			DMETimeOuts.put("Content-Type", contentType);
+			System.setProperty("AFT_LATITUDE", latitude);
+			System.setProperty("AFT_LONGITUDE", longitude);
+			System.setProperty("AFT_ENVIRONMENT",props.getProperty("AFT_ENVIRONMENT"));
+			//System.setProperty("DME2.DEBUG", "true");
+		//	System.setProperty("AFT_DME2_HTTP_EXCHANGE_TRACE_ON", "true");
+			//System.out.println("XXXXXX"+url);
+			
+			//SSL changes
+			System.setProperty("AFT_DME2_CLIENT_SSL_INCLUDE_PROTOCOLS",
+					"SSLv3,TLSv1,TLSv1.1");
+			System.setProperty("AFT_DME2_CLIENT_IGNORE_SSL_CONFIG", "false");
+			System.setProperty("AFT_DME2_CLIENT_KEYSTORE_PASSWORD", "changeit");
+			
+			//SSL changes
+			
+			sender = new DME2Client(new URI(url), 5000L);
+				
+			sender.setAllowAllHttpReturnCodes(true);
+			sender.setMethod(methodType);
+			sender.setSubContext(subContextPath);	
+			sender.setCredentials(dmeuser, dmepassword);
+			sender.setHeaders(DMETimeOuts);
+			if(handlers.equalsIgnoreCase("yes")){
+				sender.addHeader("AFT_DME2_EXCHANGE_REQUEST_HANDLERS", props.getProperty("AFT_DME2_EXCHANGE_REQUEST_HANDLERS"));
+				sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", props.getProperty("AFT_DME2_EXCHANGE_REPLY_HANDLERS"));
+				sender.addHeader("AFT_DME2_REQ_TRACE_ON", props.getProperty("AFT_DME2_REQ_TRACE_ON"));
+				}else{
+					sender.addHeader("AFT_DME2_EXCHANGE_REPLY_HANDLERS", "com.att.nsa.mr.dme.client.HeaderReplyHandler");
+				}
+		} catch (DME2Exception x) {
+			getLog().warn(x.getMessage(), x);
+			throw new DME2Exception(x.getErrorCode(),x.getErrorMessage());
+		} catch (URISyntaxException x) {
+			
+			getLog().warn(x.getMessage(), x);
+			throw new URISyntaxException(url,x.getMessage());
+		} catch (Exception x) {
+
+			getLog().warn(x.getMessage(), x);
+			throw new Exception(x.getMessage());
+		}
+	}
+	
+	private MRSimplerBatchPublisher ( Collection<String> hosts, String topic, int maxBatchSize, long maxBatchAgeMs, boolean compress) throws MalformedURLException
+	{
+		super ( hosts );
+
+		if ( topic == null || topic.length() < 1 )
+		{
+			throw new IllegalArgumentException ( "A topic must be provided." );
+		}
+		
+		fHostSelector = new HostSelector(hosts, null);
+		fClosed = false;
+		fTopic = topic;
+		fMaxBatchSize = maxBatchSize;
+		fMaxBatchAgeMs = maxBatchAgeMs;
+		fCompress = compress;
+
+		fPending = new LinkedBlockingQueue<TimestampedMessage> ();
+		fDontSendUntilMs = 0;
+		fExec = new ScheduledThreadPoolExecutor ( 1 );
+		pubResponse = new MRPublisherResponse();
+		
+	}
+	
+	private MRSimplerBatchPublisher ( Collection<String> hosts, String topic, int maxBatchSize, long maxBatchAgeMs, boolean compress, boolean allowSelfSignedCerts, int httpThreadOccurnace ) throws MalformedURLException
+	{
+		super ( hosts );
+
+		if ( topic == null || topic.length() < 1 )
+		{
+			throw new IllegalArgumentException ( "A topic must be provided." );
+		}
+		
+		fHostSelector = new HostSelector(hosts, null);
+		fClosed = false;
+		fTopic = topic;
+		fMaxBatchSize = maxBatchSize;
+		fMaxBatchAgeMs = maxBatchAgeMs;
+		fCompress = compress;
+		threadOccuranceTime=httpThreadOccurnace;
+		fPending = new LinkedBlockingQueue<TimestampedMessage> ();
+		fDontSendUntilMs = 0;
+		fExec = new ScheduledThreadPoolExecutor ( 1 );
+		fExec.scheduleAtFixedRate ( new Runnable()
+		{
+			@Override
+			public void run ()
+			{
+				send ( false );
+			}
+		}, 100, threadOccuranceTime, TimeUnit.MILLISECONDS );
+	}
+
+	private static class TimestampedMessage extends message
+	{
+		public TimestampedMessage ( message m )
+		{
+			super ( m );
+			timestamp = Clock.now();
+		}
+		public final long timestamp;
+	}
+
+	public String getUsername() {
+		return username;
+	}
+
+	public void setUsername(String username) {
+		this.username = username;
+	}
+
+	public String getPassword() {
+		return password;
+	}
+
+	public void setPassword(String password) {
+		this.password = password;
+	}
+
+	public String getHost() {
+		return host;
+	}
+
+	public void setHost(String host) {
+		this.host = host;
+	}
+
+	public String getContentType() {
+		return contentType;
+	}
+
+	public void setContentType(String contentType) {
+		this.contentType = contentType;
+	}
+
+	public String getAuthKey() {
+		return authKey;
+	}
+
+	public void setAuthKey(String authKey) {
+		this.authKey = authKey;
+	}
+
+	public String getAuthDate() {
+		return authDate;
+	}
+
+	public void setAuthDate(String authDate) {
+		this.authDate = authDate;
+	}
+	
+}