Java 类javax.jms.QueueSender 实例源码

@Test
public void testGetQueue() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueSender sender = session.createSender(queue);

    assertNotNull(sender.getQueue());
    assertSame(queue, sender.getQueue());

    sender.close();

    try {
        sender.getQueue();
        fail("Cannot read topic on closed sender");
    } catch (IllegalStateException ise) {}
}

@Test(timeout = 60000)
public void testSenderAndPublisherDest() throws Exception {
    JmsPoolXAConnectionFactory pcf = new JmsPoolXAConnectionFactory();
    pcf.setConnectionFactory(new ActiveMQXAConnectionFactory(
        "vm://test?broker.persistent=false&broker.useJmx=false"));

    QueueConnection connection = pcf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    QueueSender sender = session.createSender(session.createQueue("AA"));
    assertNotNull(sender.getQueue().getQueueName());

    connection.close();

    TopicConnection topicConnection = pcf.createTopicConnection();
    TopicSession topicSession = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    TopicPublisher topicPublisher = topicSession.createPublisher(topicSession.createTopic("AA"));
    assertNotNull(topicPublisher.getTopic().getTopicName());

    topicConnection.close();
    pcf.stop();
}

/** Send a JSON message to our notification queue.
 */
public void invokeJMS(JsonObject json) throws JMSException, NamingException {
    if (!initialized) initialize(); //gets our JMS managed resources (Q and QCF)

    QueueConnection connection = queueCF.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

    TextMessage message = session.createTextMessage(json.toString());
    System.out.println("Sending "+json.toString()+" to "+queue.getQueueName());
    QueueSender sender = session.createSender(queue);
    sender.send(message);

    sender.close();
    session.close();
    connection.close();

    System.out.println("Message sent successfully!");
}

@Test
public void processAndReply() throws JMSException {
    MessagingMessageListenerAdapter listener = createDefaultInstance(String.class);
    String body = "echo text";
    String correlationId = "link-1234";
    Destination replyDestination = new Destination() {};

    TextMessage reply = mock(TextMessage.class);
    QueueSender queueSender = mock(QueueSender.class);
    Session session = mock(Session.class);
    given(session.createTextMessage(body)).willReturn(reply);
    given(session.createProducer(replyDestination)).willReturn(queueSender);

    listener.setDefaultResponseDestination(replyDestination);
    StubTextMessage inputMessage = createSimpleJmsTextMessage(body);
    inputMessage.setJMSCorrelationID(correlationId);
    listener.onMessage(inputMessage, session);
    assertDefaultListenerMethodInvocation();

    verify(reply).setJMSCorrelationID(correlationId);
    verify(queueSender).send(reply);
    verify(queueSender).close();
}

private void processAndReplyWithSendTo(MessagingMessageListenerAdapter listener, boolean pubSubDomain) throws JMSException {
    String body = "echo text";
    String correlationId = "link-1234";
    Destination replyDestination = new Destination() {};

    DestinationResolver destinationResolver = mock(DestinationResolver.class);
    TextMessage reply = mock(TextMessage.class);
    QueueSender queueSender = mock(QueueSender.class);
    Session session = mock(Session.class);

    given(destinationResolver.resolveDestinationName(session, "replyDestination", pubSubDomain))
            .willReturn(replyDestination);
    given(session.createTextMessage(body)).willReturn(reply);
    given(session.createProducer(replyDestination)).willReturn(queueSender);

    listener.setDestinationResolver(destinationResolver);
    StubTextMessage inputMessage = createSimpleJmsTextMessage(body);
    inputMessage.setJMSCorrelationID(correlationId);
    listener.onMessage(inputMessage, session);

    verify(destinationResolver).resolveDestinationName(session, "replyDestination", pubSubDomain);
    verify(reply).setJMSCorrelationID(correlationId);
    verify(queueSender).send(reply);
    verify(queueSender).close();
}

/**
 * Sends a message to the myWMS application inside the application
 * server.
 * 
 * @param mfcMessage the message to be send
 * @throws JMSException
 * @throws NamingException
 */
