Java 类javax.jms.TemporaryQueue 实例源码

private void sendWithReplyToTemp(ConnectionFactory cf, String serviceQueue) throws JMSException, InterruptedException {
    Connection connection = cf.createConnection();
    connection.start();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryQueue tempQueue = session.createTemporaryQueue();
    TextMessage msg = session.createTextMessage("Request");
    msg.setJMSReplyTo(tempQueue);
    MessageProducer producer = session.createProducer(session.createQueue(serviceQueue));
    producer.send(msg);

    MessageConsumer consumer = session.createConsumer(tempQueue);
    Message replyMsg = consumer.receive();
    assertNotNull(replyMsg);

    LOG.debug("Reply message: {}", replyMsg);

    consumer.close();

    producer.close();
    session.close();
    connection.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 {
        responseQueue = session.createTemporaryQueue();
        producer = session.createProducer(queue);
        consumer = session.createConsumer(responseQueue);
        requestMessage.setJMSReplyTo(responseQueue);
        producer.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();
        }
    }
}

public final boolean oneIteration() throws Exception {
    if ((tempQueues) && (tempQueuePerMessage)) {
        // Close existing temporary queue
        if (messageConsumer != null) messageConsumer.close();
        if (destConsumer != null) {
            ((TemporaryQueue) destConsumer).delete();
        }

        // Open new temporary queue
        destConsumer = context.createTemporaryQueue();
        messageConsumer = context.createConsumer(destConsumer);
        outMessage.setJMSReplyTo(destConsumer);
    }

    messageProducer.send(destProducer, outMessage);             
    if ( transacted ) context.commit();

    if ((inMessage = messageConsumer.receive(timeout)) != null) {
        if ( transacted ) context.commit();
        incIterations();
    } else {
        throw new Exception("No response to message (\nID: " + outMessage.getJMSMessageID() + "\nCorrID: " + outMessage.getJMSCorrelationID() +" )");
    }
    return true;
}

public final boolean oneIteration() throws Exception {
    if ((tempQueues) && (tempQueuePerMessage)) {
        // Close temporary queue
        if (messageConsumer != null) messageConsumer.close();
        if (destConsumer != null) {
            ((TemporaryQueue) destConsumer).delete();
        }

        // Open new temporary queue
        destConsumer = session.createTemporaryQueue();
        messageConsumer = session.createConsumer(destConsumer);
        outMessage.setJMSReplyTo(destConsumer);
    }

    startResponseTimePeriod();
    messageProducer.send(outMessage, deliveryMode, priority, expiry);               
    if (transacted) session.commit();

    if ((inMessage = messageConsumer.receive(timeout))!= null) {
        if (transacted) session.commit();
        incIterations();
    } else {
        throw new Exception("No response to message (\nID: "+outMessage.getJMSMessageID()+ "\nCorrId: " + outMessage.getJMSCorrelationID() +")");
    }
    return true;
}

protected void loadTemporaryDestinationTypes(JmsOperations template) {
    if (template == null) {
        throw new IllegalArgumentException("No JmsTemplate supplied!");
    }
    template.execute(new SessionCallback<Object>() {
        public Object doInJms(Session session) throws JMSException {
            TemporaryQueue queue = session.createTemporaryQueue();
            setTemporaryQueueType(queue.getClass());

            TemporaryTopic topic = session.createTemporaryTopic();
            setTemporaryTopicType(topic.getClass());

            queue.delete();
            topic.delete();
            return null;
        }
    });
}

@Test
public void testTemporaryDestinationTypes() throws Exception {
    JmsEndpoint endpoint = getMandatoryEndpoint("activemq:test.queue", JmsEndpoint.class);
    JmsConfiguration configuration = endpoint.getConfiguration();
    JmsProviderMetadata providerMetadata = configuration.getProviderMetadata();
    assertNotNull("provider", providerMetadata);

    Class<? extends TemporaryQueue> queueType = endpoint.getTemporaryQueueType();
    Class<? extends TemporaryTopic> topicType = endpoint.getTemporaryTopicType();

    log.info("Found queue type: " + queueType);
    log.info("Found topic type: " + topicType);

    assertNotNull("queueType", queueType);
    assertNotNull("topicType", topicType);

    assertEquals("queueType", ActiveMQTempQueue.class, queueType);
    assertEquals("topicType", ActiveMQTempTopic.class, topicType);
}

