Java 类javax.jms.TemporaryTopic 实例源码

@Before
public void setUp() throws Exception {
  embeddedBroker.start();
  out = new SimpleEventBus();
  in = new SimpleEventBus();

  ActiveMQConnectionFactory connectionFactory = embeddedBroker.createConnectionFactory();
  topicConnection = connectionFactory.createTopicConnection();
  topicConnection.start();
  TopicSession topicSession = topicConnection.createTopicSession(
      true, Session.SESSION_TRANSACTED);
  TemporaryTopic topic = topicSession.createTemporaryTopic();
  consumer = topicSession.createConsumer(topic);

  publisher = new JmsPublisher(out);
  publisher.setTopic(topic);
  publisher.setConnectionFactory(connectionFactory);
  publisher.postConstruct();
  publisher.start();

  jmsMessageSource = new JmsMessageSource(this.consumer,
      new DefaultJmsMessageConverter(new XStreamSerializer()));
  jmsMessageSource.subscribe(in::publish);
}

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);
}

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

   Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   TemporaryTopic topic = session.createTemporaryTopic();

   System.out.println("topic:" + topic.getTopicName());
   MessageConsumer consumer = session.createConsumer(topic);
   MessageProducer producer = session.createProducer(topic);

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

   connection.start();

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

   assertNotNull(message);
}