public void send(MFCMessage mfcMessage)
    throws JMSException,
        NamingException
{

    // create the jms session
    QueueSession session =
        connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

    // lookup of the jms queue over jndi
    Queue outputQueue =
        (Queue) context.lookup("queue/" + OUTPUT_QUEUE_NAME);

    // create the message publisher
    QueueSender sender = session.createSender(outputQueue);

    ObjectMessage objectMessage = session.createObjectMessage();
    objectMessage.setObject(mfcMessage);
    objectMessage.setJMSTimestamp(System.currentTimeMillis());

    sender.send(objectMessage);
    sender.close();

    session.close();
}

private void acknowledge(Message jmsMessage, String msgid) throws JMSException, ServiceLocatorException {
        QueueConnection connection = null;
        QueueSession session = null;
        QueueSender sender = null;
        try {
            Queue respQueue = (Queue) jmsMessage.getJMSReplyTo();
            QueueConnectionFactory qcf = JMSServices.getInstance().getQueueConnectionFactory(null);
            connection = qcf.createQueueConnection();
            session = connection.createQueueSession(false, QueueSession.DUPS_OK_ACKNOWLEDGE);
            sender = session.createSender(respQueue);
            Message respMsg = session.createTextMessage(msgid);
//        respMsg.setJMSCorrelationID(correlationId); not used
            sender.send(respMsg);
        } finally {
            if (sender != null) sender.close();
            if (session != null) session.close();
            if (connection != null) connection.close();
        }
    }

private void acknowledge(Message jmsMessage, String msgid) throws JMSException, ServiceLocatorException {
        QueueConnection connection = null;
        QueueSession session = null;
        QueueSender sender = null;
        try {
            Queue respQueue = (Queue) jmsMessage.getJMSReplyTo();
            QueueConnectionFactory qcf = JMSServices.getInstance().getQueueConnectionFactory(null);
            connection = qcf.createQueueConnection();
            session = connection.createQueueSession(false, QueueSession.DUPS_OK_ACKNOWLEDGE);
            sender = session.createSender(respQueue);
            Message respMsg = session.createTextMessage(msgid);
//        respMsg.setJMSCorrelationID(correlationId); not used
            sender.send(respMsg);
        } finally {
            if (sender != null) sender.close();
            if (session != null) session.close();
            if (connection != null) connection.close();
        }
    }

/**
 * Send a message to testInboundQueue queue
 *
 * @throws Exception
 */
private void sendMessage() throws Exception {
    InitialContext initialContext = JmsClientHelper.getActiveMqInitialContext();
    QueueConnectionFactory connectionFactory
            = (QueueConnectionFactory) initialContext.lookup(JmsClientHelper.QUEUE_CONNECTION_FACTORY);
    QueueConnection queueConnection = connectionFactory.createQueueConnection();
    QueueSession queueSession = queueConnection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
    QueueSender sender = queueSession.createSender(queueSession.createQueue(QUEUE_NAME));

    String message = "<?xml version='1.0' encoding='UTF-8'?>" +
            "    <ser:getQuote xmlns:ser=\"http://services.samples\" xmlns:xsd=\"http://services.samples/xsd\"> " +
            "      <ser:request>" +
            "        <xsd:symbol>IBM</xsd:symbol>" +
            "      </ser:request>" +
            "    </ser:getQuote>";
    try {
        TextMessage jmsMessage = queueSession.createTextMessage(message);
        jmsMessage.setJMSType("incorrecttype");
        sender.send(jmsMessage);
    } finally {
        queueConnection.close();
    }
}

/**
 * Close a JMS {@link MessageProducer}.
 * @param messageProducer JMS Message Producer that needs to be closed.
 * @throws JMSException if an error occurs while closing the producer.
 */
public void closeProducer(MessageProducer messageProducer) throws JMSException {
    if (messageProducer != null) {
        if (logger.isDebugEnabled()) {
            logger.debug("Closing a JMS Message Producer of: " + this.connectionFactoryString);
        }
        if ((JMSConstants.JMS_SPEC_VERSION_1_1.equals(jmsSpec)) || (JMSConstants.JMS_SPEC_VERSION_2_0
                .equals(jmsSpec))) {
            messageProducer.close();
        } else {
            if (JMSConstants.JMSDestinationType.QUEUE.equals(this.destinationType)) {
                ((QueueSender) messageProducer).close();
            } else {
                ((TopicPublisher) messageProducer).close();
            }
        }
    }
}

/**
 * com.sun.ts.tests.jms.ee.all.queueconn.QueueConnTest line 171
 */