public FileClient(String filename) throws NamingException, JMSException {
    Context ctx = new InitialContext();
    ConnectionFactory factory = (ConnectionFactory) ctx.lookup("ConnectionFactory");
    Destination requestQueue = (Destination) ctx.lookup(DESTINATION);
    mConnection = factory.createConnection();
    mConnection.start();
    mSession = mConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    mMessageProducer = mSession.createProducer(requestQueue);
    TemporaryQueue replyQueue = null;
    replyQueue = mSession.createTemporaryQueue();
    mMessageConsumer = mSession.createConsumer(replyQueue);
    TextMessage request = mSession.createTextMessage();
    request.setText(filename);
    request.setJMSReplyTo(replyQueue);
    mMessageProducer.send(request);
}

@Test(timeout = 60000)
public void testReplyToUsingQueue() throws Throwable {
   Connection connection = createConnection();

   try {
      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      TemporaryQueue queue = session.createTemporaryQueue();
      MessageProducer p = session.createProducer(queue);

      TextMessage message = session.createTextMessage();
      message.setText("Message temporary");
      message.setJMSReplyTo(session.createQueue(getQueueName()));
      p.send(message);

      MessageConsumer cons = session.createConsumer(queue);
      connection.start();

      message = (TextMessage) cons.receive(5000);
      assertNotNull(message);
      Destination jmsReplyTo = message.getJMSReplyTo();
      assertNotNull(jmsReplyTo);
      assertNotNull(message);
   } finally {
      connection.close();
   }
}

@Test(timeout = 60000)
public void testReplyToUsingTempQueue() throws Throwable {
   Connection connection = createConnection();

   try {
      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      TemporaryQueue queue = session.createTemporaryQueue();
      MessageProducer p = session.createProducer(queue);

      TextMessage message = session.createTextMessage();
      message.setText("Message temporary");
      message.setJMSReplyTo(session.createTemporaryQueue());
      p.send(message);

      MessageConsumer cons = session.createConsumer(queue);
      connection.start();

      message = (TextMessage) cons.receive(5000);
      Destination jmsReplyTo = message.getJMSReplyTo();
      assertNotNull(jmsReplyTo);
      assertNotNull(message);
   } finally {
      connection.close();
   }
}

@Test(timeout = 60000)
public void testCreateTemporaryQueue() throws Throwable {
   Connection connection = createConnection();

   try {
      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      TemporaryQueue queue = session.createTemporaryQueue();
      System.out.println("queue:" + queue.getQueueName());
      MessageProducer producer = session.createProducer(queue);

      TextMessage message = session.createTextMessage();
      message.setText("Message temporary");
      producer.send(message);

      MessageConsumer consumer = session.createConsumer(queue);
      connection.start();

      message = (TextMessage) consumer.receive(5000);

      assertNotNull(message);
   } finally {
      connection.close();
   }
}

@Test
public void testTempQueues() throws Exception {
   TemporaryQueue temp = localSession.createTemporaryQueue();
   MessageProducer producer = localSession.createProducer(temp);
   producer.send(localSession.createTextMessage("test"));
   Thread.sleep(100);
   assertEquals("Destination not created", 1, remoteBroker.getAdminView().getTemporaryQueues().length);
   temp.delete();

   assertTrue("Destination not deleted", Wait.waitFor(new Wait.Condition() {
      @Override
      public boolean isSatisified() throws Exception {
         return 0 == remoteBroker.getAdminView().getTemporaryQueues().length;
      }
   }));
}

