AMQ-6974 - Merge commit '5b1412ddf'

This closes #284

activemq-broker/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java

@@ -1923,7 +1923,7 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
                 if (info == null) {
                     long deadline = System.currentTimeMillis() + unit.toMillis(timeout);
 
-                    while (!disposed.get() || System.currentTimeMillis() < deadline) {
+                    while (!disposed.get() || System.currentTimeMillis() - deadline < 0) {
                         if (slot.await(1, TimeUnit.MILLISECONDS)) {
                             break;
                         }

activemq-broker/src/main/java/org/apache/activemq/transport/vm/VMTransportFactory.java

@@ -191,7 +191,7 @@ public class VMTransportFactory extends TransportFactory {
             broker = registry.lookup(brokerName);
             if (broker == null || waitForStart > 0) {
                 final long expiry = System.currentTimeMillis() + waitForStart;
-                while ((broker == null || !broker.isStarted()) && expiry > System.currentTimeMillis()) {
+                while ((broker == null || !broker.isStarted()) && System.currentTimeMillis() - expiry < 0) {
                     long timeout = Math.max(0, expiry - System.currentTimeMillis());
                     if (broker == null) {
                         try {

activemq-client/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java

@@ -1157,7 +1157,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
                         break;
                     }
                 }
-            } while (numberNotReplayed > 0 && expiry < System.currentTimeMillis());
+            } while (numberNotReplayed > 0 && expiry - System.currentTimeMillis() < 0);
         }
     }
 

activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/ConnectionPool.java

@@ -233,7 +233,7 @@ public class ConnectionPool implements ExceptionListener {
             }
         }
 
-        if (expiryTimeout > 0 && System.currentTimeMillis() > firstUsed + expiryTimeout) {
+        if (expiryTimeout > 0 && (firstUsed + expiryTimeout) - System.currentTimeMillis() < 0) {
             hasExpired = true;
             if (referenceCount == 0) {
                 close();
@@ -243,7 +243,7 @@ public class ConnectionPool implements ExceptionListener {
 
         // Only set hasExpired here is no references, as a Connection with references is by
         // definition not idle at this time.
-        if (referenceCount == 0 && idleTimeout > 0 && System.currentTimeMillis() > lastUsed + idleTimeout) {
+        if (referenceCount == 0 && idleTimeout > 0 && (lastUsed + idleTimeout) - System.currentTimeMillis() < 0) {
             hasExpired = true;
             close();
             expired = true;

activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/ConnectionPoolTest.java

@@ -0,0 +1,62 @@
+package org.apache.activemq.jms.pool;
+
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.broker.BrokerService;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+
+import static org.junit.Assert.assertFalse;
+
+public class ConnectionPoolTest extends JmsPoolTestSupport {
+    private static final Logger LOG = LoggerFactory.getLogger(ConnectionPoolTest.class);
+
+
+    private class PooledConnectionFactoryTest extends PooledConnectionFactory {
+        ConnectionPool pool = null;
+        @Override
+        protected Connection newPooledConnection(ConnectionPool connection) {
+            connection.setIdleTimeout(Integer.MAX_VALUE);
+            this.pool = connection;
+            Connection ret = super.newPooledConnection(connection);
+            ConnectionPool cp = ((PooledConnection) ret).pool;
+            cp.decrementReferenceCount();
+            // will fail if timeout does overflow
+            assertFalse(cp.expiredCheck());
+            return ret;
+        }
+
+        public ConnectionPool getPool() {
+            return pool;
+        }
+
+    }
+
+    @Override
+    @Before
+    public void setUp() throws Exception {
+        super.setUp();
+
+        brokerService = new BrokerService();
+        brokerService.setDeleteAllMessagesOnStartup(true);
+        brokerService.setPersistent(false);
+        brokerService.setUseJmx(false);
+        brokerService.setAdvisorySupport(false);
+        brokerService.setSchedulerSupport(false);
+        brokerService.start();
+        brokerService.waitUntilStarted();
+    }
+
+    @Test(timeout = 120000)
+    public void demo() throws JMSException, InterruptedException {
+        final PooledConnectionFactoryTest pooled = new PooledConnectionFactoryTest();
+        pooled.setConnectionFactory(new ActiveMQConnectionFactory("vm://localhost?create=false"));
+        pooled.setMaxConnections(2);
+        pooled.setExpiryTimeout(Long.MAX_VALUE);
+        pooled.start();
+    }
+}

activemq-tooling/activemq-perf-maven-plugin/src/main/java/org/apache/activemq/tool/JmsConsumerClient.java

@@ -84,7 +84,7 @@ public class JmsConsumerClient extends AbstractJmsMeasurableClient {
             LOG.info("Starting to synchronously receive messages for " + duration + " ms...");
             long endTime = System.currentTimeMillis() + duration;
 
-            while (System.currentTimeMillis() < endTime) {
+            while (System.currentTimeMillis() - endTime < 0) {
                 getJmsConsumer().receive();
                 incThroughput();
                 sleep();

activemq-tooling/activemq-perf-maven-plugin/src/main/java/org/apache/activemq/tool/JmsProducerClient.java

@@ -199,7 +199,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
 
                 // Send to more than one actual destination
                 if (dest.length > 1) {
-                    while (System.currentTimeMillis() < endTime) {
+                    while (System.currentTimeMillis() - endTime < 0) {
                         for (int j = 0; j < dest.length; j++) {
                             getJmsProducer().send(dest[j], getJmsTextMessage());
                             incThroughput();
@@ -209,7 +209,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
                     }
                     // Send to only one actual destination
                 } else {
-                    while (System.currentTimeMillis() < endTime) {
+                    while (System.currentTimeMillis() - endTime < 0) {
                         getJmsProducer().send(getJmsTextMessage());
                         incThroughput();
                         sleep();
@@ -224,7 +224,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
                 // Send to more than one actual destination
                 long count = 1;
                 if (dest.length > 1) {
-                    while (System.currentTimeMillis() < endTime) {
+                    while (System.currentTimeMillis() - endTime < 0) {
                         for (int j = 0; j < dest.length; j++) {
                             getJmsProducer().send(dest[j], createJmsTextMessage("Text Message [" + count++ + "]"));
                             incThroughput();
@@ -235,7 +235,7 @@ public class JmsProducerClient extends AbstractJmsMeasurableClient {
 
                     // Send to only one actual destination
                 } else {
-                    while (System.currentTimeMillis() < endTime) {
+                    while (System.currentTimeMillis() - endTime < 0) {
 
                         getJmsProducer().send(createJmsTextMessage("Text Message [" + count++ + "]"));
                         incThroughput();

activemq-unit-tests/src/test/java/org/apache/activemq/network/DemandForwardingBridgeSupportTest.java

@@ -0,0 +1,183 @@
+/**
+ * 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.network;
+
+import junit.framework.Test;
+import org.apache.activemq.broker.StubConnection;
+import org.apache.activemq.command.*;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.util.Arrays;
+
+public class DemandForwardingBridgeSupportTest extends NetworkTestSupport {
+
+    private DemandForwardingBridge bridge;
+
+    private StubConnection producerConnection;
+
+    private ProducerInfo producerInfo;
+
+    private StubConnection consumerConnection;
+
+    private SessionInfo consumerSessionInfo;
+
+
+    public void testOverflow() throws Exception {
+        NetworkBridgeConfiguration configuration = getDefaultBridgeConfiguration();
+
+        configuration.setExcludedDestinations(Arrays.asList(ActiveMQDestination.createDestination("OTHER.>",
+                ActiveMQDestination.TOPIC_TYPE)));
+        configuration.setDynamicallyIncludedDestinations(Arrays.asList(ActiveMQDestination.createDestination(
+                "TEST", ActiveMQDestination.QUEUE_TYPE)));
+
+        configureAndStartBridge(configuration);
+        assertReceiveMessageOn("TEST", ActiveMQDestination.QUEUE_TYPE);
+        assertReceiveNoMessageOn("OTHER.T1", ActiveMQDestination.TOPIC_TYPE);
+    }
+
+    private void assertReceiveMessageOn(String destinationName, byte destinationType) throws Exception,
+            InterruptedException {
+
+        ActiveMQDestination destination = ActiveMQDestination.createDestination(destinationName, destinationType);
+
+        // Send the message to the local broker.
+        producerConnection.send(createMessage(producerInfo, destination, destinationType));
+
+        // Make sure the message was delivered via the remote.
+        Message m = createConsumerAndReceiveMessage(destination);
+
+        assertNotNull(m);
+    }
+
+    private void assertReceiveNoMessageOn(String destinationName, byte destinationType) throws Exception,
+            InterruptedException {
+
+        ActiveMQDestination destination = ActiveMQDestination.createDestination(destinationName, destinationType);
+
+        // Send the message to the local broker.
+        producerConnection.send(createMessage(producerInfo, destination, destinationType));
+
+        // Make sure the message was delivered via the remote.
+        Message m = createConsumerAndReceiveMessage(destination);
+        assertNull(m);
+    }
+
+    private Message createConsumerAndReceiveMessage(ActiveMQDestination destination) throws Exception {
+        // Now create remote consumer that should cause message to move to this
+        // remote consumer.
+        ConsumerInfo consumerInfo = createConsumerInfo(consumerSessionInfo, destination);
+        consumerConnection.send(consumerInfo);
+
+        Message m = receiveMessage(consumerConnection);
+        return m;
+    }
+
+    private void configureAndStartBridge(NetworkBridgeConfiguration configuration) throws Exception {
+        bridge = new DemandForwardingBridge(configuration, createTransport(), createRemoteTransport());
+        bridge.setBrokerService(broker);
+        bridge.setDynamicallyIncludedDestinations(configuration.getDynamicallyIncludedDestinations().toArray(
+                new ActiveMQDestination[configuration.getDynamicallyIncludedDestinations().size()]
+        ));
+        bridge.setExcludedDestinations(configuration.getExcludedDestinations().toArray(
+                new ActiveMQDestination[configuration.getExcludedDestinations().size()]
+        ));
+        bridge.setStaticallyIncludedDestinations(configuration.getStaticallyIncludedDestinations().toArray(
+                new ActiveMQDestination[configuration.getStaticallyIncludedDestinations().size()]
+        ));
+        bridge.start();
+    }
+
+    public NetworkBridgeConfiguration getDefaultBridgeConfiguration() {
+        NetworkBridgeConfiguration config = new NetworkBridgeConfiguration();
+        config.setBrokerName("local");
+        config.setDispatchAsync(false);
+        return config;
+    }
+
+    // create sockets with max waiting value accepted
+    @Override
+    protected String getLocalURI() {
+        int port = findFreePort();
+        return String.format("tcp://localhost:%d?connectionTimeout=2147483647", port);
+    }
+
+    @Override
+    protected String getRemoteURI() {
+        int port = findFreePort();
+        return String.format("tcp://localhost:%d?connectionTimeout=2147483647",port);
+    }
+
+    private static int findFreePort() {
+        ServerSocket socket = null;
+        try {
+            socket = new ServerSocket(0);
+            socket.setReuseAddress(true);
+            int port = socket.getLocalPort();
+            try {
+                socket.close();
+            } catch (IOException e) {
+                // Ignore IOException on close()
+            }
+            return port;
+        } catch (IOException e) {
+        } finally {
+            if (socket != null) {
+                try {
+                    socket.close();
+                } catch (IOException e) {
+                }
+            }
+        }
+        throw new IllegalStateException("Could not find a free TCP/IP port to start embedded Jetty HTTP Server on");
+    }
+
+
+    @Override
+    protected void setUp() throws Exception {
+        super.setUp();
+
+        producerConnection = createConnection();
+        ConnectionInfo producerConnectionInfo = createConnectionInfo();
+        SessionInfo producerSessionInfo = createSessionInfo(producerConnectionInfo);
+        producerInfo = createProducerInfo(producerSessionInfo);
+        producerConnection.send(producerConnectionInfo);
+        producerConnection.send(producerSessionInfo);
+        producerConnection.send(producerInfo);
+
+        consumerConnection = createRemoteConnection();
+        ConnectionInfo consumerConnectionInfo = createConnectionInfo();
+        consumerSessionInfo = createSessionInfo(consumerConnectionInfo);
+        consumerConnection.send(consumerConnectionInfo);
+        consumerConnection.send(consumerSessionInfo);
+    }
+
+
+    @Override
+    protected void tearDown() throws Exception {
+        super.tearDown();
+    }
+
+    public static Test suite() {
+        return suite(DemandForwardingBridgeSupportTest.class);
+    }
+
+    public static void main(String[] args) {
+        junit.textui.TestRunner.run(suite());
+    }
+
+}