@Test
public void testCreateReceiverWithMessageSelector() throws Exception {
   QueueConnection qc = null;

   try {
      qc = createQueueConnection();
      QueueSession qs = qc.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

      QueueReceiver qreceiver = qs.createReceiver(queue1, "targetMessage = TRUE");

      qc.start();

      TextMessage m = qs.createTextMessage();
      m.setText("one");
      m.setBooleanProperty("targetMessage", false);

      QueueSender qsender = qs.createSender(queue1);

      qsender.send(m);

      m.setText("two");
      m.setBooleanProperty("targetMessage", true);

      qsender.send(m);

      TextMessage rm = (TextMessage) qreceiver.receive(1000);

      ProxyAssertSupport.assertEquals("two", rm.getText());
   } finally {
      if (qc != null) {
         qc.close();
      }
      Thread.sleep(2000);
      removeAllMessages(queue1.getQueueName(), true);
      checkEmpty(queue1);
   }
}

/**
 * Create a queue sender
 *
 * @param queue The queue
 * @return The queue sender
 * @throws JMSException Thrown if an error occurs
 */
@Override
public QueueSender createSender(final Queue queue) throws JMSException {
   lock();
   try {
      QueueSession session = getQueueSessionInternal();

      if (ActiveMQRASession.trace) {
         ActiveMQRALogger.LOGGER.trace("createSender " + session + " queue=" + queue);
      }

      QueueSender result = session.createSender(queue);
      result = new ActiveMQRAQueueSender(result, this);

      if (ActiveMQRASession.trace) {
         ActiveMQRALogger.LOGGER.trace("createdSender " + session + " sender=" + result);
      }

      addProducer(result);

      return result;
   } finally {
      unlock();
   }
}

@Before
public void setUp() throws Exception {
    System.setProperty(Context.INITIAL_CONTEXT_FACTORY, TestContextFactory.class.getName());
    QueueConnectionFactory queueConnectionFactory = Mockito.mock(QueueConnectionFactory.class);
    Queue queue = Mockito.mock(Queue.class);
    Context context = Mockito.mock(Context.class);
    TestContextFactory.context = context;
    when(context.lookup(eq("jms/queueConnectionFactory"))).thenReturn(queueConnectionFactory);
    when(context.lookup(eq("jms/mailQueue"))).thenReturn(queue);
    queueSender = Mockito.mock(QueueSender.class);
    QueueConnection queueConnection = Mockito.mock(QueueConnection.class);
    when(queueConnectionFactory.createQueueConnection()).thenReturn(queueConnection);
    when(queueConnectionFactory.createQueueConnection(anyString(), anyString())).thenReturn(queueConnection);
    QueueSession queueSession = Mockito.mock(QueueSession.class);
    bytesMessage = Mockito.mock(BytesMessage.class);
    when(queueSession.createBytesMessage()).thenReturn(bytesMessage);
    when(queueConnection.createQueueSession(anyBoolean(), anyInt())).thenReturn(queueSession);
    when(queueSession.createSender(eq(queue))).thenReturn(queueSender);
    transport = new SmtpJmsTransport(Session.getDefaultInstance(new Properties()), new URLName("jms"));
    transportListener = Mockito.mock(TransportListener.class);
    transport.addTransportListener(transportListener);
}

/**
 * Overrides the superclass method to use the JMS 1.0.2 API to send a response.
 * <p>Uses the JMS pub-sub API if the given destination is a topic,
 * else uses the JMS queue API.
 */
protected void sendResponse(Session session, Destination destination, Message response) throws JMSException {
    MessageProducer producer = null;
    try {
        if (destination instanceof Topic) {
            producer = ((TopicSession) session).createPublisher((Topic) destination);
            postProcessProducer(producer, response);
            ((TopicPublisher) producer).publish(response);
        }
        else {
            producer = ((QueueSession) session).createSender((Queue) destination);
            postProcessProducer(producer, response);
            ((QueueSender) producer).send(response);
        }
    }
    finally {
        JmsUtils.closeMessageProducer(producer);
    }
}

/**
 * This implementation overrides the superclass method to use JMS 1.0.2 API.
 */
