import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
/**
 * 
 * @param config
 * @return
 * @throws Exception
 */
private static ActiveMQConnectionFactory createConnectionFactory(BrokerConfig config) throws Exception {
	Map<String, Object> params = Collections.singletonMap(
			org.apache.activemq.artemis.core.remoting.impl.invm.TransportConstants.SERVER_ID_PROP_NAME, "1");
	final ActiveMQConnectionFactory connectionFactory;
	if (config.getUrl() != null) {
		connectionFactory = ActiveMQJMSClient.createConnectionFactory(config.getUrl(), null);
	} else {
		if (config.getHost() != null) {
			params.put(TransportConstants.HOST_PROP_NAME, config.getHost());
			params.put(TransportConstants.PORT_PROP_NAME, config.getPort());
		}
		if (config.isHa()) {
			connectionFactory = ActiveMQJMSClient.createConnectionFactoryWithHA(JMSFactoryType.CF, new TransportConfiguration(config.getConnectorFactory(), params));
		} else {
			connectionFactory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(config.getConnectorFactory(), params));
		}
	}
	if (config.isSecurityEnabled()) {
		connectionFactory.setUser(config.getUsername());
		connectionFactory.setPassword(config.getPassword());
	}
	return connectionFactory.disableFinalizeChecks();
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
private <T extends ActiveMQConnectionFactory> T createEmbeddedConnectionFactory(
		Class<T> factoryClass) throws Exception {
	try {
		TransportConfiguration transportConfiguration = new TransportConfiguration(
				InVMConnectorFactory.class.getName(),
				this.properties.getEmbedded().generateTransportParameters());
		ServerLocator serviceLocator = ActiveMQClient
				.createServerLocatorWithoutHA(transportConfiguration);
		return factoryClass.getConstructor(ServerLocator.class)
				.newInstance(serviceLocator);
	}
	catch (NoClassDefFoundError ex) {
		throw new IllegalStateException("Unable to create InVM "
				+ "Artemis connection, ensure that artemis-jms-server.jar "
				+ "is in the classpath", ex);
	}
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
private <T extends ActiveMQConnectionFactory> T createNativeConnectionFactory(
		Class<T> factoryClass) throws Exception {
	Map<String, Object> params = new HashMap<String, Object>();
	params.put(TransportConstants.HOST_PROP_NAME, this.properties.getHost());
	params.put(TransportConstants.PORT_PROP_NAME, this.properties.getPort());
	TransportConfiguration transportConfiguration = new TransportConfiguration(
			NettyConnectorFactory.class.getName(), params);
	Constructor<T> constructor = factoryClass.getConstructor(boolean.class,
			TransportConfiguration[].class);
	T connectionFactory = constructor.newInstance(false,
			new TransportConfiguration[] { transportConfiguration });
	String user = this.properties.getUser();
	if (StringUtils.hasText(user)) {
		connectionFactory.setUser(user);
		connectionFactory.setPassword(this.properties.getPassword());
	}
	return connectionFactory;
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Test(timeout = 60000)
public void testSendAMQPReceiveOpenWire() throws Exception {
   server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setDefaultAddressRoutingType(RoutingType.ANYCAST));

   int nMsgs = 200;

   AmqpClient client = createAmqpClient();
   AmqpConnection connection = addConnection(client.connect());
   try {
      sendMessages(nMsgs, connection);

      int count = getMessageCount(server.getPostOffice(), testQueueName);
      assertEquals(nMsgs, count);

      ConnectionFactory factory = new org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616");
      receiveJMS(nMsgs, factory);
   } finally {
      connection.close();
   }
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
private static void createJMSObjects(final int server) throws Exception {
   // Step 1. Instantiate a JMS Connection Factory object from JNDI on server 1
   ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://127.0.0.1:" + (61616 + server));

   // Step 2. We create a JMS Connection connection
   connection = connectionFactory.createConnection();

   // Step 3. We start the connection to ensure delivery occurs
   connection.start();

   // Step 4. We create a JMS Session
   session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

   // Step 5. Look-up the JMS Queue object from JNDI
   Queue queue = session.createQueue("exampleQueue");

   // Step 6. We create a JMS MessageConsumer object
   consumer = session.createConsumer(queue);

   // Step 7. We create a JMS MessageProducer object
   producer = session.createProducer(queue);
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
public static void main(final String[] args) throws Exception {
   // Step 2. Perfom a lookup on the queue
   Queue queue = ActiveMQJMSClient.createQueue("exampleQueue");

   // Step 4.Create a JMS Context using the try-with-resources statement
   try
      (
         // Even though ConnectionFactory is not closeable it would be nice to close an ActiveMQConnectionFactory
         ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
         JMSContext jmsContext = cf.createContext()
      ) {
      // Step 5. create a jms producer
      JMSProducer jmsProducer = jmsContext.createProducer();

      // Step 6. Try sending a message, we don't have the appropriate privileges to do this so this will throw an exception
      jmsProducer.send(queue, "A Message from JMS2!");

      System.out.println("Received:" + jmsContext.createConsumer(queue).receiveBody(String.class));
   }
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Override
public synchronized ActiveMQConnectionFactory recreateCF(String name,
                                                         ConnectionFactoryConfiguration cf) throws Exception {
   List<String> bindings = connectionFactoryBindings.get(name);

   if (bindings == null) {
      throw ActiveMQJMSServerBundle.BUNDLE.cfDoesntExist(name);
   }

   String[] usedBindings = bindings.toArray(new String[bindings.size()]);

   ActiveMQConnectionFactory realCF = internalCreateCFPOJO(cf);

   if (cf.isPersisted()) {
      storage.storeConnectionFactory(new PersistedConnectionFactory(cf));
      storage.addBindings(PersistedType.ConnectionFactory, cf.getName(), usedBindings);
   }

   for (String bindingsElement : usedBindings) {
      this.bindToBindings(bindingsElement, realCF);
   }

   return realCF;
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Override
@Before
public void setUp() throws Exception {
   try {
      super.setUp();
      // Deploy a connection factory with load balancing but no failover on node0
      List<String> bindings = new ArrayList<>();
      bindings.add("StrictTCKConnectionFactory");

      getJmsServerManager().createConnectionFactory("StrictTCKConnectionFactory", false, JMSFactoryType.CF, NETTY_CONNECTOR, null, ActiveMQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, ActiveMQClient.DEFAULT_CONNECTION_TTL, ActiveMQClient.DEFAULT_CALL_TIMEOUT, ActiveMQClient.DEFAULT_CALL_FAILOVER_TIMEOUT, ActiveMQClient.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT, ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ActiveMQClient.DEFAULT_COMPRESS_LARGE_MESSAGES, ActiveMQClient.DEFAULT_CONSUMER_WINDOW_SIZE, ActiveMQClient.DEFAULT_CONSUMER_MAX_RATE, ActiveMQClient.DEFAULT_CONFIRMATION_WINDOW_SIZE, ActiveMQClient.DEFAULT_PRODUCER_WINDOW_SIZE, ActiveMQClient.DEFAULT_PRODUCER_MAX_RATE, true, true, true, ActiveMQClient.DEFAULT_AUTO_GROUP, ActiveMQClient.DEFAULT_PRE_ACKNOWLEDGE, ActiveMQClient.DEFAULT_CONNECTION_LOAD_BALANCING_POLICY_CLASS_NAME, ActiveMQClient.DEFAULT_ACK_BATCH_SIZE, ActiveMQClient.DEFAULT_ACK_BATCH_SIZE, ActiveMQClient.DEFAULT_USE_GLOBAL_POOLS, ActiveMQClient.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, ActiveMQClient.DEFAULT_THREAD_POOL_MAX_SIZE, ActiveMQClient.DEFAULT_RETRY_INTERVAL, ActiveMQClient.DEFAULT_RETRY_INTERVAL_MULTIPLIER, ActiveMQClient.DEFAULT_MAX_RETRY_INTERVAL, ActiveMQClient.DEFAULT_RECONNECT_ATTEMPTS, ActiveMQClient.DEFAULT_FAILOVER_ON_INITIAL_CONNECTION, null, "/StrictTCKConnectionFactory");

      CTSMiscellaneousTest.cf = (ActiveMQConnectionFactory) getInitialContext().lookup("/StrictTCKConnectionFactory");
   } catch (Exception e) {
      e.printStackTrace();
   }

}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
private ActiveMQConnectionFactory createConnectionFactory() throws Exception {
   Map<String, Object> params = new HashMap<>();
   params.put(org.apache.activemq.artemis.core.remoting.impl.invm.TransportConstants.SERVER_ID_PROP_NAME, "1");
   final ActiveMQConnectionFactory activeMQConnectionFactory;
   if (configuration.getUrl() != null) {
      activeMQConnectionFactory = ActiveMQJMSClient.createConnectionFactory(configuration.getUrl(), null);
   } else {
      if (configuration.getHost() != null) {
         params.put(TransportConstants.HOST_PROP_NAME, configuration.getHost());
         params.put(TransportConstants.PORT_PROP_NAME, configuration.getPort());
      }
      if (configuration.isHa()) {
         activeMQConnectionFactory = ActiveMQJMSClient.createConnectionFactoryWithHA(JMSFactoryType.CF, new TransportConfiguration(configuration.getConnectorFactory(), params));
      } else {
         activeMQConnectionFactory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF, new TransportConfiguration(configuration.getConnectorFactory(), params));
      }
   }
   if (configuration.hasAuthentication()) {
      activeMQConnectionFactory.setUser(configuration.getUsername());
      activeMQConnectionFactory.setPassword(configuration.getPassword());
   }
   // The CF will probably be GCed since it was injected, so we disable the finalize check
   return activeMQConnectionFactory.disableFinalizeChecks();
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
public void performCoreManagement(ManagementCallback<ClientMessage> cb) throws Exception {

      try (ActiveMQConnectionFactory factory = createConnectionFactory();
         ServerLocator locator = factory.getServerLocator();
           ClientSessionFactory sessionFactory = locator.createSessionFactory();
           ClientSession session = sessionFactory.createSession(user, password, false, true, true, false, ActiveMQClient.DEFAULT_ACK_BATCH_SIZE)) {
         session.start();
         ClientRequestor requestor = new ClientRequestor(session, "activemq.management");
         ClientMessage message = session.createMessage(false);

         cb.setUpInvocation(message);

         ClientMessage reply = requestor.request(message);

         if (ManagementHelper.hasOperationSucceeded(reply)) {
            cb.requestSuccessful(reply);
         } else {
            cb.requestFailed(reply);
         }
      }
   }
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Test
public void testResourceAdapterSetupNoReconnectAttemptsOverride() throws Exception {
   ActiveMQResourceAdapter qResourceAdapter = new ActiveMQResourceAdapter();
   qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
   qResourceAdapter.setConnectionParameters("server-id=0");
   qResourceAdapter.setReconnectAttempts(100);
   ActiveMQRATestBase.MyBootstrapContext ctx = new ActiveMQRATestBase.MyBootstrapContext();

   qResourceAdapter.start(ctx);
   ActiveMQActivationSpec spec = new ActiveMQActivationSpec();
   spec.setResourceAdapter(qResourceAdapter);
   spec.setUseJNDI(false);
   spec.setDestinationType("javax.jms.Queue");
   spec.setDestination(MDBQUEUE);

   ActiveMQConnectionFactory fac = qResourceAdapter.getConnectionFactory(spec);

   assertEquals(100, fac.getReconnectAttempts());

   qResourceAdapter.stop();
   assertFalse(spec.isHasBeenUpdated());
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Test
public void testResourceAdapterSetupReconnectAttemptDefault() throws Exception {
   ActiveMQResourceAdapter qResourceAdapter = new ActiveMQResourceAdapter();
   qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
   qResourceAdapter.setConnectionParameters("server-id=0");
   ActiveMQRATestBase.MyBootstrapContext ctx = new ActiveMQRATestBase.MyBootstrapContext();

   qResourceAdapter.start(ctx);
   ActiveMQActivationSpec spec = new ActiveMQActivationSpec();
   spec.setResourceAdapter(qResourceAdapter);
   spec.setUseJNDI(false);
   spec.setDestinationType("javax.jms.Queue");
   spec.setDestination(MDBQUEUE);

   ActiveMQConnectionFactory fac = qResourceAdapter.getConnectionFactory(spec);

   assertEquals(-1, fac.getReconnectAttempts());

   qResourceAdapter.stop();
   assertFalse(spec.isHasBeenUpdated());
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Test
public void testResourceAdapterSetupReconnectAttemptsOverride() throws Exception {
   ActiveMQResourceAdapter qResourceAdapter = new ActiveMQResourceAdapter();
   qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
   qResourceAdapter.setConnectionParameters("server-id=0");
   ActiveMQRATestBase.MyBootstrapContext ctx = new ActiveMQRATestBase.MyBootstrapContext();

   qResourceAdapter.start(ctx);
   ActiveMQActivationSpec spec = new ActiveMQActivationSpec();
   spec.setResourceAdapter(qResourceAdapter);
   spec.setUseJNDI(false);
   spec.setDestinationType("javax.jms.Queue");
   spec.setDestination(MDBQUEUE);
   spec.setReconnectAttempts(100);
   ActiveMQConnectionFactory fac = qResourceAdapter.getConnectionFactory(spec);

   assertEquals(100, fac.getReconnectAttempts());

   qResourceAdapter.stop();
   assertTrue(spec.isHasBeenUpdated());
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Override
@Before
public void setUp() throws Exception {
   super.setUp();

   uri = new URI(scheme + "://" + hostname + ":" + port);

   server = createServer();
   server.start();

   waitForServerToStart(server.getActiveMQServer());

   connectionFactory = createConnectionFactory();

   ((ActiveMQConnectionFactory)connectionFactory).setCompressLargeMessage(isCompressLargeMessages());

   if (isSecurityEnabled()) {
      connection = connectionFactory.createConnection("brianm", "wombats");
   } else {
      connection = connectionFactory.createConnection();
   }
   session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
   queue = session.createQueue(getQueueName());
   topic = session.createTopic(getTopicName());
   connection.start();
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
protected void setupCF() throws Exception {
   if (spec.getConnectionFactoryLookup() != null) {
      Context ctx;
      if (spec.getParsedJndiParams() == null) {
         ctx = new InitialContext();
      } else {
         ctx = new InitialContext(spec.getParsedJndiParams());
      }
      Object fac = ctx.lookup(spec.getConnectionFactoryLookup());
      if (fac instanceof ActiveMQConnectionFactory) {
         // This will clone the connection factory
         // to make sure we won't close anyone's connection factory when we stop the MDB
         factory = ActiveMQJMSClient.createConnectionFactory(((ActiveMQConnectionFactory) fac).toURI().toString(), "internalConnection");
      } else {
         factory = ra.newConnectionFactory(spec);
      }
   } else {
      factory = ra.newConnectionFactory(spec);
   }
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Test
public void testConnectionFactoryJgroupsFile() throws Exception {
   createDiscoveryFactoryJGroupsFile();
   cf = (ActiveMQConnectionFactory) namingContext.lookup("/MyConnectionFactory");

   // apparently looking up the connection factory with the org.apache.activemq.artemis.jms.tests.tools.container.InVMInitialContextFactory
   // is not enough to actually serialize it so we serialize it manually
   byte[] x = serialize(cf);
   ActiveMQConnectionFactory y = deserialize(x, ActiveMQConnectionFactory.class);
   checkEquals(cf, y);
   DiscoveryGroupConfiguration dgc = y.getDiscoveryGroupConfiguration();
   Assert.assertEquals(dgc.getName(), "dg1");
   Assert.assertEquals(dgc.getDiscoveryInitialWaitTimeout(), 5000);
   Assert.assertEquals(dgc.getRefreshTimeout(), 5000);
   Assert.assertTrue(dgc.getBroadcastEndpointFactory() instanceof JGroupsFileBroadcastEndpointFactory);
   JGroupsFileBroadcastEndpointFactory befc = (JGroupsFileBroadcastEndpointFactory) dgc.getBroadcastEndpointFactory();
   Assert.assertEquals("myChannel", befc.getChannelName());
   Assert.assertEquals("/META-INF/myfile.xml", befc.getFile());
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Test
public void testDefaultConstructor() throws Exception {
   ActiveMQConnectionFactory cf = ActiveMQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType.CF);
   assertFactoryParams(cf, null, null, null, ActiveMQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD, ActiveMQClient.DEFAULT_CONNECTION_TTL, ActiveMQClient.DEFAULT_CALL_TIMEOUT, ActiveMQClient.DEFAULT_CALL_FAILOVER_TIMEOUT, ActiveMQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE, ActiveMQClient.DEFAULT_CONSUMER_WINDOW_SIZE, ActiveMQClient.DEFAULT_CONSUMER_MAX_RATE, ActiveMQClient.DEFAULT_CONFIRMATION_WINDOW_SIZE, ActiveMQClient.DEFAULT_PRODUCER_MAX_RATE, ActiveMQClient.DEFAULT_BLOCK_ON_ACKNOWLEDGE, ActiveMQClient.DEFAULT_BLOCK_ON_DURABLE_SEND, ActiveMQClient.DEFAULT_BLOCK_ON_NON_DURABLE_SEND, ActiveMQClient.DEFAULT_AUTO_GROUP, ActiveMQClient.DEFAULT_PRE_ACKNOWLEDGE, ActiveMQClient.DEFAULT_CONNECTION_LOAD_BALANCING_POLICY_CLASS_NAME, ActiveMQClient.DEFAULT_ACK_BATCH_SIZE, ActiveMQClient.DEFAULT_ACK_BATCH_SIZE, ActiveMQClient.DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT, ActiveMQClient.DEFAULT_USE_GLOBAL_POOLS, ActiveMQClient.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, ActiveMQClient.DEFAULT_THREAD_POOL_MAX_SIZE, ActiveMQClient.DEFAULT_RETRY_INTERVAL, ActiveMQClient.DEFAULT_RETRY_INTERVAL_MULTIPLIER, ActiveMQClient.DEFAULT_RECONNECT_ATTEMPTS);
   Connection conn = null;

   try {
      conn = cf.createConnection();

      conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

      Assert.fail("Should throw exception");
   } catch (JMSException e) {
      // Ok
   }
   if (conn != null) {
      conn.close();
   }

   ActiveMQConnectionFactoryTest.log.info("Got here");

   testSettersThrowException(cf);
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Test
public void testConnectionFactoryUDP() throws Exception {
   createDiscoveryFactoryUDP();
   cf = (ActiveMQConnectionFactory) namingContext.lookup("/MyConnectionFactory");

   // apparently looking up the connection factory with the org.apache.activemq.artemis.jms.tests.tools.container.InVMInitialContextFactory
   // is not enough to actually serialize it so we serialize it manually
   byte[] x = serialize(cf);
   ActiveMQConnectionFactory y = deserialize(x, ActiveMQConnectionFactory.class);
   checkEquals(cf, y);
   DiscoveryGroupConfiguration dgc = y.getDiscoveryGroupConfiguration();
   Assert.assertEquals(dgc.getName(), "dg1");
   Assert.assertEquals(dgc.getDiscoveryInitialWaitTimeout(), 5000);
   Assert.assertEquals(dgc.getRefreshTimeout(), 5000);
   Assert.assertTrue(dgc.getBroadcastEndpointFactory() instanceof UDPBroadcastEndpointFactory);
   UDPBroadcastEndpointFactory befc = (UDPBroadcastEndpointFactory) dgc.getBroadcastEndpointFactory();
   Assert.assertEquals(Integer.parseInt(System.getProperty("org.apache.activemq.artemis.api.core.UDPBroadcastEndpointFactory.localBindPort", "-1")), befc.getLocalBindPort());
   Assert.assertEquals(System.getProperty("org.apache.activemq.artemis.api.core.UDPBroadcastEndpointFactory.localBindAddress"), befc.getLocalBindAddress());
   Assert.assertEquals(getUDPDiscoveryPort(), befc.getGroupPort());
   Assert.assertEquals(getUDPDiscoveryAddress(), befc.getGroupAddress());
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Override
protected ActiveMQConnectionFactory internalNewObject(URI uri,
                                                      Map<String, String> query,
                                                      String name) throws Exception {
   JMSConnectionOptions options = newConectionOptions(uri, query);

   List<TransportConfiguration> configurations = TCPTransportConfigurationSchema.getTransportConfigurations(uri, query, TransportConstants.ALLOWABLE_CONNECTOR_KEYS, name, NettyConnectorFactory.class.getName());

   TransportConfiguration[] tcs = new TransportConfiguration[configurations.size()];

   configurations.toArray(tcs);

   ActiveMQConnectionFactory factory;

   if (options.isHa()) {
      factory = ActiveMQJMSClient.createConnectionFactoryWithHA(options.getFactoryTypeEnum(), tcs);
   } else {
      factory = ActiveMQJMSClient.createConnectionFactoryWithoutHA(options.getFactoryTypeEnum(), tcs);
   }

   return BeanSupport.setData(uri, factory, query);
}
 

import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory; //导入依赖的package包/类
@Test
public void testPreCommitAcksSetOnConnectionFactory() throws Exception {
   ((ActiveMQConnectionFactory) cf).setPreAcknowledge(true);
   conn = cf.createConnection();

   Session session = conn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
   jBossQueue = ActiveMQJMSClient.createQueue(JmsConsumerTest.Q_NAME);
   MessageProducer producer = session.createProducer(jBossQueue);
   MessageConsumer consumer = session.createConsumer(jBossQueue);
   int noOfMessages = 100;
   for (int i = 0; i < noOfMessages; i++) {
      producer.send(session.createTextMessage("m" + i));
   }

   conn.start();
   for (int i = 0; i < noOfMessages; i++) {
      Message m = consumer.receive(500);
      Assert.assertNotNull(m);
   }

   // Messages should all have been acked since we set pre ack on the cf
   SimpleString queueName = new SimpleString(JmsConsumerTest.Q_NAME);
   Queue queue = server.locateQueue(queueName);
   Wait.assertEquals(0, queue::getDeliveringCount);
   Wait.assertEquals(0, queue::getMessageCount);
}