javax.management.modelmbean.XMLParseException#javax.jms.Destination源码实例Demo

@Test
public void testRemoveNotScheduled() throws Exception {
   Connection connection = createConnection();

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

   // Create the Browse Destination and the Reply To location
   Destination management = session.createTopic(ScheduledMessage.AMQ_SCHEDULER_MANAGEMENT_DESTINATION);

   MessageProducer producer = session.createProducer(management);

   try {

      // Send the remove request
      Message remove = session.createMessage();
      remove.setStringProperty(ScheduledMessage.AMQ_SCHEDULER_ACTION, ScheduledMessage.AMQ_SCHEDULER_ACTION_REMOVEALL);
      remove.setStringProperty(ScheduledMessage.AMQ_SCHEDULED_ID, new IdGenerator().generateId());
      producer.send(remove);
   } catch (Exception e) {
      fail("Caught unexpected exception during remove of unscheduled message.");
   }
}
 

@Test
public void testAutoStartOffDoesNotStartTheConnectionMessageConsumerSelector() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    JmsMessageConsumer consumer = Mockito.mock(JmsMessageConsumer.class);

    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    Mockito.when(session.createConsumer(any(Destination.class), anyString())).thenReturn(consumer);
    Mockito.when(session.createTemporaryTopic()).thenReturn(new JmsTemporaryTopic());

    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);
    context.setAutoStart(false);

    try {
        context.createConsumer(context.createTemporaryTopic(), "a = b");
    } finally {
        context.close();
    }

    Mockito.verify(session, Mockito.times(1)).createConsumer(any(Topic.class), anyString());
    Mockito.verify(connection, Mockito.times(0)).start();
}
 

public void sendOrder(int customerId, Date date, String... itemIds) throws Exception {
    // format the JMS message from the input parameters
    String d = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss").format(date);
    String body = customerId + "," + d;
    for (String id : itemIds) {
        body += "," + id;
    }

    // use JMS code to send the message (a bit ugly code but it works)
    Connection con = fac.createConnection();
    con.start();
    Session ses = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
    Destination dest = ses.createQueue("order");
    MessageProducer prod = ses.createProducer(dest);
    prod.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
    Message msg = ses.createTextMessage(body);
    prod.send(msg);
    prod.close();
    ses.close();
    con.close();
}
 

/**
 * Similar to other properties, {@code null} should be expected even if it seems unintuitive.
 *
 * <p>The JMS 1.1 specification 4.2.1 suggests destination details are provider specific.
 * Further, JavaDoc on {@link Queue#getQueueName()} and {@link Topic#getTopicName()} say "Clients
 * that depend upon the name are not portable." Next, such operations can raise {@link
 * JMSException} messages which this code can coerce to null. Finally, destinations are not
 * constrained to implement only one of {@link Queue} or {@link Destination}. This implies one
 * could return null while the other doesn't, such as was the case in issue #1098.
 */
@Nullable static String channelName(@Nullable Destination destination) {
  if (destination == null) return null;
  boolean isQueue = isQueue(destination);
  try {
    if (isQueue) {
      return ((Queue) destination).getQueueName();
    } else {
      return ((Topic) destination).getTopicName();
    }
  } catch (Throwable t) {
    propagateIfFatal(t);
    log(t, "error getting destination name from {0}", destination, null);
  }
  return null;
}
 

protected void drainDestination(final ConnectionFactory cf, final Destination dest) throws JMSException {
   Connection conn = null;
   try {
      conn = cf.createConnection();
      Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
      MessageConsumer cons = sess.createConsumer(dest);
      Message m = null;
      conn.start();
      log.trace("Draining messages from " + dest);
      while (true) {
         m = cons.receive(DRAIN_WAIT_TIME);
         if (m == null) {
            break;
         }
         log.trace("Drained message");
      }
   } finally {
      if (conn != null) {
         conn.close();
      }
   }
}
 

public void sendMessages(ConnectionFactory connectionFactory) throws Exception {
    for (int i = 0; i < NUM_MESSAGES; i++) {
        Connection connection = connectionFactory.createConnection();
        try {
            connection.start();

            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createQueue(QUEUE);
            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            String msgTo = "hello";
            TextMessage message = session.createTextMessage(msgTo);
            producer.send(message);
        } finally {
            connection.close();
        }
        LOG.debug("sent " + i + " messages using " + connectionFactory.getClass());
    }
}
 

