javax.jms.QueueSession#createTextMessage ( )源码实例Demo

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

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

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

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

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

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

      qc.start();

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

      QueueSender qsender = qs.createSender(queue1);

      qsender.send(m);

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

      qsender.send(m);

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

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

@Test(timeout = 5 * 60 * 1000)
public void test() throws Throwable {
   executeCommand("bundle:list");

   withinReason(new Callable<Boolean>() {
      @Override
      public Boolean call() throws Exception {
         assertTrue("artemis bundle installed", verifyBundleInstalled("artemis-server-osgi"));
         return true;
      }
   });

   Object service = waitForService("(objectClass=org.apache.activemq.artemis.core.server.ActiveMQServer)", 30000);
   assertNotNull(service);
   log.debug("have service " + service);

   executeCommand("service:list -n");

   Connection connection = null;
   try {
      JmsConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:5672");
      connection = factory.createConnection(USER, PASSWORD);
      connection.start();

      QueueSession sess = (QueueSession) connection.createSession(false, javax.jms.Session.AUTO_ACKNOWLEDGE);
      Queue queue = sess.createQueue("exampleQueue");
      MessageProducer producer = sess.createProducer(queue);
      producer.send(sess.createTextMessage("TEST"));

      // Test browsing
      try (QueueBrowser browser = sess.createBrowser(queue)) {
         Enumeration messages = browser.getEnumeration();
         while (messages.hasMoreElements()) {
            messages.nextElement();
         }
      }

      // Test management
      Queue managementQueue = sess.createQueue("activemq.management");
      QueueRequestor requestor = new QueueRequestor(sess, managementQueue);
      connection.start();
      TextMessage m = sess.createTextMessage();
      m.setStringProperty("_AMQ_ResourceName", "broker");
      m.setStringProperty("_AMQ_OperationName", "getQueueNames");
      m.setText("[\"ANYCAST\"]");
      Message reply = requestor.request(m);
      String json = ((TextMessage) reply).getText();
      JsonArray array = Json.createReader(new StringReader(json)).readArray();
      List<JsonString> queues = (List<JsonString>) array.get(0);
      assertNotNull(queues);
      assertFalse(queues.isEmpty());

      MessageConsumer consumer = sess.createConsumer(queue);
      Message msg = consumer.receive(5000);
      assertNotNull(msg);
   } finally {
      if (connection != null) {
         connection.close();
      }
   }
}
 

@Test
public void testSendEmptyMessages() throws Exception {
   Queue dest = new ActiveMQQueue(queueName);

   QueueSession defaultQueueSession =  connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
   QueueSender defaultSender = defaultQueueSession.createSender(dest);
   defaultSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
   connection.start();

   Message msg = defaultQueueSession.createMessage();
   msg.setStringProperty("testName", "testSendEmptyMessages");
   defaultSender.send(msg);

   QueueReceiver queueReceiver = defaultQueueSession.createReceiver(dest);
   assertNotNull("Didn't receive message", queueReceiver.receive(1000));

   //bytes
   BytesMessage bytesMessage = defaultQueueSession.createBytesMessage();
   bytesMessage.setStringProperty("testName", "testSendEmptyMessages");
   defaultSender.send(bytesMessage);
   assertNotNull("Didn't receive message", queueReceiver.receive(1000));

   //map
   MapMessage mapMessage = defaultQueueSession.createMapMessage();
   mapMessage.setStringProperty("testName", "testSendEmptyMessages");
   defaultSender.send(mapMessage);
   assertNotNull("Didn't receive message", queueReceiver.receive(1000));

   //object
   ObjectMessage objMessage = defaultQueueSession.createObjectMessage();
   objMessage.setStringProperty("testName", "testSendEmptyMessages");
   defaultSender.send(objMessage);
   assertNotNull("Didn't receive message", queueReceiver.receive(1000));

   //stream
   StreamMessage streamMessage = defaultQueueSession.createStreamMessage();
   streamMessage.setStringProperty("testName", "testSendEmptyMessages");
   defaultSender.send(streamMessage);
   assertNotNull("Didn't receive message", queueReceiver.receive(1000));

   //text
   TextMessage textMessage = defaultQueueSession.createTextMessage();
   textMessage.setStringProperty("testName", "testSendEmptyMessages");
   defaultSender.send(textMessage);
   assertNotNull("Didn't receive message", queueReceiver.receive(1000));
}
 

/**
    * Sends a message,and if transacted, the queueSession is committed.
    * <p>This method is intended for <b>clients</b>, as <b>server</b>s
    * will use the <code>sendReply</code>.
    * @return the correlationID of the sent message
    */
   public TextMessage sendMessage(QueueSession session, QueueSender sender, String message, Map udzMap, String bifName, byte btcData[])
       throws IfsaException {

    try {
		if (!isRequestor()) {
			throw new IfsaException(getLogPrefix()+ "Provider cannot use sendMessage, should use sendReply");
		}
        IFSATextMessage msg = (IFSATextMessage)session.createTextMessage();
        msg.setText(message);
		if (udzMap != null && msg instanceof IFSAMessage) {
			// Handle UDZs
			log.debug(getLogPrefix()+"add UDZ map to IFSAMessage");
			// process the udzMap
			Map udzObject = (Map)((IFSAMessage) msg).getOutgoingUDZObject();
			udzObject.putAll(udzMap);
		}
		String replyToQueueName="-"; 
        //Client side
        if (messageProtocol.equals(IfsaMessageProtocolEnum.REQUEST_REPLY)) {
            // set reply-to address
            Queue replyTo=getMessagingSource().getClientReplyQueue(session);
            msg.setJMSReplyTo(replyTo);
            replyToQueueName=replyTo.getQueueName();
        }
        if (messageProtocol.equals(IfsaMessageProtocolEnum.FIRE_AND_FORGET)) {
         	// not applicable
        }
		if (StringUtils.isNotEmpty(bifName)) {
			msg.setBifName(bifName);
		}
		if (btcData!=null && btcData.length>0) {
			msg.setBtcData(btcData);
		}

		if (log.isDebugEnabled()) {
			log.debug(getLogPrefix()
					+ " messageProtocol ["
					+ messageProtocol
					+ "] replyToQueueName ["
					+ replyToQueueName
					+ "] sending message ["
					+ message
					+ "]");
		} else {
			if (log.isInfoEnabled()) {
				log.info(getLogPrefix()
						+ " messageProtocol ["
						+ messageProtocol
						+ "] replyToQueueName ["
						+ replyToQueueName
						+ "] sending message");
			}
		}

        // send the message
        sender.send(msg);

        // perform commit
        if (isJmsTransacted() && !(messagingSource.isXaEnabledForSure() && JtaUtil.inTransaction())) {
            session.commit();
            log.debug(getLogPrefix()+ "committing (send) transaction");
        }

        return msg;
	    
 	} catch (Exception e) {
		throw new IfsaException(e);
	}
}
 

private void sendMessageToServer() throws JMSException, IOException {

		QueueConnection queueConn = (QueueConnection) connectionFactory.createConnection();
		queueConn.start();

		QueueSession queueSession = queueConn.createQueueSession(false, Session.DUPS_OK_ACKNOWLEDGE);

		Destination destination = queueSession.createQueue(jmsQueue);

		MessageProducer queueSender = queueSession.createProducer(destination);
		queueSender.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
		Message message = queueSession.createTextMessage(createMessage());
		queueSender.send(message);

	}