protected void doSend(MessageProducer producer, Message message) throws JMSException {
    if (isPubSubDomain()) {
        if (isExplicitQosEnabled()) {
            ((TopicPublisher) producer).publish(message, getDeliveryMode(), getPriority(), getTimeToLive());
        }
        else {
            ((TopicPublisher) producer).publish(message);
        }
    }
    else {
        if (isExplicitQosEnabled()) {
            ((QueueSender) producer).send(message, getDeliveryMode(), getPriority(), getTimeToLive());
        }
        else {
            ((QueueSender) producer).send(message);
        }
    }
}

@Test
public void testConverter() throws Exception {
    JmsTemplate102 template = createTemplate();
    template.setConnectionFactory(queueConnectionFactory);
    template.setMessageConverter(new SimpleMessageConverter());
    String s = "Hello world";

    QueueSender queueSender = mock(QueueSender.class);
    TextMessage message = mock(TextMessage.class);

    given(queueSession.createSender(this.queue)).willReturn(queueSender);
    given(queueSession.createTextMessage("Hello world")).willReturn(message);

    template.convertAndSend(queue, s);

    if (useTransactedTemplate()) {
        verify(queueSession).commit();
    }
    verify(queueSender).send(message);
    verify(queueSender).close();
    verify(queueSession).close();
    verify(queueConnection).close();
}

@Override
public QueueSender createSender(Queue queue) throws JMSException
   {
    externalAccessLock.readLock().lock();
    try
    {
        checkNotClosed();
        LocalQueueSender sender = new LocalQueueSender(this,queue,idProvider.createID());
        registerProducer(sender);
        return sender;
    }
    finally
    {
        externalAccessLock.readLock().unlock();
    }
   }

@Override
public QueueSender createSender(Queue queue) throws JMSException
   {
    externalAccessLock.readLock().lock();
    try
    {
        checkNotClosed();

        RemoteQueueSender sender =  new RemoteQueueSender(this,
                                                          DestinationTools.asRef(queue),
                                                          idProvider.createID());
        registerProducer(sender);
        return sender;
    }
    finally
    {
        externalAccessLock.readLock().unlock();
    }
   }

@Test
public void testCreateQueueSender() throws Exception {
    JmsConnectionFactory factory = new JmsConnectionFactory(getBrokerAmqpConnectionURI());
    QueueConnection connection = factory.createQueueConnection();
    assertNotNull(connection);

    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);
    Queue queue = session.createQueue(name.getMethodName());
    QueueSender sender = session.createSender(queue);
    assertNotNull(sender);

    QueueViewMBean proxy = getProxyToQueue(name.getMethodName());
    assertEquals(0, proxy.getQueueSize());
    connection.close();
}

@Test
public void testCreateQueueSender() throws Exception {
    JmsConnectionFactory factory = new JmsConnectionFactory(getBrokerAmqpConnectionURI());
    QueueConnection connection = factory.createQueueConnection();
    assertNotNull(connection);

    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    assertNotNull(session);
    Queue queue = session.createQueue(name.getMethodName());
    QueueSender sender = session.createSender(queue);
    assertNotNull(sender);

    QueueViewMBean proxy = getProxyToQueue(name.getMethodName());
    assertEquals(0, proxy.getQueueSize());
    connection.close();
}

/**
 * Intended for server-side reponse sending and implies that the received
 * message *always* contains a reply-to address.
 */
public void sendReply(QueueSession session, Message received_message, String response) throws IfsaException {
    QueueSender tqs=null;
    try {
        TextMessage answer = session.createTextMessage();
        answer.setText(response);
        Queue replyQueue = (Queue)received_message.getJMSReplyTo();
        tqs = session.createSender(replyQueue );
        if (log.isDebugEnabled()) log.debug(getLogPrefix()+ "sending reply to ["+ received_message.getJMSReplyTo()+ "]");
        ((IFSAServerQueueSender) tqs).sendReply(received_message, answer);
    } catch (Throwable t) {
        throw new IfsaException(t);
    } finally {
        if (tqs!=null) {
            try {
                tqs.close();
            } catch (JMSException e) {
                log.warn(getLogPrefix()+ "exception closing reply queue sender",e);
            }   
        }
    }
}

/**
 * {@inheritDoc}
 */
