/*
 * 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.messaginghub.pooled.jms;

import static org.junit.Assert.fail;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.jms.JMSException;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.junit.Before;
import org.junit.Test;
import org.messaginghub.pooled.jms.JmsPoolConnection;
import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PooledConnectionExpiredConnectionsUnderLoad extends ActiveMQJmsPoolTestSupport {

    private static final Logger LOG = LoggerFactory.getLogger(PooledConnectionExpiredConnectionsUnderLoad.class);

    @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 testExpiredConnectionsAreNotReturned() throws JMSException, InterruptedException {
        final CountDownLatch latch = new CountDownLatch(1);
        final AtomicBoolean done = new AtomicBoolean(false);
        final JmsPoolConnectionFactory pooled = new JmsPoolConnectionFactory();
        pooled.setConnectionFactory(new ActiveMQConnectionFactory("vm://localhost?create=false"));

        pooled.setMaxConnections(2);
        pooled.setExpiryTimeout(1L);
        Thread[] threads = new Thread[5];
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(new Runnable() {
                @Override
                public void run() {
                    while (!done.get() && latch.getCount() > 0) {
                        try {
                            final JmsPoolConnection pooledConnection = (JmsPoolConnection) pooled.createConnection();
                            if (pooledConnection.getConnection() == null) {
                                LOG.info("Found broken connection.");
                                latch.countDown();
                            }
                            pooledConnection.close();
                        } catch (JMSException e) {
                            LOG.warn("Caught Exception", e);
                        }
                    }
                }
            });
        }

        for (Thread thread : threads) {
            thread.start();
        }

        if (latch.await(3, TimeUnit.SECONDS)) {
            fail("A thread obtained broken connection");
        }

        done.set(true);

        for (Thread thread : threads) {
            thread.join();
        }
    }
}

@@ -1052,261 +1052,18 @@
 org.
-apache.activemq.ActiveMQConnectionFactory;%0Aimport org.apache.activemq.broker.BrokerService;%0Aimport org.junit.Before;%0Aimport org.junit.Test;%0Aimport org.messaginghub.pooled.jms.JmsPoolConnection;%0Aimport org.messaginghub.pooled.jms.JmsPoolConnectionFactory
+junit.Test
 ;%0Aim
@@ -1332,468 +1332,8 @@
 );%0A%0A
-    @Override%0A    @Before%0A    public void setUp() throws Exception %7B%0A        super.setUp();%0A%0A        brokerService = new BrokerService();%0A        brokerService.setDeleteAllMessagesOnStartup(true);%0A        brokerService.setPersistent(false);%0A        brokerService.setUseJmx(false);%0A        brokerService.setAdvisorySupport(false);%0A        brokerService.setSchedulerSupport(false);%0A        brokerService.start();%0A        brokerService.waitUntilStarted();%0A    %7D%0A%0A
     
@@ -1628,19 +1628,20 @@
 d = 
-new Jms
+create
 Pool
+ed
 Conn
@@ -1657,24 +1657,25 @@
 tory();%0A
+%0A
         
 pooled.s
@@ -1670,100 +1670,18 @@
     
-pooled.setConnectionFactory(new ActiveMQConnectionFactory(%22vm://localhost?create=false%22));%0A%0A
+try %7B%0A    
     
@@ -1713,32 +1713,36 @@
 ons(2);%0A        
+    
 pooled.setExpiry
@@ -1762,16 +1762,20 @@
         
+   
+ 
 Thread%5B%5D
@@ -1800,16 +1800,20 @@
 ead%5B5%5D;%0A
+    
         
@@ -1863,24 +1863,28 @@
             
+ 
+   
 threads%5Bi%5D =
@@ -1928,16 +1928,20 @@
         
+   
+ 
 @Overrid
@@ -1942,16 +1942,20 @@
 verride%0A
+    
         
@@ -1978,24 +1978,28 @@
 oid run() %7B%0A
+    
             
@@ -2068,38 +2068,46 @@
                 
+   
+ 
 try %7B%0A
+    
                 
@@ -2232,24 +2232,28 @@
             
+    
 if (pooledCo
@@ -2316,24 +2316,28 @@
             
+  
+  
 LOG.info(%22Fo
@@ -2358,24 +2358,28 @@
 nection.%22);%0A
+    
             
@@ -2437,34 +2437,40 @@
                 
-%7D%0A
+    %7D%0A  
                 
@@ -2477,24 +2477,26 @@
             
+  
 pooledConnec
@@ -2505,24 +2505,28 @@
 on.close();%0A
+    
             
@@ -2588,24 +2588,28 @@
             
+    
 LOG.warn(%22Ca
@@ -2645,34 +2645,42 @@
                 
+ 
-%7D%0A
+   %7D%0A    
                 
@@ -2693,32 +2693,36 @@
                 
+    
 %7D%0A            %7D)
@@ -2719,16 +2719,20 @@
         
+ 
+   
 %7D);%0A    
@@ -2727,35 +2727,43 @@
     %7D);%0A        
+   
+ 
 %7D%0A%0A
+    
         for (Thr
@@ -2794,24 +2794,28 @@
             
+    
 thread.start
@@ -2818,35 +2818,43 @@
 tart();%0A        
-%7D%0A%0A
+    %7D%0A%0A    
         if (latc
@@ -2897,16 +2897,20 @@
         
+    
 fail(%22A 
@@ -2946,35 +2946,43 @@
 tion%22);%0A        
-%7D%0A%0A
+    %7D%0A%0A    
         done.set
@@ -2990,32 +2990,36 @@
 true);%0A%0A        
+    
 for (Thread thre
@@ -3042,24 +3042,28 @@
             
+    
 thread.join(
@@ -3061,24 +3061,85 @@
 ead.join();%0A
+            %7D%0A        %7D finally %7B%0A            pooled.stop();%0A
         %7D%0A  

Fix test to prevent transient VM broker being left alive.