import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
private DequeueTask<ArtemisMessage, String> newTask(String sentData)
        throws Exception {
    CoreQueueConfiguration q = new CoreQueueConfiguration();
    q.setName("q");
    ClientSession sesh = mock(ClientSession.class);
    ClientConsumer receiver = mock(ClientConsumer.class);
    when(sesh.createConsumer(q.getName(), false)).thenReturn(receiver);
    
    QConnector<ArtemisMessage> connector = new ArtemisQConnector(q, sesh);
    DequeueTask<ArtemisMessage, String> task =
            new DequeueTask<>(connector, this, deserializer(sentData), true);

    verify(receiver).setMessageHandler(any());
    
    return task;
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Test
public void testBasicSendToNoQueue() throws Exception {
   SimpleString qName = new SimpleString("q1");
   AddressSettings addressSettings = new AddressSettings().setMaxDeliveryAttempts(1);
   server.getAddressSettingsRepository().addMatch(qName.toString(), addressSettings);
   clientSession.createQueue(qName, qName, null, false);
   ClientProducer producer = clientSession.createProducer(qName);
   producer.send(createTextMessage(clientSession, "heyho!"));
   clientSession.start();
   ClientConsumer clientConsumer = clientSession.createConsumer(qName);
   ClientMessage m = clientConsumer.receive(500);
   m.acknowledge();
   Assert.assertNotNull(m);
   Assert.assertEquals(m.getBodyBuffer().readString(), "heyho!");
   // force a cancel
   clientSession.rollback();
   m = clientConsumer.receiveImmediate();
   Assert.assertNull(m);
   clientConsumer.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Test
public void testSimpleSendNoXAJMSContext() throws Exception {
   Queue q = ActiveMQJMSClient.createQueue(MDBQUEUE);

   try (ClientSessionFactory sf = locator.createSessionFactory();
        ClientSession session = sf.createSession();
        ClientConsumer consVerify = session.createConsumer(MDBQUEUE);
        JMSContext jmsctx = qraConnectionFactory.createContext();
   ) {
      session.start();
      // These next 4 lines could be written in a single line however it makes difficult for debugging
      JMSProducer producer = jmsctx.createProducer();
      producer.setProperty("strvalue", "hello");
      TextMessage msgsend = jmsctx.createTextMessage("hello");
      producer.send(q, msgsend);

      ClientMessage msg = consVerify.receive(1000);
      assertNotNull(msg);
      assertEquals("hello", msg.getStringProperty("strvalue"));
   }
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Test
public void testSimpleSendNoXAJMS1() throws Exception {
   Queue q = ActiveMQJMSClient.createQueue(MDBQUEUE);
   try (ClientSessionFactory sf = locator.createSessionFactory();
        ClientSession session = sf.createSession();
        ClientConsumer consVerify = session.createConsumer(MDBQUEUE);
        Connection conn = qraConnectionFactory.createConnection();
   ) {
      Session jmsSess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
      session.start();
      MessageProducer producer = jmsSess.createProducer(q);
      // These next 4 lines could be written in a single line however it makes difficult for debugging
      TextMessage msgsend = jmsSess.createTextMessage("hello");
      msgsend.setStringProperty("strvalue", "hello");
      producer.send(msgsend);

      ClientMessage msg = consVerify.receive(1000);
      assertNotNull(msg);
      assertEquals("hello", msg.getStringProperty("strvalue"));
   }
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Override
public void individualAcknowledge(final ClientConsumer consumer, final Message message) throws ActiveMQException {
   // if we're pre-acknowledging then we don't need to do anything
   if (preAcknowledge) {
      return;
   }

   checkClosed();

   startCall();
   try {

      sessionContext.sendACK(true, blockOnAcknowledge, consumer, message);
   } finally {
      endCall();
   }
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Test
public void testWildcardRoutingHashAndStar() throws Exception {
   SimpleString addressAB = new SimpleString("a.b.c");
   SimpleString addressAC = new SimpleString("a.c");
   SimpleString address = new SimpleString("#.b.*");
   SimpleString queueName1 = new SimpleString("Q1");
   SimpleString queueName2 = new SimpleString("Q2");
   SimpleString queueName = new SimpleString("Q");
   clientSession.createQueue(addressAB, queueName1, null, false);
   clientSession.createQueue(addressAC, queueName2, null, false);
   clientSession.createQueue(address, queueName, null, false);
   ClientProducer producer = clientSession.createProducer(addressAB);
   ClientProducer producer2 = clientSession.createProducer(addressAC);
   ClientConsumer clientConsumer = clientSession.createConsumer(queueName);
   clientSession.start();
   producer.send(createTextMessage(clientSession, "m1"));
   producer2.send(createTextMessage(clientSession, "m2"));
   ClientMessage m = clientConsumer.receive(500);
   Assert.assertNotNull(m);
   Assert.assertEquals("m1", m.getBodyBuffer().readString());
   m.acknowledge();
   m = clientConsumer.receiveImmediate();
   Assert.assertNull(m);
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Test
public void testReceiveThrowsExceptionWhenHandlerSet() throws Exception {

   ClientSessionFactory cf = createSessionFactory(locator);
   ClientSession session = cf.createSession(false, true, true);
   session.createQueue(addressA, queueA, false);
   ClientConsumer cc = session.createConsumer(queueA);
   session.start();
   cc.setMessageHandler(new MessageHandler() {
      @Override
      public void onMessage(final ClientMessage message) {
      }
   });
   try {
      cc.receive();
      Assert.fail("should throw exception");
   } catch (ActiveMQIllegalStateException ise) {
      //ok
   } catch (ActiveMQException e) {
      Assert.fail("Invalid Exception type:" + e.getType());
   }
   session.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
private void checkUserReceiveNoSend(final String queue,
                                    final ClientSession connection,
                                    final ClientSession sendingConn) throws Exception {
   connection.start();
   try {
      ClientProducer prod = connection.createProducer(queue);
      ClientMessage m = connection.createMessage(false);
      try {
         prod.send(m);
         Assert.fail("should throw exception");
      } catch (ActiveMQException e) {
         // pass
      }

      prod = sendingConn.createProducer(queue);
      prod.send(m);
      ClientConsumer con = connection.createConsumer(queue);
      ClientMessage rec = con.receive(1000);
      Assert.assertNotNull(rec);
      rec.acknowledge();
   } finally {
      connection.stop();
   }
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Test
public void testClearListener() throws Exception {
   ClientSessionFactory sf = createSessionFactory(locator);

   ClientSession session = sf.createSession(false, true, true);
   session.start();

   ClientConsumer consumer = session.createConsumer(QUEUE);

   consumer.setMessageHandler(new MessageHandler() {
      @Override
      public void onMessage(final ClientMessage msg) {
      }
   });

   consumer.setMessageHandler(null);
   consumer.receiveImmediate();

   session.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
/**
 * @throws Exception
 */
private void verifyMessageOnServer(final int server, final int numberOfMessages) throws Exception {
   ServerLocator backupLocator = createInVMLocator(server);
   ClientSessionFactory factorybkp = addSessionFactory(createSessionFactory(backupLocator));
   ClientSession sessionbkp = factorybkp.createSession(false, false);
   sessionbkp.start();
   ClientConsumer consumerbkp = sessionbkp.createConsumer(ADDRESS);
   for (int i = 0; i < numberOfMessages; i++) {
      ClientMessage msg = consumerbkp.receive(1000);
      assertNotNull(msg);
      msg.acknowledge();
      sessionbkp.commit();
   }
   sessionbkp.close();
   factorybkp.close();
   backupLocator.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Test
public void testConsumeFromFormat() throws Exception {
   ActiveMQServer server = createServer(true);

   DecodeJournal.importJournal(server.getConfiguration().getJournalLocation().getAbsolutePath(), "activemq-data", "amq", 2, 102400, new StringReader(journalFile));
   DecodeJournal.importJournal(server.getConfiguration().getBindingsLocation().getAbsolutePath(), "activemq-bindings", "bindings", 2, 1048576, new StringReader(bindingsFile));
   server.start();

   ServerLocator locator = createInVMNonHALocator();
   ClientSessionFactory factory = createSessionFactory(locator);
   ClientSession session = factory.createSession();
   session.start();

   ClientConsumer consumer = session.createConsumer("A1");
   for (int i = 0; i < 10; i++) {
      ClientMessage msg = consumer.receive(5000);
      assertNotNull(msg);
      msg.acknowledge();
      assertEquals(i, msg.getIntProperty("key").intValue());
   }

   session.commit();
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Test
public void testLargeMessage() throws Exception {
   ClientProducer producer = clientSessionTxReceives.createProducer(address);
   ClientConsumer consumer = clientSessionTxReceives.createConsumer(qName1);
   SimpleString rh = new SimpleString("SMID1");

   for (int i = 0; i < 50; i++) {
      ClientMessage message = clientSession.createMessage(true);
      message.setBodyInputStream(createFakeLargeStream(300 * 1024));
      message.putStringProperty(Message.HDR_LAST_VALUE_NAME, rh);
      producer.send(message);
      clientSession.commit();
   }
   clientSessionTxReceives.start();
   ClientMessage m = consumer.receive(1000);
   Assert.assertNotNull(m);
   m.acknowledge();
   Assert.assertNull(consumer.receiveImmediate());
   clientSessionTxReceives.commit();
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
private int readMessages(final ClientSession session,
                         final ClientConsumer consumer,
                         final SimpleString queue) throws ActiveMQException {
   session.start();
   int msgs = 0;

   ClientMessage msg = null;
   do {
      msg = consumer.receive(1000);
      if (msg != null) {
         msg.acknowledge();
         if (++msgs % 10000 == 0) {
            System.out.println("received " + msgs);
            session.commit();

         }
      }
   }
   while (msg != null);

   session.commit();

   return msgs;
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
/**
 * @param numberOfMessagesExpected
 * @param queue
 * @throws ActiveMQException
 */
private void drainQueue(final int numberOfMessagesExpected, final SimpleString queue) throws ActiveMQException {
   ClientSession sess = sf.createSession(true, true);

   ClientConsumer consumer = sess.createConsumer(queue);

   sess.start();

   for (int i = 0; i < numberOfMessagesExpected; i++) {
      ClientMessage msg = consumer.receive(5000);
      Assert.assertNotNull(msg);

      if (i % 100 == 0) {
         // System.out.println("Received #" + i + "  on thread after start");
      }
      msg.acknowledge();
   }

   Assert.assertNull(consumer.receiveImmediate());

   sess.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Override
public QConsumer<ArtemisMessage> newConsumer(
        BiConsumerE<ArtemisMessage, InputStream> messageHandler)
        throws ActiveMQException {
    ClientConsumer consumer =
            session.createConsumer(config.getName(), false);
    return new ArtemisQConsumer(consumer, messageHandler);
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Override
public QConsumer<ArtemisMessage> newBrowser(
        BiConsumerE<ArtemisMessage, InputStream> messageHandler)
        throws ActiveMQException {
    ClientConsumer consumer =
            session.createConsumer(config.getName(), true);
    return new ArtemisQConsumer(consumer, messageHandler);
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
/**
 * Creates a new instance.
 * @param consumer the Artemis queue consumer.
 * @param messageHandler processes messages received by the consumer.
 * @throws NullPointerException if any argument is {@code null}.
 * @throws ActiveMQException if the message handler couldn't be connected
 * to the consumer.
 */
public ArtemisQConsumer(
        ClientConsumer consumer,
        BiConsumerE<ArtemisMessage, InputStream> messageHandler)
        throws ActiveMQException {
    requireNonNull(consumer, "consumer");
    requireNonNull(messageHandler, "messageHandler");

    this.consumer = consumer;
    this.messageHandler = messageHandler;

    consumer.setMessageHandler(this);
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Before
public void setup() throws ActiveMQException {
    mockConfig = mock(CoreQueueConfiguration.class);
    mockSession = mock(ClientSession.class);
    target = new ArtemisQConnector(mockConfig, mockSession);

    when(mockSession.createConsumer(anyString(), anyBoolean()))
            .thenReturn(mock(ClientConsumer.class));
    when(mockSession.createProducer(anyString()))
            .thenReturn(mock(ClientProducer.class));
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Before
public void setup() throws ActiveMQException {
    mockConsumer = mock(ClientConsumer.class);
    target = new ArtemisQConsumer(mockConsumer, this::handleMessage);
    receivedMessage = null;
    receivedBody = null;
}
 

import org.apache.activemq.artemis.api.core.client.ClientConsumer; //导入依赖的package包/类
@Test
public void testMulticastQueues() throws Exception {
   server.createQueue(multicastAddress, RoutingType.MULTICAST, multicastQ1, null, true, false, -1, false, true);
   server.createQueue(multicastAddress, RoutingType.MULTICAST, multicastQ2, null, true, false, -1, false, true);
   server.createQueue(multicastAddress, RoutingType.MULTICAST, multicastQ3, null, true, false, -1, false, true);

   ClientSessionFactory cf = createSessionFactory(locator);
   ClientSession session = cf.createSession();
   session.start();

   //send 3 messages
   ClientProducer producer = session.createProducer(multicastAddress);
   sendMessages(session, producer, 1);

   ClientConsumer consumer1 = session.createConsumer(toFullQN(multicastAddress, multicastQ1));
   ClientConsumer consumer2 = session.createConsumer(toFullQN(multicastAddress, multicastQ2));
   ClientConsumer consumer3 = session.createConsumer(toFullQN(multicastAddress, multicastQ3));
   session.start();

   //each consumer receives one
   ClientMessage m = consumer1.receive(2000);
   assertNotNull(m);
   m.acknowledge();
   m = consumer2.receive(2000);
   assertNotNull(m);
   m.acknowledge();
   m = consumer3.receive(2000);
   assertNotNull(m);
   m.acknowledge();

   session.commit();
   //queues are empty now
   for (SimpleString q : new SimpleString[]{multicastQ1, multicastQ2, multicastQ3}) {
      QueueQueryResult query = server.queueQuery(toFullQN(multicastAddress, q));
      assertTrue(query.isExists());
      assertEquals(multicastAddress, query.getAddress());
      assertEquals(toFullQN(multicastAddress, q), query.getName());
      assertEquals(0, query.getMessageCount());
   }
}