public void submitJob(AbstractFileManagementJob job) {
    ServiceLocator locator = ServiceLocatorFactory.getLocator();
    final QueueConnectionFactory factory = (QueueConnectionFactory) locator.lookup(DEFAULT_QUEUE_CONN_FACTORY);
    final Queue queue = (Queue) locator.lookup(FileManagementMDB.QUEUE_JNDI_NAME);
    QueueConnection queueConnection = null;
    QueueSession queueSession = null;
    QueueSender queueSender = null;
    try {
        queueConnection = factory.createQueueConnection();
        queueSession = queueConnection.createQueueSession(true, 0);
        queueSender = queueSession.createSender(queue);
        final ObjectMessage message = queueSession.createObjectMessage(job);
        queueSender.send(message);
    } catch (JMSException e) {
        LOG.error("Couldn't submit job to JMS", e);
    } finally {
        close(queueSender);
        close(queueSession);
        close(queueConnection);
    }
}

public QueueSender createSender(Queue queue) throws JMSException {
    lock();
    try {
        QueueSession session = getQueueSession();
        if (trace)
            log.trace("createSender " + session + " queue=" + queue);
        QueueSender result = session.createSender(queue);
        result = new JmsQueueSender(result, this);
        if (trace)
            log.trace("createdSender " + session + " sender=" + result);
        addProducer(result);
        return result;
    } finally {
        unlock();
    }
}

private QueueSender getQueueSender(Queue destination) throws JMSException {
    QueueSender result = null;

    if (useAnonymousProducers) {
        result = safeGetSessionHolder().getOrCreateSender();
    } else {
        result = ((QueueSession) getInternalSession()).createSender(destination);
    }

    return result;
}

public QueueSender getOrCreateSender() throws JMSException {
    if (sender == null) {
        synchronized (this) {
            if (sender == null) {
                sender = ((QueueSession) session).createSender(null);
            }
        }
    }

    return sender;
}

@Test
public void testToString() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createQueueConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueSender sender = session.createSender(queue);

    assertNotNull(sender.toString());
}

@Test
public void testSendToQueueFailsIfNotAnonymousPublisher() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createTopicConnection();
    QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    QueueSender sender = session.createSender(queue);

    try {
        sender.send(session.createTemporaryQueue(), session.createTextMessage());
        fail("Should not be able to send to alternate destination");
    } catch (UnsupportedOperationException ex) {}
}

/**
 * Realiza proceso de aviso a BackOffice para un tr�mite. Metemos en cola as�ncrona.
 * En caso de error lanza mensaje al log y permite continuar.
 * @param procedimiento
 */
private void avisoBackOffice(Procedimiento procedimiento,Date hasta) throws Exception{      

        log.debug("Aviso a backoffice procedimiento  " + procedimiento.getIdentificador() + " (hasta " + StringUtil.fechaACadena(hasta,StringUtil.FORMATO_TIMESTAMP) + ")");

        // Obtenemos entradas no procesadas                             
        TramiteBandejaDelegate tbd = (TramiteBandejaDelegate) DelegateUtil.getTramiteBandejaDelegate();

        // Obtenemos tramites del procedimiento que tienen entradas pendientes y generamos un mensaje por tramite
        String idTramites[] = tbd.obtenerIdTramitesProcedimiento(procedimiento.getIdentificador(),ConstantesBTE.ENTRADA_NO_PROCESADA,null,hasta);
        if (idTramites != null) {
            for (int i = 0; i < idTramites.length; i++) {

                String  entradas [] = tbd.obtenerNumerosEntradas(procedimiento.getIdentificador(), idTramites[i], ConstantesBTE.ENTRADA_NO_PROCESADA,null,hasta);

        log.debug("Aviso de " + entradas.length + " nuevas entradas para backoffice tr�mite " + idTramites[i] + " hasta " + StringUtil.fechaACadena(hasta,StringUtil.FORMATO_TIMESTAMP) + " (Procedimiento: " + procedimiento.getIdentificador() + ")");

        if (entradas.length > 0){
        // Dejamos entrada en la cola de avisos
        InitialContext ctx = new InitialContext();
        String colaAvisos = (String) ctx.lookup("java:comp/env/colaAvisos");
        Queue queue = (Queue) ctx.lookup(colaAvisos);        
        QueueConnectionFactory factory = (QueueConnectionFactory) ctx.lookup("java:/XAConnectionFactory");
        QueueConnection cnn = factory.createQueueConnection();
        QueueSession sess = cnn.createQueueSession(false,QueueSession.AUTO_ACKNOWLEDGE);              
        TextMessage msg = sess.createTextMessage(BteStringUtil.numeroEntradasToString(entradas));
        QueueSender sender = sess.createSender(queue);
        sender.send(msg,DeliveryMode.NON_PERSISTENT,4,0);               
        }
            }
        }           
}