public void testNoSlowConsumerAdvisory() throws Exception {
   Session s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   TemporaryQueue queue = s.createTemporaryQueue();
   MessageConsumer consumer = s.createConsumer(queue);
   consumer.setMessageListener(new MessageListener() {
      @Override
      public void onMessage(Message message) {
      }
   });
   Topic advisoryTopic = AdvisorySupport.getSlowConsumerAdvisoryTopic((ActiveMQDestination) queue);
   s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   MessageConsumer advisoryConsumer = s.createConsumer(advisoryTopic);
   // start throwing messages at the consumer
   MessageProducer producer = s.createProducer(queue);
   for (int i = 0; i < MESSAGE_COUNT; i++) {
      BytesMessage m = s.createBytesMessage();
      m.writeBytes(new byte[1024]);
      producer.send(m);
   }
   Message msg = advisoryConsumer.receive(1000);
   assertNull(msg);
}

public void testSlowConsumerAdvisory() throws Exception {
   Session s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   TemporaryQueue queue = s.createTemporaryQueue();
   MessageConsumer consumer = s.createConsumer(queue);
   assertNotNull(consumer);

   Topic advisoryTopic = AdvisorySupport.getSlowConsumerAdvisoryTopic((ActiveMQDestination) queue);
   s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   MessageConsumer advisoryConsumer = s.createConsumer(advisoryTopic);
   // start throwing messages at the consumer
   MessageProducer producer = s.createProducer(queue);
   for (int i = 0; i < MESSAGE_COUNT; i++) {
      BytesMessage m = s.createBytesMessage();
      m.writeBytes(new byte[1024]);
      producer.send(m);
   }
   Message msg = advisoryConsumer.receive(1000);
   assertNotNull(msg);
}

public void testMessageDeliveryAdvisory() throws Exception {
   Session s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   TemporaryQueue queue = s.createTemporaryQueue();
   MessageConsumer consumer = s.createConsumer(queue);
   assertNotNull(consumer);

   Topic advisoryTopic = AdvisorySupport.getMessageDeliveredAdvisoryTopic((ActiveMQDestination) queue);
   MessageConsumer advisoryConsumer = s.createConsumer(advisoryTopic);
   //start throwing messages at the consumer
   MessageProducer producer = s.createProducer(queue);

   BytesMessage m = s.createBytesMessage();
   m.writeBytes(new byte[1024]);
   producer.send(m);

   Message msg = advisoryConsumer.receive(1000);
   assertNotNull(msg);
}

public void testTempMessageConsumedAdvisory() throws Exception {
   Session s = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   TemporaryQueue queue = s.createTemporaryQueue();
   MessageConsumer consumer = s.createConsumer(queue);

   Topic advisoryTopic = AdvisorySupport.getMessageConsumedAdvisoryTopic((ActiveMQDestination) queue);
   MessageConsumer advisoryConsumer = s.createConsumer(advisoryTopic);
   //start throwing messages at the consumer
   MessageProducer producer = s.createProducer(queue);

   BytesMessage m = s.createBytesMessage();
   m.writeBytes(new byte[1024]);
   producer.send(m);
   String id = m.getJMSMessageID();
   Message msg = consumer.receive(1000);
   assertNotNull(msg);

   msg = advisoryConsumer.receive(1000);
   assertNotNull(msg);

   ActiveMQMessage message = (ActiveMQMessage) msg;
   ActiveMQMessage payload = (ActiveMQMessage) message.getDataStructure();
   String originalId = payload.getJMSMessageID();
   assertEquals(originalId, id);
}

/**
 * Test you can't delete a Destination with Active Subscribers
 *
 * @throws JMSException
 */
@Test
public void testDeleteDestinationWithSubscribersFails() throws JMSException {
   Connection connection = factory.createConnection();
   connections.add(connection);
   Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   TemporaryQueue queue = session.createTemporaryQueue();

   connection.start();

   session.createConsumer(queue);

   // This message delivery should NOT work since the temp connection is
   // now closed.
   try {
      queue.delete();
      Assert.fail("Should fail as Subscribers are active");
   } catch (JMSException e) {
      Assert.assertTrue("failed to throw an exception", true);
   }
}

