Java 类javax.jms.JMSConsumer 实例源码

private void connectInternal(Runnable command) {
    final JMSContext jmsCtx = mock(JMSContext.class);
    when(this.jmsCtxProvider.get()).thenReturn(jmsCtx);

    when(this.artemisConfig.getManagementNotificationAddress()).thenReturn(new SimpleString("notif"));
    final JMSConsumer consumer = mock(JMSConsumer.class);
    when(jmsCtx.createConsumer(any())).thenReturn(consumer);

    command.run();

    verify(this.jmsCtxProvider).get();
    verify(this.artemisConfig).getManagementNotificationAddress();
    verify(this.log).info("Connecting to broker for sourcing destination events.");
    verify(jmsCtx).createConsumer(any());
    verify(consumer).setMessageListener(this.eventProducer);
    verifyNoMoreInteractions(consumer);
}

@Test
public void testSendStreamMessage() throws JMSException, InterruptedException {
   JmsProducerCompletionListenerTest.CountingCompletionListener cl = new JmsProducerCompletionListenerTest.CountingCompletionListener(1);
   JMSProducer producer = context.createProducer();
   producer.setAsync(cl);
   StreamMessage msg = context.createStreamMessage();
   msg.setStringProperty("name", name.getMethodName());
   String bprop = "booleanProp";
   String iprop = "intProp";
   msg.setBooleanProperty(bprop, true);
   msg.setIntProperty(iprop, 42);
   msg.writeBoolean(true);
   msg.writeInt(67);
   producer.send(queue1, msg);
   JMSConsumer consumer = context.createConsumer(queue1);
   Message msg2 = consumer.receive(100);
   Assert.assertNotNull(msg2);
   Assert.assertTrue(cl.completionLatch.await(1, TimeUnit.SECONDS));
   StreamMessage sm = (StreamMessage) cl.lastMessage;
   Assert.assertEquals(true, sm.getBooleanProperty(bprop));
   Assert.assertEquals(42, sm.getIntProperty(iprop));
   Assert.assertEquals(true, sm.readBoolean());
   Assert.assertEquals(67, sm.readInt());
}

@Test
public void sharedDurableSubSimpleRoundRobin() throws Exception {
   context = cf.createContext();
   try {
      JMSConsumer con1 = context.createSharedDurableConsumer(topic1, "mySharedCon");
      JMSConsumer con2 = context.createSharedDurableConsumer(topic1, "mySharedCon");
      context.start();
      JMSProducer producer = context.createProducer();
      int numMessages = 10;
      for (int i = 0; i < numMessages; i++) {
         producer.send(topic1, "msg:" + i);
      }

      for (int i = 0; i < numMessages; i += 2) {
         String msg = con1.receiveBody(String.class, 5000);
         System.out.println("msg = " + msg);
         msg = con2.receiveBody(String.class, 5000);
         System.out.println("msg = " + msg);
      }

   } finally {
      context.close();
   }
}

@Test
public void sharedNonDurableUnsubscribeDifferentTopic() throws Exception {
   context = cf.createContext();
   try {
      JMSConsumer con1 = context.createSharedConsumer(topic1, "mySharedCon");
      JMSConsumer con2 = context.createSharedConsumer(topic1, "mySharedCon");
      con1.close();
      Binding binding = server.getPostOffice().getBinding(new SimpleString("nonDurable.mySharedCon"));
      assertNotNull(binding);
      con2.close();
      binding = server.getPostOffice().getBinding(new SimpleString("nonDurable.mySharedCon"));
      assertNull(binding);
      con1 = context.createSharedConsumer(topic2, "mySharedCon");
   } finally {
      context.close();
   }
}

@Test(timeout = 20000)
public void testCreateConsumer() throws Exception {
    try (TestAmqpPeer testPeer = new TestAmqpPeer();) {
        JMSContext context = testFixture.createJMSContext(testPeer);
        testPeer.expectBegin();
        testPeer.expectReceiverAttach();
        testPeer.expectLinkFlow();

        Queue queue = context.createQueue("test");
        JMSConsumer consumer = context.createConsumer(queue);
        assertNotNull(consumer);

        testPeer.expectEnd();
        testPeer.expectClose();
        context.close();

        testPeer.waitForAllHandlersToComplete(1000);
    }
}

