public interface Consumer {

    public enum Status{
        INITIALIZED, RUNNING, STOPPED
    }
    public Status getStatus();

    /**
     * Start the consumer with the provided ConsumerConfig class.
     *
     * @param config: A {@link ConsumerConfig} to configure this Consumer.
     * @return true for success otherwise false.
     */
    public boolean start(ConsumerConfig config) throws InvalidConfigException, InvalidStatusException;

    /**
     * Stop the consumer
     */
    public void stop() throws InvalidStatusException;

    /**
     * ReceiveImmediate checks if a value can be fetched without blocking
     * returns the value or null otherwise.
     *
     * The user is required to ack() each message if he chooses ack_mode as
     * 'CLIENT' in ConsumerConfig. Setting ack_mode as 'AUTO_CLIENT' means this
     * consumer library will acknowledge internally after each receive() call.
     *
     * @return Messsage : {@link Message} received.
     */
    public Message receiveImmediate();

    /**
     * Blocking receive upto maximum of specified timeout milliseconds.
     *
     * The user is required to ack() each message if he chooses ack_mode as
     * 'CLIENT' in ConsumerConfig. Setting ack_mode as 'AUTO_CLIENT' means this
     * consumer library will acknowledge internally after each receive() call.
     *
     * @param timeout: in ms
     *
     * @return Messsage : {@link Message} received.
     */
    public Message receive(long timeout) throws ReceiveTimeoutException;

    /**
     * Blocking receive.
     *
     * The user is required to ack() each message if he chooses ack_mode as
     * 'CLIENT' in ConsumerConfig. Setting ack_mode as 'AUTO_CLIENT' means this
     * consumer library will acknowledge internally after each receive() call.
     *
     * @return Messsage : {@link Message} received.
     */
    public Message receive();

    /**
     * Ack the last received message.
     * 
     * The user is required to ack() each message if he chooses ack_mode as
     * 'CLIENT' in ConsumerConfig. Setting ack_mode as 'AUTO_CLIENT' means this
     * consumer library will acknowledge internally after each receive() call
     * 
     * @return true or false if ack failed.
     */
    public boolean ack();

    /**
     * Ack the message given the ackId contained in Message.
     *
     * The user is required to ack() each message if he chooses ack_mode as
     * 'CLIENT' in ConsumerConfig. Setting ack_mode as 'AUTO_CLIENT' means this
     * consumer library will acknowledge internally after each receive() call
     *
     * @param ackId AckID field in Message.
     *
     * @return true or false if ack failed.
     */

    public boolean ack(String ackId);

    /**
     * Ack the last message received.
     *
     * The user is required to ack() each message if he chooses ack_mode as
     * 'CLIENT' in ConsumerConfig. Setting ack_mode as 'AUTO_CLIENT' means this
     * consumer library will acknowledge internally after each receive() call
     * The ackSafe() call blocks till it receives a confirmation from the server
     * Blocking ackSafe upto maximum of specified timeout milliseconds
     * @param ackId AckID field in Message.
     *
     * @return true or false if ack failed.
     * @throws InterruptedException 
     */

    public boolean ackSafe();


    /**
     * Ack the message given the ackId contained in Message.
     * 
     * The user is required to ack() each message if he chooses ack_mode as
     * 'CLIENT' in ConsumerConfig. Setting ack_mode as 'AUTO_CLIENT' means this
     * consumer library will acknowledge internally after each receive() call
     * The ackSafe() call blocks till it receives a confirmation from the server
     * 
     * @param ackId AckID field in Message.
     *
     * @return true or false if ack failed.
     */
    public boolean ackSafe(String ackId );
    /**
     * Ack the last message received.
     *
     * The user is required to ack() each message if he chooses ack_mode as
     * 'CLIENT' in ConsumerConfig. Setting ack_mode as 'AUTO_CLIENT' means this
     * consumer library will acknowledge internally after each receive() call
     * The ackSafe() call blocks till it receives a confirmation from the server
     *
     * @param ackId AckID field in Message.
     *
     * @return true or false if ack failed.
     * @throws InterruptedException
     */

