[jboss-cvs] JBoss Messaging SVN: r8449 - in branches/Branch_1_4: tests/src/org/jboss/test/messaging/jms and 1 other directory.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Tue Oct 18 23:17:35 EDT 2011
Author: gaohoward
Date: 2011-10-18 23:17:35 -0400 (Tue, 18 Oct 2011)
New Revision: 8449
Modified:
branches/Branch_1_4/src/main/org/jboss/jms/client/container/SessionAspect.java
branches/Branch_1_4/tests/src/org/jboss/test/messaging/jms/AcknowledgementTest.java
Log:
JBMESSAGING-1897
Modified: branches/Branch_1_4/src/main/org/jboss/jms/client/container/SessionAspect.java
===================================================================
--- branches/Branch_1_4/src/main/org/jboss/jms/client/container/SessionAspect.java 2011-10-19 03:16:06 UTC (rev 8448)
+++ branches/Branch_1_4/src/main/org/jboss/jms/client/container/SessionAspect.java 2011-10-19 03:17:35 UTC (rev 8449)
@@ -369,6 +369,7 @@
}
finally
{
+ state.clearAckMap();
state.setAutoAckInfo(null);
}
}
Modified: branches/Branch_1_4/tests/src/org/jboss/test/messaging/jms/AcknowledgementTest.java
===================================================================
--- branches/Branch_1_4/tests/src/org/jboss/test/messaging/jms/AcknowledgementTest.java 2011-10-19 03:16:06 UTC (rev 8448)
+++ branches/Branch_1_4/tests/src/org/jboss/test/messaging/jms/AcknowledgementTest.java 2011-10-19 03:17:35 UTC (rev 8449)
@@ -42,6 +42,8 @@
import org.jboss.jms.client.JBossSession;
import org.jboss.jms.client.delegate.ClientSessionDelegate;
+import org.jboss.jms.client.delegate.DelegateSupport;
+import org.jboss.jms.client.state.SessionState;
import org.jboss.test.messaging.tools.ServerManagement;
import EDU.oswego.cs.dl.util.concurrent.Latch;
@@ -70,6 +72,285 @@
// Public --------------------------------------------------------
+ public void testMemLeakInDifferentAckModes() throws Exception
+ {
+ final int NUM_MSG = 50;
+
+ Connection conn = null;
+ Session producerSess = null;
+ Session consumerSess = null;
+
+ //auto
+ log.info("Testing leak in AUTO_ACKNOWLEDGE mode...");
+ try
+ {
+ conn = cf.createConnection();
+
+ producerSess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = producerSess.createProducer(queue1);
+
+ consumerSess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageConsumer consumer = consumerSess.createConsumer(queue1);
+ conn.start();
+
+ //Send some messages
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ Message m = producerSess.createMessage();
+ producer.send(m);
+ }
+
+ log.trace("Sent messages");
+
+ int count = 0;
+
+ Message m = null;
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ m = consumer.receive(2000);
+ if (m == null) break;
+ count++;
+ }
+ assertRemainingMessages(0);
+
+ assertEquals(count, NUM_MSG);
+
+ checkSessionAckMap((JBossSession)consumerSess);
+ }
+ finally
+ {
+ if (conn != null)
+ {
+ conn.close();
+ }
+ conn = null;
+ producerSess = null;
+ consumerSess = null;
+ }
+
+ //dups
+ log.info("Testing leak in DUPS_OK_ACKNOWLEDGE mode...");
+
+ final int BATCH_SIZE = 10;
+
+ String mbeanConfig =
+ "<mbean code=\"org.jboss.jms.server.connectionfactory.ConnectionFactory\"\n" +
+ " name=\"jboss.messaging.destination:service=DupsConnectionFactory\"\n" +
+ " xmbean-dd=\"xmdesc/ConnectionFactory-xmbean.xml\">\n" +
+ " <depends optional-attribute-name=\"ServerPeer\">jboss.messaging:service=ServerPeer</depends>\n" +
+ " <depends optional-attribute-name=\"Connector\">jboss.messaging:service=Connector,transport=bisocket</depends>\n" +
+ " <attribute name=\"JNDIBindings\">\n" +
+ " <bindings>\n" +
+ " <binding>/mycf</binding>\n" +
+ " </bindings>\n" +
+ " </attribute>\n" +
+ " <attribute name=\"DupsOKBatchSize\">" + BATCH_SIZE + "</attribute>" +
+ " </mbean>";
+
+ ObjectName on = ServerManagement.deploy(mbeanConfig);
+ ServerManagement.invoke(on, "create", new Object[0], new String[0]);
+ ServerManagement.invoke(on, "start", new Object[0], new String[0]);
+
+ try
+ {
+ ConnectionFactory myCF = (ConnectionFactory)ic.lookup("/mycf");
+
+ conn = myCF.createConnection();
+
+ producerSess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = producerSess.createProducer(queue1);
+
+ consumerSess = conn.createSession(false, Session.DUPS_OK_ACKNOWLEDGE);
+ MessageConsumer consumer = consumerSess.createConsumer(queue1);
+ conn.start();
+
+ //Send some messages
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ Message m = producerSess.createMessage();
+ producer.send(m);
+ }
+
+ log.trace("Sent messages");
+
+ Message m = null;
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ m = consumer.receive(2000);
+
+ assertNotNull(m);
+ }
+
+ checkSessionAckMap((JBossSession)consumerSess);
+ }
+ finally
+ {
+ if (conn != null)
+ {
+ conn.close();
+ }
+ conn = null;
+ producerSess = null;
+ consumerSess = null;
+
+ ServerManagement.invoke(on, "stop", new Object[0], new String[0]);
+ ServerManagement.invoke(on, "destroy", new Object[0], new String[0]);
+ }
+
+ //client
+ log.info("Testing leak in CLIENT_ACKNOWLEDGE mode...");
+
+ try
+ {
+ conn = cf.createConnection();
+
+ producerSess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = producerSess.createProducer(queue1);
+
+ consumerSess = conn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+ MessageConsumer consumer = consumerSess.createConsumer(queue1);
+ conn.start();
+
+ //Send some messages
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ Message m = producerSess.createMessage();
+ producer.send(m);
+ }
+
+ log.trace("Sent messages");
+
+ Message m = null;
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ m = consumer.receive(2000);
+ }
+
+ m.acknowledge();
+ assertRemainingMessages(0);
+
+ checkSessionAckMap((JBossSession)consumerSess);
+ }
+ finally
+ {
+ if (conn != null)
+ {
+ conn.close();
+ }
+ conn = null;
+ producerSess = null;
+ consumerSess = null;
+ }
+
+ //transactional (actually the ackMap is not used in tx mode (including XA), but test anyway)
+ log.info("Testing leak in SESSION_TRANSACTED mode...");
+
+ try
+ {
+ conn = cf.createConnection();
+
+ producerSess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = producerSess.createProducer(queue1);
+
+ consumerSess = conn.createSession(true, Session.SESSION_TRANSACTED);
+ MessageConsumer consumer = consumerSess.createConsumer(queue1);
+ conn.start();
+
+ //Send some messages
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ Message m = producerSess.createMessage();
+ producer.send(m);
+ }
+
+ log.trace("Sent messages");
+
+ Message m = null;
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ m = consumer.receive(2000);
+ }
+
+ consumerSess.commit();
+ assertRemainingMessages(0);
+
+ checkSessionAckMap((JBossSession)consumerSess);
+ }
+ finally
+ {
+ if (conn != null)
+ {
+ conn.close();
+ }
+ conn = null;
+ producerSess = null;
+ consumerSess = null;
+ }
+
+ //recover
+ log.info("Testing leak in CLIENT_ACKNOWLEDGE mode with recover ...");
+
+ try
+ {
+ conn = cf.createConnection();
+
+ producerSess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = producerSess.createProducer(queue1);
+
+ consumerSess = conn.createSession(false, Session.CLIENT_ACKNOWLEDGE);
+ MessageConsumer consumer = consumerSess.createConsumer(queue1);
+ conn.start();
+
+ //Send some messages
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ Message m = producerSess.createMessage();
+ producer.send(m);
+ }
+
+ log.trace("Sent messages");
+
+ Message m = null;
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ m = consumer.receive(2000);
+ }
+
+ consumerSess.recover();
+
+ for (int i = 0; i < NUM_MSG; i++)
+ {
+ m = consumer.receive(2000);
+ }
+
+ m.acknowledge();
+ assertRemainingMessages(0);
+
+ checkSessionAckMap((JBossSession)consumerSess);
+ }
+ finally
+ {
+ if (conn != null)
+ {
+ conn.close();
+ }
+ conn = null;
+ producerSess = null;
+ consumerSess = null;
+ }
+
+ }
+
+ /**
+ * make sure the session's ackMap is cleared.
+ */
+ private void checkSessionAckMap(JBossSession session)
+ {
+ DelegateSupport support = (DelegateSupport)session.getDelegate();
+ SessionState state = (SessionState)support.getState();
+ assertEquals(0, state.getAckMap().size());
+ }
+
/* Topics shouldn't hold on to messages if there are no subscribers */
public void testPersistentMessagesForTopicDropped() throws Exception
{
More information about the jboss-cvs-commits
mailing list