import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的package包/类
private static ClientSession startSession(
        ClientSessionFactory csf,
        FunctionE<ClientSessionFactory, ClientSession> createSession)
            throws ActiveMQException {
    boolean created = false;
    try {
        ClientSession session = createSession.apply(csf);
        session.start();
        created = true;

        return session;
    } finally {
        if (!created) {    // (1)
            csf.close();
        }
    }
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的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.ClientSession; //导入依赖的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.ClientSession; //导入依赖的package包/类
protected final ClientSessionFactoryInternal createSessionFactoryAndWaitForTopology(ServerLocator locator,
                                                                                    TransportConfiguration transportConfiguration,
                                                                                    int topologyMembers) throws Exception {
   CountDownLatch countDownLatch = new CountDownLatch(topologyMembers * 2);

   locator.addClusterTopologyListener(new LatchClusterTopologyListener(countDownLatch));

   ClientSessionFactoryInternal sf = (ClientSessionFactoryInternal) locator.createSessionFactory(transportConfiguration);
   addSessionFactory(sf);

   assertTrue("topology members expected " + topologyMembers, countDownLatch.await(5, TimeUnit.SECONDS));

   closeSessionFactory(sf);

   sf = (ClientSessionFactoryInternal) locator.createSessionFactory(liveServer.getServer().getNodeID().toString());
   addSessionFactory(sf);

   if (sf2 == null) {
      sf2 = (ClientSessionFactoryInternal) locator.createSessionFactory(backupServer.getServer().getNodeID().toString());

      ClientSession session2 = createSession(sf2, false, false);
      session2.createQueue(ADDRESS, ADDRESS, null, true);
      addSessionFactory(sf2);
   }
   return sf;
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的package包/类
@Test
public void testAutClose() throws Exception {
   ServerLocator locatorx;
   ClientSession sessionx;
   ClientSessionFactory factoryx;
   try (ServerLocator locator = createInVMNonHALocator();
        ClientSessionFactory factory = locator.createSessionFactory();
        ClientSession session = factory.createSession(false, false)) {
      locatorx = locator;
      sessionx = session;
      factoryx = factory;
   }

   assertTrue(locatorx.isClosed());
   assertTrue(sessionx.isClosed());
   assertTrue(factoryx.isClosed());
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的package包/类
public void deploy(QueueDeployment queueDeployment) throws Exception {
   if (!started) {
      throw new Exception("You must start() this class instance before deploying");
   }
   String queueName = queueDeployment.getName();
   try (ClientSession session = sessionFactory.createSession(false, false, false)) {
      ClientSession.AddressQuery query = session.addressQuery(SimpleString.toSimpleString(queueName));
      if (!query.isExists()) {
         session.createAddress(SimpleString.toSimpleString(queueName), RoutingType.ANYCAST, true);
         session.createQueue(SimpleString.toSimpleString(queueName), RoutingType.ANYCAST, SimpleString.toSimpleString(queueName), queueDeployment.isDurableSend());
      } else {
         ClientSession.QueueQuery qquery = session.queueQuery(SimpleString.toSimpleString(queueName));
         if (!qquery.isExists()) {
            session.createQueue(SimpleString.toSimpleString(queueName), RoutingType.ANYCAST, SimpleString.toSimpleString(queueName), queueDeployment.isDurableSend());
         }
      }
   }

   destination.createQueueResource(queueName, queueDeployment.isDurableSend(), queueDeployment.getConsumerSessionTimeoutSeconds(), queueDeployment.isDuplicatesAllowed());

}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的package包/类
private void setupServer(final JournalType journalType) throws Exception {
   Configuration config = createDefaultNettyConfig().setJournalType(journalType).setJournalFileSize(ActiveMQDefaultConfiguration.getDefaultJournalFileSize()).setJournalMinFiles(ActiveMQDefaultConfiguration.getDefaultJournalMinFiles()).setJournalCompactMinFiles(2).setJournalCompactPercentage(50);

   server = createServer(true, config);

   server.start();

   ServerLocator locator = createNettyNonHALocator().setBlockOnDurableSend(false).setBlockOnNonDurableSend(false).setBlockOnAcknowledge(false);

   sf = createSessionFactory(locator);

   ClientSession sess = sf.createSession();

   try {
      sess.createQueue(ADDRESS, QUEUE, true);
   } catch (Exception ignored) {
   }

   sess.close();
   locator.close();
   locator = createInVMNonHALocator();
   sf = createSessionFactory(locator);
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的package包/类
@Test
public void testClosingTemporaryTopicDeletesQueue() throws JMSException, ActiveMQException {
   conn = cf.createConnection();

   Assert.assertEquals(1, server.getRemotingService().getConnections().size());

   session1 = (ActiveMQSession) conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
   ActiveMQTemporaryTopic topic = (ActiveMQTemporaryTopic) session1.createTemporaryTopic();
   String address = topic.getAddress();
   session1.close();
   conn.close();
   conn2 = cf.createConnection();
   session2 = (ActiveMQSession) conn2.createSession(false, Session.AUTO_ACKNOWLEDGE);

   ClientSession cs = session2.getCoreSession();
   try {
      cs.createConsumer(address);
      fail("the address from the TemporaryTopic still exists!");
   } catch (ActiveMQException e) {
      assertEquals("expecting 'queue does not exist'", ActiveMQExceptionType.QUEUE_DOES_NOT_EXIST, e.getType());
   }
}
 

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

   session = sessionFactory.createSession(true, false, false);

   Xid xid = newXID();

   session.start(xid, XAResource.TMNOFLAGS);

   try {
      session.start(xid, XAResource.TMRESUME);
      Assert.fail("XAException expected");
   } catch (XAException e) {
      Assert.assertEquals(XAException.XAER_PROTO, e.errorCode);
   }

   session.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的package包/类
@Test
public void testOneWaySSL() throws Exception {
   createCustomSslServer();
   String text = RandomUtil.randomString();

   tc.getParams().put(TransportConstants.SSL_ENABLED_PROP_NAME, true);
   tc.getParams().put(TransportConstants.TRUSTSTORE_PROVIDER_PROP_NAME, storeType);
   tc.getParams().put(TransportConstants.TRUSTSTORE_PATH_PROP_NAME, CLIENT_SIDE_TRUSTSTORE);
   tc.getParams().put(TransportConstants.TRUSTSTORE_PASSWORD_PROP_NAME, PASSWORD);

   ServerLocator locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(tc));
   ClientSessionFactory sf = addSessionFactory(createSessionFactory(locator));
   ClientSession session = addClientSession(sf.createSession(false, true, true));
   session.createQueue(CoreClientOverOneWaySSLTest.QUEUE, CoreClientOverOneWaySSLTest.QUEUE, false);
   ClientProducer producer = addClientProducer(session.createProducer(CoreClientOverOneWaySSLTest.QUEUE));

   ClientMessage message = createTextMessage(session, text);
   producer.send(message);

   ClientConsumer consumer = addClientConsumer(session.createConsumer(CoreClientOverOneWaySSLTest.QUEUE));
   session.start();

   ClientMessage m = consumer.receive(1000);
   Assert.assertNotNull(m);
   Assert.assertEquals(text, m.getBodyBuffer().readString());
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的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.ClientSession; //导入依赖的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.ClientSession; //导入依赖的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.ClientSession; //导入依赖的package包/类
@Test
public void testRouteToSingleTemporaryQueue() throws Exception {
   ClientSession sendSession = cf.createSession(false, true, true);
   sendSession.createTemporaryQueue(addressA, queueA);
   int numMessages = 300;
   ClientProducer p = sendSession.createProducer(addressA);
   for (int i = 0; i < numMessages; i++) {
      p.send(sendSession.createMessage(false));
   }
   ClientSession session = cf.createSession(false, true, true);
   ClientConsumer c1 = session.createConsumer(queueA);
   session.start();
   for (int i = 0; i < numMessages; i++) {
      ClientMessage m = c1.receive(5000);
      Assert.assertNotNull(m);
      m.acknowledge();
   }
   Assert.assertNull(c1.receiveImmediate());
   sendSession.close();
   session.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的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.ClientSession; //导入依赖的package包/类
@Test
public void testCreateSessionWithNullUserPass() throws Exception {
   ActiveMQServer server = createServer();
   ActiveMQJAASSecurityManager securityManager = (ActiveMQJAASSecurityManager) server.getSecurityManager();
   securityManager.getConfiguration().addUser("guest", "guest");
   securityManager.getConfiguration().setDefaultUser("guest");
   server.start();
   ClientSessionFactory cf = createSessionFactory(locator);

   try {
      ClientSession session = cf.createSession(false, true, true);

      session.close();
   } catch (ActiveMQException e) {
      Assert.fail("should not throw exception");
   }
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的package包/类
public Object invokeOperation(final Class desiredType,
                              final String operationName,
                              final Object... args) throws Exception {
   try (ClientSessionFactory sessionFactory = locator.createSessionFactory();
        ClientSession session = getSession(sessionFactory);
        ClientRequestor requestor = getClientRequestor(session)) {
      ClientMessage m = session.createMessage(false);
      ManagementHelper.putOperationInvocation(m, resourceName, operationName, args);
      ClientMessage reply = requestor.request(m);
      if (reply != null) {
         if (ManagementHelper.hasOperationSucceeded(reply)) {
            return ManagementHelper.getResult(reply, desiredType);
         } else {
            throw new Exception((String) ManagementHelper.getResult(reply));
         }
      } else {
         return null;
      }
   }
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的package包/类
@Test
public void testCreateTempQueueWithRole() 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, true, false, false, false, false, false);
   Set<Role> roles = new HashSet<>();
   roles.add(role);
   securityRepository.addMatch(SecurityTest.addressA, roles);
   securityManager.getConfiguration().addRole("auser", "arole");
   ClientSessionFactory cf = createSessionFactory(locator);
   ClientSession session = cf.createSession("auser", "pass", false, true, true, false, -1);
   session.createQueue(SecurityTest.addressA, SecurityTest.queueA, false);
   session.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的package包/类
@Test
public void testCreateTempQueueWithoutRole() 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, false, false, false, false);
   Set<Role> roles = new HashSet<>();
   roles.add(role);
   securityRepository.addMatch(SecurityTest.addressA, roles);
   securityManager.getConfiguration().addRole("auser", "arole");
   ClientSessionFactory cf = createSessionFactory(locator);
   ClientSession session = cf.createSession("auser", "pass", false, true, true, false, -1);
   try {
      session.createQueue(SecurityTest.addressA, SecurityTest.queueA, false);
      Assert.fail("should throw exception");
   } catch (ActiveMQSecurityException se) {
      //ok
   } catch (ActiveMQException e) {
      fail("Invalid Exception type:" + e.getType());
   }
   session.close();
}
 

import org.apache.activemq.artemis.api.core.client.ClientSession; //导入依赖的package包/类
/**
 * Connects to the Artemis server and starts a client session.
 * @param locator locates the Artemis server.
 * @param createSession factory method to create a new session.
 * @throws Exception if the connection could not be established or the
 * session could not be started.
 * @throws NullPointerException if any argument is {@code null}.
 */
public ServerConnector(
        ServerLocator locator,
        FunctionE<ClientSessionFactory, ClientSession> createSession)
            throws Exception {
    requireNonNull(locator, "locator");
    requireNonNull(createSession, "createSession");
    
    this.factory = locator.createSessionFactory();
    this.session = startSession(factory, createSession);
}