/**
 * Create a MessageConsumer for the given JMS Session,
 * registering a MessageListener for the specified listener.
 * @param session the JMS Session to work on
 * @return the MessageConsumer
 * @throws JMSException if thrown by JMS methods
 * @see #executeListener
 */
protected MessageConsumer createListenerConsumer(final Session session) throws JMSException {
	Destination destination = getDestination();
	if (destination == null) {
		String destinationName = getDestinationName();
		Assert.state(destinationName != null, "No destination set");
		destination = resolveDestinationName(session, destinationName);
	}
	MessageConsumer consumer = createConsumer(session, destination);

	if (this.taskExecutor != null) {
		consumer.setMessageListener(message -> this.taskExecutor.execute(() -> processMessage(message, session)));
	}
	else {
		consumer.setMessageListener(message -> processMessage(message, session));
	}

	return consumer;
}
 

@Test
public void testAutoStartOnDoesStartTheConnectionMessageConsumerSelector() throws JMSException {
    JmsConnection connection = Mockito.mock(JmsConnection.class);
    JmsSession session = Mockito.mock(JmsSession.class);
    JmsMessageConsumer consumer = Mockito.mock(JmsMessageConsumer.class);

    Mockito.when(connection.createSession(Mockito.anyInt())).thenReturn(session);
    Mockito.when(session.createConsumer(any(Destination.class), anyString())).thenReturn(consumer);
    Mockito.when(session.createTemporaryTopic()).thenReturn(new JmsTemporaryTopic());

    JmsContext context = new JmsContext(connection, JMSContext.AUTO_ACKNOWLEDGE);
    context.setAutoStart(true);

    try {
        context.createConsumer(context.createTemporaryTopic(), "a = b");
    } finally {
        context.close();
    }

    Mockito.verify(session, Mockito.times(1)).createConsumer(any(Topic.class), anyString());
    Mockito.verify(connection, Mockito.times(1)).start();
}
 

@Override
public void convertAndSend(Object message) throws JmsException {
	Destination defaultDestination = getDefaultDestination();
	if (defaultDestination != null) {
		convertAndSend(defaultDestination, message);
	}
	else {
		convertAndSend(getRequiredDefaultDestinationName(), message);
	}
}
 

@Override
public Message<?> receive() {
	Destination defaultDestination = getDefaultDestination();
	if (defaultDestination != null) {
		return receive(defaultDestination);
	}
	else {
		return receive(getRequiredDefaultDestinationName());
	}
}
 

/**
 * This is a JMS spec independent method to create a MessageProducer. Please be cautious when
 * making any changes
 *
 * @param session     JMS session
 * @param destination the Destination
 * @param isQueue     is the Destination a queue?
 * @param jmsSpec11   should we use JMS 1.1 API ?
 * @return a MessageProducer to send messages to the given Destination
 * @throws JMSException on errors, to be handled and logged by the caller
 */
public MessageProducer createProducer(Session session, Destination destination, Boolean isQueue)
        throws JMSException {
    if ("2.0".equals(jmsSpec) || "1.1".equals(jmsSpec) || isQueue == null) {
        return session.createProducer(destination);
    } else {
        if (isQueue) {
            return ((QueueSession) session).createSender((Queue) destination);
        } else {
            return ((TopicSession) session).createPublisher((Topic) destination);
        }
    }
}
 

@Override
public void send(MessageCreator messageCreator) throws JmsException {
	Destination defaultDestination = getDefaultDestination();
	if (defaultDestination != null) {
		send(defaultDestination, messageCreator);
	}
	else {
		send(getRequiredDefaultDestinationName(), messageCreator);
	}
}
 

public MessageConsumer createMessageConsumer(Session session, Destination destination) {
    MessageConsumer messageConsumer = super.createMessageConsumer(session, destination);
    if (this.cacheLevel >= JMSConstants.CACHE_CONSUMER) {
        cachedMessageConsumer = messageConsumer;
    }
    return messageConsumer;
}
 

