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

  /**
   * Start producer, this opens threads
   */
  public void start(ProducerConfig config) throws InvalidConfigException, TooManyConnectionRetryAttemptsException, InvalidStatusException;

  /**
   * Provides convenient API to refresh connection, in case connection with
   * the broker breaks during send or any other operation.
   *
   * @throws TooManyConnectionRetryAttemptsException
   *
   */
  public void refreshConnection() throws TooManyConnectionRetryAttemptsException;

  /**
   * Stop producer, throws exception
   */
  public void stop() throws BrokerConnectionCloseFailedException, InvalidStatusException;

  /**
   * Fire and forget send. Fast (1500+ QPS) but less reliable way of sending
   * data to the broker.
   * <p/>
   * The Producer thread sends data on the server connection and returns
   * instantly The Message server might not have persisted this message to any
   * durable format On server restarts/failure the user will lose messages.
   * This should be used when you have very high load and losing few messages
   * will not cause much harm.
   *
   * @param message : A {@link Message} to send.
   */
  public void send(Message message) throws TooManyConnectionRetryAttemptsException, SendFailedException;

  /**
   * Fire and forget send. Fast (1500+ QPS) but less reliable way of sending
   * data to the broker.
   * <p/>
   * The Producer thread sends data on the server connection and returns
   * instantly The Message server might not have persisted this message to any
   * durable format On server restarts/failure the user will lose messages.
   * This should be used when you have very high load and losing few messages
   * will not cause much harm.
   *
   * @param message : A {@link Message} to send.
   * @param headers : A Map<String, String> for headers to send along with the messages.
   */
  public void send(Message message, Map<String, String> headers) throws TooManyConnectionRetryAttemptsException, SendFailedException;

  /**
   * Fire and forget send. Fast (1500+ QPS) but less reliable way of sending
   * data to the broker, along with queue/topic name to send data to
   * <p/>
   * The Producer thread sends data on the server connection and returns
   * instantly The Message server might not have persisted this message to any
   * durable format On server restarts/failure the user will lose messages.
   * This should be used when you have very high load and losing few messages
   * will not cause much harm.
   *
   * @param message : A {@link Message} to send.
   * @param headers : A Map<String, String> for headers to send along with the messages.
   */
  public void send(Message message, String destinationName, Map<String, String> headers) throws TooManyConnectionRetryAttemptsException, SendFailedException;


  /**
   * Safe Send with Receipt.
   * <p/>
   * Moderately fast (150+ QPS) way of publishing data to message server. The
   * Producer thread sends data on the server connection and waits for server
   * to return a message receipt acknowledgment.
   * <p/>
   * Guarantees that messages are not lost for durable and persistent queues.
   *
   * @param message : A {@link Message} to send.
   */
  public void sendSafe(Message message) throws TooManyConnectionRetryAttemptsException, SendFailedException;

  /**
   * Safe Send with Receipt.
   * <p/>
   * Moderately fast (150+ QPS) way of publishing data to message server. The
   * Producer thread sends data on the server connection and waits for server
   * to return a message receipt acknowledgment.
   * <p/>
   * Guarantees that messages are not lost for durable and persistent queues.
   *
   * @param message : A {@link Message} to send.
   * @param headers : A Map<String, String> for headers to send along with the messages.
   */
  public void sendSafe(Message message, Map<String, String> headers) throws TooManyConnectionRetryAttemptsException, SendFailedException;


  /**
   * Safe Send with Receipt along with queue/topic name
   * <p/>
   * Moderately fast (150+ QPS) way of publishing data to message server. The
   * Producer thread sends data on the server connection and waits for server
   * to return a message receipt acknowledgment.
   * <p/>
   * Guarantees that messages are not lost for durable and persistent queues.
   *
   * @param message : A {@link Message} to send.
   * @param headers : A Map<String, String> for headers to send along with the messages.
   */
  public void sendSafe(Message message, String destinationName, Map<String, String> headers) throws TooManyConnectionRetryAttemptsException, SendFailedException;
  
  
} 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 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 ProducerConfig {

    private HostParams broker;
    private long connectionLifetime = 300000;
    private String destinationName;
    private DestinationType destinationType = DestinationType.QUEUE;
    private String userName = "guest";
    private String password = "guest";
    private int publishMaxRetryAttempts = 3;
    private boolean verboseLog = false;

    public boolean isVerboseLog() {
        return verboseLog;
    }

    public void setVerboseLog(boolean verboseLog) {
        this.verboseLog = verboseLog;
    }

	public int getPublishMaxRetryAttempts() {
        return publishMaxRetryAttempts;
    }

    public void setPublishMaxRetryAttempts(int publishMaxRetryAttempts) {
        this.publishMaxRetryAttempts = publishMaxRetryAttempts;
    }

    public HostParams getBroker() {
        return broker;
    }

    public void setBroker(HostParams broker) {
        this.broker = broker;
    }

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

} @SuppressWarnings("serial")
public class BrokerConnectionCloseFailedException extends MessageBusException {
    public BrokerConnectionCloseFailedException(Exception e){
        super(e);
    }
    
