Java 类javax.jms.QueueRequestor 实例源码

@Override
protected void setUp() throws Exception {
   super.setUp();
   context = createApplicationContext();

   createConnections();

   requestServerSession = localConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
   Queue theQueue = requestServerSession.createQueue(getClass().getName());
   requestServerConsumer = requestServerSession.createConsumer(theQueue);
   requestServerConsumer.setMessageListener(this);
   requestServerProducer = requestServerSession.createProducer(null);

   QueueSession session = remoteConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
   requestor = new QueueRequestor(session, theQueue);
}

@Override
protected void setUp() throws Exception {
   super.setUp();

   context = createApplicationContext();

   createConnections();

   requestServerSession = localConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
   theQueue = requestServerSession.createQueue(QUEUE_NAME);
   requestServerConsumer = requestServerSession.createConsumer(theQueue);
   requestServerProducer = requestServerSession.createProducer(null);

   QueueSession session = localConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
   requestor = new QueueRequestor(session, theQueue);
}

@Deployment
public static JavaArchive getDeployment() {
    return ShrinkWrap.create(JavaArchive.class, "test.jar")
            .addPackages(true, "victor.training.jee6")
            .addPackages(true, "org.mockito")
            .addPackages(true, "org.apache.tools")
            .addPackages(true, "org.objenesis")
            .addClass(QueueRequestor.class)
            .addAsManifestResource(new File("src/main/webapp/WEB-INF/beans.xml"))
            ;
}

@Deployment
public static JavaArchive getDeployment() {
    return ShrinkWrap.create(JavaArchive.class, "test.jar").addPackages(true, "victor.training.ejb")
            .addPackages(true, "org.mockito")
            .addPackages(true, "org.objenesis").addClass(QueueRequestor.class);
}

public static void main(final String[] args) throws Exception {
   QueueConnection connection = null;
   InitialContext initialContext = null;
   try {
      // Step 1. Create an initial context to perform the JNDI lookup.
      initialContext = new InitialContext();

      // Step 2. Perfom a lookup on the queue
      Queue queue = (Queue) initialContext.lookup("queue/exampleQueue");

      // Step 3. Look-up the JMS queue connection factory
      QueueConnectionFactory cf = (QueueConnectionFactory) initialContext.lookup("ConnectionFactory");

      // Step 4. Create a TextReverserService which consumes messages from the queue and sends message with reversed
      // text
      TextReverserService reverserService = new TextReverserService(cf, queue);

      // Step 5. Create a JMS QueueConnection
      connection = cf.createQueueConnection();

      // Step 6. Start the connection
      connection.start();

      // Step 7. Create a JMS queue session with AUTO_ACKNOWLEDGE mode
      QueueSession session = connection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

      // Step 8. Create a JMS queue requestor to send requests to the queue
      QueueRequestor queueRequestor = new QueueRequestor(session, queue);

      // Step 9. Create a JMS message to send as a request
      TextMessage request = session.createTextMessage("Hello, World!");

      // Step 10. Use the requestor to send the request and wait to receive a reply
      TextMessage reply = (TextMessage) queueRequestor.request(request);

      // Step 11. The reply's text contains the reversed request's text
      System.out.println("Send request: " + request.getText());
      System.out.println("Received reply:" + reply.getText());

      // Step.12 close the queue requestor
      queueRequestor.close();

      // Step 13. close the text reverser service
      reverserService.close();
   } finally {
      if (connection != null) {
         try {
            // Step 14. Be sure to close the JMS resources!
            connection.close();
         } catch (JMSException e) {
            e.printStackTrace();
         }
      }

      if (initialContext != null) {
         // Also the InitialContext
         initialContext.close();
      }
   }
}

private static int getMessageCount(final Connection connection) throws Exception {
   QueueSession session = ((QueueConnection) connection).createQueueSession(false, Session.AUTO_ACKNOWLEDGE);

   Queue managementQueue = ActiveMQJMSClient.createQueue("activemq.management");

   QueueRequestor requestor = new QueueRequestor(session, managementQueue);

   connection.start();

   Message m = session.createMessage();

   JMSManagementHelper.putAttribute(m, ResourceNames.QUEUE + "exampleQueue", "messageCount");

   Message response = requestor.request(m);

   int messageCount = (Integer) JMSManagementHelper.getResult(response, Integer.class);

   return messageCount;
}

@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.info("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();
      }
   }
}

/**
 * Constructor.
 *
 * @param session
 *            the session to use to send the message
 * @param destination
 *            the queue to send the message on
 * @throws JMSException
 *             when internally used {@link QueueRequestor} can not be
 *             constructed with <code>session</code> and
 *             <code>destination</code>
 */
public TemporaryQueueExecutor(QueueSession session, Queue destination) throws JMSException {
    requestor = new QueueRequestor(session, destination);
}

/**
 * Constructor.
 *
 * @param session
 *            the session to use to send the message
 * @param destination
 *            the queue to send the message on
 * @throws JMSException
 */
public TemporaryQueueExecutor(QueueSession session, Queue destination) throws JMSException {
    requestor = new QueueRequestor(session, destination);
}