@Test
public void testHandlerOnSendWithAnonymousProducer() throws Exception {
   final int MSG_COUNT = 750;
   ServerLocator locator = createInVMNonHALocator();
   locator.setConfirmationWindowSize(256);

   ClientSessionFactory factory = locator.createSessionFactory();
   ClientSession session = factory.createSession();
   final AtomicInteger count = new AtomicInteger(0);
   ClientProducer producer = session.createProducer();
   for (int i = 0; i < MSG_COUNT; i++) {
      ClientMessage message = session.createMessage(true);
      producer.send(address, message, message1 -> count.incrementAndGet());
   }
   Wait.assertEquals(MSG_COUNT, () -> count.get(), 2000, 100);
}
 

@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();
}
 

@Test
public void testRouteToSingleNonDurableQueue() throws Exception {
   ClientSession sendSession = cf.createSession(false, true, true);
   sendSession.createQueue(new QueueConfiguration(queueA).setAddress(addressA).setDurable(false));
   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();
}
 

public void sendMessages(final int start, final int end) throws Exception {
   try (ClientSession session = factory.createSession(false, false)) {

      try {
         session.createQueue(new QueueConfiguration(QUEUE));
      } catch (Exception ignored) {
      }

      ClientProducer prod = session.createProducer(QUEUE);

      for (int i = start; i < end; i++) {
         ClientMessage msg = session.createMessage(true);
         msg.putIntProperty(new SimpleString("key"), i);
         msg.getBodyBuffer().writeUTF("message " + i);
         prod.send(msg);
      }

      session.commit();
      session.close();
      // server.stop(); -- this test was not supposed to stop the server, it should crash
   }
}
 

@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");
   }
}
 

public static void main(final String[] args) throws Exception {
   try {
      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(OK);
   } catch (Throwable t) {
      CrashClient.log.error(t.getMessage(), t);
      System.exit(NOT_OK);
   }
}
 

private void sendMessages() throws Exception {
   ClientSession session = sf.createSession(true, true);

   String filter = null;
   session.createAddress(SimpleString.toSimpleString(ADDRESS), RoutingType.MULTICAST, false);
   session.createQueue(new QueueConfiguration(ADDRESS).setRoutingType(RoutingType.MULTICAST).setFilterString(filter));
   ClientProducer producer = session.createProducer(ADDRESS);

   for (int i = 0; i < numMessages; i++) {
      ClientMessage message = session.createMessage(true);
      setBody(i, message);
      message.putIntProperty("int", i);
      message.putShortProperty("short", (short) i);
      message.putByteProperty("byte", (byte) i);
      message.putFloatProperty("float", floatValue(i));
      message.putStringProperty(SIMPLE_STRING_KEY, new SimpleString(Integer.toString(i)));
      message.putBytesProperty("byte[]", byteArray(i));
      message.putObjectProperty("null-value", null);
      producer.send(message);
   }
   session.commit();
}
 

@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();
}
 

@Test
@BMRules(
   rules = {@BMRule(
      name = "StompProtocolManager Leak Server Rule",
      targetClass = "org.apache.activemq.artemis.core.protocol.stomp.StompProtocolManager",
      targetMethod = "onNotification(org.apache.activemq.artemis.core.server.management.Notification)",
      targetLocation = "EXIT",
      helper = "org.apache.activemq.artemis.tests.extras.byteman.StompInternalStateTest",
      action = "verifyBindingAddRemove($1, $0.destinations)")})
public void testStompProtocolManagerLeak() throws Exception {
   ClientSession session = null;
   try {
      assertNull(resultTestStompProtocolManagerLeak);
      ServerLocator locator = createNettyNonHALocator();
      ClientSessionFactory factory = createSessionFactory(locator);
      session = factory.createSession();
      session.createQueue(new QueueConfiguration(STOMP_QUEUE_NAME).setDurable(false).setTemporary(true));
      session.deleteQueue(STOMP_QUEUE_NAME);

      assertNull(resultTestStompProtocolManagerLeak);
   } finally {
      if (session != null) {
         session.close();
      }
   }
}
 