/**
 * Test that if the JMS ReplyTo header field has been set as a <code>TemporaryQueue</code>,
 * it will be rightly get also as a <code>TemporaryQueue</code>
 * (and not only as a <code>Queue</code>).
 */
@Test
public void testJMSReplyTo_2() {
   try {
      TemporaryQueue tempQueue = senderSession.createTemporaryQueue();
      Message message = senderSession.createMessage();
      message.setJMSReplyTo(tempQueue);
      sender.send(message);

      Message msg = receiver.receive(TestConfig.TIMEOUT);
      Destination dest = msg.getJMSReplyTo();
      Assert.assertTrue("JMS ReplyTo header field should be a TemporaryQueue", dest instanceof TemporaryQueue);
      Queue replyTo = (Queue) dest;
      Assert.assertEquals("JMS ReplyTo header field should be equals to the temporary queue", replyTo.getQueueName(), tempQueue.getQueueName());
   } catch (JMSException e) {
      fail(e);
   }
}

@Test
public void testTemporaryQueueShouldNotBeInJNDI() throws Exception {
   Connection producerConnection = createConnection();

   Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);

   TemporaryQueue tempQueue = producerSession.createTemporaryQueue();
   String queueName = tempQueue.getQueueName();

   try {
      ic.lookup("/queue/" + queueName);
      ProxyAssertSupport.fail("The temporary queue should not be bound to JNDI");
   } catch (NamingException e) {
      // Expected
   }
}

/**
 * https://jira.jboss.org/jira/browse/JBMESSAGING-1566
 */
@Test
public void testCanNotCreateConsumerFromAnotherConnectionForTemporaryQueue() throws Exception {
   Connection conn = createConnection();

   Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

   TemporaryQueue tempQueue = sess.createTemporaryQueue();

   Connection anotherConn = createConnection();

   Session sessFromAnotherConn = anotherConn.createSession(false, Session.AUTO_ACKNOWLEDGE);

   try {
      sessFromAnotherConn.createConsumer(tempQueue);
      ProxyAssertSupport.fail("Only temporary destination's own connection is allowed to create MessageConsumers for them.");
   } catch (JMSException e) {
   }

   conn.close();
   anotherConn.close();
}

private static byte destinationType(Destination destination) {
   if (destination instanceof Queue) {
      if (destination instanceof TemporaryQueue) {
         return TEMP_QUEUE_TYPE;
      } else {
         return QUEUE_TYPE;
      }
   } else if (destination instanceof Topic) {
      if (destination instanceof TemporaryTopic) {
         return TEMP_TOPIC_TYPE;
      } else {
         return TOPIC_TYPE;
      }
   }

   throw new IllegalArgumentException("Unknown Destination Type passed to JMS Transformer.");
}

@Override
public TemporaryQueue createTemporaryQueue() throws JMSException {
   // As per spec. section 4.11
   if (sessionType == ActiveMQSession.TYPE_TOPIC_SESSION) {
      throw new IllegalStateException("Cannot create a temporary queue using a TopicSession");
   }

   try {
      ActiveMQTemporaryQueue queue = ActiveMQDestination.createTemporaryQueue(this);

      SimpleString simpleAddress = queue.getSimpleAddress();

      session.createTemporaryQueue(simpleAddress, RoutingType.ANYCAST, simpleAddress);

      connection.addTemporaryQueue(simpleAddress);

      return queue;
   } catch (ActiveMQException e) {
      throw JMSExceptionHelper.convertFromActiveMQException(e);
   }
}