    public BrokerConnectionCloseFailedException(String message){
        super(message);
    }
} @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 SendFailedException extends MessageBusException{
    public SendFailedException(String message) {
        super(message);
    }

    public SendFailedException(Exception e) {
        super(e);
    }
} @SuppressWarnings("serial")
public class TooManyConnectionRetryAttemptsException extends BrokerConnectionFailedException {

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

    public TooManyConnectionRetryAttemptsException(Exception e) {
        super(e);
    }
} public class ProducerImpl implements Producer {

    public static final String SCHEDULED_MESSAGES_DELIVERY_TIME_MS = "scheduled_delivery_time";

    private Logger log = Logger.getLogger(ProducerImpl.class);
    private Object connectionAccessLock = new Object();

    private static final String TOPIC_NAME_PREFIX = "jms.topic.";
    private static final String QUEUE_NAME_PREFIX = "jms.queue.";
    private final int MAX_RETRY_COUNT = 3;
    private final long REST_INTERVAL = 1000;
    private final StompConnection connection;
    private ProducerConfig config;
    private Timer refreshConnectionTimer = new Timer();
    private long sendStartTime = 0;
    private Status status;

    public Status getStatus() {
        return status;
    }

    Gson gson = new Gson();

    public ProducerImpl() {
        this(new StompConnection());
    }

    public ProducerImpl(StompConnection connection) {
        this.connection = connection;
        this.status = Status.INITIALIZED;
    }

    /**
     * Starts connection with the broker.
     */
    @Override
    public void start(ProducerConfig aConfig) throws InvalidConfigException,
            TooManyConnectionRetryAttemptsException, InvalidStatusException {

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

        config = aConfig;

        validateConfigs(config);

        startConnection();

        refreshConnectionTimer.schedule(new RefreshConnectionTimerTask(this),
                config.getConnectionLifetime(), config.getConnectionLifetime());
        this.status = Status.RUNNING;
    }

    /**
     * Stops connection. Also takes care of closing TimerTask
     */
    @Override
    public void stop() throws BrokerConnectionCloseFailedException,
            InvalidStatusException {
        switch (this.status) {

        case RUNNING:
            break;
        case STOPPED:
            log.info("Producer is already stopped, nothing to do.");
            return;
        case INITIALIZED:
            throw new InvalidStatusException(
                    "Producer cannot be stopped. Status=" + this.status);
        }

        refreshConnectionTimer.cancel();
        stopConnection();
        this.status = Status.STOPPED;
    }

    /**
     * Sends message to the broker in fire and forget mode. Using this API is
     * fast but less reliable, compared with sendSafe() Returns false, if send
     * failes for any reason
     */
    @Override
    public void send(Message message)
            throws TooManyConnectionRetryAttemptsException, SendFailedException {
        sendInternal(message, null, false, null);
    }

    /**
     * Sends message to the broker in fire and forget mode. Using this API is
     * fast but less reliable, compared with sendSafe() Returns false, if send
     * failes for any reason
     */
    @Override
    public void send(Message message, Map<String, String> headers)
            throws TooManyConnectionRetryAttemptsException, SendFailedException {
        sendInternal(message, headers, false, null);
    }

    @Override
    public void send(Message message, String destinationName,
            Map<String, String> headers)
            throws TooManyConnectionRetryAttemptsException, SendFailedException {
        sendInternal(message, headers, false, destinationName);
    }

    /**
     * Reliable protocol of sending data to the broker returns false, if send
     * fails for any reason
     * 
     * @throws SendFailedException
     * @throws TooManyConnectionRetryAttemptsException
     * 
     */
    @Override
    public void sendSafe(Message message)
            throws TooManyConnectionRetryAttemptsException, SendFailedException {
        sendInternal(message, null, true, null);
    }

    /**
     * Reliable protocol of sending data to the broker returns false, if send
     * fails for any reason
     * 
     * @throws SendFailedException
     * @throws TooManyConnectionRetryAttemptsException
     * 
     */
    @Override
    public void sendSafe(Message message, Map<String, String> headers)
            throws TooManyConnectionRetryAttemptsException, SendFailedException {
        sendInternal(message, headers, true, null);
    }

    @Override
    public void sendSafe(Message message, String destinationName,
            Map<String, String> headers)
            throws TooManyConnectionRetryAttemptsException, SendFailedException {
        sendInternal(message, headers, true, destinationName);
    }

    /**
     * Refreshes connection. Also reports connection statistics.
     */
    public void refreshConnection()
            throws TooManyConnectionRetryAttemptsException {
        if (this.status != Status.RUNNING) {
            log.warn("This producer is not running, skip refreshing connection.");
            return;
        }
        try {
            log.debug("Refreshing connection with the broker "
                    + config.getBroker());
            synchronized (connectionAccessLock) {
                stopConnection();
                startConnection();
            }
        } catch (Exception e) {
            log.error("Failed to refresh connection with the broker for config:"
                    + gson.toJson(config));
        }

    }

    private void sendInternal(Message message, Map<String, String> headers,
            boolean isSafeSend, String destinationName)
            throws SendFailedException, TooManyConnectionRetryAttemptsException {
        if (this.status != Status.RUNNING) {
            throw new InvalidStatusException(
                    "This producer is not running and cannot publish. Status="
                            + this.status);

        }

        if (destinationName == null)
            destinationName = config.getDestinationName();
        int attempt = 0;
        boolean done = false;
        while (!done && attempt++ < config.getPublishMaxRetryAttempts()) {
            try {
                synchronized (connectionAccessLock) {

                    sendStartTime = System.currentTimeMillis();
                    if (config.isVerboseLog()) {
                        log.info("Publishing to destination_name="
                                + destinationName
                                + ", message_id="+ message.getMessageId()
                                + ", message contents=" + message.getMessageInternal().toString());
                    }
                    if (isSafeSend) {
                        connection.sendSafe(destinationName,
                                Utils.getThriftDataAsBytes(message), headers);
                    } else {
                        connection.send(destinationName,
                                Utils.getThriftDataAsBytes(message), headers);
                    }
                    if (config.isVerboseLog()) {
                        log.info("result=success destination_name=" + destinationName
                                + ", message_id=" + message.getMessageId()
                                + ", duration=" + (System.currentTimeMillis() - sendStartTime)
                                + "ms");
                    }
                    done = true;
                }
            } catch (Exception e) {
                log.warn(
                        "Failed attempt (" + attempt + "/"
                                + config.getPublishMaxRetryAttempts()
                                + ") sending message to the broker "
                                + config.getBroker()
                                + ". Refreshing connection.", e);
                refreshConnection();
            }
        }

        if (!done) {
            log.error("Failed to publish result=fail destination_name=" + destinationName
                    + ", message_id=" + message.getMessageId()
                    + ", duration=" + (System.currentTimeMillis() - sendStartTime)
                    + "ms" 
                    + ", message_contents=" + message.getMessageInternal());
            throw new SendFailedException("Failed to send message in "
                    + config.getPublishMaxRetryAttempts()
                    + " attempts to broker:" + config.getBroker());
        }
    }

    private void stopConnection() {
        try {
            connection.close();
        } catch (IOException e) {
            log.error("Error while closing connection with the broker "
                    + config.getBroker(), e);
        }
    }

    private void startConnection()
            throws TooManyConnectionRetryAttemptsException {
        int retryCount = 0;
        boolean done = false;
        while (!done && retryCount++ < MAX_RETRY_COUNT) {
            try {
                connection.open(config.getBroker().getHost(), config
                        .getBroker().getPort());
                connection.connect(config.getUserName(), config.getPassword());
                done = true;
            } catch (Exception e) {
                log.debug(
                        "Exception connecting to the broker " + config.getBroker()
                                + " retrying attempt:" + retryCount, e);
                Utils.sleep(REST_INTERVAL);
            }
        }

        if (!done) {
            throw new TooManyConnectionRetryAttemptsException(
                    "Max connection attempts reached. Can not connect to the broker "
                            + config.getBroker());
        }
    }

    private void validateConfigs(ProducerConfig 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);

            }
        }
    }
} public class MessageInternal implements TBase<MessageInternal, MessageInternal._Fields>, java.io.Serializable, Cloneable {
  private static final TStruct STRUCT_DESC = new TStruct("MessageInternal");

  private static final TField MESSAGE_ID_FIELD_DESC = new TField("messageId", TType.STRING, (short)1);
  private static final TField PAYLOAD_FIELD_DESC = new TField("payload", TType.STRUCT, (short)2);
  private static final TField PROPERTIES_FIELD_DESC = new TField("properties", TType.MAP, (short)3);

  public String messageId;
  public MessagePayload payload;
  public Map<String,String> properties;

  /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
  public enum _Fields implements TFieldIdEnum {
    MESSAGE_ID((short)1, "messageId"),
    PAYLOAD((short)2, "payload"),
    PROPERTIES((short)3, "properties");

    private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();

    static {
      for (_Fields field : EnumSet.allOf(_Fields.class)) {
        byName.put(field.getFieldName(), field);
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, or null if its not found.
     */
    public static _Fields findByThriftId(int fieldId) {
      switch(fieldId) {
        case 1: // MESSAGE_ID
          return MESSAGE_ID;
        case 2: // PAYLOAD
          return PAYLOAD;
        case 3: // PROPERTIES
          return PROPERTIES;
        default:
          return null;
      }
    }

    /**
     * Find the _Fields constant that matches fieldId, throwing an exception
     * if it is not found.
     */
    public static _Fields findByThriftIdOrThrow(int fieldId) {
      _Fields fields = findByThriftId(fieldId);
      if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
      return fields;
    }

    /**
     * Find the _Fields constant that matches name, or null if its not found.
     */
    public static _Fields findByName(String name) {
      return byName.get(name);
    }

    private final short _thriftId;
    private final String _fieldName;

    _Fields(short thriftId, String fieldName) {
      _thriftId = thriftId;
      _fieldName = fieldName;
    }

    public short getThriftFieldId() {
      return _thriftId;
    }

    public String getFieldName() {
      return _fieldName;
    }
  }

  // isset id assignments

  public static final Map<_Fields, FieldMetaData> metaDataMap;
  static {
    Map<_Fields, FieldMetaData> tmpMap = new EnumMap<_Fields, FieldMetaData>(_Fields.class);
    tmpMap.put(_Fields.MESSAGE_ID, new FieldMetaData("messageId", TFieldRequirementType.REQUIRED, 
        new FieldValueMetaData(TType.STRING)));
    tmpMap.put(_Fields.PAYLOAD, new FieldMetaData("payload", TFieldRequirementType.REQUIRED, 
        new StructMetaData(TType.STRUCT, MessagePayload.class)));
    tmpMap.put(_Fields.PROPERTIES, new FieldMetaData("properties", TFieldRequirementType.OPTIONAL, 
        new MapMetaData(TType.MAP, 
            new FieldValueMetaData(TType.STRING), 
            new FieldValueMetaData(TType.STRING))));
    metaDataMap = Collections.unmodifiableMap(tmpMap);
    FieldMetaData.addStructMetaDataMap(MessageInternal.class, metaDataMap);
  }

  public MessageInternal() {
  }

  public MessageInternal(
    String messageId,
    MessagePayload payload)
  {
    this();
    this.messageId = messageId;
    this.payload = payload;
  }

  /**
   * Performs a deep copy on <i>other</i>.
   */
  public MessageInternal(MessageInternal other) {
    if (other.isSetMessageId()) {
      this.messageId = other.messageId;
    }
    if (other.isSetPayload()) {
      this.payload = new MessagePayload(other.payload);
    }
    if (other.isSetProperties()) {
      Map<String,String> __this__properties = new HashMap<String,String>();
      for (Map.Entry<String, String> other_element : other.properties.entrySet()) {

        String other_element_key = other_element.getKey();
        String other_element_value = other_element.getValue();

        String __this__properties_copy_key = other_element_key;

        String __this__properties_copy_value = other_element_value;

        __this__properties.put(__this__properties_copy_key, __this__properties_copy_value);
      }
      this.properties = __this__properties;
    }
  }

  public MessageInternal deepCopy() {
    return new MessageInternal(this);
  }

  @Override
  public void clear() {
    this.messageId = null;
    this.payload = null;
    this.properties = null;
  }

  public String getMessageId() {
    return this.messageId;
  }

  public MessageInternal setMessageId(String messageId) {
    this.messageId = messageId;
    return this;
  }

  public void unsetMessageId() {
    this.messageId = null;
  }

  /** Returns true if field messageId is set (has been asigned a value) and false otherwise */
  public boolean isSetMessageId() {
    return this.messageId != null;
  }

  public void setMessageIdIsSet(boolean value) {
    if (!value) {
      this.messageId = null;
    }
  }

  public MessagePayload getPayload() {
    return this.payload;
  }

  public MessageInternal setPayload(MessagePayload payload) {
    this.payload = payload;
    return this;
  }

  public void unsetPayload() {
    this.payload = null;
  }

  /** Returns true if field payload is set (has been asigned a value) and false otherwise */
  public boolean isSetPayload() {
    return this.payload != null;
  }

  public void setPayloadIsSet(boolean value) {
    if (!value) {
      this.payload = null;
    }
  }

  public int getPropertiesSize() {
    return (this.properties == null) ? 0 : this.properties.size();
  }

  public void putToProperties(String key, String val) {
    if (this.properties == null) {
      this.properties = new HashMap<String,String>();
    }
    this.properties.put(key, val);
  }

  public Map<String,String> getProperties() {
    return this.properties;
  }

  public MessageInternal setProperties(Map<String,String> properties) {
    this.properties = properties;
    return this;
  }

  public void unsetProperties() {
    this.properties = null;
  }

  /** Returns true if field properties is set (has been asigned a value) and false otherwise */
  public boolean isSetProperties() {
    return this.properties != null;
  }

  public void setPropertiesIsSet(boolean value) {
    if (!value) {
      this.properties = null;
    }
  }

  public void setFieldValue(_Fields field, Object value) {
    switch (field) {
    case MESSAGE_ID:
      if (value == null) {
        unsetMessageId();
      } else {
        setMessageId((String)value);
      }
      break;

    case PAYLOAD:
      if (value == null) {
        unsetPayload();
      } else {
        setPayload((MessagePayload)value);
      }
      break;

    case PROPERTIES:
      if (value == null) {
        unsetProperties();
      } else {
        setProperties((Map<String,String>)value);
      }
      break;

    }
  }

  public Object getFieldValue(_Fields field) {
    switch (field) {
    case MESSAGE_ID:
      return getMessageId();

    case PAYLOAD:
      return getPayload();

    case PROPERTIES:
      return getProperties();

    }
    throw new IllegalStateException();
  }

  /** Returns true if field corresponding to fieldID is set (has been asigned a value) and false otherwise */
  public boolean isSet(_Fields field) {
    if (field == null) {
      throw new IllegalArgumentException();
    }

    switch (field) {
    case MESSAGE_ID:
      return isSetMessageId();
    case PAYLOAD:
      return isSetPayload();
    case PROPERTIES:
      return isSetProperties();
    }
    throw new IllegalStateException();
  }

  @Override
  public boolean equals(Object that) {
    if (that == null)
      return false;
    if (that instanceof MessageInternal)
      return this.equals((MessageInternal)that);
    return false;
  }

  public boolean equals(MessageInternal that) {
    if (that == null)
      return false;

    boolean this_present_messageId = true && this.isSetMessageId();
    boolean that_present_messageId = true && that.isSetMessageId();
    if (this_present_messageId || that_present_messageId) {
      if (!(this_present_messageId && that_present_messageId))
        return false;
      if (!this.messageId.equals(that.messageId))
        return false;
    }

    boolean this_present_payload = true && this.isSetPayload();
    boolean that_present_payload = true && that.isSetPayload();
    if (this_present_payload || that_present_payload) {
      if (!(this_present_payload && that_present_payload))
        return false;
      if (!this.payload.equals(that.payload))
        return false;
    }

    boolean this_present_properties = true && this.isSetProperties();
    boolean that_present_properties = true && that.isSetProperties();
    if (this_present_properties || that_present_properties) {
      if (!(this_present_properties && that_present_properties))
        return false;
      if (!this.properties.equals(that.properties))
        return false;
    }

    return true;
  }

  @Override
  public int hashCode() {
    return 0;
  }

  public int compareTo(MessageInternal other) {
    if (!getClass().equals(other.getClass())) {
      return getClass().getName().compareTo(other.getClass().getName());
    }

    int lastComparison = 0;
    MessageInternal typedOther = (MessageInternal)other;

    lastComparison = Boolean.valueOf(isSetMessageId()).compareTo(typedOther.isSetMessageId());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetMessageId()) {
      lastComparison = TBaseHelper.compareTo(this.messageId, typedOther.messageId);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    lastComparison = Boolean.valueOf(isSetPayload()).compareTo(typedOther.isSetPayload());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetPayload()) {
      lastComparison = TBaseHelper.compareTo(this.payload, typedOther.payload);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    lastComparison = Boolean.valueOf(isSetProperties()).compareTo(typedOther.isSetProperties());
    if (lastComparison != 0) {
      return lastComparison;
    }
    if (isSetProperties()) {
      lastComparison = TBaseHelper.compareTo(this.properties, typedOther.properties);
      if (lastComparison != 0) {
        return lastComparison;
      }
    }
    return 0;
  }

  public _Fields fieldForId(int fieldId) {
    return _Fields.findByThriftId(fieldId);
  }

  public void read(TProtocol iprot) throws TException {
    TField field;
    iprot.readStructBegin();
    while (true)
    {
      field = iprot.readFieldBegin();
      if (field.type == TType.STOP) { 
        break;
      }
      switch (field.id) {
        case 1: // MESSAGE_ID
          if (field.type == TType.STRING) {
            this.messageId = iprot.readString();
          } else { 
            TProtocolUtil.skip(iprot, field.type);
          }
          break;
        case 2: // PAYLOAD
          if (field.type == TType.STRUCT) {
            this.payload = new MessagePayload();
            this.payload.read(iprot);
          } else { 
            TProtocolUtil.skip(iprot, field.type);
          }
          break;
        case 3: // PROPERTIES
          if (field.type == TType.MAP) {
            {
              TMap _map0 = iprot.readMapBegin();
              this.properties = new HashMap<String,String>(2*_map0.size);
              for (int _i1 = 0; _i1 < _map0.size; ++_i1)
              {
                String _key2;
                String _val3;
                _key2 = iprot.readString();
                _val3 = iprot.readString();
                this.properties.put(_key2, _val3);
              }
              iprot.readMapEnd();
            }
          } else { 
            TProtocolUtil.skip(iprot, field.type);
          }
          break;
        default:
          TProtocolUtil.skip(iprot, field.type);
      }
      iprot.readFieldEnd();
    }
    iprot.readStructEnd();

    // check for required fields of primitive type, which can't be checked in the validate method
    validate();
  }

  public void write(TProtocol oprot) throws TException {
    validate();

    oprot.writeStructBegin(STRUCT_DESC);
    if (this.messageId != null) {
      oprot.writeFieldBegin(MESSAGE_ID_FIELD_DESC);
      oprot.writeString(this.messageId);
      oprot.writeFieldEnd();
    }
    if (this.payload != null) {
      oprot.writeFieldBegin(PAYLOAD_FIELD_DESC);
      this.payload.write(oprot);
      oprot.writeFieldEnd();
    }
    if (this.properties != null) {
      if (isSetProperties()) {
        oprot.writeFieldBegin(PROPERTIES_FIELD_DESC);
        {
          oprot.writeMapBegin(new TMap(TType.STRING, TType.STRING, this.properties.size()));
          for (Map.Entry<String, String> _iter4 : this.properties.entrySet())
          {
            oprot.writeString(_iter4.getKey());
            oprot.writeString(_iter4.getValue());
          }
          oprot.writeMapEnd();
        }
        oprot.writeFieldEnd();
      }
    }
    oprot.writeFieldStop();
    oprot.writeStructEnd();
  }

  @Override
  public String toString() {
    StringBuilder sb = new StringBuilder("MessageInternal(");
    boolean first = true;

    sb.append("messageId:");
    if (this.messageId == null) {
      sb.append("null");
    } else {
      sb.append(this.messageId);
    }
    first = false;
    if (!first) sb.append(", ");
    sb.append("payload:");
    if (this.payload == null) {
      sb.append("null");
    } else {
      sb.append(this.payload);
    }
    first = false;
    if (isSetProperties()) {
      if (!first) sb.append(", ");
      sb.append("properties:");
      if (this.properties == null) {
        sb.append("null");
      } else {
        sb.append(this.properties);
      }
      first = false;
    }
    sb.append(")");
    return sb.toString();
  }

  public void validate() throws TException {
    // check for required fields
    if (messageId == null) {
      throw new TProtocolException("Required field 'messageId' was not present! Struct: " + toString());
    }
    if (payload == null) {
      throw new TProtocolException("Required field 'payload' was not present! Struct: " + toString());
    }
  }

} public class StompConnection {

    // Set this to appropriate value for now set to infinity
    public static final long RECEIVE_TIMEOUT = 30000;

    private Logger log = Logger.getLogger(StompConnection.class);
    private boolean connected = false;
    private Socket stompSocket;
    private ByteArrayOutputStream inputBuffer = new ByteArrayOutputStream();

    public void open(String host, int port) throws IOException,
            UnknownHostException {
        Socket sock = new Socket(host, port);
        sock.setTcpNoDelay(true);
        open(sock);
    }

    public void open(Socket socket) {
        stompSocket = socket;
        connected = true;

    }

    public void close() throws IOException {
        if (stompSocket != null) {
            if (!stompSocket.isClosed())
                stompSocket.close();
            stompSocket = null;
            connected = false;
        }
    }

    public void sendFrame(String data) throws IOException {
        byte[] bytes = data.getBytes("UTF-8");
        OutputStream outputStream = stompSocket.getOutputStream();
        outputStream.write(bytes);
        outputStream.write(0);
        outputStream.flush();
    }

    public void sendFrame(String frame, byte[] data) throws IOException {
        byte[] bytes = frame.getBytes("UTF-8");
        OutputStream outputStream = stompSocket.getOutputStream();
        outputStream.write(bytes);
        outputStream.write(data);
        outputStream.write(0);
        outputStream.flush();
    }

    public StompFrame receive() throws IOException {
        return receive(RECEIVE_TIMEOUT);
    }

    public StompFrame receive(long timeOut) throws IOException {
        stompSocket.setSoTimeout((int) timeOut);
        InputStream is = stompSocket.getInputStream();
        StompWireFormat wf = new StompWireFormat();
        DataInputStream dis = new DataInputStream(is);
        return (StompFrame) wf.unmarshal(dis);
    }

    public String receiveFrame() throws IOException {
        return receiveFrame(RECEIVE_TIMEOUT);
    }

    public String receiveFrame(long timeOut) throws IOException {
        stompSocket.setSoTimeout((int) timeOut);
        InputStream is = stompSocket.getInputStream();
        int c = 0;
        for (;;) {
            c = is.read();
            if (c < 0) {
                throw new IOException("socket closed.");
            } else if (c == 0) {
                c = is.read();
                if (c == '\n') {
                    // end of frame
                    return stringFromBuffer(inputBuffer);
                } else {
                    inputBuffer.write(0);
                    inputBuffer.write(c);
                }
            } else {
                inputBuffer.write(c);
            }
        }
    }

    public boolean isConnected() {
        if (stompSocket == null || !stompSocket.isConnected()) {
            connected = false;
        }

        return connected;
    }

    private String stringFromBuffer(ByteArrayOutputStream inputBuffer)
            throws IOException {
        byte[] ba = inputBuffer.toByteArray();
        inputBuffer.reset();
        return new String(ba, "UTF-8");
    }

    public Socket getStompSocket() {
        return stompSocket;
    }

    public void setStompSocket(Socket stompSocket) {
        this.stompSocket = stompSocket;
    }

    public void connect(String username, String password) throws IOException,
            MessageBusException {
        connect(username, password, null);
    }

    public void connect(String username, String password, String client)
            throws IOException, MessageBusException {
        Map<String, String> headers = new HashMap();
        headers.put("login", username);
        headers.put("passcode", password);
        if (client != null) {
            headers.put("client-id", client);
        }
        StompFrame frame = new StompFrame("CONNECT", headers);
        sendFrame(frame.format());

        StompFrame connect = receive();
        if (!connect.getAction().equals(Stomp.Responses.CONNECTED)) {
            throw new MessageBusException("Not connected to server: "
                    + connect.getBody());
        }
    }

    public void disconnect() throws IOException {

        StompFrame frame = new StompFrame("DISCONNECT");
        sendFrame(frame.format());
    }

    public void sendSafe(String destination, String message,
            Map<String, String> headers) throws IOException,
            SendFailedException {
        sendSafe(destination, message.getBytes(), headers);
    }

    public void sendSafe(String destination, byte[] message,
            Map<String, String> headers) throws IOException,
            SendFailedException {
        // Add receipt header to the headers.
        if (headers == null) {
            headers = new HashMap<String, String>();
        }
        headers.put("receipt", "1");

        // send the message.
        send(destination, message, headers);

        // Wait for the
        StompFrame connect = receive();
        if (!connect.getAction().equals(Stomp.Responses.RECEIPT)) {
            throw new SendFailedException("Failed to receive RECEIPT: "
                    + connect.getBody());
        }
        log.info("Message sent receipt acknowledged for destination:"
                + destination);
    }

    public void send(String destination, String message,
            Map<String, String> headers) throws IOException {
        send(destination, message.getBytes(), null, headers);
    }

    public void send(String destination, byte[] message,
            Map<String, String> headers) throws IOException {
        send(destination, message, null, headers);
    }

    public void send(String destination, byte[] message, String transaction,
            Map<String, String> headers) throws IOException {
        if (headers == null) {
            headers = new HashMap<String, String>();
        }
        headers.put("destination", destination);
        headers.put("persistent", "true");
        if (transaction != null) {
            headers.put("transaction", transaction);
        }

        StompFrame frame = new StompFrame("SEND", headers, message);
        sendFrame(frame.format());
    }

    public void subscribe(String destination) throws IOException {
        subscribe(destination, null, null);
    }

    public void subscribe(String destination, String ack) throws IOException {
        subscribe(destination, ack, new HashMap<String, String>());
    }

    public void subscribe(String destination, String ack,
            Map<String, String> headers) throws IOException{
        if (headers == null) {
            headers = new HashMap<String, String>();
        }
        headers.put("destination", destination);

        if (ack != null) {
            headers.put("ack", ack);
        }
        StompFrame frame = new StompFrame("SUBSCRIBE", headers);
        sendFrame(frame.format());
    }

    public void unsubscribe(String destination) throws IOException {
        unsubscribe(destination, null);
    }

    public void unsubscribe(String destination, Map<String, String> headers)
            throws IOException {
        if (headers == null) {
            headers = new HashMap<String, String>();
        }
        headers.put("destination", destination);

        StompFrame frame = new StompFrame("UNSUBSCRIBE", headers);
        sendFrame(frame.format());
    }

    public void begin(String transaction) throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("transaction", transaction);
        StompFrame frame = new StompFrame("BEGIN", headers);
        sendFrame(frame.format());
    }

    public void abort(String transaction) throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("transaction", transaction);
        StompFrame frame = new StompFrame("ABORT", headers);
        sendFrame(frame.format());
    }

    public void commit(String transaction) throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("transaction", transaction);
        StompFrame frame = new StompFrame("COMMIT", headers);
        sendFrame(frame.format());
    }

    public void ack(String messageId) throws IOException {
        ack(messageId, null, null, null, null);
    }

    public void ack(String messageId, String receiptId) throws IOException {
        ack(messageId, null, null, null, receiptId);
    }

    public void ack(String messageId, String transaction,
            String subscriptionId, String connectionId, String receiptId)
            throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("message-id", messageId);
        log.debug("acking message-id: " + messageId);
        if (transaction != null)
            headers.put("transaction", transaction);
        if (subscriptionId != null && !subscriptionId.equals(""))
            headers.put("subscription", subscriptionId);

        if (connectionId != null && !connectionId.equals(Utils.NULL_STRING))
            headers.put("connection-id", connectionId);

        if (receiptId != null && !receiptId.equals(Utils.NULL_STRING))
            headers.put("receipt", receiptId);

        StompFrame frame = new StompFrame("ACK", headers);
        sendFrame(frame.format());
    }

    public void keepAlive() throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        StompFrame frame = new StompFrame("KEEPALIVE", headers);
        sendFrame(frame.format());
    }

    public void credit(StompFrame frame) throws IOException {
        credit(frame.getHeaders().get("message-id"));
    }

    public void credit(String messageId) throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("message-id", messageId);
        StompFrame frame = new StompFrame("CREDIT", headers);
        sendFrame(frame.format());
    }

    protected String appendHeaders(HashMap<String, Object> headers) {
        StringBuffer result = new StringBuffer();
        for (String key : headers.keySet()) {
            result.append(key + ":" + headers.get(key) + "\n");
        }
        result.append("\n");
        return result.toString();
    }

    public void nack(String messageId, String subscriptionId)
            throws IOException {
        Map<String, String> headers = new HashMap<String, String>();
        headers.put("message-id", messageId);
        if (subscriptionId != null && !subscriptionId.equals(""))
            headers.put("subscription", subscriptionId);

        StompFrame frame = new StompFrame("NACK", headers);
        sendFrame(frame.format());
    }

}
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.Map;
import junit.framework.TestCase;
import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.BasicConfigurator;
import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
import org.junit.Test;
import com.groupon.messagebus.api.Producer;
import com.groupon.messagebus.api.HostParams;
import com.groupon.messagebus.api.Message;
import com.groupon.messagebus.api.ProducerConfig;
import com.groupon.messagebus.api.exceptions.BrokerConnectionCloseFailedException;
import com.groupon.messagebus.api.exceptions.InvalidConfigException;
import com.groupon.messagebus.api.exceptions.InvalidStatusException;
import com.groupon.messagebus.api.exceptions.SendFailedException;
import com.groupon.messagebus.api.exceptions.TooManyConnectionRetryAttemptsException;
import com.groupon.messagebus.client.ProducerImpl;
import com.groupon.messagebus.thrift.api.MessageInternal;
import com.groupon.stomp.StompConnection;
package com.groupon.messagebus.client.test;
/*
 * 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 ProducerImplTest extends TestCase {
    private ProducerConfig config = new ProducerConfig();
    private HostParams dummyHostParam = new HostParams("localhost", 61613);
    private String MESSAGE_ID = "id-1";
    private byte[] MESSAGE_DATA = "my sweeet message.".getBytes();
    private byte[] MESSAGE_DATA2 = "my failing message.".getBytes();

    void setDefaultProducerConfigs() {
        config = new ProducerConfig();
        config.setConnectionLifetime(300000);
        config.setDestinationName("jms.queue.TestQueue1");
        config.setBroker(dummyHostParam);
    }

    public void setUp() {
        setDefaultProducerConfigs();
        BasicConfigurator.configure();
    }

    private StompConnection mockStompConnection() throws TException,
            IOException, SendFailedException {
        StompConnection connection = mock(StompConnection.class);

        when(connection.isConnected()).thenReturn(true);

        // for simple send do nothing.
        doNothing().when(connection).send(config.getDestinationName(),
                getMessageAsBytes(MESSAGE_ID, MESSAGE_DATA), null);

        // for simple sendSafe do Nothing.
        doNothing().when(connection).sendSafe(config.getDestinationName(),
                getMessageAsBytes(MESSAGE_ID, MESSAGE_DATA), null);

        // for messageData2 throw receipt not found exception.
        doThrow(new SendFailedException("Receipt failed")).when(connection)
                .sendSafe(config.getDestinationName(),
                        getMessageAsBytes(MESSAGE_ID, MESSAGE_DATA2), null);

        return connection;
    }
    
    private StompConnection mockStompConnectionNoMsgId() throws TException,
    IOException, SendFailedException {
        StompConnection connection = mock(StompConnection.class);

        when(connection.isConnected()).thenReturn(true);

        //for simple send do nothing.
        doNothing().when(connection).send(config.getDestinationName(),
        getMessageAsBytes(MESSAGE_DATA), null);

        // for simple sendSafe do Nothing.
        doNothing().when(connection).sendSafe(config.getDestinationName(),
        getMessageAsBytes(MESSAGE_DATA), null);

        // for messageData2 throw receipt not found exception.
        doThrow(new SendFailedException("Receipt failed")).when(connection)
        .sendSafe(config.getDestinationName(),
                getMessageAsBytes(MESSAGE_DATA2), null);

        return connection;
}   

    private byte[] getMessageAsBytes(String messageId, byte[] messageData)
            throws TException, UnsupportedEncodingException {
        MessageInternal message = Message.createBinaryMessage(messageId,