@Test
@BMRules(
   rules = {@BMRule(
      name = "trace ClientBootstrap.connect",
      targetClass = "org.jboss.netty.bootstrap.ClientBootstrap",
      targetMethod = "connect",
      targetLocation = "ENTRY",
      action = "org.apache.activemq.artemis.tests.extras.byteman.LatencyTest.debugLog(\"netty connecting\")"), @BMRule(
      name = "sleep OioWorker.run",
      targetClass = "org.jboss.netty.channel.socket.oio.OioWorker",
      targetMethod = "run",
      targetLocation = "ENTRY",
      action = "Thread.sleep(500)")})
public void testLatency() throws Exception {
   ActiveMQServer server = createServer(createDefaultNettyConfig());
   server.start();
   ServerLocator locator = createNettyNonHALocator();
   ClientSessionFactory factory = createSessionFactory(locator);
   ClientSession session = factory.createSession();
   session.close();
   server.stop();
}
 

@Test
public void testStart() throws Exception {
   cf = createSessionFactory(locator);
   ClientSession clientSession = cf.createSession(false, true, true);
   if (legacyCreateQueue) {
      clientSession.createQueue(queueName, queueName, false);
   } else {
      clientSession.createQueue(new QueueConfiguration(queueName).setDurable(false));
   }
   clientSession.start();
   clientSession.close();
}
 

@Override
public MessageQueue create(String name, MessageHandler messageHandler) throws Exception {
    final String queueName = format(QUEUE_NAME_FORMAT,this.clusterName,name);
    ClientSession clientSession =  clientSessionFactory.createSession(activeMQUsername, activeMQPassword, false, true, true, false, serverLocator.getAckBatchSize());
    ensureQueueExists(clientSession, queueName, name);
    return new RemoteMessageQueue(queueName, name, clientSession, clientSession.createProducer(format(EA_ADDRESS_FORMAT, clusterName)));
}
 

@Test
public void testJAASSecurityManagerAuthentication() throws Exception {
   ActiveMQJAASSecurityManager securityManager = new ActiveMQJAASSecurityManager();
   ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig().setSecurityEnabled(true).addAcceptorConfiguration("netty", "tcp://127.0.0.1:61616?securityDomain=PropertiesLogin"), ManagementFactory.getPlatformMBeanServer(), securityManager, false));
   server.start();
   ClientSessionFactory cf = createSessionFactory(locator);

   try {
      ClientSession session = cf.createSession("first", "secret", false, true, true, false, 0);
      session.close();
   } catch (ActiveMQException e) {
      e.printStackTrace();
      Assert.fail("should not throw exception");
   }
}
 

@Test
public void testDefaultConsumerWindowSize() throws Exception {
   ActiveMQServer messagingService = createServer(false, isNetty());

   messagingService.start();
   messagingService.createQueue(new QueueConfiguration(queueA).setRoutingType(RoutingType.ANYCAST));

   ClientSessionFactory cf = createSessionFactory(locator);
   ClientSession session = cf.createSession(false, true, true);
   ClientConsumerImpl consumer = (ClientConsumerImpl) session.createConsumer(queueA);

   consumer.start();

   assertEquals(ActiveMQClient.DEFAULT_CONSUMER_WINDOW_SIZE / 2, consumer.getClientWindowSize());
}
 

@Test
public void testBasicReceive() throws Exception {
   ClientSessionFactory cf = createSessionFactory(locator);
   ClientSession sendSession = cf.createSession(false, true, true);
   ClientProducer cp = sendSession.createProducer(addressA);
   ClientSession session = cf.createSession(false, true, true);
   session.createQueue(new QueueConfiguration(queueA).setAddress(addressA).setDurable(false));
   ClientConsumer cc = session.createConsumer(queueA);
   session.start();
   cp.send(sendSession.createMessage(false));
   Assert.assertNotNull(cc.receive());
   session.close();
   sendSession.close();
}
 

@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())));
}
 

@Test
public void testQueueQueryNoQ() throws Exception {
   server.getAddressSettingsRepository().addMatch("#", new AddressSettings().setAutoCreateQueues(false));
   cf = createSessionFactory(locator);
   ClientSession clientSession = cf.createSession(false, true, true);
   QueueQuery resp = clientSession.queueQuery(new SimpleString(queueName));
   Assert.assertFalse(resp.isExists());
   Assert.assertFalse(resp.isAutoCreateQueues());
   Assert.assertEquals(queueName, resp.getAddress().toString());
   clientSession.close();
}
 