public void testTemporaryQueueLifecycle() throws Exception
{
    Session session;
    TextMessage msg;
    MessageProducer producer;
    MessageConsumer consumer;

    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryQueue tempQueue = session.createTemporaryQueue();

    msg = session.createTextMessage("foobar");
    producer = session.createProducer(tempQueue);
    producer.send(msg, DeliveryMode.NON_PERSISTENT, 0, 0);
    producer.close();

    consumer = session.createConsumer(tempQueue);
    connection.start();
    msg = (TextMessage) consumer.receive(RECV_TIMEOUT);
    assertNotNull(msg);
    assertEquals("foobar", msg.getText());

    tempQueue.delete();
    session.close();
}

@Override
public TemporaryQueue createTemporaryQueue() throws JMSException
   {    
    externalAccessLock.readLock().lock();
    try
    {
        checkNotClosed();
        String queueName = "TEMP-QUEUE-"+UUIDProvider.getInstance().getShortUUID();
        engine.createTemporaryQueue(queueName);
        connection.registerTemporaryQueue(queueName);

        return new TemporaryQueueRef(connection,queueName);
    }
    finally
    {
        externalAccessLock.readLock().unlock();
    }
   }

@Override
public TemporaryQueue createTemporaryQueue() throws JMSException
   {
    externalAccessLock.readLock().lock();
    try
    {
        checkNotClosed();

        CreateTemporaryQueueQuery query = new CreateTemporaryQueueQuery();
        query.setSessionId(id);
        CreateTemporaryQueueResponse response = (CreateTemporaryQueueResponse)transportEndpoint.blockingRequest(query);

        return new TemporaryQueueRef(connection,response.getQueueName());
    }
    finally
    {
        externalAccessLock.readLock().unlock();
    }
   }

public TemporaryQueue createTemporaryQueue() throws JMSException
{
    checkNotClosed();
    try
    {
        AMQTemporaryQueue result = new AMQTemporaryQueue(this);

        // this is done so that we can produce to a temporary queue before we create a consumer
        result.setQueueName(result.getRoutingKey());
        createQueue(result.getAMQQueueName(), result.isAutoDelete(),
                    result.isDurable(), result.isExclusive());
        bindQueue(result.getAMQQueueName(), result.getRoutingKey(),
                new FieldTable(), result.getExchangeName(), result);
        return result;
    }
    catch (Exception e)
    {
       JMSException jmse = new JMSException("Cannot create temporary queue");
       jmse.setLinkedException(e);
       jmse.initCause(e);
       throw jmse;
    }
}

/**
 * Tests that a connection with a 'prefixes' set on its does not alter the
 * address for a temporary queue in the to/reply-to fields for incoming messages.
 *
 * @throws Exception if an error occurs during the test.
 */
@Test(timeout = 20000)
public void testReceivedMessageWithTemporaryQueueDestinationsOnConnectionWithPrefixes() throws Exception {
    Class<? extends Destination> destType = TemporaryQueue.class;
    String destPrefix = "q12321-";
    String destName = "temp-queue://myTempQueue";
    String replyName = "temp-queue://myReplyTempQueue";
    String destAddress = destName; // We won't manipulate the temporary addresses generated by the broker
    String replyAddress = replyName; // We won't manipulate the temporary addresses generated by the broker
    String annotationName = AmqpDestinationHelper.JMS_DEST_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
    Byte annotationValue = AmqpDestinationHelper.TEMP_QUEUE_TYPE;
    String replyAnnotationName = AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
    Byte replyAnnotationValue = AmqpDestinationHelper.TEMP_QUEUE_TYPE;

    doReceivedMessageOnConnectionWithPrefixTestImpl(destType, destPrefix, destName, replyName,
                                                    destAddress, replyAddress, annotationName,
                                                    annotationValue, replyAnnotationName, replyAnnotationValue);
}

@Test(timeout = 20000)
public void testCreateTemporaryQueue() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        Connection connection = testFixture.establishConnecton(testPeer);
        connection.start();

        testPeer.expectBegin();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        String dynamicAddress = "myTempQueueAddress";
        testPeer.expectTempQueueCreationAttach(dynamicAddress);

        TemporaryQueue tempQueue = session.createTemporaryQueue();
        assertNotNull("TemporaryQueue object was null", tempQueue);
        assertNotNull("TemporaryQueue queue name was null", tempQueue.getQueueName());
        assertEquals("TemporaryQueue name not as expected", dynamicAddress, tempQueue.getQueueName());

        testPeer.expectClose();
        connection.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}