/**
   * Tras producirse una entrada en el BackOffice realizamos aviso inmediato si procede  
   * @param entrada
   */
  private void avisoBackOffice(String numeroEntrada) throws Exception{

    // TODO Destino cola parametrizable por properties

    // Recuperamos informacion tramite
    TramiteBandejaDelegate td = DelegateUtil.getTramiteBandejaDelegate();
TramiteBandeja tramiteBandeja = td.obtenerTramiteBandeja(numeroEntrada);

// Comprobamos si tiene habilitado el aviso inmediato
if (tramiteBandeja.getProcedimiento().avisosEnabled() &&
    tramiteBandeja.getProcedimiento().getInmediata() == 'S'){

    // Dejamos entrada en la cola de avisos inmediatos
    InitialContext ctx = new InitialContext();
    String colaAvisos = (String) ctx.lookup("java:comp/env/colaAvisos");
    Queue queue = (Queue) ctx.lookup(colaAvisos);                   
    QueueConnectionFactory factory = (QueueConnectionFactory) ctx.lookup("java:/JmsXA");
    QueueConnection cnn = factory.createQueueConnection();

    // Creamos sesion transacted para que hasta q no se haga commit no se env�e el mensaje
    QueueSession sess = cnn.createQueueSession(true,0);         

    TextMessage msg = sess.createTextMessage(numeroEntrada);
    QueueSender sender = sess.createSender(queue);
    sender.send(msg);

    cnn.close();

}

  }

/**
 * Build a dynamic proxy that reflectively adapts to JMS 2.0 API methods, if necessary.
 * Otherwise simply return this CachedMessageProducer instance itself.
 */
public MessageProducer getProxyIfNecessary() {
    if (completionListenerClass != null) {
        return (MessageProducer) Proxy.newProxyInstance(CachedMessageProducer.class.getClassLoader(),
                new Class<?>[] {MessageProducer.class, QueueSender.class, TopicPublisher.class},
                new Jms2MessageProducerInvocationHandler());
    }
    else {
        return this;
    }
}

@Test
public void testWithResponsiveMessageDelegateWithDefaultDestination_SendsReturnTextMessageWhenSessionSupplied() throws Exception {
    Queue destination = mock(Queue.class);
    TextMessage sentTextMessage = mock(TextMessage.class);
    // correlation ID is queried when response is being created...
    given(sentTextMessage.getJMSCorrelationID()).willReturn(
            CORRELATION_ID);
    // Reply-To is queried when response is being created...
    given(sentTextMessage.getJMSReplyTo()).willReturn(null); // we want to fall back to the default...

    TextMessage responseTextMessage = mock(TextMessage.class);

    QueueSender queueSender = mock(QueueSender.class);
    Session session = mock(Session.class);
    given(session.createTextMessage(RESPONSE_TEXT)).willReturn(responseTextMessage);
    given(session.createProducer(destination)).willReturn(queueSender);

    ResponsiveMessageDelegate delegate = mock(ResponsiveMessageDelegate.class);
    given(delegate.handleMessage(sentTextMessage)).willReturn(RESPONSE_TEXT);

    MessageListenerAdapter adapter = new MessageListenerAdapter(delegate) {
        @Override
        protected Object extractMessage(Message message) {
            return message;
        }
    };
    adapter.setDefaultResponseDestination(destination);
    adapter.onMessage(sentTextMessage, session);

    verify(responseTextMessage).setJMSCorrelationID(CORRELATION_ID);
    verify(queueSender).send(responseTextMessage);
    verify(queueSender).close();
    verify(delegate).handleMessage(sentTextMessage);
}

