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;
    }

} public class Log {
    public static void log(String message){
        System.out.println(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;
    }
} @SuppressWarnings("serial")
public class InvalidConfigException extends MessageBusException{
    public InvalidConfigException(String message){
        super(message);
    }
} @SuppressWarnings("serial")
public class InvalidStatusException extends RuntimeException {

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

} 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 {

        log.debug("consumer " + this.toString() + " starting.");

        switch (this.status) {
        case INITIALIZED:
            break;
        default:
            throw new InvalidStatusException(
                    "Consumer cannot be started. Status=" + this.status);
        }

        config = aConfig;

        validateConfigs(config);

        Set<HostParams> hostsList = null;
                
        //If useDynamicServerList and url is null, generate it.
        if(config.useDynamicServerList() && null == config.getDynamicServerListFetchURL() && 
           null != config.getHostParams() && config.getHostParams().size() > 0){
            HostParams param = config.getHostParams().iterator().next();

            try {
                config.setDynamicServerListFetchURL(DynamicServerListGetter.buildDynamicServersURL(param.getHost(), 8081));
            } catch (URISyntaxException e) {
                log.error("Error creating dynamic server url from " + param, e);
            }
        }
        
        if (null != config.getDynamicServerListFetchURL()) {

            try {
                hostsList = fetchHostList();
            } catch (MalformedURLException e) {
                throw new InvalidConfigException(
                        "Invalid dynamic server list fetcher url "
                                + config.getDynamicServerListFetchURL());
            } catch (IOException e) {
                log.error(
                        "IOException when getting broker list. Aborting consumer start.",
                        e);
                
                return false;
            } catch (Exception e) {
                log.error(
                        "Something was wrong when getting broker list. Aborting consumer start.",
                        e);
                return false;
            }

            refreshServerListTimer.schedule(
                    new RefreshServerListTimerTask(this),
                    config.getConnectionLifetime(),
                    config.getConnectionLifetime());

        } else {
            hostsList = config.getHostParams();
        }

        executor = Executors
                .newFixedThreadPool(hostsList.size() > 0 ? hostsList.size() : 1);

        for (HostParams host : hostsList) {
            try {
                startAndRegisterConnection(host);
            
            } catch (BrokerConnectionFailedException e) {
                log.error("Not able to connect to broker " + host.toString()
                        + e.getStackTrace());
            }
        }

        this.status = Status.RUNNING;
        return true;
    }

    @Override
    public void stop() throws InvalidStatusException {
        switch (this.status) {

        case RUNNING:
            break;
        case STOPPED:
            log.info("Consumer is already stopped, nothing to do.");
            return;
        default:
            throw new InvalidStatusException(
                    "Consumer cannot be stopped. Status=" + this.status);
        }
        for (StompServerFetcher client : serverList) {
            client.close();
        }
        if (executor != null)
            executor.shutdownNow();
        serverList.clear();
        currentServers.clear();
        refreshServerListTimer.cancel();
        log.debug("Consumer " + this.toString() + " stopped successfully");
        this.status = Status.STOPPED;
    }

    @Override
    public boolean ack() {
        if (config.getAckType() == ConsumerAckType.AUTO_CLIENT_ACK) {
            log.warn("This consumer has auto ack type. No need to explicitly ack.");
            return false;
        }

        if (this.status != Status.RUNNING) {
            log.warn("This consumer is not running, can not ack.");
            return false;
        }

        if (lastSentServer != null) {
            try {
                lastSentServer.ack();
                lastSentServer = null;
                return true;
            } catch (AckFailedException e) {
                log.error("Ack failed to server " + lastSentServer, e);
                return false;
            }
        }
        return false;
    }

    public boolean ack(String ackId) {
        if (config.getAckType() == ConsumerAckType.AUTO_CLIENT_ACK) {
            log.warn("This consumer has auto ack type. No need to explicitly ack.");
            return false;
        }

        if (this.status != Status.RUNNING) {
            log.warn("This consumer is not running, can not ack.");
            return false;
        }

        try {
            String clearText = Utils.decode(ackId);

            StringTokenizer st = new StringTokenizer(clearText, ":");
            if (st.countTokens() != 4)
                throw new AckFailedException(new Exception(
                        "Wrong ack id format."));
            String hostname = st.nextToken();
            int port = Integer.parseInt(st.nextToken());
            String messageId = st.nextToken();
            String connectionId = st.nextToken();

            HostParams hostInfo = new HostParams(hostname, port);
            List<StompServerFetcher> connections = currentServers.get(hostInfo);
            if (connections == null || connections.size() == 0) {
                connections = new ArrayList<StompServerFetcher>();
                StompServerFetcher server = new StompServerFetcher(hostname,
                        port, config);
                connections.add(server);
                currentServers.put(hostInfo, connections);
                serverList.add(server);
            }

            // We should send ack from one connection only.
            try {
                connections.get(0).ack(messageId, connectionId);
                // nullifies lastSentServer so default ack() won't double-ack
                // this message
                lastSentServer = null;
                return true;
            } catch (AckFailedException e) {
                log.error("Ack failed to messageId " + messageId, e);
            }
        } catch (AckFailedException e) {
            log.error("Ack failed to ackId " + ackId, e);
            return false;
        }
        return false;
    }

    @Override
    public Message receiveImmediate() {
        return receiveImpl(false);

    }

    @Override
    public Message receive() {
        return receiveImpl(true);
    }

    @Override
    public Message receive(long timeout) throws ReceiveTimeoutException {
        if (this.status != Status.RUNNING) {
            log.warn("This consumer is not running, can not receive. Status="
                    + this.status);
            return null;
        }

        Callable<Message> task = new Callable<Message>() {
            public Message call() {
                return receiveImpl(true);
            }
        };

        Future<Message> future = executor.submit(task);
        try {

            return future.get(timeout, TimeUnit.MILLISECONDS);

        } catch (TimeoutException tx) {
            log.debug("Receive request timed out", tx);
            throw new ReceiveTimeoutException(tx);
        } catch (InterruptedException ix) {
            log.error(ix.getMessage(), ix);
            throw new ReceiveTimeoutException(ix);
        } catch (ExecutionException ex) {
            log.error(ex.getMessage(), ex);
            throw new ReceiveTimeoutException(ex);
        } finally {
            future.cancel(true);
        }

    }

    @Override
    public boolean nack() {
        if (this.status != Status.RUNNING) {
            log.warn("This consumer is not running, can not nack.");
            return false;
        }
        if (lastSentServer != null) {
            try {
                lastSentServer.nack();
                lastSentServer = null;
                return true;
            } catch (NackFailedException e) {
                log.error("Nack failed to server " + lastSentServer, e);
                return false;
            }
        }
        return false;
    }

    @Override
    public boolean nack(String ackId) {
        if (this.status != Status.RUNNING) {
            log.warn("This consumer is not running, can not nack.");
            return false;
        }
        try {
            String clearText = Utils.decode(ackId);

            StringTokenizer st = new StringTokenizer(clearText, ":");
            if (st.countTokens() != 4)
                throw new NackFailedException(new Exception(
                        "Wrong nack id format."));
            String hostname = st.nextToken();
            int port = Integer.parseInt(st.nextToken());
            String messageId = st.nextToken();

            HostParams hostInfo = new HostParams(hostname, port);
            List<StompServerFetcher> connections = currentServers.get(hostInfo);
            if (connections == null || connections.size() == 0) {
                connections = new ArrayList<StompServerFetcher>();
                StompServerFetcher server = new StompServerFetcher(hostname,
                        port, config);
                connections.add(server);
                currentServers.put(hostInfo, connections);
                serverList.add(server);
            }

            // We should send nack from one connection only.
            try {
                connections.get(0).nack(messageId);
                // nullifies lastSentServer so default nack() won't double-ack
                // this message
                lastSentServer = null;
                return true;
            } catch (NackFailedException e) {
                log.error("Nack failed to messageId " + messageId, e);
            }
        } catch (NackFailedException e) {
            log.error("nack failed to nackId " + ackId, e);
            return false;
        }
        return false;
    }

    @Override
    public boolean keepAlive() {
        boolean keepAlive = true;
        if (this.status != Status.RUNNING) {
            log.warn("This consumer is not running, can not send keep-alive.");
            return false;
        }
        try {
            Iterator<Entry<HostParams, List<StompServerFetcher>>> hostIterator = currentServers
                    .entrySet().iterator();
            while (hostIterator.hasNext()) {
                Map.Entry<HostParams, List<StompServerFetcher>> pairs = hostIterator
                        .next();
                List<StompServerFetcher> connections = currentServers.get(pairs
                        .getKey());

                // Send KeepAlive
                try {
                    connections.get(0).keepAlive();
                } catch (KeepAliveFailedException e) {
                    log.error(
                            "Keepalive failed for broker "
                                    + connections.get(0).getHost(), e);
                    keepAlive = false;
                }
            }

        } catch (Exception e) {
            log.warn("KeepAlive failed", e);
            keepAlive = false;
        }
        // Send heartbeat to all brokers
        return keepAlive;
    }

    @Override
    public boolean ackSafe(String ackId) {
        return ackSafe(ackId, ACKSAFE_TIMEOUT);
    }

    @Override
    public boolean ackSafe(String ackId, long timeout) {
        if (config.getAckType() == ConsumerAckType.AUTO_CLIENT_ACK) {
            log.warn("This consumer has auto ack type. No need to explicitly ack.");
            return false;
        }

        if (this.status != Status.RUNNING) {
            log.warn("This consumer is not running, can not ack.");
            return false;
        }

        try {
            String clearText = Utils.decode(ackId);

            StringTokenizer st = new StringTokenizer(clearText, ":");
            if (st.countTokens() != 4)
                throw new AckFailedException(new Exception(
                        "Wrong ack id format."));
            String hostname = st.nextToken();
            int port = Integer.parseInt(st.nextToken());
            String messageId = st.nextToken();
            String connectionId = st.nextToken();

            HostParams hostInfo = new HostParams(hostname, port);
            List<StompServerFetcher> connections = currentServers.get(hostInfo);
            if (connections == null || connections.size() == 0) {
                connections = new ArrayList<StompServerFetcher>();
                StompServerFetcher server = new StompServerFetcher(hostname,
                        port, config);
                connections.add(server);
                currentServers.put(hostInfo, connections);
                serverList.add(server);
            }

            // We should send ack from one connection only.
            try {
                connections.get(0).ackSafe(messageId, connectionId, timeout);
                // nullifies lastSentServer so default ackSafe() won't
                // double-ack this message
                lastSentServer = null;
                return true;
            } catch (AckFailedException e) {
                log.error("Ack failed to messageId " + messageId, e);
                return false;
            }
        } catch (AckFailedException e) {
            log.error("Ack failed to ackId " + ackId, e);
            return false;
        }
    }

    @Override
    public boolean ackSafe() {
        return ackSafe(ACKSAFE_TIMEOUT);
    }

    @Override
    public boolean ackSafe(long timeout) {
        if (config.getAckType() == ConsumerAckType.AUTO_CLIENT_ACK) {
            log.warn("This consumer has auto ack type. No need to explicitly ack.");
            return false;
        }

        if (this.status != Status.RUNNING) {
            log.warn("This consumer is not running, can not ack.");
            return false;
        }

        if (lastSentServer != null) {
            try {
                lastSentServer.ackSafe(timeout);
                lastSentServer = null;
                return true;
            } catch (AckFailedException e) {
                log.error("Ack failed to server " + lastSentServer, e);
                return false;
            } catch (Exception e) {
                log.error("Ack failed", e);
                return false;
            }
        }
        return false;
    }

    private Message receiveImpl(boolean blocking) {
        if (this.status != Status.RUNNING) {
            log.warn("This consumer is not running, can not receive. Status="
                    + this.status);
            return null;
        }

        StompFrame tmpFrame = null;

        do {
            for (int idx = 0; idx < serverList.size(); idx++) {
                int serverIdx = (lastContactedServerIdx + 1 + idx)
                        % serverList.size();
                StompServerFetcher server = serverList.get(serverIdx);
                tmpFrame = server.receiveLast();
                if (tmpFrame != null) {
                    lastContactedServerIdx = serverIdx;
                    lastSentServer = server;
                    Message message = Utils.getMessageFromBytes(tmpFrame
                            .getBody().getBytes());
                    log.debug("Received message: " + message);
                    log.debug("received message-id: "
                            + message.getMessageId());
                    // set ackId hashed from hostname and port

                    if (message != null) {
                        StringBuilder clearText = new StringBuilder(
                                server.getHost() + ":");
                        clearText.append("" + server.getPort() + ":");
                        clearText.append(tmpFrame.getHeaders()
                                .get("message-id") + ":");

                        // bbansal: For backward compatibility connectionId is
                        // not a required field here.
                        String connectionId = (tmpFrame.getHeaders()
                                .containsKey("connection-id")) ? tmpFrame
                                .getHeaders().get("connection-id")
                                : Utils.NULL_STRING;
                        clearText.append(connectionId);

                        message.setAckId(Utils.encode(clearText.toString()));
                        message.setMessageProperties(tmpFrame.getHeaders());
                    }
                    return message;
                }
            }

            try {
                if (blocking) {
                    Thread.sleep(config.getReceiveSleepInterval());
                } else {
                    return null;
                }

            } catch (InterruptedException e) {
                log.debug("receiveImpl blocking timed out.");
                return null;
            }

        } while (blocking);

        return null; // make compiler happy
    }

    private void startAndRegisterConnection(HostParams aHost)
            throws InvalidDestinationException, BrokerConnectionFailedException {

        String threadName = aHost.getHost();

        StompServerFetcher connection = getStompServerFetcher(aHost, config);

        connection.refreshConnection();

        log.debug("opening connection for " + aHost.getHost() + " Port="
                + aHost.getPort());

        serverList.add(connection);
        Thread thread = new Thread(connection, threadName);
        thread.setDaemon(true);
        thread.start();

        if (!currentServers.containsKey(aHost)) {
            currentServers.put(aHost, new ArrayList<StompServerFetcher>());
        }

        // Add connection to currentServers connection list.
        currentServers.get(aHost).add(connection);

    }

    public StompServerFetcher getStompServerFetcher(HostParams aHost,
            ConsumerConfig aConfig) {
        return new StompServerFetcher(aHost.getHost(), aHost.getPort(), aConfig);
    }

    private void stopHostAndUnregisterConnection(HostParams aHost) {
        List<StompServerFetcher> tmpList = currentServers.get(aHost);
        // remove all
        if (tmpList != null) {
            for (StompServerFetcher server : tmpList) {
                serverList.remove(server);
                server.close();
                log.debug("Connection with the broker " + server.toString()
                        + " closed successfully");
            }
        }
        currentServers.remove(aHost);
        lastContactedServerIdx = -1; // reset this guy
    }

    private void validateConfigs(ConsumerConfig aConfig)
            throws InvalidConfigException {
        
        if (aConfig.getDestinationName() == null) {
            throw new InvalidConfigException("Destination name can not be null");

        }
        if (aConfig.getDestinationType() == DestinationType.QUEUE) {
            if (config.getDestinationName().indexOf(QUEUE_NAME_PREFIX) != 0) {
                String message = "Invalid destination/queue name: "
                        + config.getDestinationName()
                        + ". Queue name must start with " + QUEUE_NAME_PREFIX;
                log.error(message);
                throw new InvalidConfigException(message);
            }
        }
        if (aConfig.getDestinationType() == DestinationType.TOPIC) {
            if (config.getDestinationName().indexOf(TOPIC_NAME_PREFIX) != 0) {
                String message = "Invalid destination/topic name: "
                        + config.getDestinationName()
                        + ". Topic name must start with " + TOPIC_NAME_PREFIX;
                log.error(message);
                throw new InvalidConfigException(message);

            }

            if (config.getSubscriptionId() == null) {
                log.error("When consuming topics, subscription-id must be specified");
                throw new InvalidConfigException(
                        "When consuming for topics, subscription id must be specified");
            }

        }
    }

    public void refreshServers() {

        try {
            Set<HostParams> newServerSet = fetchHostList();
            StringBuffer logMessageBuffer = new StringBuffer(
                    "New server list includes");

            // find servers to stop and remove
            HashSet<HostParams> removedServers = new HashSet<HostParams>();
            for (HostParams aHost : currentServers.keySet()) {
                if (!newServerSet.contains(aHost)) {
                    removedServers.add(aHost);
                }
            }

            for (HostParams bHost : removedServers) {
                log.debug("Stopping " + bHost);
                this.stopHostAndUnregisterConnection(bHost);
            }

            // find servers to add
            for (HostParams aHost : newServerSet) {
                if (null == currentServers.get(aHost)) {
                    log.debug("Starting new broker connection with " + aHost);
                    try {
                        startAndRegisterConnection(aHost);
                    } catch (BrokerConnectionFailedException e) {
                        log.error("Failed to connect to " + aHost.toString()
                                + ".\n" + e.getStackTrace());
                    }
                }
                logMessageBuffer.append(", " + aHost);
            }

            log.debug("Dynamically refreshing Server List");
            log.debug(logMessageBuffer.toString());
        } catch (MalformedURLException me) {
            log.error(
                    "Incorrect host list url is provided. Abort refresh servers.",
                    me);
        } catch (IOException ie) {
            log.warn(
                    "IOException when trying to get server list. Aborting refresh servers.",
                    ie);
        } catch (Exception e) {
            log.warn(
                    "Exception when trying to get server list. Aborting refresh servers.",
                    e);
        }

    }

    public Set<HostParams> fetchHostList() throws MalformedURLException,
            IOException {
        return DynamicServerListGetter.fetchHostList(config
                .getDynamicServerListFetchURL());
    }

}
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Set;
import org.apache.log4j.BasicConfigurator;
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.Log;
import com.groupon.messagebus.api.Message;
import com.groupon.messagebus.api.exceptions.InvalidConfigException;
import com.groupon.messagebus.api.exceptions.InvalidStatusException;
import com.groupon.messagebus.client.ConsumerImpl;
package com.groupon.messagebus.client.examples;
/*
 * 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 DLQHandleExample {

    /**
     * @param args
     * @throws InterruptedException 
     * @throws InvalidStatusException 
     */
    public static void main(String[] args) throws InterruptedException, InvalidStatusException {
        // TODO Auto-generated method stub
        ConsumerConfig config = new ConsumerConfig();
        config.setConnectionLifetime(300000);

        // default host to connect to
        HostParams host = new HostParams("localhost", 6661);
        Set<HostParams> hostsList = new HashSet<HostParams>();
        hostsList.add(host);

        //config.setDynamicServerListFetchURL("http://mbus-consumers-list.tm:8081/jmx?command=get_attribute&args=org.hornetq%3Amodule%3DCore%2Ctype%3DServer%20ListOfBrokers");
        config.setDestinationType(DestinationType.QUEUE);
        config.setHostParams(hostsList);
        config.setDestinationName("jms.queue.DLQ");
        config.setAckType(ConsumerAckType.CLIENT_ACK);
        config.setConnectionLifetime(300000);
        config.setSubscriptionId("client-xxx");

        BasicConfigurator.configure();

        Consumer con = new ConsumerImpl();
        try {
            con.start(config);
        } catch (InvalidConfigException e) {