Java 类javax.jms.Topic 实例源码

public MessageConsumer createTopicConsumer(String selector) throws JMSException {
    if (isQueue) {
        throw new IllegalArgumentException("Only for topic, not queue");
    }
    String consumerId = "consumer-" + UUID.randomUUID();
    topicConnection = startConnection(consumerId);
    Session session = topicConnection.createSession(isTransacted, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTopic(destinationName);
    if (isDurable) {
        if (selector != null) {
            return session.createDurableSubscriber(topic, consumerId, selector, true);
        } else {
            return session.createDurableSubscriber(topic, consumerId);
        }
    } else {
        if (selector != null) {
            return session.createConsumer(topic, selector);
        } else {
            return session.createConsumer(topic);
        }
    }
}

@Test
public void validateJmsHeadersAndPropertiesAreTransferredFromFFAttributes() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    Map<String, String> flowFileAttributes = new HashMap<>();
    flowFileAttributes.put("foo", "foo");
    flowFileAttributes.put(JmsHeaders.REPLY_TO, "myTopic");
    publisher.publish(destinationName, "hellomq".getBytes(), flowFileAttributes);

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    assertEquals("foo", receivedMessage.getStringProperty("foo"));
    assertTrue(receivedMessage.getJMSReplyTo() instanceof Topic);
    assertEquals("myTopic", ((Topic) receivedMessage.getJMSReplyTo()).getTopicName());

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

@Before
public void setUp() throws Exception {
  eventBus = new SimpleEventBus();
  cut = new JmsPublisher(eventBus);
  connectionFactory = mock(TopicConnectionFactory.class);
  publisher = mock(TopicPublisher.class);
  topic = mock(Topic.class);
  converter = mock(JmsMessageConverter.class);
  cut.setConnectionFactory(connectionFactory);
  cut.setTopic(topic);
  cut.setTransacted(true);
  cut.setMessageConverter(converter);
  cut.setPersistent(false);
  cut.postConstruct();
  cut.start();
}

/**
 * Create a topic
 *
 * @param topicName the topic name
 */
@PublicAtsApi
public void createTopic(

                         final String topicName ) {

    try {
        final Session session = loadSession(false, Session.AUTO_ACKNOWLEDGE);
        final Topic topic = session.createTopic(topicName);
        session.createConsumer(topic).close();
    } catch (JMSException e) {
        throw new JmsMessageException("Could not start listening for messages on topic " + topicName,
                                      e);
    } finally {
        releaseSession(false);
    }
}

@Test
public void testGetNoLocal() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createTopicConnection();
    TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTemporaryTopic();
    TopicSubscriber subscriber = session.createDurableSubscriber(topic, "name", "color = red", true);

    assertTrue(subscriber.getNoLocal());

    subscriber.close();

    try {
        subscriber.getNoLocal();
        fail("Cannot read state on closed subscriber");
    } catch (IllegalStateException ise) {}
}