@Override
public JMSConsumer createConsumer(Destination destination) {
    try {
        return startIfNeeded(new JmsPoolJMSConsumer((JmsPoolMessageConsumer) getSession().createConsumer(destination)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createConsumer(Destination destination, String selector) {
    try {
        return startIfNeeded(new JmsPoolJMSConsumer((JmsPoolMessageConsumer) getSession().createConsumer(destination, selector)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

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

@Override
public JMSConsumer createDurableConsumer(Topic topic, String name) {
    try {
        return startIfNeeded(new JmsPoolJMSConsumer((JmsPoolMessageConsumer) getSession().createDurableConsumer(topic, name)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(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 createSharedConsumer(Topic topic, String name) {
    try {
        return startIfNeeded(new JmsPoolJMSConsumer((JmsPoolMessageConsumer) getSession().createSharedConsumer(topic, name)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

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

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

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

private JMSConsumer startIfNeeded(JMSConsumer consumer) throws JMSException {
    if (getAutoStart()) {
        connection.start();
    }

    return consumer;
}

@Override
public JMSConsumer createConsumer(Destination destination) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createConsumer(destination)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createConsumer(Destination destination, String selector) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createConsumer(destination, selector)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createConsumer(Destination destination, String selector, boolean noLocal) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createConsumer(destination, selector, noLocal)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@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 createDurableConsumer(Topic topic, String name, String selector, boolean noLocal) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createDurableConsumer(topic, name, selector, noLocal)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createSharedConsumer(Topic topic, String name) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createSharedConsumer(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 JMSConsumer createSharedDurableConsumer(Topic topic, String name) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createSharedDurableConsumer(topic, name)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}

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

@Test
public void testCloseMoreThanOnce() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue());

    consumer.close();
    consumer.close();
}

@Test
public void testReceive() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue());

    assertNull(consumer.receive());

    consumer.close();

    try {
        consumer.receive();
        fail("Should not be able to interact with closed consumer");
    } catch (IllegalStateRuntimeException ise) {}
}

@Test
public void testReceiveNoWait() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue());

    assertNull(consumer.receiveNoWait());

    consumer.close();

    try {
        consumer.receiveNoWait();
        fail("Should not be able to interact with closed consumer");
    } catch (IllegalStateRuntimeException ise) {}
}

@Test
public void testReceiveTimed() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue());

    assertNull(consumer.receive(1));

    consumer.close();

    try {
        consumer.receive(1);
        fail("Should not be able to interact with closed consumer");
    } catch (IllegalStateRuntimeException ise) {}
}

@Test
public void testGetMessageSelector() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue(), "Color = Red");

    assertNotNull(consumer.getMessageSelector());
    assertEquals("Color = Red", consumer.getMessageSelector());

    consumer.close();

    try {
        consumer.getMessageSelector();
        fail("Should not be able to interact with closed consumer");
    } catch (IllegalStateRuntimeException ise) {}
}

@Test
public void testReceiveBody() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue());

    try {
        consumer.receiveBody(String.class);
        fail("Should not be able to interact with closed consumer");
    } catch (JMSRuntimeException ise) {}
}

@Test
public void testReceiveBodyNoWait() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue());

    try {
        consumer.receiveBodyNoWait(String.class);
        fail("Should not be able to interact with closed consumer");
    } catch (JMSRuntimeException ise) {}
}

@Test
public void testReceiveBodyTimed() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue());

    try {
        consumer.receiveBody(String.class, 1);
        fail("Should not be able to interact with closed consumer");
    } catch (JMSRuntimeException ise) {}
}

@Test
@RunAsClient
public void testListener() throws JMSException, NamingException, InterruptedException {
    final Properties env = new Properties();
    env.put(INITIAL_CONTEXT_FACTORY, "org.jboss.naming.remote.client.InitialContextFactory");
    env.put(PROVIDER_URL, "http-remoting://127.0.0.1:8080");

    Context namingContext = new InitialContext(env);

    ConnectionFactory connectionFactory = (ConnectionFactory) namingContext.lookup(CONNECTION_FACTORY);
    logger.info("Got ConnectionFactory " + CONNECTION_FACTORY);

    Destination destination = (Destination) namingContext.lookup(REMOTE_QUEUE_LOOKUP);
    logger.info("Got JMS Endpoint " + REMOTE_QUEUE_LOOKUP);

    String question = "how many components in your family?";
    String response = "they are four";

    // Create the JMS context
    JMSContext context = connectionFactory.createContext(USER_NAME, USER_PASSWORD);
    Questionary questionary = new Questionary();
    questionary.setQuestion(question);
    questionary.setResponse(response);
    assertFalse("questionary is not approved", questionary.isApproved());
    JMSProducer producer = context.createProducer();
    producer.send(destination, questionary);

    JMSConsumer consumer = context.createConsumer(destination);
    MessageQueueListener messageQueueListener = new MessageQueueListener();
    consumer.setMessageListener(messageQueueListener);
    sleep(100);
    questionary = messageQueueListener.getQuestionary();
    assertEquals("the question is: ", question, questionary.getQuestion());
    assertEquals("the response is: ", response, questionary.getResponse());
    assertTrue("the message is approved: ", questionary.isApproved());
}

@Override
public JMSConsumer createConsumer(Destination destination) {
    try {
        return startIfNeeded(new JMSConsumerImpl(this, getSession().createConsumer(destination)));
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

@Override
public JMSConsumer createConsumer(Destination destination, String selector) {
    try {
        return startIfNeeded(new JMSConsumerImpl(this, getSession().createConsumer(destination, selector)));
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

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

@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 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 JMSConsumer createSharedConsumer(Topic topic, String name) {
    try {
        return startIfNeeded(new JMSConsumerImpl(this, getSession().createSharedConsumer(topic, name)));
    } catch (JMSException jmse) {
        throw Utils.convertToRuntimeException(jmse);
    }
}

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