class DefaultJMSMessageConverter implements JMSMessageConverter {

  @Override
  public List<Event> convert(Message message) throws JMSException {
    Event event = new SimpleEvent();
    Map<String, String> headers = event.getHeaders();
    @SuppressWarnings("rawtypes")
    Enumeration propertyNames = message.getPropertyNames();
    while (propertyNames.hasMoreElements()) {
      String name = propertyNames.nextElement().toString();
      String value = message.getStringProperty(name);
      headers.put(name, value);
    }
    if (message instanceof BytesMessage) {
      BytesMessage bytesMessage = (BytesMessage)message;
      long length = bytesMessage.getBodyLength();
      if (length > 0L) {
        if (length > Integer.MAX_VALUE) {
          throw new JMSException("Unable to process message " + "of size "
              + length);
        }
        byte[] body = new byte[(int)length];
        int count = bytesMessage.readBytes(body);
        if (count != length) {
          throw new JMSException("Unable to read full message. " +
              "Read " + count + " of total " + length);
        }
        event.setBody(body);
      }
    } else if (message instanceof TextMessage) {
      TextMessage textMessage = (TextMessage)message;
      String text = textMessage.getText();
      if (text != null) {
        event.setBody(text.getBytes(charset));
      }
    } else if (message instanceof ObjectMessage) {
      ObjectMessage objectMessage = (ObjectMessage)message;
      Object object = objectMessage.getObject();
      if (object != null) {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        ObjectOutput out = null;
        try {
          out = new ObjectOutputStream(bos);
          out.writeObject(object);
          event.setBody(bos.toByteArray());
        } catch (IOException e) {
          throw new FlumeException("Error serializing object", e);
        } finally {
          try {
            if (out != null) {
              out.close();
            }
          } catch (IOException e) {
            throw new FlumeException("Error closing ObjectOutputStream", e);
          }
          try {
            if (bos != null) {
              bos.close();
            }
          } catch (IOException e) {
            throw new FlumeException("Error closing ByteArrayOutputStream", e);
          }
        }
      }

    }
    List<Event> events = new ArrayList<Event>(1);
    events.add(event);
    return events;
  }

  private  DefaultJMSMessageConverter(String charset);

}