@Test
public void testWithResponsiveMessageDelegateWhenReturnTypeIsAJMSMessageAndNoMessageConverterIsSupplied() throws Exception {
    Queue destination = mock(Queue.class);
    final TextMessage sentTextMessage = mock(TextMessage.class);
    // correlation ID is queried when response is being created...
    given(sentTextMessage.getJMSCorrelationID()).willReturn(CORRELATION_ID);
    // Reply-To is queried when response is being created...
    given(sentTextMessage.getJMSReplyTo()).willReturn(destination);

    TextMessage responseMessage = mock(TextMessage.class);
    QueueSender queueSender = mock(QueueSender.class);

    Session session = mock(Session.class);
    given(session.createProducer(destination)).willReturn(queueSender);

    ResponsiveJmsTextMessageReturningMessageDelegate delegate = mock(ResponsiveJmsTextMessageReturningMessageDelegate.class);
    given(delegate.handleMessage(sentTextMessage)).willReturn(responseMessage);

    final MessageListenerAdapter adapter = new MessageListenerAdapter(delegate) {
        @Override
        protected Object extractMessage(Message message) {
            return message;
        }
    };
    adapter.setMessageConverter(null);
    adapter.onMessage(sentTextMessage, session);

    verify(responseMessage).setJMSCorrelationID(CORRELATION_ID);
    verify(queueSender).send(responseMessage);
    verify(queueSender).close();
}

public String sendTextMessage(String queueName, String msg) throws Exception {

        if (StringUtils.isBlank(queueName)) {
            throw new IllegalArgumentException("Queue can not be null!");
        }

        if (logger.isDebugEnabled()) {

            StringBuilder sb = new StringBuilder();
            sb.append("sendTextMessage :: Setting params... \n");
            sb.append("\t- queueManagerName = '{}' \n");
            sb.append("\t- queueName = '{}' \n");
            sb.append("\t- replyToQueueName = '{}' \n");
            sb.append("\t- hostname = '{}' \n");
            sb.append("\t- port = '{}' \n");
            sb.append("\t- channel = '{}' \n");

            LoggerUtils.logDebug(logger, sb.toString(), queueManagerName, queueName, responseQueueName, hostname, port,
                channel);
        }

        QueueConnection connection = null;
        TextMessage requestMessage;
        try {

            QueueConnectionFactory qcf = new MQQueueConnectionFactory();
            ((MQQueueConnectionFactory) qcf).setIntProperty(WMQConstants.WMQ_CONNECTION_MODE,
                WMQConstants.WMQ_CM_CLIENT);
            ((MQQueueConnectionFactory) qcf).setHostName(hostname);
            ((MQQueueConnectionFactory) qcf).setPort(port);
            ((MQQueueConnectionFactory) qcf).setQueueManager(queueManagerName);
            if (StringUtils.isNotBlank(channel)) {
                ((MQQueueConnectionFactory) qcf).setChannel(channel);
            }
            // ((MQQueueConnectionFactory) qcf).setCCSID(500);

            connection = qcf.createQueueConnection(" ", " ");
            connection.start();

            // Create a session
            MQQueueSession session = (MQQueueSession) connection.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);

            // Get a request queue
            Queue queue = ((MQQueueSession) session).createQueue("queue://" + queueManagerName + "/" + queueName);

            // Create message sender
            QueueSender sender = session.createSender(queue);
            requestMessage = session.createTextMessage(msg);
            // m1.setIntProperty("JMS_", MQC.MQENC_NATIVE);

            // Setting reply-to queue
            Queue queueResp = ((MQQueueSession) session)
                .createQueue("queue://" + queueManagerName + "/" + responseQueueName);
            requestMessage.setJMSReplyTo(queueResp);
            LoggerUtils.logDebug(logger, "sendTextMessage :: message \n{}", requestMessage.toString());

            sender.send(requestMessage);
            LoggerUtils.logDebug(logger, "sendTextMessage :: Message Sent! ID: {} \n",
                requestMessage.getJMSMessageID());

            return requestMessage.getJMSMessageID();

        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException je) {
                    je.printStackTrace();
                }
            }
        }
    }

private void closeResources(QueueConnection pConn, QueueSession pSession, QueueSender pSender)
        throws JMSException {
    if (pSender != null) {
        pSender.close();
    }
    if (pSession != null) {
        pSession.close();
    }
    if (pConn != null) {
        pConn.close();
    }
}