/**
 * Setup a session
 *
 * @param cf
 * @return The connection
 * @throws Exception Thrown if an error occurs
 */
protected ClientSession setupSession(ClientSessionFactory cf) throws Exception {
   ClientSession result = null;

   try {
      result = ra.createSession(cf, spec.getAcknowledgeModeInt(), spec.getUser(), spec.getPassword(), ra.getPreAcknowledge(), ra.getDupsOKBatchSize(), ra.getTransactionBatchSize(), isDeliveryTransacted, spec.isUseLocalTx(), spec.getTransactionTimeout());

      result.addMetaData("resource-adapter", "inbound");
      result.addMetaData(ClientSession.JMS_SESSION_IDENTIFIER_PROPERTY, "");
      String clientID = ra.getClientID() == null ? spec.getClientID() : ra.getClientID();
      if (clientID != null) {
         result.addMetaData(ActiveMQConnection.JMS_SESSION_CLIENT_ID_PROPERTY, clientID);
      }

      logger.debug("Using queue connection " + result);

      return result;
   } catch (Throwable t) {
      try {
         if (result != null) {
            result.close();
         }
      } catch (Exception e) {
         logger.trace("Ignored error closing connection", e);
      }
      if (t instanceof Exception) {
         throw (Exception) t;
      }
      throw new RuntimeException("Error configuring connection", t);
   }
}
 

@Test
public void testCreateQueue() throws Exception {
   liveJMSServer.createQueue(true, "queue1", null, true, "/queue/queue1");
   assertNotNull(ctx1.lookup("/queue/queue1"));

   ActiveMQConnectionFactory jbcf = ActiveMQJMSClient.createConnectionFactoryWithHA(JMSFactoryType.CF, livetc);

   jbcf.setReconnectAttempts(-1);

   Connection conn = null;

   try {
      conn = JMSUtil.createConnectionAndWaitForTopology(jbcf, 2, 5);

      Session sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

      ClientSession coreSession = ((ActiveMQSession) sess).getCoreSession();

      JMSUtil.crash(liveServer, coreSession);

      assertNotNull(ctx2.lookup("/queue/queue1"));
   } finally {
      if (conn != null) {
         conn.close();
      }
   }
}
 

@Test
@BMRules(
   rules = {@BMRule(
      name = "trace clientsessionimpl commit",
      targetClass = "org.apache.activemq.artemis.core.client.impl.ClientSessionImpl",
      targetMethod = "commit",
      targetLocation = "ENTRY",
      action = "org.apache.activemq.artemis.tests.extras.byteman.BMFailoverTest.serverToStop.getServer().stop(true)")})
public void testFailoverOnReceiveCommit() throws Exception {
   serverToStop = liveServer;
   locator = getServerLocator();
   createSessionFactory();
   ClientSession session = createSessionAndQueue();

   ClientSession sendSession = createSession(sf, true, true);

   ClientProducer producer = addClientProducer(sendSession.createProducer(FailoverTestBase.ADDRESS));

   sendMessages(sendSession, producer, 10);

   ClientConsumer consumer = session.createConsumer(FailoverTestBase.ADDRESS);
   session.start();
   for (int i = 0; i < 10; i++) {
      ClientMessage m = consumer.receive(500);
      assertNotNull(m);
      m.acknowledge();
   }
   try {
      session.commit();
      fail("should have thrown an exception");
   } catch (ActiveMQTransactionOutcomeUnknownException e) {
      //pass
   } catch (ActiveMQTransactionRolledBackException e1) {
      //pass
   }
   Queue bindable = (Queue) backupServer.getServer().getPostOffice().getBinding(FailoverTestBase.ADDRESS).getBindable();
   assertEquals(10, getMessageCount(bindable));

}
 