    public boolean ackSafe(long timeout);


    /**
     * Ack the message given the ackId contained in Message.
     *
     * The user is required to ack() each message if he chooses ack_mode as
     * 'CLIENT' in ConsumerConfig. Setting ack_mode as 'AUTO_CLIENT' means this
     * consumer library will acknowledge internally after each receive() call
     * The ackSafe() call blocks till it receives a confirmation from the server
     *
     * Blocking ackSafe upto maximum of specified timeout milliseconds.
     * @param ackId AckID field in Message.
     *
     * @return true or false if ack failed.
     */
    public boolean ackSafe(String ackId, long timeout);

    /**
     * Nack the last received message
     *
     * @return true or false if nack failed.
     */

    public boolean nack();

    /**
     * Nack the message given the msgId contained in Message.
     *
     * @param msgId MsgID field in Message.
     *
     * @return true or false if nack failed.
     */

    public boolean nack(String msgId);

    /**
     * Keepalive sends a heart beat to all the servers
     * @return true or false if keepalive failed.
     */
    public boolean keepAlive();
} public enum ConsumerAckType{
    AUTO_CLIENT_ACK,
    CLIENT_ACK;
} public class ConsumerConfig {

    // Set of servers to consume from.
    private Set<HostParams> hostParamsSet = null;

    // Consumer destination information.
    private String destinationName;
    private DestinationType destinationType = DestinationType.QUEUE;
    private ConsumerAckType ackType = ConsumerAckType.CLIENT_ACK;

    // Connection lifetime. Servers drop consumer connections if inactive for
    // this time period.
    // also used for refreshing consumer connection internally.
    private long connectionLifetime = 300000;

    // For topics (only) different subscription id means different subscriptions
    // (each gets same message)
    // specifying same subscription ids means for that subscriptions data will
    // be load-balanced.
    // For queues setting subscription id is a noop and all consumers get
    // load-balanced.
    private String subscriptionId;

    // URL to fetch active server list dynamically.
    private String dynamicServerListFetchURL;
    private boolean useDynamicServerList = true;



    private String userName = "guest";
    private String password = "guest";

    private long receiveSleepInterval = 1;
    // Consumer maintains internal thread pool. This property sets the size of
    // the thread pool. Default value is 4. We recommend setting this less than
    // half the number of cores in the box
    private int threadPoolSize = 4;

    /******
     * 
     * @deprecated Number of threads per server is now always 1
     * 
     * @return Always 1.
     */
    @Deprecated
    public int getNumOfThreadsPerServer() {
        return 1;
    }

    /***
     * 
     * @deprecated Number of threads per server is now always 1. This function
     *             does nothing
     * 
     * @param numOfThreadsPerServer
     * 
     */
    @Deprecated
    public void setNumOfThreadsPerServer(int numOfThreadsPerServer) {

    }

    public boolean useDynamicServerList() {
        return useDynamicServerList;
    }

    public void setUseDynamicServerList(boolean useDynamicServerList) {
        this.useDynamicServerList = useDynamicServerList;
    }
    
    public int getThreadPoolSize() {
        return threadPoolSize;
    }

    public void setThreadPoolSize(int threadPoolSize) {
        this.threadPoolSize = threadPoolSize;
    }

    public Set<HostParams> getHostParams() {
        return hostParamsSet;
    }

    public void setHostParams(Set<HostParams> aHostList) {
        this.hostParamsSet = aHostList;
    }

    public long getConnectionLifetime() {
        return connectionLifetime;
    }

    public void setConnectionLifetime(long connectionLifetime) {
        this.connectionLifetime = connectionLifetime;
    }

    public String getDestinationName() {
        return destinationName;
    }

    public void setDestinationName(String destinationName) {
        this.destinationName = destinationName;
    }

    public DestinationType getDestinationType() {
        return destinationType;
    }

    public void setDestinationType(DestinationType destinationType) {
        this.destinationType = destinationType;
    }

    public ConsumerAckType getAckType() {
        return ackType;
    }

    public void setAckType(ConsumerAckType ackType) {
        this.ackType = ackType;
    }

    public String getSubscriptionId() {
        return subscriptionId;
    }

    public void setSubscriptionId(String subscriptionId) {
        this.subscriptionId = subscriptionId;
    }

    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 getDynamicServerListFetchURL() {
        return dynamicServerListFetchURL;
    }

    public void setDynamicServerListFetchURL(String dynamicServerListFetchURL) {
        this.dynamicServerListFetchURL = dynamicServerListFetchURL;
    }

    public long getReceiveSleepInterval() {
        return receiveSleepInterval;
    }

    public void setReceiveSleepInterval(long receiveSleepInterval) {
        this.receiveSleepInterval = receiveSleepInterval;
    }

} public enum DestinationType{
        QUEUE,
        TOPIC
} public class HostParams {
    private String host;
    private int port;

    public HostParams(String aHost, int aPort) {
        host = aHost;
        port = aPort;
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public String toString() {
        return host + ":" + port;
    }

    @Override
    public int hashCode() {
        String hostName = host + ":" + port;
        return hostName.hashCode();
    }

    @Override
    public boolean equals(Object o) {

        if (o instanceof HostParams) {
            HostParams other = (HostParams) o;

            return ((host + ":" + port).equalsIgnoreCase((other.getHost() + ":" + other.getPort())));
        }

        return false;
    }

} @SuppressWarnings("serial")
public class AckFailedException extends MessageBusException {
    public AckFailedException(Exception e){
        super(e);
    }

    public AckFailedException(String message) {
        super(message);
    }
} @SuppressWarnings("serial")
public class BrokerConnectionFailedException extends MessageBusException {
    public BrokerConnectionFailedException(String message) {
        super(message);
    }
    
    public BrokerConnectionFailedException(Exception e){
        super(e);
    }
} @SuppressWarnings("serial")
public class InvalidConfigException extends MessageBusException{
    public InvalidConfigException(String message){
        super(message);
    }
} @SuppressWarnings("serial")
public class InvalidDestinationException extends RuntimeException{
    public InvalidDestinationException(Exception e){
        super(e);
    }

    public InvalidDestinationException(String msg) {
        super(msg);
    }
} @SuppressWarnings("serial")
public class InvalidStatusException extends RuntimeException {

    public InvalidStatusException(String message) {
        super(message);
    }

} @SuppressWarnings("serial")
public class KeepAliveFailedException extends MessageBusException{

    public KeepAliveFailedException(Exception e) {
        super(e);
    }

} @SuppressWarnings("serial")
public class NackFailedException extends MessageBusException{

    public NackFailedException(Exception e) {
        super(e);
    }

} @SuppressWarnings("serial")
public class ReceiveTimeoutException extends MessageBusException{
    public ReceiveTimeoutException(Exception e) {
        super(e);
    }

    public ReceiveTimeoutException(String message){
        super(message);
    }


} public class Message {

    private final MessageInternal messageInternal;
    // Used to encode host address and message-id. Only used by consumer.
    private String ackId;

    public String getAckId() {
        return ackId;
    }

    public void setAckId(String ackId) {
        this.ackId = ackId;
    }

    private Logger log = Logger.getLogger(Message.class);
    
    private static String getSaltedMessageId(){
        String ret = "";
        try{
            String seed = UUID.randomUUID().toString();
            byte[] bytes =MessageDigest.getInstance("MD5").digest( seed.getBytes());
            StringBuilder sb = new StringBuilder();
            for (byte b : bytes) {
                sb.append(String.format("%02x", b));
            }
            ret = sb.toString();
        }
        catch(NoSuchAlgorithmException e){
            ret = UUID.randomUUID().toString();
        }
        return ret;
    }
    
    public static Message createStringMessage(String messagePayload) {
        return createStringMessage(getSaltedMessageId(), messagePayload);
    }

    public static Message createStringMessage(String messageId,
            String messagePayload) {
        // Create a MessagePayload with string payload type.
        MessagePayload payload = new MessagePayload(MessagePayloadType.STRING);
        payload.setStringPayload(messagePayload);

        return new Message(new MessageInternal(messageId, payload));
    }

    public static Message createBinaryMessage(byte[] binaryPayload) {
        
        return createBinaryMessage(getSaltedMessageId(), binaryPayload);
    }
    
    public static Message createBinaryMessage(String messageId,
            byte[] binaryPayload) {
        // Create a MessagePayload with binary payload type.
        MessagePayload payload = new MessagePayload(MessagePayloadType.BINARY);
        payload.setBinaryPayload(binaryPayload);

        return new Message(new MessageInternal(messageId, payload));
    }

    public static Message createJsonMessage(Object objectPayload){
        return createJsonMessage(getSaltedMessageId(), objectPayload);
    }
            
    
    public static Message createJsonMessage(String messageId,
            Object objectPayload) {
        // convert the passed object to its json representation using
        // google-gson library.
        Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'")
                .create();
        String jsonData = gson.toJson(objectPayload);

        // Create a MessagePayload with json string payload type.
        MessagePayload payload = new MessagePayload(MessagePayloadType.JSON);
        payload.setStringPayload(jsonData);

        return new Message(new MessageInternal(messageId, payload));
    }
    
    public static Message createJsonStringMessage(String jsonString){
        return createJsonStringMessage(getSaltedMessageId(), jsonString);
    }

    public static Message createJsonStringMessage(String messageId,
            String jsonString) {
        // Create a MessagePayload with json string payload type.
        MessagePayload payload = new MessagePayload(MessagePayloadType.JSON);
        payload.setStringPayload(jsonString);

        return new Message(new MessageInternal(messageId, payload));
    }

    public Message(MessageInternal messageInternal) {
        this.messageInternal = messageInternal;
    }

    public String getMessageId() {
        return messageInternal.getMessageId();
    }

    public MessagePayloadType getMessagePayloadType() {
        return messageInternal.getPayload().getMessageFormat();
    }

    public String getJSONStringPayload() throws MessageBusException {
        MessagePayloadType payloadFormat = messageInternal.getPayload()
                .getMessageFormat();
        if (!MessagePayloadType.JSON.equals(payloadFormat)) {
            throw new MessageBusException("Cannot get KeyValue payload from "
                    + payloadFormat + " payload type message.");
        }

        return messageInternal.getPayload().getStringPayload();
    }

    public String getStringPayload() throws MessageBusException {
        MessagePayloadType payloadFormat = messageInternal.getPayload()
                .getMessageFormat();
        if (!MessagePayloadType.STRING.equals(payloadFormat)) {
            throw new MessageBusException("Cannot get String payload from "
                    + payloadFormat + " payload type message.");
        }

        return messageInternal.getPayload().getStringPayload();
    }

    public byte[] getBinaryPayload() throws MessageBusException {
        MessagePayloadType payloadFormat = messageInternal.getPayload()
                .getMessageFormat();
        if (!MessagePayloadType.BINARY.equals(payloadFormat)) {
            throw new MessageBusException("Cannot get Binary payload from "
                    + payloadFormat + " payload type message.");
        }

        return messageInternal.getPayload().getBinaryPayload();
    }

    public void setMessageProperties(Map<String, String> properties) {
        messageInternal.setProperties(properties);
    }

    public Map<String, String> getMessageProperties() {
        return messageInternal.getProperties();
    }

    public MessageInternal getMessageInternal() {
        return messageInternal;
    }
} public class DynamicServerListGetter {

    private static Logger log = Logger.getLogger(DynamicServerListGetter.class);
    private static final int MAX_READ_TIMEOUT = 5000;
    private static final int MAX_CONNECT_TIMEOUT = 1000;
    
    public static String buildDynamicServersURL(String hostname, int port) throws URISyntaxException{        
        URIBuilder builder =  new URIBuilder();
        builder.setHost(hostname);
        builder.setPort(port);
        builder.setPath("/jmx");
        builder.addParameter("command", "get_attribute");
        builder.addParameter("args", "org.hornetq:module=Core,type=Server ListOfBrokers");
        builder.setScheme("http");
        return builder.build().toASCIIString();        
    }
    
    protected static String fetch(String aURL) throws MalformedURLException, IOException {
        String content = null;
        URLConnection connection = null;
            
        connection = new URL(aURL).openConnection();
        connection.setConnectTimeout(MAX_CONNECT_TIMEOUT);
        connection.setReadTimeout(MAX_READ_TIMEOUT);
        Scanner scanner = new Scanner(connection.getInputStream());
        scanner.useDelimiter("\\Z");
        content = scanner.next();


        return content;
    }

    public static Set<HostParams> parseAndReturnHosts(String content) {
        Set<HostParams> serverSet = new HashSet<HostParams>();

        if (content == null || content.trim().length() == 0){
            return serverSet;
        }

        String[] servers = content.split(",");
        for (String host : servers) {                    
            String[] hostParam = host.split(":");
            serverSet.add(new HostParams(hostParam[0], Integer.parseInt(hostParam[1])));
        }

        return serverSet;
    }

    /**
     * Fetches XML from given URL, 
     * parses list of brokers and returns the brokers
     * as set of HostParams objects
     * @param aURL
     * @return
     * @throws IOException 
     * @throws MalformedURLException 
     */
    public static Set<HostParams> fetchHostList(String aURL) throws MalformedURLException, IOException {
        return parseAndReturnHosts(fetch(aURL));                              
    }

    public static void main(String[] args) throws MalformedURLException, IOException {


        String aURL = "http://localhost:18081/jmx?command=get_attribute&args=org.hornetq%3Amodule%3DCore%2Ctype%3DServer%20ListOfBrokers";
        BasicConfigurator.configure();


         Set<HostParams> hosts = parseAndReturnHosts(DynamicServerListGetter.fetch(aURL));


         if(hosts!=null){
             for(HostParams host: hosts){
                 System.out.println(host);
             }
         }

         if (hosts.size() == 0 || hosts == null)
             System.out.println("No servers found");


         }


} public class StompFrame implements Command {

    public static final byte[] NO_DATA = new byte[] {};

    private String action;
    private Map<String, String> headers = new HashMap<String, String>();
    private byte[] content = NO_DATA;

    public StompFrame(String command) {
        this(command, null, null);
    }

    public StompFrame(String command, Map<String, String> headers) {
        this(command, headers, null);
    }

    public StompFrame(String command, Map<String, String> headers, byte[] data) {
        this.action = command;
        if (headers != null)
            this.headers = headers;
        if (data != null)
            this.content = data;
    }

    public StompFrame() {
    }

    public String getAction() {
        return action;
    }

    public void setAction(String command) {
        this.action = command;
    }

    public byte[] getContent() {
        return content;
    }

    public String getBody() {
        try {
            return new String(content, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            return new String(content);
        }
    }

    public void setContent(byte[] data) {
        this.content = data;
    }

    public Map<String, String> getHeaders() {
        return headers;
    }

    public void setHeaders(Map<String, String> headers) {
        this.headers = headers;
    }

    //
    // Methods in the Command interface
    //
    public int getCommandId() {
        return 0;
    }

    public Endpoint getFrom() {
        return null;
    }

    public Endpoint getTo() {
        return null;
    }

    public boolean isBrokerInfo() {
        return false;
    }

    public boolean isMessage() {
        return false;
    }

    public boolean isMessageAck() {
        return false;
    }

    public boolean isMessageDispatch() {
        return false;
    }

    public boolean isMessageDispatchNotification() {
        return false;
    }

    public boolean isResponse() {
        return false;
    }

    public boolean isResponseRequired() {
        return false;
    }

    public boolean isShutdownInfo() {
        return false;
    }

    public boolean isConnectionControl() {
        return false;
    }

    public boolean isWireFormatInfo() {
        return false;
    }

    public void setCommandId(int value) {
    }

    public void setFrom(Endpoint from) {
    }

    public void setResponseRequired(boolean responseRequired) {
    }

    public void setTo(Endpoint to) {
    }

    /*
    public Response visit(CommandVisitor visitor) throws Exception {
        return null;
    }
    */

    public byte getDataStructureType() {
        return 0;
    }

    public boolean isMarshallAware() {
        return false;
    }

    public String toString() {
        return format(true);
    }

    public String format() {
        return format(false);
    }

    public String format(boolean forLogging) {
        StringBuffer buffer = new StringBuffer();
        buffer.append(getAction());
        buffer.append("\n");
        Map headers = getHeaders();
        for (Iterator iter = headers.entrySet().iterator(); iter.hasNext();) {
            Map.Entry entry = (Map.Entry)iter.next();
            buffer.append(entry.getKey());
            buffer.append(":");
            if (forLogging && entry.getKey().toString().toLowerCase().contains(Stomp.Headers.Connect.PASSCODE)) {
                buffer.append("*****");
            } else {
                buffer.append(entry.getValue());
            }
            buffer.append("\n");
        }
        buffer.append("\n");
        if (getContent() != null) {
            try {
                String contentString = new String(getContent(), "UTF-8");
                if (forLogging) {
                    contentString = MarshallingSupport.truncate64(contentString);
                }
                buffer.append(contentString);
            } catch (Throwable e) {
                buffer.append(Arrays.toString(getContent()));
            }
        }
        return buffer.toString();
    }
}
import java.io.IOException;
import java.net.MalformedURLException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.apache.http.client.utils.URIBuilder;
import org.apache.log4j.Logger;
import com.groupon.messagebus.api.Consumer;
import com.groupon.messagebus.api.ConsumerAckType;
import com.groupon.messagebus.api.ConsumerConfig;
import com.groupon.messagebus.api.DestinationType;
import com.groupon.messagebus.api.HostParams;
import com.groupon.messagebus.api.exceptions.AckFailedException;
import com.groupon.messagebus.api.exceptions.BrokerConnectionFailedException;
import com.groupon.messagebus.api.exceptions.InvalidConfigException;
import com.groupon.messagebus.api.exceptions.InvalidDestinationException;
import com.groupon.messagebus.api.exceptions.InvalidStatusException;
import com.groupon.messagebus.api.exceptions.KeepAliveFailedException;
import com.groupon.messagebus.api.exceptions.NackFailedException;
import com.groupon.messagebus.api.exceptions.ReceiveTimeoutException;
import com.groupon.messagebus.api.Message;
import com.groupon.messagebus.util.DynamicServerListGetter;
import com.groupon.stomp.StompFrame;
package com.groupon.messagebus.client;



/**
 * Actual implementation of Consumer This class provides following: Starts new
 * threads with all the configured brokers. Every time receive() is called, this
 * class round robins between these threads to access next message. In a way its
 * a load balancer for these threads.
 * 
 * Copyright (c) 2013, Groupon, Inc.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are
 * met:
 *
 * Redistributions of source code must retain the above copyright notice,
 * this list of conditions and the following disclaimer.
 *
 * Redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the following disclaimer in the
 * documentation and/or other materials provided with the distribution.
 *
 * Neither the name of GROUPON nor the names of its contributors may be
 * used to endorse or promote products derived from this software without
 * specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
 * IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
 * TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
 * PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
 * HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED
 * TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
public class ConsumerImpl implements Consumer {

    private Logger log = Logger.getLogger(ConsumerImpl.class);
    private Timer refreshServerListTimer = new Timer();

    private List<StompServerFetcher> serverList = Collections
            .synchronizedList(new ArrayList<StompServerFetcher>());
    private Map<HostParams, List<StompServerFetcher>> currentServers = new HashMap<HostParams, List<StompServerFetcher>>();
    private StompServerFetcher lastSentServer = null;
    private int lastContactedServerIdx = -1;

    private ConsumerConfig config;
    private static final String TOPIC_NAME_PREFIX = "jms.topic.";
    private static final String QUEUE_NAME_PREFIX = "jms.queue.";
    private static final long ACKSAFE_TIMEOUT = 1000;
    private Status status;

    public Status getStatus() {
        return status;
    }

    ExecutorService executor;

    public ConsumerImpl() {
        this.status = Status.INITIALIZED;
    }

    @Override
    public boolean start(ConsumerConfig aConfig) throws InvalidConfigException,
            InvalidStatusException {