git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1195045 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Timothy A. Bish 2011-10-29 21:42:49 +00:00
parent f0ed5d0c72
commit 83fde0b2a0
2 changed files with 114 additions and 8 deletions

View File

@ -107,7 +107,6 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
public static final String DEFAULT_BROKER_URL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL; public static final String DEFAULT_BROKER_URL = ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
private static final Logger LOG = LoggerFactory.getLogger(ActiveMQConnection.class); private static final Logger LOG = LoggerFactory.getLogger(ActiveMQConnection.class);
private static final IdGenerator CONNECTION_ID_GENERATOR = new IdGenerator();
public final ConcurrentHashMap<ActiveMQTempDestination, ActiveMQTempDestination> activeTempDestinations = new ConcurrentHashMap<ActiveMQTempDestination, ActiveMQTempDestination>(); public final ConcurrentHashMap<ActiveMQTempDestination, ActiveMQTempDestination> activeTempDestinations = new ConcurrentHashMap<ActiveMQTempDestination, ActiveMQTempDestination>();
@ -599,7 +598,12 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
* cause this exception to be thrown. * cause this exception to be thrown.
*/ */
public void close() throws JMSException { public void close() throws JMSException {
// Store the interrupted state and clear so that cleanup happens without
// leaking connection resources. Reset in finally to preserve state.
boolean interrupted = Thread.interrupted();
try { try {
// If we were running, lets stop first. // If we were running, lets stop first.
if (!closed.get() && !transportFailed.get()) { if (!closed.get() && !transportFailed.get()) {
stop(); stop();
@ -663,8 +667,6 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
doAsyncSendPacket(new ShutdownInfo()); doAsyncSendPacket(new ShutdownInfo());
} }
ServiceSupport.dispose(this.transport);
started.set(false); started.set(false);
// TODO if we move the TaskRunnerFactory to the connection // TODO if we move the TaskRunnerFactory to the connection
@ -680,13 +682,19 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
} }
} finally { } finally {
try { try {
if (executor != null){ if (executor != null) {
executor.shutdown(); executor.shutdown();
} }
}catch(Throwable e) { } catch (Throwable e) {
LOG.error("Error shutting down thread pool " + e,e); LOG.error("Error shutting down thread pool " + e, e);
} }
ServiceSupport.dispose(this.transport);
factoryStats.removeConnection(this); factoryStats.removeConnection(this);
if (interrupted) {
Thread.currentThread().interrupt();
}
} }
} }

View File

@ -0,0 +1,98 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.bugs;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.fail;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
public class AMQ3529Test {
private ConnectionFactory connectionFactory;
private Connection connection;
private Session session;
private BrokerService broker;
private String connectionUri;
@Before
public void startBroker() throws Exception {
broker = new BrokerService();
broker.setDeleteAllMessagesOnStartup(true);
broker.setPersistent(false);
broker.setUseJmx(false);
broker.addConnector("tcp://0.0.0.0:0");
broker.start();
broker.waitUntilStarted();
connectionUri = broker.getTransportConnectors().get(0).getPublishableConnectString();
connectionFactory = new ActiveMQConnectionFactory(connectionUri);
}
@After
public void stopBroker() throws Exception {
broker.stop();
broker.waitUntilStopped();
}
@Test(timeout = 60000)
public void testInterruptionAffects() throws Exception {
ThreadGroup tg = new ThreadGroup("tg");
assertEquals(0, tg.activeCount());
Thread client = new Thread(tg, "client") {
@Override
public void run() {
try {
connection = connectionFactory.createConnection();
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
assertNotNull(session);
} catch (JMSException e) {
fail(e.getMessage());
}
// next line is the nature of the test, if I remove this line, everything works OK
Thread.currentThread().interrupt();
try {
connection.close();
} catch (JMSException e) {
}
}
};
client.start();
client.join();
Thread.sleep(2000);
Thread[] remainThreads = new Thread[tg.activeCount()];
tg.enumerate(remainThreads);
for (Thread t : remainThreads) {
if (t.isAlive() && !t.isDaemon())
fail("Remaining thread: " + t.toString());
}
}
}