From 6595dc4a62d53744a4ba3fa5ff3841280ecbd1ab Mon Sep 17 00:00:00 2001 From: "Timothy A. Bish" Date: Thu, 24 Feb 2011 16:51:59 +0000 Subject: [PATCH] Apply patch for https://issues.apache.org/jira/browse/AMQ-3185 Confirmed the current code allows for two calls to stop before the disposed flag is triggered. git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1074209 13f79535-47bb-0310-9956-ffa450edef68 --- .../activemq/transport/vm/VMTransport.java | 2 +- .../transport/vm/VMTransportServer.java | 14 +- .../activemq/bugs/VMTransportClosureTest.java | 131 ++++++++++++++++++ 3 files changed, 139 insertions(+), 8 deletions(-) create mode 100644 activemq-core/src/test/java/org/apache/activemq/bugs/VMTransportClosureTest.java diff --git a/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java b/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java index 5484eb7930..3e05d4a8d8 100755 --- a/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java +++ b/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransport.java @@ -59,7 +59,7 @@ public class VMTransport implements Transport, Task { private TaskRunner taskRunner; private final Object lazyInitMutext = new Object(); private final Valve enqueueValve = new Valve(true); - private final AtomicBoolean stopping = new AtomicBoolean(); + protected final AtomicBoolean stopping = new AtomicBoolean(); private volatile int receiveCounter; public VMTransport(URI location) { diff --git a/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransportServer.java b/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransportServer.java index a60da5e25a..1313427a4f 100755 --- a/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransportServer.java +++ b/activemq-core/src/main/java/org/apache/activemq/transport/vm/VMTransportServer.java @@ -75,13 +75,13 @@ public class VMTransportServer implements TransportServer { connectionCount.incrementAndGet(); VMTransport client = new VMTransport(location) { public void stop() throws Exception { - if (disposed) { - return; - } - super.stop(); - if (connectionCount.decrementAndGet() == 0 && disposeOnDisconnect) { - VMTransportServer.this.stop(); - } + if (stopping.compareAndSet(false, true) && !disposed) { + super.stop(); + if (connectionCount.decrementAndGet() == 0 + && disposeOnDisconnect) { + VMTransportServer.this.stop(); + } + } }; }; diff --git a/activemq-core/src/test/java/org/apache/activemq/bugs/VMTransportClosureTest.java b/activemq-core/src/test/java/org/apache/activemq/bugs/VMTransportClosureTest.java new file mode 100644 index 0000000000..6a96a147a5 --- /dev/null +++ b/activemq-core/src/test/java/org/apache/activemq/bugs/VMTransportClosureTest.java @@ -0,0 +1,131 @@ +/** + * 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 java.io.IOException; + +import javax.jms.MessageProducer; +import javax.jms.Session; + +import org.apache.activemq.ActiveMQConnection; +import org.apache.activemq.EmbeddedBrokerTestSupport; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.command.ShutdownInfo; +import org.apache.activemq.transport.Transport; +import org.apache.activemq.transport.TransportFactory; +import org.apache.activemq.transport.TransportListener; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +public class VMTransportClosureTest extends EmbeddedBrokerTestSupport { + private static final Log LOG = LogFactory + .getLog(VMTransportClosureTest.class); + private static final long MAX_TEST_TIME_MILLIS = 300000; // 5min + private static final int NUM_ATTEMPTS = 100000; + + public void setUp() throws Exception { + setAutoFail(true); + setMaxTestTime(MAX_TEST_TIME_MILLIS); + super.setUp(); + } + + /** + * EmbeddedBrokerTestSupport.createBroker() binds the broker to a VM + * transport address, which results in a call to + * VMTransportFactory.doBind(location): + *

+ * + * public TransportServer doBind(URI location) throws IOException { + * return bind(location, false); + *} + * + *

+ * As a result, VMTransportServer.disposeOnDisconnect is false. + * To expose the bug, we need to have VMTransportServer.disposeOnDisconnect + * true, which is the case when the VMTransportServer is not + * already bound when the first connection is made. + */ + @Override + protected BrokerService createBroker() throws Exception { + BrokerService answer = new BrokerService(); + answer.setPersistent(isPersistent()); + // answer.addConnector(bindAddress); + return answer; + } + + /** + * This test demonstrates how the "disposeOnDisonnect" feature of + * VMTransportServer can incorrectly close all VM connections to the local + * broker. + */ + public void testPrematureClosure() throws Exception { + + // Open a persistent connection to the local broker. The persistent + // connection is maintained through the test and should prevent the + // VMTransportServer from stopping itself when the local transport is + // closed. + ActiveMQConnection persistentConn = (ActiveMQConnection) createConnection(); + persistentConn.start(); + Session session = persistentConn.createSession(true, + Session.SESSION_TRANSACTED); + MessageProducer producer = session.createProducer(destination); + + for (int i = 0; i < NUM_ATTEMPTS; i++) { + LOG.info("Attempt: " + i); + + // Open and close a local transport connection. As is done by by + // most users of the transport, ensure that the transport is stopped + // when closed by the peer (via ShutdownInfo). Closing the local + // transport should not affect the persistent connection. + final Transport localTransport = TransportFactory.connect(broker + .getVmConnectorURI()); + localTransport.setTransportListener(new TransportListener() { + public void onCommand(Object command) { + if (command instanceof ShutdownInfo) { + try { + localTransport.stop(); + } catch (Exception ex) { + throw new RuntimeException(ex); + } + } + } + + public void onException(IOException error) { + // ignore + } + + public void transportInterupted() { + // ignore + } + + public void transportResumed() { + // ignore + } + }); + + localTransport.start(); + localTransport.stop(); + + // Ensure that the persistent connection is still usable. + producer.send(session.createMessage()); + session.rollback(); + } + + persistentConn.close(); + } +}