public void testLoadTempAdvisoryTopics() throws Exception {
   for (int i = 0; i < MESSAGE_COUNT; i++) {
      TemporaryTopic tempTopic = session.createTemporaryTopic();
      MessageConsumer consumer = session.createConsumer(tempTopic);
      MessageProducer producer = session.createProducer(tempTopic);
      consumer.close();
      producer.close();
      tempTopic.delete();
   }

   AdvisoryBroker ab = (AdvisoryBroker) broker.getBroker().getAdaptor(AdvisoryBroker.class);
   assertTrue(ab.getAdvisoryDestinations().size() == 0);
   assertTrue(ab.getAdvisoryConsumers().size() == 0);
   assertTrue(ab.getAdvisoryProducers().size() == 0);
   RegionBroker rb = (RegionBroker) broker.getBroker().getAdaptor(RegionBroker.class);

   for (Destination dest : rb.getDestinationMap().values()) {
      LOG.debug("Destination: {}", dest);
   }

   // there should be at least 2 destinations - advisories -
   // 1 for the connection + 1 generic ones
   assertTrue("Should be at least 2 destinations", rb.getDestinationMap().size() > 2);
}

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

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

   TemporaryTopic tempTopic = producerSession.createTemporaryTopic();
   String topicName = tempTopic.getTopicName();

   try {
      ic.lookup("/topic/" + topicName);
      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 testCanNotCreateConsumerFromAnotherCnnectionForTemporaryTopic() throws Exception {
   Connection conn = createConnection();

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

   TemporaryTopic tempTopic = sess.createTemporaryTopic();

   Connection anotherConn = createConnection();

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

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

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 TemporaryTopic createTemporaryTopic() throws JMSException {
   // As per spec. section 4.11
   if (sessionType == ActiveMQSession.TYPE_QUEUE_SESSION) {
      throw new IllegalStateException("Cannot create a temporary topic on a QueueSession");
   }

   try {
      ActiveMQTemporaryTopic topic = ActiveMQDestination.createTemporaryTopic(this);

      SimpleString simpleAddress = topic.getSimpleAddress();

      // We create a dummy subscription on the topic, that never receives messages - this is so we can perform JMS
      // checks when routing messages to a topic that
      // does not exist - otherwise we would not be able to distinguish from a non existent topic and one with no
      // subscriptions - core has no notion of a topic

      session.createTemporaryQueue(simpleAddress, simpleAddress, ActiveMQSession.REJECTING_FILTER);

      connection.addTemporaryQueue(simpleAddress);

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

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

    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryTopic tempTopic = session.createTemporaryTopic();

    consumer = session.createConsumer(tempTopic);
    connection.start();

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

    msg = (TextMessage) consumer.receive(RECV_TIMEOUT);
    assertNotNull(msg);
    assertEquals("foobar", msg.getText());

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

@Override
public TemporaryTopic createTemporaryTopic() throws JMSException
   {
    externalAccessLock.readLock().lock();
    try
    {
        checkNotClosed();
        String topicName = "TEMP-TOPIC-"+UUIDProvider.getInstance().getShortUUID();
        engine.createTemporaryTopic(topicName);
        connection.registerTemporaryTopic(topicName);

        return new TemporaryTopicRef(connection,topicName);
    }
    finally
    {
        externalAccessLock.readLock().unlock();
    }
   }

@Override
public TemporaryTopic createTemporaryTopic() throws JMSException
   {
    externalAccessLock.readLock().lock();
    try
    {
        checkNotClosed();

        CreateTemporaryTopicQuery query = new CreateTemporaryTopicQuery();
        query.setSessionId(id);
        CreateTemporaryTopicResponse response = 
            (CreateTemporaryTopicResponse)transportEndpoint.blockingRequest(query);

        return new TemporaryTopicRef(connection,response.getTopicName());
    }
    finally
    {
        externalAccessLock.readLock().unlock();
    }
   }

public void testSendingSameMessage() throws Exception
{
    AMQConnection conn = (AMQConnection) getConnection("guest", "guest");
    TopicSession session = conn.createTopicSession(true, Session.AUTO_ACKNOWLEDGE);
    TemporaryTopic topic = session.createTemporaryTopic();
    assertNotNull(topic);
    TopicPublisher producer = session.createPublisher(topic);
    MessageConsumer consumer = session.createConsumer(topic);
    conn.start();
    TextMessage sentMessage = session.createTextMessage("Test Message");
    producer.send(sentMessage);
    session.commit();
    TextMessage receivedMessage = (TextMessage) consumer.receive(2000);
    assertNotNull(receivedMessage);
    assertEquals(sentMessage.getText(), receivedMessage.getText());
    producer.send(sentMessage);
    session.commit();
    receivedMessage = (TextMessage) consumer.receive(2000);
    assertNotNull(receivedMessage);
    assertEquals(sentMessage.getText(), receivedMessage.getText());
    session.commit();
    conn.close();

}

/**
 * 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 testReceivedMessageWithTemporaryTopicDestinationsOnConnectionWithPrefixes() throws Exception {
    Class<? extends Destination> destType = TemporaryTopic.class;
    String destPrefix = "q12321-";
    String destName = "temp-topic://myTempTopic";
    String replyName = "temp-topic://myReplyTempTopic";
    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_TOPIC_TYPE;
    String replyAnnotationName = AmqpDestinationHelper.JMS_REPLY_TO_TYPE_MSG_ANNOTATION_SYMBOL_NAME;
    Byte replyAnnotationValue = AmqpDestinationHelper.TEMP_TOPIC_TYPE;

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

@Test(timeout = 20000)
public void testCreateTemporaryTopic() 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 = "myTempTopicAddress";
        testPeer.expectTempTopicCreationAttach(dynamicAddress);

        TemporaryTopic tempTopic = session.createTemporaryTopic();
        assertNotNull("TemporaryTopic object was null", tempTopic);
        assertNotNull("TemporaryTopic name was null", tempTopic.getTopicName());
        assertEquals("TemporaryTopic name not as expected", dynamicAddress, tempTopic.getTopicName());

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

        testPeer.waitForAllHandlersToComplete(1000);
    }
}

@Test(timeout = 20000)
public void testCreateAndDeleteTemporaryTopic() 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 = "myTempTopicAddress";
        testPeer.expectTempTopicCreationAttach(dynamicAddress);
        TemporaryTopic tempTopic = session.createTemporaryTopic();

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

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

        testPeer.waitForAllHandlersToComplete(1000);
    }
}

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

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryTopic tempTopic = session.createTemporaryTopic();
    assertNotNull(tempTopic);

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

@Test(timeout=30000)
public void testDeleteTemporaryTopic() throws Exception {
    connection = createAmqpConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTemporaryTopic();
    assertNotNull(topic);
    assertTrue(topic instanceof TemporaryTopic);

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

    TemporaryTopic tempTopic = (TemporaryTopic) topic;
    tempTopic.delete();

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

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

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

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryTopic tempTopic = session.createTemporaryTopic();
    session.createConsumer(tempTopic);

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

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

    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryTopic tempTopic = session.createTemporaryTopic();
    MessageConsumer consumer = session.createConsumer(tempTopic);

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

    consumer.close();

    // Now it should be allowed
    tempTopic.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 TemporaryTopic createTemporaryTopic() throws JMSException {
    if (info.getType() == JmsConnectionFactory.QUEUE) {
        throw new IllegalStateException("Cannot create temporary topic for javax.jms.QueueSession");
    }

    lock();
    try {
        Session session = getSession();
        if (trace)
            log.trace("createTemporaryTopic " + session);
        TemporaryTopic temp = session.createTemporaryTopic();
        if (trace)
            log.trace("createdTemporaryTopic " + session + " temp=" + temp);
        sf.addTemporaryTopic(temp);
        return temp;
    } finally {
        unlock();
    }
}

@Override
public TemporaryTopic createTemporaryTopic() throws JMSException {
    TemporaryTopic result;

    result = getInternalSession().createTemporaryTopic();

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

    return result;
}

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

protected TemporaryTopic createTemporaryTopic() throws JMSException {
    String destinationName = connectionId.toString() + ":" + tempDestIdGenerator.incrementAndGet();
    MockJMSTemporaryTopic topic = new MockJMSTemporaryTopic(destinationName);
    signalCreateTemporaryDestination(topic);
    tempDestinations.put(topic, topic);
    topic.setConnection(this);
    stats.temporaryDestinationCreated(topic);
    return topic;
}

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

@Test(timeout = 60000)
public void testCreateTemporaryTopic() throws Exception {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    TemporaryTopic topic = session.createTemporaryTopic();
    assertNotNull(topic);
    assertTrue(topic instanceof MockJMSTemporaryTopic);
}

@Override
public TemporaryTopic createTemporaryTopic() {
    try {
        return getSession().createTemporaryTopic();
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

/**
 * 
 * @param d
 * @return
 * @throws JMSException
 */
public String fromDestination(Destination d) throws JMSException {
    if (d == null)  return null;

    if (d instanceof TemporaryTopic || d instanceof TemporaryQueue) {
        throw new IllegalArgumentException("Temporary destinations are not supported! [" + d + "]");
    }

    if (d instanceof Topic) {
        final Topic topic = (Topic) d;
        return new StringBuilder("topic/").append(topic.getTopicName()).toString();
    }
    final Queue queue = (Queue) d;
    return new StringBuilder("queue/").append(queue.getQueueName()).toString();
}

/**
 * Closes the {@code Requestor}.
 *  
 * @throws JMSException if the JMS provider fails to close the {@code Requestor} due to some internal error.
 */
@Override
public void close() throws JMSException {
    this.consumer.close();
    if (this.tempDest instanceof TemporaryTopic) {
        ((TemporaryTopic) this.tempDest).delete();
    } else {
        ((TemporaryQueue) this.tempDest).delete();
    }
}

@Test
public void topic() throws JMSException {
    final Destination dest = Mockito.mock(Topic.class);
    final TemporaryTopic tempDest = mock(TemporaryTopic.class);
    when(this.context.createTemporaryTopic()).thenReturn(tempDest);

    test(dest, tempDest);

    verify(this.context).createTemporaryTopic();
    verify(tempDest).delete();
    verifyNoMoreInteractions(dest);
}

/**
 * Sends the specified text to the listener queue and waits for a response (until timeout
 * reached). Expects a TextMessage reply back from the listener. The listener should
 * respond to the Message replyTo.
 */
@SuppressWarnings("unchecked")
@Override
public String sendRequestToQueue(final String text, final String jmsListenerQueue, final long timeout) {
  if (text == null) {
    throw new NullPointerException("Attempting to send a null text message.");
  }
  return jmsTemplate.execute(session -> {
    String returnString = null;
    TemporaryTopic replyTopic = session.createTemporaryTopic();
    TextMessage textMessage = session.createTextMessage();
    textMessage.setText(text);
    textMessage.setJMSReplyTo(replyTopic);

    MessageConsumer consumer = session.createConsumer(replyTopic);

    Destination requestDestination = new ActiveMQQueue(jmsListenerQueue);
    MessageProducer messageProducer = session.createProducer(requestDestination);
    messageProducer.send(textMessage);

    // Wait for reply
    Message replyMessage = consumer.receive(timeout);
    if (replyMessage != null) {
      if (replyMessage instanceof TextMessage) {
        returnString = ((TextMessage) replyMessage).getText();
      } else {
        LOGGER.warn("Non-text message received as JMS reply from ActiveMQ - unable to process");
      }
    }
    return returnString;
  }, true);
}

/**
 * Lazily loads the temporary topic type if one has not been explicitly configured
 * via calling the {@link #setTemporaryTopicType(Class)}
 */
public Class<? extends TemporaryTopic> getTemporaryTopicType(JmsOperations template) {
    Class<? extends TemporaryTopic> answer = getTemporaryTopicType();
    if (answer == null) {
        loadTemporaryDestinationTypes(template);
        answer = getTemporaryTopicType();
    }
    return answer;
}

/**
 * Returns a new JMS endpoint for the given JMS destination
 */
public static JmsEndpoint newInstance(Destination destination) throws JMSException {
    if (destination instanceof TemporaryQueue) {
        return new JmsTemporaryQueueEndpoint((TemporaryQueue) destination);
    } else if (destination instanceof TemporaryTopic) {
        return new JmsTemporaryTopicEndpoint((TemporaryTopic) destination);
    } else if (destination instanceof Queue) {
        return new JmsQueueEndpoint((Queue) destination);
    } else {
        return new JmsEndpoint((Topic) destination);
    }
}

public <T extends Destination> T createDestination(String name, Class<T> kind) {
    if( kind == Queue.class ) {
        return kind.cast(new ActiveMQQueue(name));
    }
    if( kind == Topic.class ) {
        return kind.cast(new ActiveMQTopic(name));
    }
    if( kind == TemporaryQueue.class ) {
        return kind.cast(new ActiveMQTempQueue(name));
    }
    if( kind == TemporaryTopic.class ) {
        return kind.cast(new ActiveMQTempTopic(name));
    }
    return kind.cast(ActiveMQDestination.createDestination(name, ActiveMQDestination.QUEUE_TYPE));
}

public static ActiveMQDestination transform(Destination dest) throws JMSException {
    if (dest == null) {
        return null;
    }
    if (dest instanceof ActiveMQDestination) {
        return (ActiveMQDestination)dest;
    }

    if (dest instanceof Queue && dest instanceof Topic) {
        String queueName = ((Queue) dest).getQueueName();
        String topicName = ((Topic) dest).getTopicName();
        if (queueName != null && topicName == null) {
            return new ActiveMQQueue(queueName);
        } else if (queueName == null && topicName != null) {
            return new ActiveMQTopic(topicName);
        } else {
            return unresolvableDestinationTransformer.transform(dest);
        }
    }
    if (dest instanceof TemporaryQueue) {
        return new ActiveMQTempQueue(((TemporaryQueue)dest).getQueueName());
    }
    if (dest instanceof TemporaryTopic) {
        return new ActiveMQTempTopic(((TemporaryTopic)dest).getTopicName());
    }
    if (dest instanceof Queue) {
        return new ActiveMQQueue(((Queue)dest).getQueueName());
    }
    if (dest instanceof Topic) {
        return new ActiveMQTopic(((Topic)dest).getTopicName());
    }
    throw new JMSException("Could not transform the destination into a ActiveMQ destination: " + dest);
}

@Test
public void testTempTopicDelete() throws Exception {
   connection.start();
   TopicSession topicSession = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

   TemporaryTopic tempTopic = topicSession.createTemporaryTopic();

   ActiveMQConnection newConn = (ActiveMQConnection) factory.createConnection();

   try {
      TopicSession newTopicSession = newConn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
      TopicPublisher publisher = newTopicSession.createPublisher(tempTopic);

      TextMessage msg = newTopicSession.createTextMessage("Test Message");

      publisher.publish(msg);

      try {
         TopicSubscriber consumer = newTopicSession.createSubscriber(tempTopic);
         fail("should have gotten exception but got consumer: " + consumer);
      } catch (JMSException ex) {
         //correct
      }

      connection.close();

      try {
         Message newMsg = newTopicSession.createMessage();
         publisher.publish(newMsg);
      } catch (JMSException e) {
         //ok
      }

   } finally {
      newConn.close();
   }
}

@Test
public void testTempTopicLeak() throws Exception {
   Connection connection = null;

   try {
      connection = factory.createConnection();
      connection.start();

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

      TemporaryTopic temporaryTopic = session.createTemporaryTopic();
      temporaryTopic.delete();

      Object[] queueResources = server.getManagementService().getResources(QueueControl.class);

      for (Object queueResource : queueResources) {

         if (((QueueControl) queueResource).getAddress().equals("ActiveMQ.Advisory.TempTopic")) {
            QueueControl queueControl = (QueueControl) queueResource;
            Wait.waitFor(() -> queueControl.getMessageCount() == 0);
            assertNotNull("addressControl for temp advisory", queueControl);

            Wait.assertEquals(0, queueControl::getMessageCount);
            Wait.assertEquals(2, queueControl::getMessagesAdded);
         }
      }
   } finally {
      if (connection != null) {
         connection.close();
      }
   }
}

protected void deleteTemporaryDestination(Destination dest) throws JMSException {
   if (topic) {
      ((TemporaryTopic) dest).delete();
   } else {
      System.out.println("Deleting: " + dest);
      ((TemporaryQueue) dest).delete();
   }
}