import javax.jms.JMSConsumer; //导入依赖的package包/类
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);
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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());
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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();
   }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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();
   }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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);
    }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@Override
public JMSConsumer createConsumer(Destination destination) {
    try {
        return startIfNeeded(new JmsPoolJMSConsumer((JmsPoolMessageConsumer) getSession().createConsumer(destination)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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);
    }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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);
    }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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);
    }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
private JMSConsumer startIfNeeded(JMSConsumer consumer) throws JMSException {
    if (getAutoStart()) {
        connection.start();
    }

    return consumer;
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@Override
public JMSConsumer createConsumer(Destination destination) {
    try {
        return startIfNeeded(new MockJMSConsumer(getSession(), (MockJMSMessageConsumer) getSession().createConsumer(destination)));
    } catch (JMSException jmse) {
        throw JMSExceptionSupport.createRuntimeException(jmse);
    }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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);
    }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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);
    }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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);
    }
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@Test
public void testCloseMoreThanOnce() throws JMSException {
    JMSConsumer consumer = context.createConsumer(context.createTemporaryQueue());

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

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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) {}
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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) {}
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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) {}
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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) {}
}
 

import javax.jms.JMSConsumer; //导入依赖的package包/类
@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) {}
}