@Test
public void testProperties() throws Exception {
   final String testAddress = "testAddress";
   final SimpleString queue = SimpleString.toSimpleString("queue");
   final int COUNT = 25;

   ActiveMQServer server = addServer(ActiveMQServers.newActiveMQServer(createDefaultInVMConfig(), false));
   server.start();

   server.createQueue(new QueueConfiguration(queue).setAddress(testAddress + (COUNT)).setRoutingType(RoutingType.ANYCAST));
   for (int i = 0; i < COUNT; i++) {
      server.deployDivert(new DivertConfiguration()
                             .setName("divert" + i)
                             .setAddress(testAddress + i)
                             .setForwardingAddress(testAddress + (i + 1)));
   }

   ServerLocator locator = createInVMNonHALocator();
   ClientSessionFactory sf = createSessionFactory(locator);
   ClientSession session = sf.createSession(false, true, true);
   session.start();

   ClientProducer producer = session.createProducer(new SimpleString(testAddress + "0"));
   ClientConsumer consumer1 = session.createConsumer(queue);
   ClientMessage message = session.createMessage(false);
   producer.send(message);

   message = consumer1.receive(DivertTest.TIMEOUT);
   Assert.assertNotNull(message);
   message.acknowledge();
   Assert.assertEquals("testAddress" + COUNT, message.getAddress());
   Assert.assertEquals("testAddress" + (COUNT - 1), message.getStringProperty(Message.HDR_ORIGINAL_ADDRESS));
}
 

@Test
public void testProperReloadWhenAddingUserViaManagement() throws Exception {
   Run.setEmbedded(true);
   File instance1 = new File(temporaryFolder.getRoot(), "instance_user");
   System.setProperty("java.security.auth.login.config", instance1.getAbsolutePath() + "/etc/login.config");
   Artemis.main("create", instance1.getAbsolutePath(), "--silent", "--no-autotune", "--no-web", "--no-amqp-acceptor", "--no-mqtt-acceptor", "--no-stomp-acceptor", "--no-hornetq-acceptor", "--require-login");
   System.setProperty("artemis.instance", instance1.getAbsolutePath());
   Object result = Artemis.internalExecute("run");
   ActiveMQServer activeMQServer = ((Pair<ManagementContext, ActiveMQServer>)result).getB();
   ActiveMQServerControl activeMQServerControl = activeMQServer.getActiveMQServerControl();

   ServerLocator serverLocator = ActiveMQClient.createServerLocator("tcp://localhost:61616");
   ClientSessionFactory sessionFactory = serverLocator.createSessionFactory();

   try {
      // this will force a properties "reload" event (i.e. initial loading)
      sessionFactory.createSession("foo", "bar", false, false, false, false, 0);
      fail("Should have failed to create session here due to security");
   } catch (Exception e) {
      // ignore
   }

   try {
      activeMQServerControl.createAddress("myAddress", RoutingType.ANYCAST.toString());
      activeMQServerControl.addSecuritySettings("myAddress", "myRole", "myRole", "myRole", "myRole", "myRole", "myRole", "myRole", "myRole", "myRole", "myRole");
      // change properties files which should cause another "reload" event
      activeMQServerControl.addUser("foo", "bar", "myRole", true);
      ClientSession session = sessionFactory.createSession("foo", "bar", false, false, false, false, 0);
      session.createQueue("myAddress", RoutingType.ANYCAST, "myQueue", true);
      ClientProducer producer = session.createProducer("myAddress");
      producer.send(session.createMessage(true));
      session.close();
   } finally {
      sessionFactory.close();
      serverLocator.close();
      stopServer();
   }
}
 