/**
 * Validate the given Destination object, checking whether it matches
 * the expected type.
 * @param destination the Destination object to validate
 * @param destinationName the name of the destination
 * @param pubSubDomain {@code true} if a Topic is expected,
 * {@code false} in case of a Queue
 */
protected void validateDestination(Destination destination, String destinationName, boolean pubSubDomain) {
	Class<?> targetClass = Queue.class;
	if (pubSubDomain) {
		targetClass = Topic.class;
	}
	if (!targetClass.isInstance(destination)) {
		throw new DestinationResolutionException(
				"Destination [" + destinationName + "] is not of expected type [" + targetClass.getName() + "]");
	}
}
 

private String getDestName(Message message) throws JMSException {
    Destination replyTo = message.getJMSReplyTo();
    if (replyTo instanceof Queue) {
        return ((Queue)replyTo).getQueueName();
    } else if (replyTo instanceof Topic) {
        return ((Topic)replyTo).getTopicName();
    }
    return null;
}
 

@Test
public void attemptToReadDisallowedReplyToPropertyIsNotFatal() throws JMSException {
	javax.jms.Message jmsMessage = new StubTextMessage() {
		@Override
		public Destination getJMSReplyTo() throws JMSException {
			throw new JMSException("illegal property");
		}
	};
	assertAttemptReadDisallowedPropertyIsNotFatal(jmsMessage, JmsHeaders.REPLY_TO);
}
 

private void sendMessages() throws Exception {
   Connection connection = connectionFactory.createConnection();
   Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   Destination destination = session.createQueue(destinationName);
   MessageProducer producer = session.createProducer(destination);
   for (int i = 0; i < MSG_COUNT; ++i) {
      producer.send(session.createTextMessage("" + i));
   }
}
 

@Override
public void send(Destination destination,
                 Message message,
                 CompletionListener completionListener) throws JMSException {
   if (ActiveMQRALogger.LOGGER.isTraceEnabled()) {
      ActiveMQRALogger.LOGGER.trace("send(" + destination + ", " + message + ", " + completionListener + ")");
   }
   producer.send(destination, message, completionListener);
}
 

@Test
public void testTransformDestinationDestinationWithNoNameThrowsJMSEx() throws JMSException {
    Destination destination = Mockito.mock(Destination.class);
    try {
        transformer.transform(destination);
        fail("Should throw a JMSException here");
    } catch (JMSException ex) {
    }
}
 

@Override
public <T> T receiveAndConvert(Class<T> targetClass) {
	Destination defaultDestination = getDefaultDestination();
	if (defaultDestination != null) {
		return receiveAndConvert(defaultDestination, targetClass);
	}
	else {
		return receiveAndConvert(getRequiredDefaultDestinationName(), targetClass);
	}
}
 

public Destination createDestination(String destinationName)
		throws JmsException {
	Destination dest = null;
	Session session = null;
	try {
		session = createSession(false, Session.AUTO_ACKNOWLEDGE);
		dest = session.createQueue(destinationName);
	} catch (Exception e) {
		throw new JmsException("cannot create destination", e);
	} finally {
		releaseSession(session);
	}
	return dest;
}
 

@Test
public void testDynamicQueue() throws NamingException, JMSException {
   Hashtable<String, String> props = new Hashtable<>();
   props.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.activemq.artemis.jndi.ActiveMQInitialContextFactory");
   Context ctx = new InitialContext(props);

   Destination destination = (Destination) ctx.lookup("dynamicQueues/myQueue");
   Assert.assertTrue(destination instanceof Queue);
}
 

@Test
public void messageSentToTopicComesBackWithTheSameJMSDestination() throws Exception
{
    Topic topic = createTopic(getTestName());
    Connection connection = getConnection();
    try
    {
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        MessageConsumer consumer = session.createConsumer(topic);

        Utils.sendMessages(session, topic, 1);

        connection.start();

        Message receivedMessage = consumer.receive(getReceiveTimeout());
        assertNotNull("Message should not be null", receivedMessage);

        Destination receivedDestination = receivedMessage.getJMSDestination();

        assertNotNull("JMSDestination should not be null", receivedDestination);
        assertTrue("Unexpected destination type", receivedDestination instanceof Topic);
        assertEquals("Unexpected destination name",
                     topic.getTopicName(),
                     ((Topic) receivedDestination).getTopicName());
    }
    finally
    {
        connection.close();
    }
}
 

