import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
protected void initMocks() throws ActiveMQException {
    CoreQueueConfiguration q = new CoreQueueConfiguration();
    q.setName("q");
    ClientSession sesh = mock(ClientSession.class);
    
    producer = mock(ClientProducer.class);
    when(sesh.createProducer(q.getAddress())).thenReturn(producer);
    
    msgToQueue = mock(ClientMessage.class);
    boolean durable = true;
    when(sesh.createMessage(durable)).thenReturn(msgToQueue);
    
    msgBody = mock(ActiveMQBuffer.class);
    when(msgToQueue.getBodyBuffer()).thenReturn(msgBody);
    
    connector = new ArtemisQConnector(q, sesh);
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
@Test
public void testExpireHalf() throws Exception {
   ClientProducer producer = clientSession.createProducer(qName);
   int numMessages = 100;
   long expiration = System.currentTimeMillis();
   for (int i = 0; i < numMessages; i++) {
      ClientMessage m = createTextMessage(clientSession, "m" + i);
      if (i % 2 == 0) {
         m.setExpiration(expiration);
      }
      producer.send(m);
   }
   Thread.sleep(1600);
   Assert.assertEquals(numMessages / 2, ((Queue) server.getPostOffice().getBinding(qName).getBindable()).getMessageCount());
   Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(qName).getBindable()).getDeliveringCount());
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
@Test
public void testCanNotUseAClosedProducer() throws Exception {
   final ClientProducer producer = session.createProducer(RandomUtil.randomSimpleString());

   Assert.assertFalse(producer.isClosed());

   producer.close();

   Assert.assertTrue(producer.isClosed());

   ActiveMQTestBase.expectActiveMQException(ActiveMQExceptionType.OBJECT_CLOSED, new ActiveMQAction() {
      @Override
      public void run() throws ActiveMQException {
         producer.send(session.createMessage(false));
      }
   });
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
@Test
public void testSendToDLAWhenNoRoute() throws Exception {
   SimpleString dla = new SimpleString("DLA");
   SimpleString address = new SimpleString("empty_address");
   AddressSettings addressSettings = new AddressSettings().setDeadLetterAddress(dla).setSendToDLAOnNoRoute(true);
   server.getAddressSettingsRepository().addMatch(address.toString(), addressSettings);
   SimpleString dlq = new SimpleString("DLQ1");
   clientSession.createQueue(dla, dlq, null, false);
   ClientProducer producer = clientSession.createProducer(address);
   producer.send(createTextMessage(clientSession, "heyho!"));
   clientSession.start();
   ClientConsumer clientConsumer = clientSession.createConsumer(dlq);
   ClientMessage m = clientConsumer.receive(500);
   m.acknowledge();
   Assert.assertNotNull(m);
   Assert.assertEquals(m.getBodyBuffer().readString(), "heyho!");
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
@Test
public void testClientOutgoingInterceptorRejectPacketOnBlockingSend() throws Exception {
   // must make the call block to exercise the right logic
   locator.setBlockOnNonDurableSend(true);
   ClientSessionFactory sf = createSessionFactory(locator);

   MyOutgoingInterceptor4 interceptor = new MyOutgoingInterceptor4();

   sf.getServerLocator().addOutgoingInterceptor(interceptor);

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

   session.createQueue(QUEUE, QUEUE, null, false);

   ClientProducer producer = session.createProducer(QUEUE);

   ClientMessage message = session.createMessage(false);

   try {
      producer.send(message);
      Assert.fail();
   } catch (ActiveMQException e) {
      // expected exception
      Assert.assertTrue(e.getType().getCode() == ActiveMQExceptionType.INTERCEPTOR_REJECTED_PACKET.getCode());
   }
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
@Test
public void testListMessagesWithEmptyFilter() throws Exception {
   SimpleString address = RandomUtil.randomSimpleString();
   SimpleString queue = RandomUtil.randomSimpleString();

   session.createQueue(address, queue, null, false);
   QueueControl queueControl = createManagementControl(address, queue);

   ClientProducer producer = session.createProducer(address);
   producer.send(session.createMessage(false));
   producer.send(session.createMessage(false));

   Map<String, Object>[] messages = queueControl.listMessages("");
   Assert.assertEquals(2, messages.length);

   consumeMessages(2, session, queue);

   messages = queueControl.listMessages("");
   Assert.assertEquals(0, messages.length);

   session.deleteQueue(queue);
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
@Test
public void testMulticastMessageRoutingExclusivity() throws Exception {
   ClientSession sendSession = cf.createSession(false, true, true);
   Set<RoutingType> routingTypes = new HashSet<>();
   routingTypes.add(RoutingType.ANYCAST);
   routingTypes.add(RoutingType.MULTICAST);
   sendSession.createAddress(addressA, routingTypes, false);
   sendSession.createQueue(addressA, RoutingType.ANYCAST, queueA);
   sendSession.createQueue(addressA, RoutingType.MULTICAST, queueB);
   sendSession.createQueue(addressA, RoutingType.MULTICAST, queueC);
   ClientProducer p = sendSession.createProducer(addressA);
   ClientMessage message = sendSession.createMessage(false);
   message.setRoutingType(RoutingType.MULTICAST);
   p.send(message);
   sendSession.close();
   assertTrue(Wait.waitFor(() -> server.locateQueue(queueA).getMessageCount() == 0));
   assertTrue(Wait.waitFor(() -> server.locateQueue(queueB).getMessageCount() + server.locateQueue(queueC).getMessageCount() == 2));
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
/**
 * @throws ActiveMQException
 */
private void sendMessages(int msgs) throws ActiveMQException {
   ClientProducer producer = prodSession.createProducer(ADDRESS);

   for (int i = 0; i < msgs; i++) {
      ClientMessage message = prodSession.createMessage(true);
      message.putIntProperty("prodNR", i % nThreads);
      producer.send(message);

      if (i % 100 == 0) {
         prodSession.commit();
      }
   }
   prodSession.commit();

   producer.close();
}
 

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

   server.start();
   HierarchicalRepository<Set<Role>> securityRepository = server.getSecurityRepository();
   ActiveMQJAASSecurityManager securityManager = (ActiveMQJAASSecurityManager) server.getSecurityManager();
   securityManager.getConfiguration().addUser("auser", "pass");
   Role role = new Role("arole", false, false, false, false, false, false, true, false, false, false);
   Set<Role> roles = new HashSet<>();
   roles.add(role);
   securityRepository.addMatch(configuration.getManagementAddress().toString(), roles);
   securityManager.getConfiguration().addRole("auser", "arole");
   locator.setBlockOnNonDurableSend(true);
   ClientSessionFactory cf = createSessionFactory(locator);
   ClientSession session = cf.createSession("auser", "pass", false, true, true, false, -1);
   ClientProducer cp = session.createProducer(configuration.getManagementAddress());
   cp.send(session.createMessage(false));
   session.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
@Test
public void testRemoveMessagesWithEmptyFilter() throws Exception {
   SimpleString address = RandomUtil.randomSimpleString();
   SimpleString queue = RandomUtil.randomSimpleString();

   session.createQueue(address, queue, null, false);
   ClientProducer producer = session.createProducer(address);

   // send on queue
   producer.send(session.createMessage(false));
   producer.send(session.createMessage(false));

   QueueControl queueControl = createManagementControl(address, queue);
   Assert.assertEquals(2, getMessageCount(queueControl));

   // removed matching messages to otherQueue
   int removedMatchedMessagesCount = queueControl.removeMessages("");
   Assert.assertEquals(2, removedMatchedMessagesCount);
   Assert.assertEquals(0, getMessageCount(queueControl));

   session.deleteQueue(queue);
}
 

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

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

   session.createQueue(QUEUE, QUEUE, null, true);

   ClientConsumer consumer = session.createConsumer(QUEUE);

   ClientProducer producer = session.createProducer(QUEUE);
   ClientMessage message = session.createMessage(Message.TEXT_TYPE, true, 0, System.currentTimeMillis(), (byte) 4);
   message.getBodyBuffer().writeString("hi");
   message.putStringProperty("hello", "elo");
   producer.send(message);

   ActiveMQServerImpl impl = (ActiveMQServerImpl) server;
   JournalStorageManager journal = (JournalStorageManager) impl.getStorageManager();
   JDBCJournalImpl journalimpl = (JDBCJournalImpl) journal.getMessageJournal();
   journalimpl.handleException(null, new Exception("failure"));

   Wait.waitFor(() -> !server.isStarted());

   Assert.assertFalse(server.isStarted());

}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
/**
 * @throws ActiveMQException
 */
private void addBogusData(final int nmessages, final String queue) throws ActiveMQException {
   ClientSession session = sf.createSession(false, false);
   try {
      session.createQueue(queue, queue, true);
   } catch (Exception ignored) {
   }

   ClientProducer prod = session.createProducer(queue);
   for (int i = 0; i < nmessages; i++) {
      ClientMessage msg = session.createMessage(true);
      msg.getBodyBuffer().writeBytes(new byte[1024]);
      prod.send(msg);
   }
   session.commit();

   session.start();

   ClientConsumer cons = session.createConsumer(queue);
   Assert.assertNotNull(cons.receive(1000));
   session.rollback();
   session.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
@Test
public void testConsumeFromDurableQueue() throws Exception {
   SimpleString queue = RandomUtil.randomSimpleString();
   SimpleString address = RandomUtil.randomSimpleString();

   session.createQueue(address, queue, true);

   ClientProducer producer = session.createProducer(address);
   producer.send(session.createMessage(false));

   session.start();
   ClientConsumer consumer = session.createConsumer(queue);
   ClientMessage message = consumer.receive(500);
   Assert.assertNotNull(message);
   message.acknowledge();

   consumer.close();
   session.deleteQueue(queue);
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的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.ClientProducer; //导入依赖的package包/类
public static void main(final String[] args) throws Exception {
   try {
      CrashClient.log.debug("args = " + Arrays.asList(args));

      ServerLocator locator = ActiveMQClient.createServerLocatorWithoutHA(new TransportConfiguration(NettyConnectorFactory.class.getName()));
      locator.setClientFailureCheckPeriod(ClientCrashTest.PING_PERIOD);
      locator.setConnectionTTL(ClientCrashTest.CONNECTION_TTL);
      ClientSessionFactory sf = locator.createSessionFactory();

      ClientSession session = sf.createSession(false, true, true);
      ClientProducer producer = session.createProducer(ClientCrashTest.QUEUE);

      // it has to be durable otherwise it may race dying before the client is killed
      ClientMessage message = session.createMessage(ActiveMQTextMessage.TYPE, true, 0, System.currentTimeMillis(), (byte) 1);
      message.getBodyBuffer().writeString(ClientCrashTest.MESSAGE_TEXT_FROM_CLIENT);

      producer.send(message);

      // exit without closing the session properly
      System.exit(9);
   } catch (Throwable t) {
      CrashClient.log.error(t.getMessage(), t);
      System.exit(1);
   }
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
/**
 * Creates a new instance.
 * @param producer the Artemis producer bound to the queue where we're
 *                going to put messages.
 * @param msgFactory the factory to create Artemis messages.
 * @throws ActiveMQException if the session couldn't create a producer.
 * @throws NullPointerException if any argument is {@code null}.
 */
public ArtemisQProducer(ClientProducer producer,
                        QMsgFactory<ArtemisMessage> msgFactory)
        throws ActiveMQException {
    requireNonNull(producer, "producer");
    requireNonNull(msgFactory, "msgFactory");

    this.producer = producer;
    this.msgFactory = msgFactory;
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
@Before
@SuppressWarnings("unchecked")
public void setUp() throws Exception {
    mockProducer = mock(ClientProducer.class);
    mockMsgFactory = mock(QMsgFactory.class);
    target = new ArtemisQProducer(mockProducer, mockMsgFactory);
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的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.ClientProducer; //导入依赖的package包/类
@Test
public void testConsumerAckImmediateCloseSession() throws Exception {

   ClientSessionFactory sf = createSessionFactory(locator);

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

   ClientProducer producer = session.createProducer(QUEUE);

   final int numMessages = 100;

   for (int i = 0; i < numMessages; i++) {
      ClientMessage message = createTextMessage(session, "m" + i);
      producer.send(message);
   }

   ClientConsumer consumer = session.createConsumer(QUEUE);
   session.start();
   for (int i = 0; i < numMessages; i++) {
      ClientMessage message2 = consumer.receive(1000);

      Assert.assertEquals("m" + i, message2.getBodyBuffer().readString());
      if (i < 50) {
         message2.acknowledge();
      }
   }
   // assert that all the messages are there and none have been acked
   Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getDeliveringCount());
   Assert.assertEquals(0, getMessageCount(((Queue) server.getPostOffice().getBinding(QUEUE).getBindable())));

   session.close();

   Assert.assertEquals(0, ((Queue) server.getPostOffice().getBinding(QUEUE).getBindable()).getDeliveringCount());
   Assert.assertEquals(0, getMessageCount(((Queue) server.getPostOffice().getBinding(QUEUE).getBindable())));
}
 

import org.apache.activemq.artemis.api.core.client.ClientProducer; //导入依赖的package包/类
@Test
public void testMultipleConsumersMessageOrder() throws Exception {
   ClientSession sendSession = cf.createSession(false, true, true);
   ClientSession recSession = cf.createSession(false, true, true);
   sendSession.createQueue(addressA, queueA, false);
   int numReceivers = 100;
   AtomicInteger count = new AtomicInteger(0);
   int numMessage = 10000;
   ClientConsumer[] clientConsumers = new ClientConsumer[numReceivers];
   Receiver[] receivers = new Receiver[numReceivers];
   CountDownLatch latch = new CountDownLatch(numMessage);
   for (int i = 0; i < numReceivers; i++) {
      clientConsumers[i] = recSession.createConsumer(queueA);
      receivers[i] = new Receiver(latch);
      clientConsumers[i].setMessageHandler(receivers[i]);
   }
   recSession.start();
   ClientProducer clientProducer = sendSession.createProducer(addressA);
   for (int i = 0; i < numMessage; i++) {
      ClientMessage cm = sendSession.createMessage(false);
      cm.getBodyBuffer().writeInt(count.getAndIncrement());
      clientProducer.send(cm);
   }
   Assert.assertTrue(latch.await(10, TimeUnit.SECONDS));
   for (Receiver receiver : receivers) {
      Assert.assertFalse("" + receiver.lastMessage, receiver.failed);
   }
   sendSession.close();
   recSession.close();
}