@Test
public void testMulticastRoutingBackwardsCompat() throws Exception {

   SimpleString sendAddress = new SimpleString("test.address");

   List<String> testAddresses = Arrays.asList("test.address", "test.#", "test.*");

   for (String consumeAddress : testAddresses) {

      // For each address, create 2 Queues with the same address, assert both queues receive message
      Queue q1 = server.createQueue(new QueueConfiguration(new SimpleString(consumeAddress + ".1")).setAddress(new SimpleString(consumeAddress)).setRoutingType(RoutingType.MULTICAST));
      Queue q2 = server.createQueue(new QueueConfiguration(new SimpleString(consumeAddress + ".2")).setAddress(new SimpleString(consumeAddress)).setRoutingType(RoutingType.MULTICAST));

      ClientSession session = sessionFactory.createSession();
      session.start();

      ClientConsumer consumer1 = session.createConsumer(q1.getName());
      ClientConsumer consumer2 = session.createConsumer(q2.getName());

      ClientProducer producer = session.createProducer(sendAddress);
      ClientMessage m = session.createMessage(ClientMessage.TEXT_TYPE, true);
      m.getBodyBuffer().writeString("TestMessage");

      producer.send(m);

      assertNotNull(consumer1.receive(2000));
      assertNotNull(consumer2.receive(2000));

      q1.deleteQueue();
      q2.deleteQueue();
   }
}
 

