[jboss-cvs] JBoss Messaging SVN: r6167 - in trunk: tests/src/org/jboss/messaging/tests/integration/client and 3 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Wed Mar 25 22:47:40 EDT 2009
Author: clebert.suconic at jboss.com
Date: 2009-03-25 22:47:40 -0400 (Wed, 25 Mar 2009)
New Revision: 6167
Added:
trunk/tests/src/org/jboss/messaging/tests/integration/client/HeuristicXATest.java
trunk/tests/src/org/jboss/messaging/tests/integration/client/PagingTest.java
Removed:
trunk/tests/src/org/jboss/messaging/tests/integration/paging/PagingServiceIntegrationTest.java
Modified:
trunk/src/main/org/jboss/messaging/core/server/Messaging.java
trunk/tests/src/org/jboss/messaging/tests/integration/client/SelfExpandingBufferTest.java
trunk/tests/src/org/jboss/messaging/tests/integration/management/ManagementTestBase.java
trunk/tests/src/org/jboss/messaging/tests/util/ServiceTestBase.java
Log:
Adding (and moving) PagingTests & HeuristicXA Tests
Modified: trunk/src/main/org/jboss/messaging/core/server/Messaging.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/server/Messaging.java 2009-03-25 19:50:12 UTC (rev 6166)
+++ trunk/src/main/org/jboss/messaging/core/server/Messaging.java 2009-03-26 02:47:40 UTC (rev 6167)
@@ -146,7 +146,7 @@
JBMSecurityManager securityManager = new JBMSecurityManagerImpl();
- ManagementService managementService = new ManagementServiceImpl(ManagementFactory.getPlatformMBeanServer(), false);
+ ManagementService managementService = new ManagementServiceImpl(ManagementFactory.getPlatformMBeanServer(), config.isJMXManagementEnabled());
remotingService.setManagementService(managementService);
@@ -165,5 +165,32 @@
return new MessagingServiceImpl(server, storageManager, remotingService);
}
+ public static MessagingServiceImpl newMessagingService(final Configuration config, final MBeanServer mbeanService)
+ {
+ StorageManager storageManager = new JournalStorageManager(config);
+ RemotingService remotingService = new RemotingServiceImpl(config);
+
+ JBMSecurityManager securityManager = new JBMSecurityManagerImpl();
+
+ ManagementService managementService = new ManagementServiceImpl(mbeanService, config.isJMXManagementEnabled());
+
+ remotingService.setManagementService(managementService);
+
+ MessagingServer server = new MessagingServerImpl();
+
+ server.setConfiguration(config);
+
+ server.setStorageManager(storageManager);
+
+ server.setRemotingService(remotingService);
+
+ server.setSecurityManager(securityManager);
+
+ server.setManagementService(managementService);
+
+ return new MessagingServiceImpl(server, storageManager, remotingService);
+ }
+
+
}
Added: trunk/tests/src/org/jboss/messaging/tests/integration/client/HeuristicXATest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/client/HeuristicXATest.java (rev 0)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/client/HeuristicXATest.java 2009-03-26 02:47:40 UTC (rev 6167)
@@ -0,0 +1,187 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005-2009, Red Hat Middleware LLC, and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jboss.messaging.tests.integration.client;
+
+import java.util.HashMap;
+
+import javax.management.MBeanServer;
+import javax.management.MBeanServerFactory;
+import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+
+import org.jboss.messaging.core.client.*;
+import org.jboss.messaging.core.config.Configuration;
+import org.jboss.messaging.core.management.MessagingServerControlMBean;
+import org.jboss.messaging.core.server.MessagingService;
+import org.jboss.messaging.core.server.Queue;
+import org.jboss.messaging.core.settings.impl.AddressSettings;
+import org.jboss.messaging.core.transaction.impl.XidImpl;
+import org.jboss.messaging.tests.integration.management.ManagementControlHelper;
+import org.jboss.messaging.tests.util.ServiceTestBase;
+import org.jboss.messaging.utils.*;
+
+/**
+ * A HeuristicXATest
+ *
+ * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ *
+ */
+public class HeuristicXATest extends ServiceTestBase
+{
+
+ // Constants -----------------------------------------------------
+
+ final SimpleString ADDRESS = new SimpleString("ADDRESS");
+
+ // Attributes ----------------------------------------------------
+
+ MBeanServer mbeanServer;
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ // Public --------------------------------------------------------
+
+ public void testHerusticCommit() throws Exception
+ {
+ internalTest(true);
+ }
+
+ public void testHerusticRollback() throws Exception
+ {
+ internalTest(false);
+ }
+
+ private void internalTest(boolean isCommit) throws Exception
+ {
+ Configuration configuration = createDefaultConfig();
+ configuration.setJMXManagementEnabled(true);
+
+ MessagingService service = createService(false,
+ configuration,
+ mbeanServer,
+ new HashMap<String, AddressSettings>());
+ try
+ {
+ service.start();
+ Xid xid = newXID();
+
+ ClientSessionFactory sf = createInVMFactory();
+
+ ClientSession session = sf.createSession(true, false, false);
+
+ session.createQueue(ADDRESS, ADDRESS, true);
+
+ session.start(xid, XAResource.TMNOFLAGS);
+
+ ClientProducer producer = session.createProducer(ADDRESS);
+
+ ClientMessage msg = session.createClientMessage(true);
+
+ msg.getBody().writeBytes(new byte[123]);
+
+ producer.send(msg);
+
+ session.end(xid, XAResource.TMSUCCESS);
+
+ session.prepare(xid);
+
+ session.close();
+
+ MessagingServerControlMBean jmxServer = ManagementControlHelper.createMessagingServerControl(mbeanServer);
+
+ String preparedTransactions[] = jmxServer.listPreparedTransactions();
+
+ assertEquals(1, preparedTransactions.length);
+
+ System.out.println(preparedTransactions[0]);
+
+ if (isCommit)
+ {
+ jmxServer.commitPreparedTransaction(XidImpl.toBase64String(xid));
+ }
+ else
+ {
+ jmxServer.rollbackPreparedTransaction(XidImpl.toBase64String(xid));
+ }
+
+ preparedTransactions = jmxServer.listPreparedTransactions();
+ assertEquals(0, preparedTransactions.length);
+
+ if (isCommit)
+ {
+ assertEquals(1,
+ ((Queue)service.getServer().getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+
+ session = sf.createSession(false, false, false);
+
+ session.start();
+ ClientConsumer consumer = session.createConsumer(ADDRESS);
+ msg = consumer.receive(1000);
+ assertNotNull(msg);
+ msg.acknowledge();
+ assertEquals(123, msg.getBodySize());
+
+ session.commit();
+ session.close();
+ }
+ else
+ {
+ assertEquals(0,
+ ((Queue)service.getServer().getPostOffice().getBinding(ADDRESS).getBindable()).getMessageCount());
+ }
+
+ }
+ finally
+ {
+ if (service.isStarted())
+ {
+ service.stop();
+ }
+ }
+
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ protected void tearDown() throws Exception
+ {
+ MBeanServerFactory.releaseMBeanServer(mbeanServer);
+ super.tearDown();
+ }
+
+ protected void setUp() throws Exception
+ {
+ super.setUp();
+ mbeanServer = MBeanServerFactory.createMBeanServer();
+ }
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Copied: trunk/tests/src/org/jboss/messaging/tests/integration/client/PagingTest.java (from rev 6166, trunk/tests/src/org/jboss/messaging/tests/integration/paging/PagingServiceIntegrationTest.java)
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/client/PagingTest.java (rev 0)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/client/PagingTest.java 2009-03-26 02:47:40 UTC (rev 6167)
@@ -0,0 +1,1323 @@
+/*
+ * JBoss, Home of Professional Open Source
+ * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
+ * by the @authors tag. See the copyright.txt in the distribution for a
+ * full listing of individual contributors.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+
+package org.jboss.messaging.tests.integration.client;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import junit.framework.AssertionFailedError;
+
+import org.jboss.messaging.core.buffers.ChannelBuffers;
+import org.jboss.messaging.core.client.ClientConsumer;
+import org.jboss.messaging.core.client.ClientMessage;
+import org.jboss.messaging.core.client.ClientProducer;
+import org.jboss.messaging.core.client.ClientSession;
+import org.jboss.messaging.core.client.ClientSessionFactory;
+import org.jboss.messaging.core.config.Configuration;
+import org.jboss.messaging.core.logging.Logger;
+import org.jboss.messaging.core.message.impl.MessageImpl;
+import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
+import org.jboss.messaging.core.server.MessagingService;
+import org.jboss.messaging.core.server.Queue;
+import org.jboss.messaging.core.settings.impl.AddressSettings;
+import org.jboss.messaging.tests.util.ServiceTestBase;
+import org.jboss.messaging.utils.DataConstants;
+import org.jboss.messaging.utils.SimpleString;
+
+/**
+ * A PagingTest
+ *
+ * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
+ *
+ * Created Dec 5, 2008 8:25:58 PM
+ *
+ *
+ */
+public class PagingTest extends ServiceTestBase
+{
+
+ // Constants -----------------------------------------------------
+ private static final Logger log = Logger.getLogger(PagingTest.class);
+
+ private static final int RECEIVE_TIMEOUT = 30000;
+
+ // Attributes ----------------------------------------------------
+
+ // Static --------------------------------------------------------
+
+ static final SimpleString ADDRESS = new SimpleString("SimpleAddress");
+
+ // Constructors --------------------------------------------------
+
+ // Public --------------------------------------------------------
+
+ public void testSendReceivePaging() throws Exception
+ {
+ clearData();
+
+ Configuration config = createDefaultConfig();
+
+ config.setPagingMaxGlobalSizeBytes(100 * 1024);
+ config.setPagingGlobalWatermarkSize(10 * 1024);
+
+ MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
+
+ messagingService.start();
+
+ final int numberOfIntegers = 256;
+
+ final int numberOfMessages = 10000;
+
+ try
+ {
+ ClientSessionFactory sf = createInVMFactory();
+
+ sf.setBlockOnNonPersistentSend(true);
+ sf.setBlockOnPersistentSend(true);
+ sf.setBlockOnAcknowledge(true);
+
+ ClientSession session = sf.createSession(null, null, false, true, true, false, 0);
+
+ session.createQueue(ADDRESS, ADDRESS, null, true, false);
+
+ ClientProducer producer = session.createProducer(ADDRESS);
+
+ ClientMessage message = null;
+
+ byte[] body = null;
+
+ for (int i = 0; i < numberOfMessages; i++)
+ {
+ message = session.createClientMessage(true);
+
+ MessagingBuffer bodyLocal = message.getBody();
+
+ for (int j = 1; j <= numberOfIntegers; j++)
+ {
+ bodyLocal.writeInt(j);
+ }
+
+ if (body == null)
+ {
+ body = bodyLocal.array();
+ }
+
+ message.putIntProperty(new SimpleString("id"), i);
+
+ producer.send(message);
+ }
+
+ session.close();
+
+ messagingService.stop();
+
+ messagingService = createService(true, config, new HashMap<String, AddressSettings>());
+ messagingService.start();
+
+ sf = createInVMFactory();
+
+ assertTrue(messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize() > 0);
+
+ session = sf.createSession(null, null, false, true, true, false, 0);
+
+ ClientConsumer consumer = session.createConsumer(ADDRESS);
+
+ session.start();
+
+ for (int i = 0; i < numberOfMessages; i++)
+ {
+ ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull(message2);
+
+ assertEquals(i, ((Integer)message2.getProperty(new SimpleString("id"))).intValue());
+
+ message2.acknowledge();
+
+ assertNotNull(message2);
+
+ try
+ {
+ assertEqualsByteArrays(body.length, body, message2.getBody().array());
+ }
+ catch (AssertionFailedError e)
+ {
+ log.info("Expected buffer:" + dumbBytesHex(body, 40));
+ log.info("Arriving buffer:" + dumbBytesHex(message2.getBody().array(), 40));
+ throw e;
+ }
+ }
+
+ consumer.close();
+
+ session.close();
+
+ assertEquals(0, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
+
+ }
+ finally
+ {
+ try
+ {
+ messagingService.stop();
+ }
+ catch (Throwable ignored)
+ {
+ }
+ }
+
+ }
+
+ /**
+ * - Make a destination in page mode
+ * - Add stuff to a transaction
+ * - Consume the entire destination (not in page mode any more)
+ * - Add stuff to a transaction again
+ * - Check order
+ *
+ */
+ public void testDepageDuringTransaction() throws Exception
+ {
+ clearData();
+
+ Configuration config = createDefaultConfig();
+
+ config.setPagingMaxGlobalSizeBytes(100 * 1024);
+ config.setPagingGlobalWatermarkSize(10 * 1024);
+
+ MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
+
+ messagingService.start();
+
+ final int numberOfIntegers = 256;
+
+ try
+ {
+ ClientSessionFactory sf = createInVMFactory();
+
+ sf.setBlockOnNonPersistentSend(true);
+ sf.setBlockOnPersistentSend(true);
+ sf.setBlockOnAcknowledge(true);
+
+ ClientSession session = sf.createSession(null, null, false, true, true, false, 0);
+
+ session.createQueue(ADDRESS, ADDRESS, null, true, false);
+
+ ClientProducer producer = session.createProducer(ADDRESS);
+
+ MessagingBuffer bodyLocal = ChannelBuffers.buffer(DataConstants.SIZE_INT * numberOfIntegers);
+
+ ClientMessage message = null;
+
+ int numberOfMessages = 0;
+ while (true)
+ {
+ message = session.createClientMessage(true);
+ message.setBody(bodyLocal);
+
+ // Stop sending message as soon as we start paging
+ if (messagingService.getServer().getPostOffice().getPagingManager().isPaging(ADDRESS))
+ {
+ break;
+ }
+ numberOfMessages++;
+
+ producer.send(message);
+ }
+
+ assertTrue(messagingService.getServer().getPostOffice().getPagingManager().isPaging(ADDRESS));
+
+ session.start();
+
+ ClientSession sessionTransacted = sf.createSession(null, null, false, false, false, false, 0);
+
+ ClientProducer producerTransacted = sessionTransacted.createProducer(ADDRESS);
+
+ for (int i = 0; i < 10; i++)
+ {
+ message = session.createClientMessage(true);
+ message.setBody(bodyLocal);
+ message.putIntProperty(new SimpleString("id"), i);
+
+ // Consume messages to force an eventual out of order delivery
+ if (i == 5)
+ {
+ ClientConsumer consumer = session.createConsumer(ADDRESS);
+ for (int j = 0; j < numberOfMessages; j++)
+ {
+ ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT);
+ msg.acknowledge();
+ assertNotNull(msg);
+ }
+
+ assertNull(consumer.receive(100));
+ consumer.close();
+ }
+
+ Integer messageID = (Integer)message.getProperty(new SimpleString("id"));
+ assertNotNull(messageID);
+ assertEquals(messageID.intValue(), i);
+
+ producerTransacted.send(message);
+ }
+
+ ClientConsumer consumer = session.createConsumer(ADDRESS);
+
+ assertNull(consumer.receive(100));
+
+ sessionTransacted.commit();
+
+ sessionTransacted.close();
+
+ for (int i = 0; i < 10; i++)
+ {
+ message = consumer.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull(message);
+
+ Integer messageID = (Integer)message.getProperty(new SimpleString("id"));
+
+ assertNotNull(messageID);
+ assertEquals("message received out of order", messageID.intValue(), i);
+
+ message.acknowledge();
+ }
+
+ assertNull(consumer.receive(100));
+
+ consumer.close();
+
+ session.close();
+
+ assertEquals(0, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
+
+ }
+ finally
+ {
+ try
+ {
+ messagingService.stop();
+ }
+ catch (Throwable ignored)
+ {
+ }
+ }
+
+ }
+
+ public void testPageOnSchedulingNoRestart() throws Exception
+ {
+ internalTestPageOnScheduling(false);
+ }
+
+ public void testPageOnSchedulingRestart() throws Exception
+ {
+ internalTestPageOnScheduling(true);
+ }
+
+ public void internalTestPageOnScheduling(final boolean restart) throws Exception
+ {
+ clearData();
+
+ Configuration config = createDefaultConfig();
+
+ config.setPagingMaxGlobalSizeBytes(100 * 1024);
+ config.setPagingGlobalWatermarkSize(10 * 1024);
+
+ MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
+
+ messagingService.start();
+
+ final int numberOfIntegers = 256;
+
+ final int numberOfMessages = 10000;
+
+ try
+ {
+ ClientSessionFactory sf = createInVMFactory();
+
+ sf.setBlockOnNonPersistentSend(true);
+ sf.setBlockOnPersistentSend(true);
+ sf.setBlockOnAcknowledge(true);
+
+ ClientSession session = sf.createSession(null, null, false, true, true, false, 0);
+
+ session.createQueue(ADDRESS, ADDRESS, null, true, false);
+
+ ClientProducer producer = session.createProducer(ADDRESS);
+
+ ClientMessage message = null;
+
+ byte[] body = null;
+
+ long scheduledTime = System.currentTimeMillis() + 5000;
+
+ for (int i = 0; i < numberOfMessages; i++)
+ {
+ message = session.createClientMessage(true);
+
+ MessagingBuffer bodyLocal = message.getBody();
+
+ for (int j = 1; j <= numberOfIntegers; j++)
+ {
+ bodyLocal.writeInt(j);
+ }
+
+
+ if (body == null)
+ {
+ body = bodyLocal.array();
+ }
+
+ message.setBody(bodyLocal);
+ message.putIntProperty(new SimpleString("id"), i);
+
+ // Worse scenario possible... only schedule what's on pages
+ if (messagingService.getServer().getPostOffice().getPagingManager().isPaging(ADDRESS))
+ {
+ message.putLongProperty(MessageImpl.HDR_SCHEDULED_DELIVERY_TIME, scheduledTime);
+ }
+
+ producer.send(message);
+ }
+
+ if (restart)
+ {
+ session.close();
+
+ messagingService.stop();
+
+ messagingService = createService(true, config, new HashMap<String, AddressSettings>());
+ messagingService.start();
+
+ sf = createInVMFactory();
+
+ session = sf.createSession(null, null, false, true, true, false, 0);
+ }
+
+ ClientConsumer consumer = session.createConsumer(ADDRESS);
+
+ session.start();
+
+ for (int i = 0; i < numberOfMessages; i++)
+ {
+ ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull(message2);
+
+ message2.acknowledge();
+
+ assertNotNull(message2);
+
+ Long scheduled = (Long)message2.getProperty(MessageImpl.HDR_SCHEDULED_DELIVERY_TIME);
+ if (scheduled != null)
+ {
+ assertTrue("Scheduling didn't work", System.currentTimeMillis() >= scheduledTime);
+ }
+
+ try
+ {
+ assertEqualsByteArrays(body.length, body, message2.getBody().array());
+ }
+ catch (AssertionFailedError e)
+ {
+ log.info("Expected buffer:" + dumbBytesHex(body, 40));
+ log.info("Arriving buffer:" + dumbBytesHex(message2.getBody().array(), 40));
+ throw e;
+ }
+ }
+
+ consumer.close();
+
+ session.close();
+
+ assertEquals(0, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
+
+ }
+ finally
+ {
+ try
+ {
+ messagingService.stop();
+ }
+ catch (Throwable ignored)
+ {
+ }
+ }
+
+ }
+
+ public void testRollbackOnSend() throws Exception
+ {
+ clearData();
+
+ Configuration config = createDefaultConfig();
+
+ config.setPagingMaxGlobalSizeBytes(100 * 1024);
+ config.setPagingGlobalWatermarkSize(10 * 1024);
+
+ MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
+
+ messagingService.start();
+
+ final int numberOfIntegers = 256;
+
+ final int numberOfMessages = 10;
+
+ try
+ {
+ ClientSessionFactory sf = createInVMFactory();
+
+ sf.setBlockOnNonPersistentSend(true);
+ sf.setBlockOnPersistentSend(true);
+ sf.setBlockOnAcknowledge(true);
+
+ ClientSession session = sf.createSession(null, null, false, false, true, false, 0);
+
+ session.createQueue(ADDRESS, ADDRESS, null, true, false);
+
+ ClientProducer producer = session.createProducer(ADDRESS);
+
+ long initialSize = messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize();
+
+ ClientMessage message = null;
+
+ for (int i = 0; i < numberOfMessages; i++)
+ {
+ message = session.createClientMessage(true);
+
+ MessagingBuffer bodyLocal = message.getBody();
+
+ for (int j = 1; j <= numberOfIntegers; j++)
+ {
+ bodyLocal.writeInt(j);
+ }
+
+
+ message.putIntProperty(new SimpleString("id"), i);
+
+ producer.send(message);
+ }
+
+ session.rollback();
+
+ ClientConsumer consumer = session.createConsumer(ADDRESS);
+
+ session.start();
+
+ assertNull(consumer.receive(100));
+
+ session.close();
+
+ assertEquals(initialSize, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
+ }
+ finally
+ {
+ try
+ {
+ messagingService.stop();
+ }
+ catch (Throwable ignored)
+ {
+ }
+ }
+
+ }
+
+ public void testCommitOnSend() throws Exception
+ {
+ clearData();
+
+ Configuration config = createDefaultConfig();
+
+ config.setPagingMaxGlobalSizeBytes(100 * 1024);
+ config.setPagingGlobalWatermarkSize(10 * 1024);
+
+ MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
+
+ messagingService.start();
+
+ final int numberOfIntegers = 10;
+
+ final int numberOfMessages = 10;
+
+ try
+ {
+ ClientSessionFactory sf = createInVMFactory();
+
+ sf.setBlockOnNonPersistentSend(true);
+ sf.setBlockOnPersistentSend(true);
+ sf.setBlockOnAcknowledge(true);
+
+ ClientSession session = sf.createSession(null, null, false, false, false, false, 0);
+
+ session.createQueue(ADDRESS, ADDRESS, null, true, false);
+
+ ClientProducer producer = session.createProducer(ADDRESS);
+
+ long initialSize = messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize();
+
+ ClientMessage message = null;
+
+ for (int i = 0; i < numberOfMessages; i++)
+ {
+ message = session.createClientMessage(true);
+
+ MessagingBuffer bodyLocal = message.getBody();
+
+ for (int j = 1; j <= numberOfIntegers; j++)
+ {
+ bodyLocal.writeInt(j);
+ }
+
+ message.putIntProperty(new SimpleString("id"), i);
+
+ producer.send(message);
+ }
+
+ session.commit();
+
+ ClientConsumer consumer = session.createConsumer(ADDRESS);
+
+ session.start();
+ for (int i = 0; i < numberOfMessages; i++)
+ {
+ ClientMessage msg = consumer.receive(500);
+ assertNotNull(msg);
+ msg.acknowledge();
+ }
+
+ session.commit();
+
+ session.close();
+
+ assertEquals(initialSize, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
+ }
+ finally
+ {
+ try
+ {
+ messagingService.stop();
+ }
+ catch (Throwable ignored)
+ {
+ }
+ }
+
+ }
+
+ public void testPageMultipleDestinations() throws Exception
+ {
+ internalTestPageMultipleDestinations(false);
+ }
+
+ public void testPageMultipleDestinationsTransacted() throws Exception
+ {
+ internalTestPageMultipleDestinations(true);
+ }
+
+ public void testDropMessagesQueueMax() throws Exception
+ {
+ testDropMessages(false);
+ }
+
+ public void testDropMessagesGlobalMax() throws Exception
+ {
+ testDropMessages(true);
+ }
+
+ private void testDropMessages(boolean global) throws Exception
+ {
+ clearData();
+
+ Configuration config = createDefaultConfig();
+
+ HashMap<String, AddressSettings> settings = new HashMap<String, AddressSettings>();
+
+ AddressSettings set = new AddressSettings();
+ set.setDropMessagesWhenFull(true);
+
+ settings.put(ADDRESS.toString(), set);
+
+ if (global)
+ {
+ set.setMaxSizeBytes(-1);
+ config.setPagingMaxGlobalSizeBytes(10 * 1024);
+ }
+ else
+ {
+ config.setPagingMaxGlobalSizeBytes(-1);
+ set.setMaxSizeBytes(10 * 1024);
+ }
+
+ config.setPagingGlobalWatermarkSize(10 * 1024);
+
+ MessagingService messagingService = createService(true, config, settings);
+
+ messagingService.start();
+
+ final int numberOfMessages = 1000;
+
+ try
+ {
+ ClientSessionFactory sf = createInVMFactory();
+
+ sf.setBlockOnNonPersistentSend(true);
+ sf.setBlockOnPersistentSend(true);
+ sf.setBlockOnAcknowledge(true);
+
+ ClientSession session = sf.createSession(null, null, false, true, true, false, 0);
+
+ session.createQueue(ADDRESS, ADDRESS, null, true, false);
+
+ ClientProducer producer = session.createProducer(ADDRESS);
+
+ ClientMessage message = null;
+
+ for (int i = 0; i < numberOfMessages; i++)
+ {
+ MessagingBuffer bodyLocal = ChannelBuffers.wrappedBuffer(new byte[1024]);
+
+ message = session.createClientMessage(true);
+ message.setBody(bodyLocal);
+
+ producer.send(message);
+ }
+
+ session = sf.createSession(null, null, false, true, true, false, 0);
+
+ ClientConsumer consumer = session.createConsumer(ADDRESS);
+
+ session.start();
+
+ for (int i = 0; i < 9; i++)
+ {
+ ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull(message2);
+
+ message2.acknowledge();
+ }
+
+ assertNull(consumer.receive(100));
+
+ assertEquals(0, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
+ assertEquals(0, messagingService.getServer()
+ .getPostOffice()
+ .getPagingManager()
+ .getPageStore(ADDRESS)
+ .getAddressSize());
+
+ for (int i = 0; i < numberOfMessages; i++)
+ {
+ MessagingBuffer bodyLocal = ChannelBuffers.wrappedBuffer(new byte[1024]);
+
+ message = session.createClientMessage(true);
+ message.setBody(bodyLocal);
+
+ producer.send(message);
+ }
+
+ for (int i = 0; i < 9; i++)
+ {
+ ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull(message2);
+
+ message2.acknowledge();
+ }
+
+ assertNull(consumer.receive(100));
+
+ session.close();
+
+ session = sf.createSession(false, true, true);
+
+ producer = session.createProducer(ADDRESS);
+
+ for (int i = 0; i < numberOfMessages; i++)
+ {
+ MessagingBuffer bodyLocal = ChannelBuffers.wrappedBuffer(new byte[1024]);
+
+ message = session.createClientMessage(true);
+ message.setBody(bodyLocal);
+
+ producer.send(message);
+ }
+
+ session.commit();
+
+ consumer = session.createConsumer(ADDRESS);
+
+ session.start();
+
+ for (int i = 0; i < 9; i++)
+ {
+ ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull(message2);
+
+ message2.acknowledge();
+ }
+
+ session.commit();
+
+ assertNull(consumer.receive(100));
+
+ session.close();
+
+ assertEquals(0, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
+ assertEquals(0, messagingService.getServer()
+ .getPostOffice()
+ .getPagingManager()
+ .getPageStore(ADDRESS)
+ .getAddressSize());
+
+ }
+ finally
+ {
+ try
+ {
+ messagingService.stop();
+ }
+ catch (Throwable ignored)
+ {
+ }
+ }
+
+ }
+
+ private void internalTestPageMultipleDestinations(boolean transacted) throws Exception
+ {
+ Configuration config = createDefaultConfig();
+
+ final int NUMBER_OF_BINDINGS = 100;
+
+ int NUMBER_OF_MESSAGES = 2;
+
+ config.setPagingMaxGlobalSizeBytes(100 * 1024);
+ config.setPagingGlobalWatermarkSize(10 * 1024);
+
+ MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
+
+ messagingService.start();
+
+ try
+ {
+ ClientSessionFactory sf = createInVMFactory();
+
+ sf.setBlockOnNonPersistentSend(true);
+ sf.setBlockOnPersistentSend(true);
+ sf.setBlockOnAcknowledge(true);
+
+ ClientSession session = sf.createSession(null, null, false, !transacted, true, false, 0);
+
+ for (int i = 0; i < NUMBER_OF_BINDINGS; i++)
+ {
+ session.createQueue(ADDRESS, new SimpleString("someQueue" + i), null, true, false);
+ }
+
+ ClientProducer producer = session.createProducer(ADDRESS);
+
+ ClientMessage message = null;
+
+ MessagingBuffer bodyLocal = ChannelBuffers.wrappedBuffer(new byte[1024]);
+
+ message = session.createClientMessage(true);
+ message.setBody(bodyLocal);
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
+ {
+ producer.send(message);
+
+ if (transacted)
+ {
+ session.commit();
+ }
+ }
+
+ session.close();
+
+ messagingService.stop();
+
+ messagingService = createService(true, config, new HashMap<String, AddressSettings>());
+ messagingService.start();
+
+ sf = createInVMFactory();
+
+ assertTrue(messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize() > 0);
+
+ session = sf.createSession(null, null, false, true, true, false, 0);
+
+ session.start();
+
+ for (int msg = 0; msg < NUMBER_OF_MESSAGES; msg++)
+ {
+
+ for (int i = 0; i < NUMBER_OF_BINDINGS; i++)
+ {
+ ClientConsumer consumer = session.createConsumer(new SimpleString("someQueue" + i));
+
+ ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
+
+ assertNotNull(message2);
+
+ message2.acknowledge();
+
+ assertNotNull(message2);
+
+ consumer.close();
+
+ }
+ }
+
+ session.close();
+
+ for (int i = 0; i < NUMBER_OF_BINDINGS; i++)
+ {
+ Queue queue = (Queue)messagingService.getServer()
+ .getPostOffice()
+ .getBinding(new SimpleString("someQueue" + i))
+ .getBindable();
+
+ assertEquals("Queue someQueue" + i + " was supposed to be empty", 0, queue.getMessageCount());
+ assertEquals("Queue someQueue" + i + " was supposed to be empty", 0, queue.getDeliveringCount());
+ }
+
+ assertEquals("There are pending messages on the server", 0, messagingService.getServer()
+ .getPostOffice()
+ .getPagingManager()
+ .getGlobalSize());
+
+ }
+ finally
+ {
+ try
+ {
+ messagingService.stop();
+ }
+ catch (Throwable ignored)
+ {
+ }
+ }
+
+ }
+
+ public void testPagingOneDestinationOnly() throws Exception
+ {
+ SimpleString PAGED_ADDRESS = new SimpleString("paged");
+ SimpleString NON_PAGED_ADDRESS = new SimpleString("non-paged");
+
+ Configuration configuration = createDefaultConfig();
+
+ configuration.setPagingMaxGlobalSizeBytes(0);
+ configuration.setPagingGlobalWatermarkSize(0);
+
+ Map<String, AddressSettings> addresses = new HashMap<String, AddressSettings>();
+
+ addresses.put("#", new AddressSettings());
+
+ AddressSettings pagedDestination = new AddressSettings();
+ pagedDestination.setPageSizeBytes(1024);
+ pagedDestination.setMaxSizeBytes(10 * 1024);
+
+ addresses.put(PAGED_ADDRESS.toString(), pagedDestination);
+
+ MessagingService service = createService(true, configuration, addresses);
+
+ try
+ {
+ service.start();
+
+ ClientSessionFactory sf = createInVMFactory();
+
+ ClientSession session = sf.createSession(false, true, false);
+
+ session.createQueue(PAGED_ADDRESS, PAGED_ADDRESS, true);
+
+ session.createQueue(NON_PAGED_ADDRESS, NON_PAGED_ADDRESS, true);
+
+ ClientProducer producerPaged = session.createProducer(PAGED_ADDRESS);
+ ClientProducer producerNonPaged = session.createProducer(NON_PAGED_ADDRESS);
+
+ int NUMBER_OF_MESSAGES = 100;
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
+ {
+ ClientMessage msg = session.createClientMessage(true);
+ msg.getBody().writeBytes(new byte[512]);
+
+ producerPaged.send(msg);
+ producerNonPaged.send(msg);
+ }
+
+ session.close();
+
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS).isPaging());
+ assertFalse(service.getServer().getPostOffice().getPagingManager().getPageStore(NON_PAGED_ADDRESS).isPaging());
+
+ session = sf.createSession(false, true, false);
+
+ session.start();
+
+ ClientConsumer consumerNonPaged = session.createConsumer(NON_PAGED_ADDRESS);
+ ClientConsumer consumerPaged = session.createConsumer(PAGED_ADDRESS);
+
+ ClientMessage ackList[] = new ClientMessage[NUMBER_OF_MESSAGES];
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
+ {
+ ClientMessage msg = consumerNonPaged.receive(5000);
+ assertNotNull(msg);
+ ackList[i] = msg;
+ }
+
+ assertNull(consumerNonPaged.receiveImmediate());
+
+ consumerNonPaged.close();
+
+ for (ClientMessage ack : ackList)
+ {
+ ack.acknowledge();
+ }
+
+ session.commit();
+
+ ackList = null;
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
+ {
+ ClientMessage msg = consumerPaged.receive(5000);
+ assertNotNull(msg);
+ msg.acknowledge();
+ session.commit();
+ }
+
+ assertNull(consumerPaged.receiveImmediate());
+
+ assertFalse(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS).isPaging());
+ assertFalse(service.getServer().getPostOffice().getPagingManager().getPageStore(NON_PAGED_ADDRESS).isPaging());
+
+ session.close();
+
+ }
+ finally
+ {
+ if (service.isStarted())
+ {
+ service.stop();
+ }
+ }
+ }
+
+ public void testPagingDifferentSizes() throws Exception
+ {
+ SimpleString PAGED_ADDRESS_A = new SimpleString("paged-a");
+ SimpleString PAGED_ADDRESS_B = new SimpleString("paged-b");
+
+ Configuration configuration = createDefaultConfig();
+
+ configuration.setPagingMaxGlobalSizeBytes(0);
+ configuration.setPagingGlobalWatermarkSize(0);
+
+ Map<String, AddressSettings> addresses = new HashMap<String, AddressSettings>();
+
+ addresses.put("#", new AddressSettings());
+
+ AddressSettings pagedDestinationA = new AddressSettings();
+ pagedDestinationA.setPageSizeBytes(1024);
+ pagedDestinationA.setMaxSizeBytes(10 * 1024);
+
+ int NUMBER_MESSAGES_BEFORE_PAGING = 20;
+
+ addresses.put(PAGED_ADDRESS_A.toString(), pagedDestinationA);
+
+ AddressSettings pagedDestinationB = new AddressSettings();
+ pagedDestinationB.setPageSizeBytes(2024);
+ pagedDestinationB.setMaxSizeBytes(20 * 1024);
+
+ addresses.put(PAGED_ADDRESS_B.toString(), pagedDestinationB);
+
+ MessagingService service = createService(true, configuration, addresses);
+
+ try
+ {
+ service.start();
+
+ ClientSessionFactory sf = createInVMFactory();
+
+ ClientSession session = sf.createSession(false, true, false);
+
+ session.createQueue(PAGED_ADDRESS_A, PAGED_ADDRESS_A, true);
+
+ session.createQueue(PAGED_ADDRESS_B, PAGED_ADDRESS_B, true);
+
+ ClientProducer producerA = session.createProducer(PAGED_ADDRESS_A);
+ ClientProducer producerB = session.createProducer(PAGED_ADDRESS_B);
+
+ int NUMBER_OF_MESSAGES = 100;
+
+ for (int i = 0; i < NUMBER_MESSAGES_BEFORE_PAGING; i++)
+ {
+ ClientMessage msg = session.createClientMessage(true);
+ msg.getBody().writeBytes(new byte[512]);
+
+ producerA.send(msg);
+ producerB.send(msg);
+ }
+
+ session.commit(); // commit was called to clean the buffer only (making sure everything is on the server side)
+
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_A).isPaging());
+ assertFalse(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_B).isPaging());
+
+ for (int i = 0; i < NUMBER_MESSAGES_BEFORE_PAGING; i++)
+ {
+ ClientMessage msg = session.createClientMessage(true);
+ msg.getBody().writeBytes(new byte[512]);
+
+ producerA.send(msg);
+ producerB.send(msg);
+ }
+
+ session.commit(); // commit was called to clean the buffer only (making sure everything is on the server side)
+
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_A).isPaging());
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_B).isPaging());
+
+ for (int i = NUMBER_MESSAGES_BEFORE_PAGING * 2; i < NUMBER_OF_MESSAGES; i++)
+ {
+ ClientMessage msg = session.createClientMessage(true);
+ msg.getBody().writeBytes(new byte[512]);
+
+ producerA.send(msg);
+ producerB.send(msg);
+ }
+
+ session.close();
+
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_A).isPaging());
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_B).isPaging());
+
+ session = sf.createSession(null, null, false, true, true, false, 0);
+
+ session.start();
+
+ ClientConsumer consumerA = session.createConsumer(PAGED_ADDRESS_A);
+
+ ClientConsumer consumerB = session.createConsumer(PAGED_ADDRESS_B);
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
+ {
+ ClientMessage msg = consumerA.receive(5000);
+ assertNotNull("Couldn't receive a message on consumerA, iteration = " + i, msg);
+ msg.acknowledge();
+ }
+
+ assertNull(consumerA.receiveImmediate());
+
+ consumerA.close();
+
+ assertFalse(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_A).isPaging());
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_B).isPaging());
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
+ {
+ ClientMessage msg = consumerB.receive(5000);
+ assertNotNull(msg);
+ msg.acknowledge();
+ session.commit();
+ }
+
+ assertNull(consumerB.receiveImmediate());
+
+ consumerB.close();
+
+ session.close();
+
+ }
+ finally
+ {
+ if (service.isStarted())
+ {
+ service.stop();
+ }
+ }
+ }
+
+
+ public void testPagingDifferentSizesAndGlobal() throws Exception
+ {
+ SimpleString PAGED_ADDRESS_A = new SimpleString("paged-a");
+ SimpleString PAGED_ADDRESS_B = new SimpleString("paged-b");
+ SimpleString PAGED_ADDRESS_GLOBAL = new SimpleString("paged-global");
+
+
+ Configuration configuration = createDefaultConfig();
+
+ configuration.setPagingMaxGlobalSizeBytes(30 * 1024);
+ configuration.setPagingGlobalWatermarkSize(1024);
+
+ Map<String, AddressSettings> addresses = new HashMap<String, AddressSettings>();
+
+ addresses.put("#", new AddressSettings());
+
+ AddressSettings pagedDestinationA = new AddressSettings();
+ pagedDestinationA.setPageSizeBytes(1024);
+ pagedDestinationA.setMaxSizeBytes(10 * 1024);
+
+ int NUMBER_MESSAGES_BEFORE_PAGING = 20;
+
+ addresses.put(PAGED_ADDRESS_A.toString(), pagedDestinationA);
+
+ AddressSettings pagedDestinationB = new AddressSettings();
+ pagedDestinationB.setPageSizeBytes(2024);
+ pagedDestinationB.setMaxSizeBytes(20 * 1024);
+
+ addresses.put(PAGED_ADDRESS_B.toString(), pagedDestinationB);
+
+ MessagingService service = createService(true, configuration, addresses);
+
+ try
+ {
+ service.start();
+
+ ClientSessionFactory sf = createInVMFactory();
+
+ ClientSession session = sf.createSession(false, true, false);
+
+ session.createQueue(PAGED_ADDRESS_A, PAGED_ADDRESS_A, true);
+
+ session.createQueue(PAGED_ADDRESS_B, PAGED_ADDRESS_B, true);
+
+ session.createQueue(PAGED_ADDRESS_GLOBAL, PAGED_ADDRESS_GLOBAL, true);
+
+ ClientProducer producerA = session.createProducer(PAGED_ADDRESS_A);
+ ClientProducer producerB = session.createProducer(PAGED_ADDRESS_B);
+
+ int NUMBER_OF_MESSAGES = 100;
+
+ for (int i = 0; i < NUMBER_MESSAGES_BEFORE_PAGING; i++)
+ {
+ ClientMessage msg = session.createClientMessage(true);
+ msg.getBody().writeBytes(new byte[512]);
+
+ producerA.send(msg);
+ producerB.send(msg);
+ }
+
+ session.commit(); // commit was called to clean the buffer only (making sure everything is on the server side)
+
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_A).isPaging());
+ assertFalse(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_B).isPaging());
+
+ for (int i = 0; i < NUMBER_MESSAGES_BEFORE_PAGING; i++)
+ {
+ ClientMessage msg = session.createClientMessage(true);
+ msg.getBody().writeBytes(new byte[512]);
+
+ producerA.send(msg);
+ producerB.send(msg);
+ }
+
+ session.commit(); // commit was called to clean the buffer only (making sure everything is on the server side)
+
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_A).isPaging());
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_B).isPaging());
+
+ for (int i = NUMBER_MESSAGES_BEFORE_PAGING * 2; i < NUMBER_OF_MESSAGES; i++)
+ {
+ ClientMessage msg = session.createClientMessage(true);
+ msg.getBody().writeBytes(new byte[512]);
+
+ producerA.send(msg);
+ producerB.send(msg);
+ }
+
+ session.close();
+
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_A).isPaging());
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_B).isPaging());
+
+ session = sf.createSession(null, null, false, true, true, false, 0);
+
+ session.start();
+
+ ClientConsumer consumerA = session.createConsumer(PAGED_ADDRESS_A);
+
+ ClientConsumer consumerB = session.createConsumer(PAGED_ADDRESS_B);
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
+ {
+ ClientMessage msg = consumerA.receive(5000);
+ assertNotNull("Couldn't receive a message on consumerA, iteration = " + i, msg);
+ msg.acknowledge();
+ }
+
+ assertNull(consumerA.receiveImmediate());
+
+ consumerA.close();
+
+ session.commit();
+
+ assertFalse(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_A).isPaging());
+ assertTrue(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_B).isPaging());
+
+ for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
+ {
+ ClientMessage msg = consumerB.receive(5000);
+ assertNotNull(msg);
+ msg.acknowledge();
+ session.commit();
+ }
+
+ assertNull(consumerB.receiveImmediate());
+
+ consumerB.close();
+
+ assertFalse(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_A).isPaging());
+ assertFalse(service.getServer().getPostOffice().getPagingManager().getPageStore(PAGED_ADDRESS_B).isPaging());
+
+ session.close();
+
+ }
+ finally
+ {
+ if (service.isStarted())
+ {
+ service.stop();
+ }
+ }
+ }
+
+
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Property changes on: trunk/tests/src/org/jboss/messaging/tests/integration/client/PagingTest.java
___________________________________________________________________
Name: svn:mergeinfo
+
Modified: trunk/tests/src/org/jboss/messaging/tests/integration/client/SelfExpandingBufferTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/client/SelfExpandingBufferTest.java 2009-03-25 19:50:12 UTC (rev 6166)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/client/SelfExpandingBufferTest.java 2009-03-26 02:47:40 UTC (rev 6167)
@@ -71,7 +71,7 @@
public void testSelfExpandingBuffer(boolean netty) throws Exception
{
- setUpService(true);
+ setUpService(netty);
ClientSessionFactory factory;
Modified: trunk/tests/src/org/jboss/messaging/tests/integration/management/ManagementTestBase.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/management/ManagementTestBase.java 2009-03-25 19:50:12 UTC (rev 6166)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/management/ManagementTestBase.java 2009-03-26 02:47:40 UTC (rev 6167)
@@ -92,6 +92,13 @@
mbeanServer = MBeanServerFactory.createMBeanServer();
}
+
+ protected void tearDown() throws Exception
+ {
+ MBeanServerFactory.releaseMBeanServer(mbeanServer);
+
+ super.tearDown();
+ }
protected void checkNoResource(ObjectName on)
{
Deleted: trunk/tests/src/org/jboss/messaging/tests/integration/paging/PagingServiceIntegrationTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/paging/PagingServiceIntegrationTest.java 2009-03-25 19:50:12 UTC (rev 6166)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/paging/PagingServiceIntegrationTest.java 2009-03-26 02:47:40 UTC (rev 6167)
@@ -1,933 +0,0 @@
-/*
- * JBoss, Home of Professional Open Source
- * Copyright 2005-2008, Red Hat Middleware LLC, and individual contributors
- * by the @authors tag. See the copyright.txt in the distribution for a
- * full listing of individual contributors.
- *
- * This is free software; you can redistribute it and/or modify it
- * under the terms of the GNU Lesser General Public License as
- * published by the Free Software Foundation; either version 2.1 of
- * the License, or (at your option) any later version.
- *
- * This software is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
- * Lesser General Public License for more details.
- *
- * You should have received a copy of the GNU Lesser General Public
- * License along with this software; if not, write to the Free
- * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
- * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
- */
-
-package org.jboss.messaging.tests.integration.paging;
-
-import java.util.HashMap;
-
-import junit.framework.AssertionFailedError;
-
-import org.jboss.messaging.core.buffers.ChannelBuffers;
-import org.jboss.messaging.core.client.ClientConsumer;
-import org.jboss.messaging.core.client.ClientMessage;
-import org.jboss.messaging.core.client.ClientProducer;
-import org.jboss.messaging.core.client.ClientSession;
-import org.jboss.messaging.core.client.ClientSessionFactory;
-import org.jboss.messaging.core.config.Configuration;
-import org.jboss.messaging.core.logging.Logger;
-import org.jboss.messaging.core.message.impl.MessageImpl;
-import org.jboss.messaging.core.remoting.spi.MessagingBuffer;
-import org.jboss.messaging.core.server.MessagingService;
-import org.jboss.messaging.core.server.Queue;
-import org.jboss.messaging.core.settings.impl.AddressSettings;
-import org.jboss.messaging.tests.util.ServiceTestBase;
-import org.jboss.messaging.utils.DataConstants;
-import org.jboss.messaging.utils.SimpleString;
-
-/**
- * A PagingServiceIntegrationTest
- *
- * @author <a href="mailto:clebert.suconic at jboss.org">Clebert Suconic</a>
- *
- * Created Dec 5, 2008 8:25:58 PM
- *
- *
- */
-public class PagingServiceIntegrationTest extends ServiceTestBase
-{
-
- // Constants -----------------------------------------------------
- private static final Logger log = Logger.getLogger(PagingServiceIntegrationTest.class);
-
- private static final int RECEIVE_TIMEOUT = 30000;
-
- // Attributes ----------------------------------------------------
-
- // Static --------------------------------------------------------
-
- static final SimpleString ADDRESS = new SimpleString("SimpleAddress");
-
- // Constructors --------------------------------------------------
-
- // Public --------------------------------------------------------
-
- public void testSendReceivePaging() throws Exception
- {
- clearData();
-
- Configuration config = createDefaultConfig();
-
- config.setPagingMaxGlobalSizeBytes(100 * 1024);
- config.setPagingGlobalWatermarkSize(10 * 1024);
-
- MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
-
- messagingService.start();
-
- final int numberOfIntegers = 256;
-
- final int numberOfMessages = 10000;
-
- try
- {
- ClientSessionFactory sf = createInVMFactory();
-
- sf.setBlockOnNonPersistentSend(true);
- sf.setBlockOnPersistentSend(true);
- sf.setBlockOnAcknowledge(true);
-
- ClientSession session = sf.createSession(null, null, false, true, true, false, 0);
-
- session.createQueue(ADDRESS, ADDRESS, null, true, false);
-
- ClientProducer producer = session.createProducer(ADDRESS);
-
- ClientMessage message = null;
-
- byte[] body = null;
-
- for (int i = 0; i < numberOfMessages; i++)
- {
- message = session.createClientMessage(true);
-
- MessagingBuffer bodyLocal = message.getBody();
-
- for (int j = 1; j <= numberOfIntegers; j++)
- {
- bodyLocal.writeInt(j);
- }
-
- if (body == null)
- {
- body = bodyLocal.array();
- }
-
- message.putIntProperty(new SimpleString("id"), i);
-
- producer.send(message);
- }
-
- session.close();
-
- messagingService.stop();
-
- messagingService = createService(true, config, new HashMap<String, AddressSettings>());
- messagingService.start();
-
- sf = createInVMFactory();
-
- assertTrue(messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize() > 0);
-
- session = sf.createSession(null, null, false, true, true, false, 0);
-
- ClientConsumer consumer = session.createConsumer(ADDRESS);
-
- session.start();
-
- for (int i = 0; i < numberOfMessages; i++)
- {
- ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
-
- assertNotNull(message2);
-
- assertEquals(i, ((Integer)message2.getProperty(new SimpleString("id"))).intValue());
-
- message2.acknowledge();
-
- assertNotNull(message2);
-
- try
- {
- assertEqualsByteArrays(body.length, body, message2.getBody().array());
- }
- catch (AssertionFailedError e)
- {
- log.info("Expected buffer:" + dumbBytesHex(body, 40));
- log.info("Arriving buffer:" + dumbBytesHex(message2.getBody().array(), 40));
- throw e;
- }
- }
-
- consumer.close();
-
- session.close();
-
- assertEquals(0, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
-
- }
- finally
- {
- try
- {
- messagingService.stop();
- }
- catch (Throwable ignored)
- {
- }
- }
-
- }
-
- /**
- * - Make a destination in page mode
- * - Add stuff to a transaction
- * - Consume the entire destination (not in page mode any more)
- * - Add stuff to a transaction again
- * - Check order
- *
- */
- public void testDepageDuringTransaction() throws Exception
- {
- clearData();
-
- Configuration config = createDefaultConfig();
-
- config.setPagingMaxGlobalSizeBytes(100 * 1024);
- config.setPagingGlobalWatermarkSize(10 * 1024);
-
- MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
-
- messagingService.start();
-
- final int numberOfIntegers = 256;
-
- try
- {
- ClientSessionFactory sf = createInVMFactory();
-
- sf.setBlockOnNonPersistentSend(true);
- sf.setBlockOnPersistentSend(true);
- sf.setBlockOnAcknowledge(true);
-
- ClientSession session = sf.createSession(null, null, false, true, true, false, 0);
-
- session.createQueue(ADDRESS, ADDRESS, null, true, false);
-
- ClientProducer producer = session.createProducer(ADDRESS);
-
- MessagingBuffer bodyLocal = ChannelBuffers.buffer(DataConstants.SIZE_INT * numberOfIntegers);
-
- ClientMessage message = null;
-
- int numberOfMessages = 0;
- while (true)
- {
- message = session.createClientMessage(true);
- message.setBody(bodyLocal);
-
- // Stop sending message as soon as we start paging
- if (messagingService.getServer().getPostOffice().getPagingManager().isPaging(ADDRESS))
- {
- break;
- }
- numberOfMessages++;
-
- producer.send(message);
- }
-
- assertTrue(messagingService.getServer().getPostOffice().getPagingManager().isPaging(ADDRESS));
-
- session.start();
-
- ClientSession sessionTransacted = sf.createSession(null, null, false, false, false, false, 0);
-
- ClientProducer producerTransacted = sessionTransacted.createProducer(ADDRESS);
-
- for (int i = 0; i < 10; i++)
- {
- message = session.createClientMessage(true);
- message.setBody(bodyLocal);
- message.putIntProperty(new SimpleString("id"), i);
-
- // Consume messages to force an eventual out of order delivery
- if (i == 5)
- {
- ClientConsumer consumer = session.createConsumer(ADDRESS);
- for (int j = 0; j < numberOfMessages; j++)
- {
- ClientMessage msg = consumer.receive(RECEIVE_TIMEOUT);
- msg.acknowledge();
- assertNotNull(msg);
- }
-
- assertNull(consumer.receive(100));
- consumer.close();
- }
-
- Integer messageID = (Integer)message.getProperty(new SimpleString("id"));
- assertNotNull(messageID);
- assertEquals(messageID.intValue(), i);
-
- producerTransacted.send(message);
- }
-
- ClientConsumer consumer = session.createConsumer(ADDRESS);
-
- assertNull(consumer.receive(100));
-
- sessionTransacted.commit();
-
- sessionTransacted.close();
-
- for (int i = 0; i < 10; i++)
- {
- message = consumer.receive(RECEIVE_TIMEOUT);
-
- assertNotNull(message);
-
- Integer messageID = (Integer)message.getProperty(new SimpleString("id"));
-
- assertNotNull(messageID);
- assertEquals("message received out of order", messageID.intValue(), i);
-
- message.acknowledge();
- }
-
- assertNull(consumer.receive(100));
-
- consumer.close();
-
- session.close();
-
- assertEquals(0, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
-
- }
- finally
- {
- try
- {
- messagingService.stop();
- }
- catch (Throwable ignored)
- {
- }
- }
-
- }
-
- public void testPageOnSchedulingNoRestart() throws Exception
- {
- internalTestPageOnScheduling(false);
- }
-
- public void testPageOnSchedulingRestart() throws Exception
- {
- internalTestPageOnScheduling(true);
- }
-
- public void internalTestPageOnScheduling(final boolean restart) throws Exception
- {
- clearData();
-
- Configuration config = createDefaultConfig();
-
- config.setPagingMaxGlobalSizeBytes(100 * 1024);
- config.setPagingGlobalWatermarkSize(10 * 1024);
-
- MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
-
- messagingService.start();
-
- final int numberOfIntegers = 256;
-
- final int numberOfMessages = 10000;
-
- try
- {
- ClientSessionFactory sf = createInVMFactory();
-
- sf.setBlockOnNonPersistentSend(true);
- sf.setBlockOnPersistentSend(true);
- sf.setBlockOnAcknowledge(true);
-
- ClientSession session = sf.createSession(null, null, false, true, true, false, 0);
-
- session.createQueue(ADDRESS, ADDRESS, null, true, false);
-
- ClientProducer producer = session.createProducer(ADDRESS);
-
- ClientMessage message = null;
-
- byte[] body = null;
-
- long scheduledTime = System.currentTimeMillis() + 5000;
-
- for (int i = 0; i < numberOfMessages; i++)
- {
- message = session.createClientMessage(true);
-
- MessagingBuffer bodyLocal = message.getBody();
-
- for (int j = 1; j <= numberOfIntegers; j++)
- {
- bodyLocal.writeInt(j);
- }
-
-
- if (body == null)
- {
- body = bodyLocal.array();
- }
-
- message.setBody(bodyLocal);
- message.putIntProperty(new SimpleString("id"), i);
-
- // Worse scenario possible... only schedule what's on pages
- if (messagingService.getServer().getPostOffice().getPagingManager().isPaging(ADDRESS))
- {
- message.putLongProperty(MessageImpl.HDR_SCHEDULED_DELIVERY_TIME, scheduledTime);
- }
-
- producer.send(message);
- }
-
- if (restart)
- {
- session.close();
-
- messagingService.stop();
-
- messagingService = createService(true, config, new HashMap<String, AddressSettings>());
- messagingService.start();
-
- sf = createInVMFactory();
-
- session = sf.createSession(null, null, false, true, true, false, 0);
- }
-
- ClientConsumer consumer = session.createConsumer(ADDRESS);
-
- session.start();
-
- for (int i = 0; i < numberOfMessages; i++)
- {
- ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
-
- assertNotNull(message2);
-
- message2.acknowledge();
-
- assertNotNull(message2);
-
- Long scheduled = (Long)message2.getProperty(MessageImpl.HDR_SCHEDULED_DELIVERY_TIME);
- if (scheduled != null)
- {
- assertTrue("Scheduling didn't work", System.currentTimeMillis() >= scheduledTime);
- }
-
- try
- {
- assertEqualsByteArrays(body.length, body, message2.getBody().array());
- }
- catch (AssertionFailedError e)
- {
- log.info("Expected buffer:" + dumbBytesHex(body, 40));
- log.info("Arriving buffer:" + dumbBytesHex(message2.getBody().array(), 40));
- throw e;
- }
- }
-
- consumer.close();
-
- session.close();
-
- assertEquals(0, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
-
- }
- finally
- {
- try
- {
- messagingService.stop();
- }
- catch (Throwable ignored)
- {
- }
- }
-
- }
-
- public void testRollbackOnSend() throws Exception
- {
- clearData();
-
- Configuration config = createDefaultConfig();
-
- config.setPagingMaxGlobalSizeBytes(100 * 1024);
- config.setPagingGlobalWatermarkSize(10 * 1024);
-
- MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
-
- messagingService.start();
-
- final int numberOfIntegers = 256;
-
- final int numberOfMessages = 10;
-
- try
- {
- ClientSessionFactory sf = createInVMFactory();
-
- sf.setBlockOnNonPersistentSend(true);
- sf.setBlockOnPersistentSend(true);
- sf.setBlockOnAcknowledge(true);
-
- ClientSession session = sf.createSession(null, null, false, false, true, false, 0);
-
- session.createQueue(ADDRESS, ADDRESS, null, true, false);
-
- ClientProducer producer = session.createProducer(ADDRESS);
-
- long initialSize = messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize();
-
- ClientMessage message = null;
-
- for (int i = 0; i < numberOfMessages; i++)
- {
- message = session.createClientMessage(true);
-
- MessagingBuffer bodyLocal = message.getBody();
-
- for (int j = 1; j <= numberOfIntegers; j++)
- {
- bodyLocal.writeInt(j);
- }
-
-
- message.putIntProperty(new SimpleString("id"), i);
-
- producer.send(message);
- }
-
- session.rollback();
-
- ClientConsumer consumer = session.createConsumer(ADDRESS);
-
- session.start();
-
- assertNull(consumer.receive(100));
-
- session.close();
-
- assertEquals(initialSize, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
- }
- finally
- {
- try
- {
- messagingService.stop();
- }
- catch (Throwable ignored)
- {
- }
- }
-
- }
-
- public void testCommitOnSend() throws Exception
- {
- clearData();
-
- Configuration config = createDefaultConfig();
-
- config.setPagingMaxGlobalSizeBytes(100 * 1024);
- config.setPagingGlobalWatermarkSize(10 * 1024);
-
- MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
-
- messagingService.start();
-
- final int numberOfIntegers = 10;
-
- final int numberOfMessages = 10;
-
- try
- {
- ClientSessionFactory sf = createInVMFactory();
-
- sf.setBlockOnNonPersistentSend(true);
- sf.setBlockOnPersistentSend(true);
- sf.setBlockOnAcknowledge(true);
-
- ClientSession session = sf.createSession(null, null, false, false, false, false, 0);
-
- session.createQueue(ADDRESS, ADDRESS, null, true, false);
-
- ClientProducer producer = session.createProducer(ADDRESS);
-
- long initialSize = messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize();
-
- ClientMessage message = null;
-
- for (int i = 0; i < numberOfMessages; i++)
- {
- message = session.createClientMessage(true);
-
- MessagingBuffer bodyLocal = message.getBody();
-
- for (int j = 1; j <= numberOfIntegers; j++)
- {
- bodyLocal.writeInt(j);
- }
-
- message.putIntProperty(new SimpleString("id"), i);
-
- producer.send(message);
- }
-
- session.commit();
-
- ClientConsumer consumer = session.createConsumer(ADDRESS);
-
- session.start();
- for (int i = 0; i < numberOfMessages; i++)
- {
- ClientMessage msg = consumer.receive(500);
- assertNotNull(msg);
- msg.acknowledge();
- }
-
- session.commit();
-
- session.close();
-
- assertEquals(initialSize, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
- }
- finally
- {
- try
- {
- messagingService.stop();
- }
- catch (Throwable ignored)
- {
- }
- }
-
- }
-
- public void testPageMultipleDestinations() throws Exception
- {
- internalTestPageMultipleDestinations(false);
- }
-
- public void testPageMultipleDestinationsTransacted() throws Exception
- {
- internalTestPageMultipleDestinations(true);
- }
-
- public void testDropMessagesQueueMax() throws Exception
- {
- testDropMessages(false);
- }
-
- public void testDropMessagesGlobalMax() throws Exception
- {
- testDropMessages(true);
- }
-
- private void testDropMessages(boolean global) throws Exception
- {
- clearData();
-
- Configuration config = createDefaultConfig();
-
- HashMap<String, AddressSettings> settings = new HashMap<String, AddressSettings>();
-
- AddressSettings set = new AddressSettings();
- set.setDropMessagesWhenFull(true);
-
- settings.put(ADDRESS.toString(), set);
-
- if (global)
- {
- set.setMaxSizeBytes(-1);
- config.setPagingMaxGlobalSizeBytes(10 * 1024);
- }
- else
- {
- config.setPagingMaxGlobalSizeBytes(-1);
- set.setMaxSizeBytes(10 * 1024);
- }
-
- config.setPagingGlobalWatermarkSize(10 * 1024);
-
- MessagingService messagingService = createService(true, config, settings);
-
- messagingService.start();
-
- final int numberOfMessages = 1000;
-
- try
- {
- ClientSessionFactory sf = createInVMFactory();
-
- sf.setBlockOnNonPersistentSend(true);
- sf.setBlockOnPersistentSend(true);
- sf.setBlockOnAcknowledge(true);
-
- ClientSession session = sf.createSession(null, null, false, true, true, false, 0);
-
- session.createQueue(ADDRESS, ADDRESS, null, true, false);
-
- ClientProducer producer = session.createProducer(ADDRESS);
-
- ClientMessage message = null;
-
- for (int i = 0; i < numberOfMessages; i++)
- {
- MessagingBuffer bodyLocal = ChannelBuffers.wrappedBuffer(new byte[1024]);
-
- message = session.createClientMessage(true);
- message.setBody(bodyLocal);
-
- producer.send(message);
- }
-
- session = sf.createSession(null, null, false, true, true, false, 0);
-
- ClientConsumer consumer = session.createConsumer(ADDRESS);
-
- session.start();
-
- for (int i = 0; i < 9; i++)
- {
- ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
-
- assertNotNull(message2);
-
- message2.acknowledge();
- }
-
- assertNull(consumer.receive(100));
-
- assertEquals(0, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
- assertEquals(0, messagingService.getServer()
- .getPostOffice()
- .getPagingManager()
- .getPageStore(ADDRESS)
- .getAddressSize());
-
- for (int i = 0; i < numberOfMessages; i++)
- {
- MessagingBuffer bodyLocal = ChannelBuffers.wrappedBuffer(new byte[1024]);
-
- message = session.createClientMessage(true);
- message.setBody(bodyLocal);
-
- producer.send(message);
- }
-
- for (int i = 0; i < 9; i++)
- {
- ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
-
- assertNotNull(message2);
-
- message2.acknowledge();
- }
-
- assertNull(consumer.receive(100));
-
- session.close();
-
- session = sf.createSession(false, true, true);
-
- producer = session.createProducer(ADDRESS);
-
- for (int i = 0; i < numberOfMessages; i++)
- {
- MessagingBuffer bodyLocal = ChannelBuffers.wrappedBuffer(new byte[1024]);
-
- message = session.createClientMessage(true);
- message.setBody(bodyLocal);
-
- producer.send(message);
- }
-
- session.commit();
-
- consumer = session.createConsumer(ADDRESS);
-
- session.start();
-
- for (int i = 0; i < 9; i++)
- {
- ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
-
- assertNotNull(message2);
-
- message2.acknowledge();
- }
-
- session.commit();
-
- assertNull(consumer.receive(100));
-
- session.close();
-
- assertEquals(0, messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize());
- assertEquals(0, messagingService.getServer()
- .getPostOffice()
- .getPagingManager()
- .getPageStore(ADDRESS)
- .getAddressSize());
-
- }
- finally
- {
- try
- {
- messagingService.stop();
- }
- catch (Throwable ignored)
- {
- }
- }
-
- }
-
- private void internalTestPageMultipleDestinations(boolean transacted) throws Exception
- {
- Configuration config = createDefaultConfig();
-
- final int NUMBER_OF_BINDINGS = 100;
-
- int NUMBER_OF_MESSAGES = 2;
-
- config.setPagingMaxGlobalSizeBytes(100 * 1024);
- config.setPagingGlobalWatermarkSize(10 * 1024);
-
- MessagingService messagingService = createService(true, config, new HashMap<String, AddressSettings>());
-
- messagingService.start();
-
- try
- {
- ClientSessionFactory sf = createInVMFactory();
-
- sf.setBlockOnNonPersistentSend(true);
- sf.setBlockOnPersistentSend(true);
- sf.setBlockOnAcknowledge(true);
-
- ClientSession session = sf.createSession(null, null, false, !transacted, true, false, 0);
-
- for (int i = 0; i < NUMBER_OF_BINDINGS; i++)
- {
- session.createQueue(ADDRESS, new SimpleString("someQueue" + i), null, true, false);
- }
-
- ClientProducer producer = session.createProducer(ADDRESS);
-
- ClientMessage message = null;
-
- MessagingBuffer bodyLocal = ChannelBuffers.wrappedBuffer(new byte[1024]);
-
- message = session.createClientMessage(true);
- message.setBody(bodyLocal);
-
- for (int i = 0; i < NUMBER_OF_MESSAGES; i++)
- {
- producer.send(message);
-
- if (transacted)
- {
- session.commit();
- }
- }
-
- session.close();
-
- messagingService.stop();
-
- messagingService = createService(true, config, new HashMap<String, AddressSettings>());
- messagingService.start();
-
- sf = createInVMFactory();
-
- assertTrue(messagingService.getServer().getPostOffice().getPagingManager().getGlobalSize() > 0);
-
- session = sf.createSession(null, null, false, true, true, false, 0);
-
- session.start();
-
- for (int msg = 0; msg < NUMBER_OF_MESSAGES; msg++)
- {
-
- for (int i = 0; i < NUMBER_OF_BINDINGS; i++)
- {
- ClientConsumer consumer = session.createConsumer(new SimpleString("someQueue" + i));
-
- ClientMessage message2 = consumer.receive(RECEIVE_TIMEOUT);
-
- assertNotNull(message2);
-
- message2.acknowledge();
-
- assertNotNull(message2);
-
- consumer.close();
-
- }
- }
-
- session.close();
-
- for (int i = 0; i < NUMBER_OF_BINDINGS; i++)
- {
- Queue queue = (Queue)messagingService.getServer()
- .getPostOffice()
- .getBinding(new SimpleString("someQueue" + i))
- .getBindable();
-
- assertEquals("Queue someQueue" + i + " was supposed to be empty", 0, queue.getMessageCount());
- assertEquals("Queue someQueue" + i + " was supposed to be empty", 0, queue.getDeliveringCount());
- }
-
- assertEquals("There are pending messages on the server", 0, messagingService.getServer()
- .getPostOffice()
- .getPagingManager()
- .getGlobalSize());
-
- }
- finally
- {
- try
- {
- messagingService.stop();
- }
- catch (Throwable ignored)
- {
- }
- }
-
- }
-
- // Package protected ---------------------------------------------
-
- // Protected -----------------------------------------------------
-
- // Private -------------------------------------------------------
-
- // Inner classes -------------------------------------------------
-
-}
Modified: trunk/tests/src/org/jboss/messaging/tests/util/ServiceTestBase.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/util/ServiceTestBase.java 2009-03-25 19:50:12 UTC (rev 6166)
+++ trunk/tests/src/org/jboss/messaging/tests/util/ServiceTestBase.java 2009-03-26 02:47:40 UTC (rev 6167)
@@ -27,6 +27,8 @@
import java.util.HashMap;
import java.util.Map;
+import javax.management.MBeanServer;
+
import org.jboss.messaging.core.client.ClientMessage;
import org.jboss.messaging.core.client.ClientSession;
import org.jboss.messaging.core.client.ClientSessionFactory;
@@ -131,7 +133,6 @@
final Configuration configuration,
final Map<String, AddressSettings> settings)
{
-
MessagingService service;
if (realFiles)
@@ -156,6 +157,37 @@
return service;
}
+
+ protected MessagingService createService(final boolean realFiles,
+ final Configuration configuration,
+ final MBeanServer mbeanServer,
+ final Map<String, AddressSettings> settings)
+ {
+
+ MessagingService service;
+
+ if (realFiles)
+ {
+ service = Messaging.newMessagingService(configuration, mbeanServer);
+ }
+ else
+ {
+ service = Messaging.newNullStorageMessagingService(configuration, mbeanServer);
+ }
+
+ for (Map.Entry<String, AddressSettings> setting : settings.entrySet())
+ {
+ service.getServer().getAddressSettingsRepository().addMatch(setting.getKey(), setting.getValue());
+ }
+
+ AddressSettings defaultSetting = new AddressSettings();
+ defaultSetting.setPageSizeBytes(configuration.getPagingGlobalWatermarkSize());
+
+ service.getServer().getAddressSettingsRepository().addMatch("#", defaultSetting);
+
+ return service;
+ }
+
protected MessagingService createService(final boolean realFiles)
{
return createService(realFiles, createDefaultConfig(), new HashMap<String, AddressSettings>());
@@ -230,6 +262,7 @@
{
Configuration configuration = new ConfigurationImpl();
configuration.setSecurityEnabled(false);
+ configuration.setJMXManagementEnabled(false);
configuration.setBindingsDirectory(getBindingsDir());
configuration.setJournalMinFiles(2);
configuration.setJournalDirectory(getJournalDir());
More information about the jboss-cvs-commits
mailing list