MQConsumerTask.java

abstract class MQConsumerTask extends Thread {

  private static final Logger logger = Logger.getLogger("ETP");

  private final String queueName;
  private final MQConnectionWrapper connection;

  private volatile boolean isStopped;

  public MQConsumerTask(String name, String queueName, MQConnectionWrapper connection) {
    super(name);
    this.queueName = queueName;
    this.connection = connection;
  }

  private String readMessage(Message m) throws JMSException, UnsupportedEncodingException {

    logger.info("New message received.");
    if (logger.isDebugEnabled()) {
      logger.debug("Message: \n" + m);
    }

    if (m instanceof TextMessage) {
      return ((TextMessage) m).getText();

    } else if (m instanceof BytesMessage) {
      return readText((BytesMessage) m, "utf-8");

    } else {
      logger.error("New message received with unsupported type: " + m.getClass());
      return null;
    }
  }

  private String readText(BytesMessage m,
                          String charset) throws JMSException, UnsupportedEncodingException {

    final ByteArrayOutputStream os = new ByteArrayOutputStream();

    final long len = m.getBodyLength();
    for (int i = 0; i < len; i++) {
      os.write(m.readByte());
    }

    return new String(os.toByteArray(), charset);
  }

  public final void onMessage(Message message) {
    try {
      final String textMessage = readMessage(message);
      if (logger.isDebugEnabled()) {
        logger.debug("Decoded message body: " + textMessage);
      }

      if (textMessage != null) {
        onMessage(message, textMessage);
      }

    } catch (JMSException e) {
      logger.error("Error reading message: \n" + message);
      onError(e);

    } catch (UnsupportedEncodingException e) {
      logger.fatal("UTF-8 Encoding is not supported. Incoming message will be skipped.");
      onError(e);

    } catch (Exception e) {
      logger.error(e.getMessage(), e);
      onError(e);
    }
  }

  @Override
  public final void run() {
    logger.info("MQ consumer started: [" + getClass().getSimpleName() + "]");

    while (!isStopped) {

      final Message m;
      try {
        m = connection.receive(queueName);

      } catch (InterruptedException e) {
        logger.info("Interrupted while reading [" + queueName + "], exiting");
        return;
      }

      try {
        if (m != null) {
          onMessage(m);

        } else {
          logger.info("NULL message. Is the queue shutting down?");
        }

      } catch (Exception e) {
        logger.error(e.getMessage(), e);
      }
    }
    logger.info("MQ consumer stopped: [" + getClass().getSimpleName() + "]");
  }

  public void shutdown() {
    isStopped = true;
  }

  public void doJoin() throws InterruptedException {
    this.interrupt();
    this.join();
  }

  /**
   * @param original  Incoming message.
   * @param content   Pre-extracted text message content, never {@code null}.
   *                  In case of binary message, data is interpreted as UTF-8 symbols.
   */
  protected abstract void onMessage(Message original, String content) throws Exception;

  protected abstract void onError(Exception e);
}