@Override
public void setJMSDestination(Destination destination) throws JMSException {
    if (destination instanceof WeEventTopic) {
        this.weEventTopic = (WeEventTopic) destination;
        return;
    }

    throw new JMSException(WeEventConnectionFactory.NotSupportTips);
}
 

/**
 * check messages are actuallly sent on a tx rollback
 *
 * @throws Exception
 */

public void testTransactionRollbackOnSend() throws Exception {
   Destination destination = createDestination(getClass().getName());
   Connection connection = createConnection();
   connection.setClientID(idGen.generateId());
   connection.start();
   Session consumerSession = connection.createSession(true, Session.CLIENT_ACKNOWLEDGE);
   MessageConsumer consumer = consumerSession.createConsumer(destination);
   Session producerSession = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
   MessageProducer producer = producerSession.createProducer(destination);
   producer.setDeliveryMode(deliveryMode);

   TextMessage sentMsg = producerSession.createTextMessage();
   sentMsg.setText("msg1");
   producer.send(sentMsg);
   producerSession.commit();

   Message recMsg = consumer.receive(RECEIVE_TIMEOUT);
   consumerSession.commit();
   assertTrue(recMsg.equals(sentMsg));

   sentMsg = producerSession.createTextMessage();
   sentMsg.setText("msg2");
   producer.send(sentMsg);
   producerSession.rollback();

   sentMsg = producerSession.createTextMessage();
   sentMsg.setText("msg3");
   producer.send(sentMsg);
   producerSession.commit();

   recMsg = consumer.receive(RECEIVE_TIMEOUT);
   assertTrue(recMsg.equals(sentMsg));
   consumerSession.commit();

   connection.close();
}
 

private MessageConsumer getOrCreateQueueConsumer(Destination destination) throws JMSException {
    MessageConsumer consumer = consumerCache.get(destination);
    if (consumer == null) {
        consumer = session.createConsumer(destination);
        consumerCache.put(destination, consumer);
    }
    return consumer;
}
 

@Override
public void onMessage(Message msg) {
   try {
      TextMessage textMsg = (TextMessage) msg;
      String payload = "REPLY: " + textMsg.getText();
      Destination replyTo;
      replyTo = msg.getJMSReplyTo();
      textMsg.clearBody();
      textMsg.setText(payload);
      requestServerProducer.send(replyTo, textMsg);
   } catch (JMSException e) {
      e.printStackTrace();
   }
}
 

@Override
public JMSProducer send(Destination destination, Serializable body) {
    try {
        ObjectMessage message = session.createObjectMessage();
        message.setObject(body);
        doSend(destination, message);
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }

    return this;
}
 

/**
 * @see org.apache.activemq.test.JmsResourceProvider#createConsumer(javax.jms.Session,
 *      javax.jms.Destination)
 */
public MessageConsumer createConsumer(Session session, Destination destination) throws JMSException {
   if (isDurableSubscriber()) {
      return session.createDurableSubscriber((Topic) destination, durableName);
   }
   return session.createConsumer(destination);
}
 

public void testCloseSendConnection() throws Exception {
   String brokerName = "closeSend";
   BrokerService broker = BrokerFactory.createBroker(new URI("broker:(tcp://localhost:0)/" + brokerName));
   broker.start();
   broker.waitUntilStarted();
   ActiveMQXAConnectionFactory cf = new ActiveMQXAConnectionFactory(broker.getTransportConnectors().get(0).getConnectUri());
   XAConnection connection = (XAConnection) cf.createConnection();
   connection.start();
   XASession session = connection.createXASession();
   XAResource resource = session.getXAResource();
   Destination dest = new ActiveMQQueue(getName());

   // publish a message
   Xid tid = createXid();
   resource.start(tid, XAResource.TMNOFLAGS);
   MessageProducer producer = session.createProducer(dest);
   ActiveMQTextMessage message = new ActiveMQTextMessage();
   message.setText(getName());
   producer.send(message);

   connection.close();

   //comment out this check as it doesn't apply to artemis
   //assertTransactionGoneFromBroker(tid);

   broker.stop();
}