@Test
public void testGetTopic() throws JMSException {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createTopicConnection();
    TopicSession session = connection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTemporaryTopic();
    TopicPublisher publisher = session.createPublisher(topic);

    assertNotNull(publisher.getTopic());
    assertSame(topic, publisher.getTopic());

    publisher.close();

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

@Test
public void validateJmsHeadersAndPropertiesAreTransferredFromFFAttributesOverJNDI() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsJndiTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    Map<String, String> flowFileAttributes = new HashMap<>();
    flowFileAttributes.put("foo", "foo");
    flowFileAttributes.put(JmsHeaders.REPLY_TO, "myTopic");
    publisher.publish(destinationName, "hellomq".getBytes(), flowFileAttributes);

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    assertEquals("foo", receivedMessage.getStringProperty("foo"));
    assertTrue(receivedMessage.getJMSReplyTo() instanceof Topic);
    assertEquals("myTopic", ((Topic) receivedMessage.getJMSReplyTo()).getTopicName());

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

@Test
public void validateJmsHeadersAndPropertiesAreTransferredFromFFAttributes() throws Exception {
    final String destinationName = "testQueue";
    JmsTemplate jmsTemplate = CommonTest.buildJmsTemplateForDestination(false);

    JMSPublisher publisher = new JMSPublisher(jmsTemplate, mock(ComponentLog.class));
    Map<String, String> flowFileAttributes = new HashMap<>();
    flowFileAttributes.put("foo", "foo");
    flowFileAttributes.put(JmsHeaders.REPLY_TO, "myTopic");
    publisher.publish(destinationName, "hellomq".getBytes(), flowFileAttributes);

    Message receivedMessage = jmsTemplate.receive(destinationName);
    assertTrue(receivedMessage instanceof BytesMessage);
    assertEquals("foo", receivedMessage.getStringProperty("foo"));
    assertTrue(receivedMessage.getJMSReplyTo() instanceof Topic);
    assertEquals("myTopic", ((Topic) receivedMessage.getJMSReplyTo()).getTopicName());

    ((CachingConnectionFactory) jmsTemplate.getConnectionFactory()).destroy();
}

@Override
public MessageConsumer createDurableConsumer(
        final Topic topic,
        final String name,
        final String messageSelector,
        final boolean noLocal
) throws JMSException {
    return notImplemented();
}

@Override
public TopicSubscriber createDurableSubscriber(
        final Topic topic,
        final String name,
        final String messageSelector,
        final boolean noLocal
) throws JMSException {
    return notImplemented();
}

private TopicPublisher getTopicPublisher(Topic destination) throws JMSException {
    TopicPublisher result = null;

    if (useAnonymousProducers) {
        result = safeGetSessionHolder().getOrCreatePublisher();
    } else {
        result = ((TopicSession) getInternalSession()).createPublisher(destination);
    }

    return result;
}

@Override
public JMSConsumer createSharedDurableConsumer(Topic topic, String name) {
    try {
        return startIfNeeded(new JMSConsumerImpl(this, getSession().createSharedDurableConsumer(topic, name)));
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createDurableConsumer(Topic topic, String name, String selector, boolean noLocal) {
    try {
        return startIfNeeded(new JmsPoolJMSConsumer((JmsPoolMessageConsumer) getSession().createDurableConsumer(topic, name, selector, noLocal)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createDurableConsumer(Topic topic, String name, String selector, boolean noLocal) {
    try {
        return startIfNeeded(new JMSConsumerImpl(this, getSession().createDurableConsumer(topic, name, selector, noLocal)));
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

@Override
public TopicSubscriber createSubscriber(
                                         Topic topic,
                                         String messageSelector,
                                         boolean noLocal ) throws JMSException {

    return addConsumer( ((TopicSession) session).createSubscriber(topic, messageSelector, noLocal));
}

@Before
public void setup() throws Exception {
  beforeSetup();
  connectionFactory = mock(ConnectionFactory.class);
  connection = mock(Connection.class);
  session = mock(Session.class);
  queue = mock(Queue.class);
  topic = mock(Topic.class);
  messageConsumer = mock(MessageConsumer.class);
  message = mock(TextMessage.class);
  when(message.getPropertyNames()).thenReturn(new Enumeration<Object>() {
    @Override
    public boolean hasMoreElements() {
      return false;
    }
    @Override
    public Object nextElement() {
      throw new UnsupportedOperationException();
    }
  });
  when(message.getText()).thenReturn(TEXT);
  when(connectionFactory.createConnection(USERNAME, PASSWORD)).thenReturn(connection);
  when(connection.createSession(true, Session.SESSION_TRANSACTED)).thenReturn(session);
  when(session.createQueue(destinationName)).thenReturn(queue);
  when(session.createConsumer(any(Destination.class), anyString())).thenReturn(messageConsumer);
  when(messageConsumer.receiveNoWait()).thenReturn(message);
  when(messageConsumer.receive(anyLong())).thenReturn(message);
  destinationName = DESTINATION_NAME;
  destinationType = JMSDestinationType.QUEUE;
  destinationLocator = JMSDestinationLocator.CDI;
  messageSelector = SELECTOR;
  batchSize = 10;
  pollTimeout = 500L;
  context = new Context();
  converter = new DefaultJMSMessageConverter.Builder().build(context);
  event = converter.convert(message).iterator().next();
  userName = Optional.of(USERNAME);
  password = Optional.of(PASSWORD);
  afterSetup();
}

@Override
public MessageConsumer createConsumer(Destination destination, String messageSelector) throws JMSException {
    if (destination instanceof Topic) {
        throw new IllegalStateException("Operation not supported by a QueueSession");
    }
    return super.createConsumer(destination, messageSelector);
}

@Override
public MessageProducer createProducer(Destination destination) throws JMSException {
    if (destination instanceof Topic) {
        throw new IllegalStateException("Operation not supported by a QueueSession");
    }
    return super.createProducer(destination);
}

@Override
public JMSConsumer createDurableConsumer(Topic topic, String name) {
    try {
        return startIfNeeded(new JMSConsumerImpl(this, getSession().createDurableConsumer(topic, name)));
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createSharedDurableConsumer(Topic topic, String name, String selector) {
    try {
        return startIfNeeded(new JMSConsumerImpl(this, getSession().createSharedDurableConsumer(topic, name, selector)));
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

@Override
public TopicSubscriber createDurableSubscriber(
        final Topic topic,
        final String name
) throws JMSException {
    return notImplemented();
}

@Override
public Object resolve(final InjectionPoint injectionPoint) {
    final Resource annotation = injectionPoint.getAnnotated().getAnnotation(Resource.class);
    if (null == annotation) {
        return null;
    }
    if (Topic.class != injectionPoint.getType()) {
        return null;
    }
    return new TestEEfiTopic(annotation.mappedName());
}

@Override
public MessageConsumer createSharedDurableConsumer(
        final Topic topic,
        final String name,
        final String messageSelector
) throws JMSException {
    return notImplemented();
}

@Override
public ConnectionConsumer createSharedDurableConnectionConsumer(Topic arg0,
        String arg1, String arg2, ServerSessionPool arg3, int arg4)
                throws JMSException {
    // TODO Auto-generated method stub
    return null;
}

@Override
public JMSConsumer createDurableConsumer(Topic topic, String name) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createDurableConsumer(topic, name)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createSharedConsumer(Topic topic, String name, String selector) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createSharedConsumer(topic, name, selector)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public ConnectionConsumer createSharedConnectionConsumer(Topic arg0,
        String arg1, String arg2, ServerSessionPool arg3, int arg4)
                throws JMSException {
    // TODO Auto-generated method stub
    return null;
}

/**
 *
 */
private String retrieveDestinationName(Destination destination, String headerName) {
    String destinationName = null;
    if (destination != null) {
        try {
            destinationName = (destination instanceof Queue) ? ((Queue) destination).getQueueName()
                    : ((Topic) destination).getTopicName();
        } catch (JMSException e) {
            this.processLog.warn("Failed to retrieve Destination name for '" + headerName + "' header", e);
        }
    }
    return destinationName;
}

@Test(timeout = 60000)
public void testCreateTopic() throws Exception {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createConnection();
    Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Topic topic = session.createTopic(getTestName());
    assertNotNull(topic);
    assertEquals(getTestName(), topic.getTopicName());
    assertTrue(topic instanceof MockJMSTopic);
}

@Test(timeout = 60000)
public void testCreateSubscriber() throws Exception {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createConnection();
    JmsPoolSession session = (JmsPoolSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    Topic topic = session.createTopic(getTestName());
    assertNotNull(session.createSubscriber(topic));

    session.close();
    try {
        session.createSubscriber(topic);
        fail("Should not be able to createSubscriber when closed");
    } catch (JMSException ex) {}
}

@Test(timeout = 60000)
public void testCreateSubscriberWithSelectorAndNoLocal() throws Exception {
    JmsPoolConnection connection = (JmsPoolConnection) cf.createConnection();
    JmsPoolSession session = (JmsPoolSession) connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

    Topic topic = session.createTopic(getTestName());
    assertNotNull(session.createSubscriber(topic, "color = red", false));

    session.close();
    try {
        session.createSubscriber(topic, "color = greean", true);
        fail("Should not be able to createSubscriber when closed");
    } catch (JMSException ex) {}
}

@Test(timeout = 30000)
public void testSharedCreateConsumer() {
    JMSContext context = cf.createContext();
    Topic topic = context.createTopic(getTestName());
    assertNotNull(context.createSharedConsumer(topic, "name"));

    context.close();
    try {
        context.createSharedConsumer(topic, "name");
        fail("Should not be able to create resource when context is closed");
    } catch (IllegalStateRuntimeException isre) {}
}

public MessageConsumer createConsumer(Destination destination,
        ProxySession session) throws JMSException {
    String destinationName = destination.toString();
    ProxyMessageConsumer messageConsumer = new ProxyMessageConsumer(session);
    messageConsumer.setDestination(destination);

    if (destination instanceof Topic) {
        this.messageHandler.registerToTopic(destinationName,
                messageConsumer.getId());
    }

    return messageConsumer;
}

@Override
public ConnectionConsumer createSharedDurableConnectionConsumer(
        final Topic topic,
        final String subscriptionName,
        final String messageSelector,
        final ServerSessionPool sessionPool,
        final int maxMessages
) throws JMSException {
    return null;
}

@Override
public TopicSubscriber createSubscriber(
                                         Topic topic,
                                         String messageSelector,
                                         boolean noLocal ) throws JMSException {

    return addConsumer(topicSession.createSubscriber(topic, messageSelector, noLocal));
}

/**
 * @param topicConnection
 * @param chatTopic
 * @param commandLineChat
 * @throws JMSException
 */
void subscribe(TopicConnection topicConnection, Topic chatTopic,
        BasicJMSChat commandLineChat) throws JMSException {
    TopicSession tsession = topicConnection.createTopicSession(false,
            Session.AUTO_ACKNOWLEDGE);
    TopicSubscriber ts = tsession.createSubscriber(chatTopic);
    ts.setMessageListener(commandLineChat);
}

@Override
public Topic createTopic(String topicName) {
    try {
        return getSession().createTopic(topicName);
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

@Test(timeout = 60000)
public void testTemporaryTopicLeakAfterConnectionClose() throws Exception {
    Connection pooledConnection = null;
    Session session = null;
    Topic tempTopic = null;
    for (int i = 0; i < 2; i++) {
        pooledConnection = pooledFactory.createConnection();
        session = pooledConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        tempTopic = session.createTemporaryTopic();
        LOG.info("Created topic named: " + tempTopic.getTopicName());
        pooledConnection.close();
    }

    assertEquals(0, countBrokerTemporaryTopics());
}

public ConnectionConsumer createDurableConnectionConsumer(Topic topic,
        String subscriptionName, String messageSelector,
        ServerSessionPool sessionPool, int maxMessages) throws JMSException {
    this.checkStatus();

    return null;
}

@Override
public MessageConsumer createSharedConsumer(
        final Topic topic,
        final String sharedSubscriptionName
) throws JMSException {
    return notImplemented();
}