public void ignoreOnMessage()
throws Exception
{
    Email email = new Email();
    email.sendMail(to, from, subject, "PRE TEST");

    Debugger.println(this,"Hello word");

    QueueConnection cnn = null;
      QueueSender sender = null;
      QueueSession session = null;
      InitialContext ctx = new InitialContext();
      Queue queue = (Queue) ctx.lookup("queue/tutorial/example");
      QueueConnectionFactory factory = (QueueConnectionFactory) ctx.lookup("ConnectionFactory");
      try {
         cnn = factory.createQueueConnection();
         session = cnn.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);

         TextMessage msg = session.createTextMessage("Hello World");

         msg.setStringProperty("to", to);
         msg.setStringProperty("from", from);
         msg.setStringProperty("subject", subject);
         sender = session.createSender(queue);
         sender.send(msg);
         System.out.println("Message sent successfully to remote queue.");
      } finally {
         // Cleanup
         // close the connection
         if (cnn != null)
         {
            cnn.close();
         }
      }
}

@Test
public void testTempQueueDelete() throws Exception {
   connection.start();
   QueueSession queueSession = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

   TemporaryQueue tempQueue = queueSession.createTemporaryQueue();

   ActiveMQConnection newConn = (ActiveMQConnection) factory.createConnection();
   try {
      QueueSession newQueueSession = newConn.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
      QueueSender queueSender = newQueueSession.createSender(tempQueue);

      Message msg = queueSession.createMessage();
      queueSender.send(msg);

      try {
         QueueReceiver consumer = newQueueSession.createReceiver(tempQueue);
         fail("should have gotten exception but got consumer: " + consumer);
      } catch (JMSException ex) {
         //correct
      }

      connection.close();

      try {
         Message newMsg = newQueueSession.createMessage();
         queueSender.send(newMsg);
      } catch (JMSException e) {
         //ok
      }

   } finally {
      newConn.close();
   }
}

/**
 * Create a new wrapper
 *
 * @param producer the producer
 * @param session  the session
 */
public ActiveMQRAQueueSender(final QueueSender producer, final ActiveMQRASession session) {
   super(producer, session);

   if (ActiveMQRAQueueSender.trace) {
      ActiveMQRALogger.LOGGER.trace("constructor(" + producer + ", " + session + ")");
   }
}

/**
 * Get queue
 *
 * @return The queue
 * @throws JMSException Thrown if an error occurs
 */
@Override
public Queue getQueue() throws JMSException {
   if (ActiveMQRAQueueSender.trace) {
      ActiveMQRALogger.LOGGER.trace("getQueue()");
   }

   return ((QueueSender) producer).getQueue();
}

public void sendAMessage(String msg) throws JMSException {

        QueueSender send = session.createSender(que);
        TextMessage tm = session.createTextMessage(msg);

        send.send(tm);
        send.close();
    }

/**
 * Actually execute the given request, sending the invoker request message
 * to the specified target queue and waiting for a corresponding response.
 * <p>The default implementation is based on standard JMS send/receive,
 * using a {@link javax.jms.TemporaryQueue} for receiving the response.
 * @param session the JMS Session to use
 * @param queue the resolved target Queue to send to
 * @param requestMessage the JMS Message to send
 * @return the RemoteInvocationResult object
 * @throws JMSException in case of JMS failure
 */
protected Message doExecuteRequest(Session session, Queue queue, Message requestMessage) throws JMSException {
    TemporaryQueue responseQueue = null;
    MessageProducer producer = null;
    MessageConsumer consumer = null;
    try {
        if (jms11Available) {
            // Standard JMS 1.1 API usage...
            responseQueue = session.createTemporaryQueue();
            producer = session.createProducer(queue);
            consumer = session.createConsumer(responseQueue);
            requestMessage.setJMSReplyTo(responseQueue);
            producer.send(requestMessage);
        }
        else {
            // Perform all calls on QueueSession reference for JMS 1.0.2 compatibility...
            // DEPRECATED but kept around with the deprecated JmsTemplate102 etc classes for the time being.
            QueueSession queueSession = (QueueSession) session;
            responseQueue = queueSession.createTemporaryQueue();
            QueueSender sender = queueSession.createSender(queue);
            producer = sender;
            consumer = queueSession.createReceiver(responseQueue);
            requestMessage.setJMSReplyTo(responseQueue);
            sender.send(requestMessage);
        }
        long timeout = getReceiveTimeout();
        return (timeout > 0 ? consumer.receive(timeout) : consumer.receive());
    }
    finally {
        JmsUtils.closeMessageConsumer(consumer);
        JmsUtils.closeMessageProducer(producer);
        if (responseQueue != null) {
            responseQueue.delete();
        }
    }
}