[jboss-cvs] JBoss Messaging SVN: r4958 - in trunk: src/main/org/jboss/messaging/core/remoting/impl and 3 other directories.
jboss-cvs-commits at lists.jboss.org
jboss-cvs-commits at lists.jboss.org
Tue Sep 16 17:33:07 EDT 2008
Author: clebert.suconic at jboss.com
Date: 2008-09-16 17:33:07 -0400 (Tue, 16 Sep 2008)
New Revision: 4958
Added:
trunk/tests/src/org/jboss/messaging/tests/integration/remoting/DestroyConsumerTest.java
Modified:
trunk/.classpath
trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java
trunk/tests/src/org/jboss/messaging/tests/integration/base/IntegrationTestBase.java
trunk/tests/src/org/jboss/messaging/tests/integration/remoting/PingTest.java
trunk/tests/src/org/jboss/messaging/tests/stress/paging/MultipleDestinationPagingTest.java
Log:
https://jira.jboss.org/jira/browse/JBMESSAGING-1421 - Calling listeners on destroy as well
Modified: trunk/.classpath
===================================================================
--- trunk/.classpath 2008-09-16 18:36:19 UTC (rev 4957)
+++ trunk/.classpath 2008-09-16 21:33:07 UTC (rev 4958)
@@ -63,6 +63,6 @@
<classpathentry kind="lib" path="thirdparty/cglib/lib/cglib.jar"/>
<classpathentry kind="lib" path="tests/tmpfiles"/>
<classpathentry kind="lib" path="thirdparty/apache-mina/lib/mina-core-2.0.0-M3-20080730.120633-1.jar" sourcepath="thirdparty/apache-mina/lib/mina-core-2.0.0-M3-20080730.120633-1-sources.jar"/>
- <classpathentry kind="lib" path="thirdparty/netty/lib/netty-3.0.0.CR3.jar"/>
+ <classpathentry kind="lib" path="thirdparty/netty/lib/netty-3.0.0.CR3.jar" sourcepath="thirdparty/netty/lib/netty-3.0.0.CR3-sources.jar"/>
<classpathentry kind="output" path="eclipse-output"/>
</classpath>
Modified: trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java 2008-09-16 18:36:19 UTC (rev 4957)
+++ trunk/src/main/org/jboss/messaging/core/remoting/impl/RemotingConnectionImpl.java 2008-09-16 21:33:07 UTC (rev 4958)
@@ -379,25 +379,10 @@
log.warn(me.getMessage());
- destroy();
+ internalClose();
// Then call the listeners
- final Set<FailureListener> listenersClone = new HashSet<FailureListener>(failureListeners);
-
- for (final FailureListener listener : listenersClone)
- {
- try
- {
- listener.connectionFailed(me);
- }
- catch (final Throwable t)
- {
- // Failure of one listener to execute shouldn't prevent others
- // from
- // executing
- log.error("Failed to execute failure listener", t);
- }
- }
+ callListeners(me);
}
}
@@ -410,22 +395,10 @@
return;
}
- if (future != null)
- {
- future.cancel(false);
- }
+ internalClose();
- pingChannel.close();
-
- destroyed = true;
-
- // We close the underlying transport connection
- transportConnection.close();
-
- if (replicatingConnection != null)
- {
- replicatingConnection.destroy();
- }
+ // We need to inform Listeners about the connection being closed
+ callListeners(null);
}
}
@@ -444,7 +417,7 @@
{
stopPinging = true;
}
-
+
// Buffer Handler implementation
// ----------------------------------------------------
@@ -471,6 +444,46 @@
// Private
// --------------------------------------------------------------------------------------
+ private void callListeners(final MessagingException me)
+ {
+ final Set<FailureListener> listenersClone = new HashSet<FailureListener>(failureListeners);
+
+ for (final FailureListener listener : listenersClone)
+ {
+ try
+ {
+ listener.connectionFailed(me);
+ }
+ catch (final Throwable t)
+ {
+ // Failure of one listener to execute shouldn't prevent others
+ // from
+ // executing
+ log.error("Failed to execute failure listener", t);
+ }
+ }
+ }
+
+ private void internalClose()
+ {
+ if (future != null)
+ {
+ future.cancel(false);
+ }
+
+ pingChannel.close();
+
+ destroyed = true;
+
+ // We close the underlying transport connection
+ transportConnection.close();
+
+ if (replicatingConnection != null)
+ {
+ replicatingConnection.destroy();
+ }
+ }
+
private void doWrite(final Packet packet)
{
if (destroyed)
Modified: trunk/tests/src/org/jboss/messaging/tests/integration/base/IntegrationTestBase.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/base/IntegrationTestBase.java 2008-09-16 18:36:19 UTC (rev 4957)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/base/IntegrationTestBase.java 2008-09-16 21:33:07 UTC (rev 4958)
@@ -34,6 +34,10 @@
import org.jboss.messaging.core.config.Configuration;
import org.jboss.messaging.core.config.TransportConfiguration;
import org.jboss.messaging.core.config.impl.ConfigurationImpl;
+import org.jboss.messaging.core.remoting.impl.invm.InVMAcceptorFactory;
+import org.jboss.messaging.core.remoting.impl.invm.InVMConnectorFactory;
+import org.jboss.messaging.core.remoting.impl.netty.NettyAcceptorFactory;
+import org.jboss.messaging.core.remoting.impl.netty.NettyConnectorFactory;
import org.jboss.messaging.core.server.MessagingService;
import org.jboss.messaging.core.server.impl.MessagingServiceImpl;
import org.jboss.messaging.core.settings.impl.QueueSettings;
@@ -55,9 +59,12 @@
// Attributes ----------------------------------------------------
- protected static final String ACCEPTOR_FACTORY = "org.jboss.messaging.core.remoting.impl.invm.InVMAcceptorFactory";
- protected static final String CONNECTOR_FACTORY = "org.jboss.messaging.core.remoting.impl.invm.InVMConnectorFactory";
+ protected static final String INVM_ACCEPTOR_FACTORY = InVMAcceptorFactory.class.getCanonicalName();
+ protected static final String INVM_CONNECTOR_FACTORY = InVMConnectorFactory.class.getCanonicalName();
+ protected static final String NETTY_ACCEPTOR_FACTORY = NettyAcceptorFactory.class.getCanonicalName();
+ protected static final String NETTY_CONNECTOR_FACTORY = NettyConnectorFactory.class.getCanonicalName();
+
protected String journalDir = System.getProperty("java.io.tmpdir", "/tmp") + "/integration-test/journal";
protected String bindingsDir = System.getProperty("java.io.tmpdir", "/tmp") + "/integration-test/bindings";
protected String pageDir = System.getProperty("java.io.tmpdir", "/tmp") + "/integration-test/page";
@@ -88,12 +95,28 @@
}
- protected MessagingService createService(Configuration configuration, Map<String, QueueSettings> settings)
+ protected MessagingService createService(boolean realFiles, boolean netty, Configuration configuration, Map<String, QueueSettings> settings)
{
- TransportConfiguration transportConfig = new TransportConfiguration(ACCEPTOR_FACTORY);
+ TransportConfiguration transportConfig = new TransportConfiguration(INVM_ACCEPTOR_FACTORY);
configuration.getAcceptorConfigurations().add(transportConfig);
- MessagingService service = MessagingServiceImpl.newNioStorageMessagingServer(configuration, journalDir, bindingsDir);
+ if (netty)
+ {
+ configuration.getAcceptorConfigurations().add(new TransportConfiguration(NETTY_ACCEPTOR_FACTORY));
+ }
+
+ MessagingService service;
+
+ if (realFiles)
+ {
+ service = MessagingServiceImpl.newNioStorageMessagingServer(configuration, journalDir, bindingsDir);
+ }
+ else
+ {
+ service = MessagingServiceImpl.newNullStorageMessagingServer(configuration);
+ }
+
+
for (Map.Entry<String, QueueSettings> setting: settings.entrySet())
{
service.getServer().getQueueSettingsRepository().addMatch(setting.getKey(), setting.getValue());
@@ -103,9 +126,9 @@
return service;
}
- protected MessagingService createService()
+ protected MessagingService createService(boolean realFiles)
{
- return createService(createDefaultConfig(), new HashMap<String, QueueSettings>());
+ return createService(realFiles, false, createDefaultConfig(), new HashMap<String, QueueSettings>());
}
@@ -120,11 +143,16 @@
}
- protected ClientSessionFactory createFactory()
+ protected ClientSessionFactory createInVMFactory()
{
- return new ClientSessionFactoryImpl(new TransportConfiguration(CONNECTOR_FACTORY));
+ return new ClientSessionFactoryImpl(new TransportConfiguration(INVM_CONNECTOR_FACTORY));
}
+ protected ClientSessionFactory createNettyFactory()
+ {
+ return new ClientSessionFactoryImpl(new TransportConfiguration(NETTY_CONNECTOR_FACTORY));
+ }
+
protected ClientMessage createTextMessage(ClientSession session, String s)
{
return createTextMessage(session, s, true);
Added: trunk/tests/src/org/jboss/messaging/tests/integration/remoting/DestroyConsumerTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/remoting/DestroyConsumerTest.java (rev 0)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/remoting/DestroyConsumerTest.java 2008-09-16 21:33:07 UTC (rev 4958)
@@ -0,0 +1,92 @@
+/*
+ * 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.remoting;
+
+import java.util.HashMap;
+
+import org.jboss.messaging.core.client.ClientConsumer;
+import org.jboss.messaging.core.client.ClientSession;
+import org.jboss.messaging.core.client.ClientSessionFactory;
+import org.jboss.messaging.core.client.impl.ClientSessionImpl;
+import org.jboss.messaging.core.postoffice.Binding;
+import org.jboss.messaging.core.server.MessagingService;
+import org.jboss.messaging.core.settings.impl.QueueSettings;
+import org.jboss.messaging.tests.integration.base.IntegrationTestBase;
+import org.jboss.messaging.util.SimpleString;
+
+public class DestroyConsumerTest extends IntegrationTestBase
+{
+
+ // Constants -----------------------------------------------------
+
+ // Attributes ----------------------------------------------------
+
+ // Static --------------------------------------------------------
+
+ // Constructors --------------------------------------------------
+
+ // Public --------------------------------------------------------
+
+ public void testDestroyConsumer() throws Exception
+ {
+ MessagingService service = createService(false, false, createDefaultConfig(), new HashMap<String, QueueSettings>());
+ service.start();
+
+ SimpleString queue = new SimpleString("add1");
+
+ ClientSessionFactory factory = createInVMFactory();
+
+ ClientSession session = factory.createSession(false, false, false, -1, false);
+
+ session.createQueue(queue, queue, null, false, false);
+
+ ClientConsumer consumer = session.createConsumer(queue);
+
+ session.start();
+
+ Binding binding = service.getServer().getPostOffice().getBindingsForAddress(queue).get(0);
+
+ assertEquals(1, binding.getQueue().getConsumerCount());
+
+ ClientSessionImpl impl = (ClientSessionImpl) session;
+
+ // Simulating a CTRL-C what would close the Socket but not the ClientSession
+ impl.cleanUp();
+
+
+ assertEquals(0, binding.getQueue().getConsumerCount());
+
+
+
+ }
+
+ // Package protected ---------------------------------------------
+
+ // Protected -----------------------------------------------------
+
+ // Private -------------------------------------------------------
+
+ // Inner classes -------------------------------------------------
+
+}
Modified: trunk/tests/src/org/jboss/messaging/tests/integration/remoting/PingTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/integration/remoting/PingTest.java 2008-09-16 18:36:19 UTC (rev 4957)
+++ trunk/tests/src/org/jboss/messaging/tests/integration/remoting/PingTest.java 2008-09-16 21:33:07 UTC (rev 4958)
@@ -264,7 +264,7 @@
registry.returnConnection(conn2.getID());
}
-
+
/*
* Test the client triggering failure due to no pong received in time
*/
Modified: trunk/tests/src/org/jboss/messaging/tests/stress/paging/MultipleDestinationPagingTest.java
===================================================================
--- trunk/tests/src/org/jboss/messaging/tests/stress/paging/MultipleDestinationPagingTest.java 2008-09-16 18:36:19 UTC (rev 4957)
+++ trunk/tests/src/org/jboss/messaging/tests/stress/paging/MultipleDestinationPagingTest.java 2008-09-16 21:33:07 UTC (rev 4958)
@@ -92,10 +92,10 @@
settings.put("page-adr", setting);
}
- service = createService(config, settings);
+ service = createService(true, false, config, settings);
service.start();
- ClientSessionFactory factory = createFactory();
+ ClientSessionFactory factory = createInVMFactory();
ClientSession session = null;
try
More information about the jboss-cvs-commits
mailing list