@Test(timeout = 20000)
public void testCreateAndDeleteTemporaryQueue() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        Connection connection = testFixture.establishConnecton(testPeer);
        connection.start();

        testPeer.expectBegin();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        String dynamicAddress = "myTempQueueAddress";
        testPeer.expectTempQueueCreationAttach(dynamicAddress);
        TemporaryQueue tempQueue = session.createTemporaryQueue();

        // Deleting the TemporaryQueue will be achieved by closing its creating link.
        testPeer.expectDetach(true, true, true);
        tempQueue.delete();

        testPeer.expectClose();
        connection.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}

@Test(timeout=30000)
public void testDeleteOfTempQueueOnClosedConnection() throws JMSException, IOException {
    connection = new JmsConnection(connectionInfo, provider);
    connection.start();

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryQueue tempQueue = session.createTemporaryQueue();
    assertNotNull(tempQueue);

    connection.close();
    try {
        tempQueue.delete();
        fail("Should have thrown an IllegalStateException");
    } catch (IllegalStateException ex) {
    }
}

@Test(timeout = 10000)
public void testCannotCreateConsumerOnDeletedTemporaryDestination() throws JMSException {
    JmsSession session = (JmsSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryQueue tempQueue = session.createTemporaryQueue();
    MessageProducer producer = session.createProducer(tempQueue);

    try {
        producer.send(session.createMessage());
    } catch (Exception ex) {
        fail("Should be able to send to this temporary destination");
    }

    tempQueue.delete();

    try {
        producer.send(session.createMessage());
        fail("Should not be able to send to this temporary destination");
    } catch (IllegalStateException ise) {}
}

@Test(timeout=30000)
public void testDeleteTemporaryQueue() throws Exception {
    connection = createAmqpConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Queue queue = session.createTemporaryQueue();
    assertNotNull(queue);
    assertTrue(queue instanceof TemporaryQueue);

    final BrokerViewMBean broker = getProxyToBroker();
    assertEquals(1, broker.getTemporaryQueues().length);

    TemporaryQueue tempQueue = (TemporaryQueue) queue;
    tempQueue.delete();

    assertTrue("Temp Queue should be deleted.", Wait.waitFor(new Wait.Condition() {

        @Override
        public boolean isSatisified() throws Exception {
            return broker.getTemporaryQueues().length == 0;
        }
    }, TimeUnit.SECONDS.toMillis(30), TimeUnit.MILLISECONDS.toMillis(50)));
}

@Test(timeout = 60000)
public void testCantConsumeFromTemporaryQueueCreatedOnAnotherConnection() throws Exception {
    connection = createAmqpConnection();
    connection.start();

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryQueue tempQueue = session.createTemporaryQueue();
    session.createConsumer(tempQueue);

    Connection connection2 = createAmqpConnection();
    try {
        Session session2 = connection2.createSession(false, Session.AUTO_ACKNOWLEDGE);
        try {
            session2.createConsumer(tempQueue);
            fail("should not be able to consumer from temporary queue from another connection");
        } catch (InvalidDestinationException ide) {
            // expected
        }
    } finally {
        connection2.close();
    }
}

@Test(timeout = 60000)
public void testCantDeleteTemporaryQueueWithConsumers() throws Exception {
    connection = createAmqpConnection();
    connection.start();

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryQueue tempQueue = session.createTemporaryQueue();
    MessageConsumer consumer = session.createConsumer(tempQueue);

    try {
        tempQueue.delete();
        fail("should not be able to delete temporary queue with active consumers");
    } catch (IllegalStateException ide) {
        // expected
    }

    consumer.close();

    // Now it should be allowed
    tempQueue.delete();
}

/**
 * Creates a an available JMS message from another provider.
 *
 * @param destination
 *        - Destination to be converted into Jms's implementation.
 * @return JmsDestination - Jms's implementation of the
 *         destination.
 * @throws JMSException
 * @throws JMSException
 *         if an error occurs
 */
public static JmsDestination transformDestination(JmsConnection connection, Destination destination) throws JMSException {
    JmsDestination result = null;

    if (destination != null) {
        if (destination instanceof JmsDestination) {
            return (JmsDestination) destination;

        } else {
            if (destination instanceof TemporaryQueue) {
                result = new JmsTemporaryQueue(((TemporaryQueue) destination).getQueueName());
            } else if (destination instanceof TemporaryTopic) {
                result = new JmsTemporaryTopic(((TemporaryTopic) destination).getTopicName());
            } else if (destination instanceof Queue) {
                result = new JmsQueue(((Queue) destination).getQueueName());
            } else if (destination instanceof Topic) {
                result = new JmsTopic(((Topic) destination).getTopicName());
            }
        }
    }

    return result;
}

@Override
public <T extends Destination> T createDestination(String name, Class<T> kind) {
    if (kind == Queue.class) {
        return kind.cast(new JmsQueue(name));
    }
    if (kind == Topic.class) {
        return kind.cast(new JmsTopic(name));
    }
    if (kind == TemporaryQueue.class) {
        return kind.cast(new JmsTemporaryQueue(name));
    }
    if (kind == TemporaryTopic.class) {
        return kind.cast(new JmsTemporaryTopic(name));
    }

    return kind.cast(new JmsQueue(name));
}

public void run() throws Exception {
    TemporaryQueue responseQ = session.createTemporaryQueue();
    MessageProducer requester = session.createProducer(destination);
    MessageConsumer responseListener = session.createConsumer(responseQ);
    responseListener.setMessageListener(this);

    for (int i = 0; i < NUM_REQUESTS; i++) {
        TextMessage request = session.createTextMessage("Job Request");
        request.setJMSReplyTo(responseQ);
        request.setJMSCorrelationID("request: " + i);
        System.out.println(request.getText());
        requester.send(request);
    }

    if (done.await(10, TimeUnit.MINUTES)) {
        System.out.println("Woohoo! Work's all done!");
    } else {
        System.out.println("Doh!! Work didn't get done.");
    }
}

public TemporaryQueue createTemporaryQueue() throws JMSException {
    if (info.getType() == JmsConnectionFactory.TOPIC) {
        throw new IllegalStateException("Cannot create temporary queue for javax.jms.TopicSession");

    }

    lock();
    try {
        Session session = getSession();
        if (trace)
            log.trace("createTemporaryQueue " + session);
        TemporaryQueue temp = session.createTemporaryQueue();
        if (trace)
            log.trace("createdTemporaryQueue " + session + " temp=" + temp);
        sf.addTemporaryQueue(temp);
        return temp;
    } finally {
        unlock();
    }
}

@Override
public TemporaryQueue createTemporaryQueue() throws JMSException {
    TemporaryQueue result;

    result = getInternalSession().createTemporaryQueue();

    // Notify all of the listeners of the created temporary Queue.
    for (JmsPoolSessionEventListener listener : this.sessionEventListeners) {
        listener.onTemporaryQueueCreate(result);
    }

    return result;
}

@Override
public TemporaryQueue createTemporaryQueue() {
    try {
        return getSession().createTemporaryQueue();
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

protected TemporaryQueue createTemporaryQueue() throws JMSException {
    String destinationName = connectionId.toString() + ":" + tempDestIdGenerator.incrementAndGet();
    MockJMSTemporaryQueue queue = new MockJMSTemporaryQueue(destinationName);
    signalCreateTemporaryDestination(queue);
    tempDestinations.put(queue, queue);
    queue.setConnection(this);
    stats.temporaryDestinationCreated(queue);
    return queue;
}