@Test
public void testOneWaySSLVerifyHost() throws Exception {
   createCustomSslServer(true);
   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, "verified-" + CLIENT_SIDE_TRUSTSTORE);
   tc.getParams().put(TransportConstants.TRUSTSTORE_PASSWORD_PROP_NAME, PASSWORD);
   tc.getParams().put(TransportConstants.VERIFY_HOST_PROP_NAME, true);

   ServerLocator locator = addServerLocator(ActiveMQClient.createServerLocatorWithoutHA(tc));
   ClientSessionFactory sf = addSessionFactory(createSessionFactory(locator));
   ClientSession session = addClientSession(sf.createSession(false, true, true));
   session.createQueue(new QueueConfiguration(CoreClientOverOneWaySSLTest.QUEUE).setDurable(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();

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

@Test
public void testRollbackThenSend() throws Exception {
   ClientSession session = sf.createSession(false, false, false);

   session.start();

   final SimpleString queueName = new SimpleString("DuplicateDetectionTestQueue");

   session.createQueue(new QueueConfiguration(queueName).setDurable(false));

   ClientProducer producer = session.createProducer(queueName);

   ClientConsumer consumer = session.createConsumer(queueName);

   ClientMessage message = createMessage(session, 0);
   SimpleString dupID1 = new SimpleString("abcdefg");
   message.putBytesProperty(Message.HDR_DUPLICATE_DETECTION_ID, dupID1.getData());
   message.putStringProperty("key", dupID1.toString());
   producer.send(message);

   session.rollback();

   message = createMessage(session, 0);
   message.putBytesProperty(Message.HDR_DUPLICATE_DETECTION_ID, dupID1.getData());
   message.putStringProperty("key", dupID1.toString());
   producer.send(message);

   session.commit();

   message = consumer.receive(5000);
   assertNotNull(message);
   assertTrue(message.getStringProperty("key").equals(dupID1.toString()));
}
 

@Test
public void testXARollback() throws Exception {
   setupDLQ(10);
   ActiveMQResourceAdapter qResourceAdapter = newResourceAdapter();
   qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
   MyBootstrapContext ctx = new MyBootstrapContext();
   qResourceAdapter.start(ctx);
   ActiveMQActivationSpec spec = new ActiveMQActivationSpec();
   spec.setResourceAdapter(qResourceAdapter);
   spec.setMaxSession(1);
   spec.setUseJNDI(false);
   spec.setDestinationType("javax.jms.Queue");
   spec.setDestination(MDBQUEUE);
   qResourceAdapter.setConnectorClassName(INVM_CONNECTOR_FACTORY);
   CountDownLatch latch = new CountDownLatch(1);
   XADummyEndpoint endpoint = new XADummyEndpoint(latch);
   DummyMessageEndpointFactory endpointFactory = new DummyMessageEndpointFactory(endpoint, true);
   qResourceAdapter.endpointActivation(endpointFactory, spec);
   ClientSession session = locator.createSessionFactory().createSession();
   ClientProducer clientProducer = session.createProducer(MDBQUEUEPREFIXED);
   ClientMessage message = session.createMessage(true);
   message.getBodyBuffer().writeString("teststring");
   clientProducer.send(message);
   session.close();
   latch.await(5, TimeUnit.SECONDS);

   assertNotNull(endpoint.lastMessage);
   assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "teststring");
   latch = new CountDownLatch(1);
   endpoint.reset(latch);
   endpoint.rollback();
   latch.await(5, TimeUnit.SECONDS);

   assertNotNull(endpoint.lastMessage);
   assertEquals(endpoint.lastMessage.getCoreMessage().getBodyBuffer().readString(), "teststring");
   qResourceAdapter.endpointDeactivation(endpointFactory, spec);
   qResourceAdapter.stop();
}
 

/**
 * @param adr
 * @param nMessages
 * @param messageSize
 * @param factory
 * @throws ActiveMQException
 */
private void sendInitialBatch(final SimpleString adr,
                              final int nMessages,
                              final int messageSize,
                              final ClientSessionFactory factory) throws ActiveMQException {
   ClientSession session = factory.createSession(false, true, true);
   ClientProducer producer = session.createProducer(adr);
   ClientMessage msg = session.createMessage(true);

   msg.getBodyBuffer().writeBytes(new byte[messageSize]);

   sendMessages(nMessages, producer, msg);
}
 

@Test
public void testAsyncConsumerAck() throws Exception {
   ActiveMQServer server = createServer(false);
   server.start();
   ServerLocator locator = createInVMNonHALocator().setBlockOnAcknowledge(true).setAckBatchSize(0);
   ClientSessionFactory cf = createSessionFactory(locator);
   ClientSession sendSession = cf.createSession(false, true, true);
   final ClientSession session = cf.createSession(false, true, true);
   sendSession.createQueue(new QueueConfiguration(queueA).setAddress(addressA).setDurable(false));
   ClientProducer cp = sendSession.createProducer(addressA);
   ClientConsumer cc = session.createConsumer(queueA);
   int numMessages = 100;
   for (int i = 0; i < numMessages; i++) {
      cp.send(sendSession.createMessage(false));
   }
   final CountDownLatch latch = new CountDownLatch(numMessages);
   session.start();
   cc.setMessageHandler(new MessageHandler() {
      @Override
      public void onMessage(final ClientMessage message) {
         try {
            message.acknowledge();
         } catch (ActiveMQException e) {
            try {
               session.close();
            } catch (ActiveMQException e1) {
               e1.printStackTrace();
            }
         }
         latch.countDown();
      }
   });
   Assert.assertTrue(latch.await(5, TimeUnit.SECONDS));
   Queue q = (Queue) server.getPostOffice().getBinding(queueA).getBindable();
   Assert.assertEquals(0, q.getDeliveringCount());
   sendSession.close();
   session.close();
}
 

/**
 * This would force a journal duplication on bindings even with the scenario that generated fixed,
 * the server shouldn't hold of from starting
 *
 * @throws Exception
 */
@Test
public void testForceDuplicationOnBindings() throws Exception {
   queue = server.createQueue(new QueueConfiguration(QUEUE).setRoutingType(RoutingType.ANYCAST));

   ClientSessionFactory factory = locator.createSessionFactory();
   ClientSession session = factory.createSession(false, false, false);

   ClientProducer producer = session.createProducer(QUEUE);

   producer.send(session.createMessage(true));
   session.commit();

   long queueID = server.getStorageManager().generateID();
   long txID = server.getStorageManager().generateID();

   // Forcing a situation where the server would unexpectedly create a duplicated queue. The server should still start normally
   LocalQueueBinding newBinding = new LocalQueueBinding(QUEUE,
                                                        new QueueImpl(queueID, QUEUE, QUEUE, null, null, true, false,
                                                                      false, null, null, null, null, null, null, null),
                                                        server.getNodeID());
   server.getStorageManager().addQueueBinding(txID, newBinding);
   server.getStorageManager().commitBindings(txID);

   server.stop();

   // a duplicate binding would impede the server from starting
   server.start();
   waitForServerToStart(server);

   server.stop();

}
 

protected ClientMessage createTextMessage(final ClientSession session, final boolean durable, final int numChars) {
   ClientMessage message = session.createMessage(Message.TEXT_TYPE,
             durable,
             0,
             System.currentTimeMillis(),
             (byte)4);
   StringBuilder builder = new StringBuilder();
   for (int i = 0; i < numChars; i++) {
      builder.append('a');
   }
   message.getBodyBuffer().writeString(builder.toString());
   return message;
}