/**
 * Copyright (C) 2011 - present by OpenGamma Inc. and the OpenGamma group of companies
 * 
 * Please see distribution for license.
 */
package com.opengamma.transport.jms;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.core.JmsTemplate;

/**
 * A message sender that uses JMS. Messages are sent in the background, preserving their order.
 */
public class QueueingJmsByteArrayMessageSender extends JmsByteArrayMessageSender {

  private static final Logger s_logger = LoggerFactory.getLogger(JmsByteArrayMessageSender.class);

  private final AtomicBoolean _isShutdown = new AtomicBoolean(false);
  private final BlockingQueue<byte[]> _messageQueue = new LinkedBlockingQueue<byte[]>();
  private final Thread _senderThread;

  /**
   * Creates an instance associated with a destination and template.
   * 
   * @param destinationName  the destination name, not null
   * @param jmsTemplate  the template, not null
   */
  public QueueingJmsByteArrayMessageSender(final String destinationName, final JmsTemplate jmsTemplate) {
    super(destinationName, jmsTemplate);
    
    _senderThread = new Thread(new Runnable() {
      @Override
      public void run() {
        while (!_isShutdown.get()) {
          
          byte[] nextMessage;
          try {
            nextMessage = _messageQueue.take();
            sendSync(nextMessage);
          } catch (InterruptedException e) {
            s_logger.warn("Interrupted while waiting for next message to send", e);
          } catch (Exception e) {
            //PLAT-1213: note that interrupting the thread can end up here as a JMSException, hence _isShutdown
            s_logger.error("Failed to send message asynchronously", e);
          }
        }
        s_logger.info("Shutdown %s", destinationName);
      }
    }, String.format("QueueingJmsByteArrayMessageSender %s", destinationName));
    _senderThread.setDaemon(true);
    _senderThread.start();
  }

  //-------------------------------------------------------------------------
  @Override
  public void send(byte[] message) {
    _messageQueue.add(message);
  }
  
  public void shutdown() {
    _isShutdown.set(true);
    _senderThread.interrupt();
  }
  
  //-------------------------------------------------------------------------
  private void sendSync(byte[] message) {
    super.send(message);
  }
  
}

@@ -420,16 +420,60 @@
 plate;%0A%0A
+import com.opengamma.util.ArgumentChecker;%0A%0A
 /**%0A * A
@@ -750,16 +750,83 @@
 .class);
+%0A  %0A  private static final byte%5B%5D s_shutdownSentinel = new byte%5B0%5D;
 %0A%0A  priv
@@ -1473,40 +1473,15 @@
 le (
-!_isShutdown.get()) %7B%0A          
+true) %7B
 %0A   
@@ -1587,157 +1587,100 @@
     
-sendSync(nextMessage);%0A          %7D catch (InterruptedException e) %7B%0A            s_logger.warn(%22Interrupted while waiting for 
+if (_isShutdown.get()) %7B%0A              break;%0A            %7D%0A            sendSync(
 next
- m
+M
 essage
- to send%22, e
 );%0A 
@@ -1716,120 +1716,8 @@
 ) %7B%0A
-            //PLAT-1213: note that interrupting the thread can end up here as a JMSException, hence _isShutdown%0A
     
@@ -1810,63 +1810,8 @@
   %7D%0A
-        s_logger.info(%22Shutdown %25s%22, destinationName);%0A
     
@@ -2162,61 +2162,187 @@
     
-_isShutdown.set(true);%0A    _senderThread.interrupt();
+//NOTE: PLAT-1236 ActiveMQ doesn't like us interrupting the thread.%0A    _isShutdown.set(true);%0A    _messageQueue.add(s_shutdownSentinel); // Make sure that the take doesn't block 
 %0A  %7D
@@ -2461,24 +2461,73 @@
  message) %7B%0A
+    ArgumentChecker.notNull(message, %22message%22);%0A
     super.se

stop the QueueingJmsByteArrayMessageSender without interrupting the thread