Track connections left open in tests

Author: acogoluegnes

src/test/java/com/rabbitmq/client/test/AMQConnectionTest.java

@@ -51,14 +51,14 @@ public class AMQConnectionTest {
     private ConnectionFactory factory;
     private MyExceptionHandler exceptionHandler;
 
-    @Before public void setUp() throws Exception {
+    @Before public void setUp() {
         _mockFrameHandler = new MockFrameHandler();
         factory = TestUtils.connectionFactory();
         exceptionHandler = new MyExceptionHandler();
         factory.setExceptionHandler(exceptionHandler);
     }
 
-    @After public void tearDown() throws Exception {
+    @After public void tearDown() {
         factory = null;
         _mockFrameHandler = null;
     }

src/test/java/com/rabbitmq/client/test/BrokenFramesTest.java

@@ -42,12 +42,12 @@ public class BrokenFramesTest {
     private MyFrameHandler myFrameHandler;
     private ConnectionFactory factory;
 
-    @Before public void setUp() throws Exception {
+    @Before public void setUp() {
         myFrameHandler = new MyFrameHandler();
         factory = TestUtils.connectionFactory();
     }
 
-    @After public void tearDown() throws Exception {
+    @After public void tearDown() {
         factory = null;
         myFrameHandler = null;
     }

src/test/java/com/rabbitmq/client/test/BrokerTestCase.java

@@ -20,7 +20,6 @@
 import com.rabbitmq.client.impl.nio.NioParams;
 import com.rabbitmq.tools.Host;
 import org.junit.After;
-import org.junit.Assume;
 import org.junit.Before;
 import org.junit.Rule;
 import org.junit.rules.TestRule;
@@ -32,7 +31,6 @@
 import javax.net.ssl.SSLContext;
 import java.io.IOException;
 import java.security.NoSuchAlgorithmException;
-import java.util.Arrays;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.TimeoutException;
@@ -184,7 +182,6 @@ public void checkShutdownSignal(int expectedCode, ShutdownSignalException sse) {
         Method method = sse.getReason();
         channel = null;
         if (sse.isHardError()) {
-            connection = null;
             AMQP.Connection.Close closeMethod = (AMQP.Connection.Close) method;
             assertEquals(expectedCode, closeMethod.getReplyCode());
         } else {

src/test/java/com/rabbitmq/client/test/ChannelAsyncCompletableFutureTest.java

@@ -25,10 +25,7 @@
 
 import java.io.IOException;
 import java.util.UUID;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 
 import static org.junit.Assert.assertTrue;
 
@@ -39,13 +36,17 @@ public class ChannelAsyncCompletableFutureTest extends BrokerTestCase {
     String queue;
     String exchange;
 
-    @Before public void init() {
+    @Override
+    protected void createResources() throws IOException, TimeoutException {
+        super.createResources();
         executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
         queue = UUID.randomUUID().toString();
         exchange = UUID.randomUUID().toString();
     }
 
-    @After public void tearDown() throws IOException {
+    @Override
+    protected void releaseResources() throws IOException {
+        super.releaseResources();
         executor.shutdownNow();
         channel.queueDelete(queue);
         channel.exchangeDelete(exchange);

src/test/java/com/rabbitmq/client/test/ChannelRpcTimeoutIntegrationTest.java

@@ -37,12 +37,12 @@ public class ChannelRpcTimeoutIntegrationTest {
     ConnectionFactory factory;
 
     @Before
-    public void setUp() throws Exception {
+    public void setUp() {
         factory = TestUtils.connectionFactory();
     }
 
     @After
-    public void tearDown() throws Exception {
+    public void tearDown() {
         factory = null;
     }
 

src/test/java/com/rabbitmq/client/test/ConnectionFactoryTest.java

@@ -27,6 +27,7 @@
 import com.rabbitmq.client.impl.CredentialsProvider;
 import com.rabbitmq.client.impl.FrameHandler;
 import com.rabbitmq.client.impl.FrameHandlerFactory;
+import org.junit.AfterClass;
 import org.junit.Test;
 
 import java.io.IOException;

src/test/java/com/rabbitmq/client/test/RecoveryAwareAMQConnectionFactoryTest.java

@@ -28,7 +28,6 @@
 
 import java.io.IOException;
 import java.util.Arrays;
-import java.util.List;
 import java.util.Queue;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.TimeoutException;

src/test/java/com/rabbitmq/client/test/RpcTopologyRecordingTest.java

@@ -17,8 +17,6 @@
 
 import com.rabbitmq.client.*;
 import com.rabbitmq.client.impl.AMQImpl;
-import org.junit.After;
-import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
@@ -57,8 +55,9 @@ protected ConnectionFactory newConnectionFactory() {
         return connectionFactory;
     }
 
-    @Before
-    public void init() {
+    @Override
+    protected void createResources() throws IOException, TimeoutException {
+        super.createResources();
         queue = UUID.randomUUID().toString();
         exchange = UUID.randomUUID().toString();
         routingKey = UUID.randomUUID().toString();
@@ -67,8 +66,9 @@ public void init() {
         routingKey2 = "e2e-" + UUID.randomUUID().toString();
     }
 
-    @After
-    public void tearDown() throws IOException {
+    @Override
+    protected void releaseResources() throws IOException {
+        super.releaseResources();
         channel.exchangeDelete(exchange);
         channel.exchangeDelete(exchange2);
     }

src/test/java/com/rabbitmq/client/test/TestUtils.java

@@ -67,6 +67,12 @@ public static void close(Connection connection) {
         }
     }
 
+    public static void abort(Connection connection) {
+        if (connection != null) {
+            connection.abort();
+        }
+    }
+
     public static SSLContext getSSLContext() throws NoSuchAlgorithmException {
         SSLContext c = null;
 

src/test/java/com/rabbitmq/client/test/functional/ConnectionOpen.java

@@ -43,13 +43,12 @@
  */
 public class ConnectionOpen {
     @Test public void correctProtocolHeader() throws IOException {
-        ConnectionFactory factory = TestUtils.connectionFactory();
         SocketFrameHandler fh = new SocketFrameHandler(SocketFactory.getDefault().createSocket("localhost", AMQP.PROTOCOL.PORT));
         fh.sendHeader();
         AMQCommand command = new AMQCommand();
         while (!command.handleFrame(fh.readFrame())) { }
         Method m = command.getMethod();
-        //    System.out.println(m.getClass());
+
         assertTrue("First command must be Connection.start",
                 m instanceof AMQP.Connection.Start);
         AMQP.Connection.Start start = (AMQP.Connection.Start) m;

src/test/java/com/rabbitmq/client/test/functional/Heartbeat.java

@@ -16,6 +16,7 @@
 
 package com.rabbitmq.client.test.functional;
 
+import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
 import com.rabbitmq.client.test.BrokerTestCase;
 import org.junit.Test;
@@ -26,19 +27,19 @@
 
 public class Heartbeat extends BrokerTestCase {
 
-    public Heartbeat()
-    {
-        super();
-        connectionFactory.setRequestedHeartbeat(1);
+    @Override
+    protected ConnectionFactory newConnectionFactory() {
+        ConnectionFactory cf = super.newConnectionFactory();
+        cf.setRequestedHeartbeat(1);
+        return cf;
     }
 
-    @Test public void heartbeat()
-        throws IOException, InterruptedException
-    {
+    @Test
+    public void heartbeat() throws InterruptedException {
         assertEquals(1, connection.getHeartbeat());
         Thread.sleep(3100);
         assertTrue(connection.isOpen());
-        ((AutorecoveringConnection)connection).getDelegate().setHeartbeat(0);
+        ((AutorecoveringConnection) connection).getDelegate().setHeartbeat(0);
         assertEquals(0, connection.getHeartbeat());
         Thread.sleep(3100);
         assertFalse(connection.isOpen());

src/test/java/com/rabbitmq/client/test/functional/UnexpectedFrames.java

@@ -17,7 +17,6 @@
 
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.client.DefaultSocketConfigurator;
 import com.rabbitmq.client.SocketConfigurators;
 import com.rabbitmq.client.impl.*;
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
@@ -35,7 +34,7 @@
 public class UnexpectedFrames extends BrokerTestCase {
 
     private interface Confuser {
-        public Frame confuse(Frame frame) throws IOException;
+        Frame confuse(Frame frame) throws IOException;
     }
 
     private static class ConfusedFrameHandler extends SocketFrameHandler {

src/test/java/com/rabbitmq/client/test/functional/UserIDHeader.java

@@ -20,6 +20,8 @@
 import java.io.IOException;
 import java.util.concurrent.TimeoutException;
 
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
 import org.junit.Test;
 
 import com.rabbitmq.client.AMQP;
@@ -48,17 +50,18 @@ public class UserIDHeader extends BrokerTestCase {
 
     @Test public void impersonatedUserId() throws IOException, TimeoutException {
         Host.rabbitmqctl("set_user_tags guest administrator impersonator");
-        connection = null;
-        channel = null;
-        setUp();
-        try {
-            publish(BAD);
+        try (Connection c = connectionFactory.newConnection()){
+            publish(BAD, c.createChannel());
         } finally {
             Host.rabbitmqctl("set_user_tags guest administrator");
         }
     }
 
     private void publish(AMQP.BasicProperties properties) throws IOException {
+        publish(properties, this.channel);
+    }
+
+    private void publish(AMQP.BasicProperties properties, Channel channel) throws IOException {
         channel.basicPublish("amq.fanout", "", properties, "".getBytes());
         channel.queueDeclare(); // To flush the channel
     }

src/test/java/com/rabbitmq/client/test/server/BlockedConnection.java

@@ -50,7 +50,12 @@ protected void releaseResources() throws IOException {
         block();
         publish(connection);
 
-        assertTrue(latch.await(10, TimeUnit.SECONDS));
+        try {
+            assertTrue(latch.await(10, TimeUnit.SECONDS));
+        } finally {
+            TestUtils.abort(connection);
+        }
+
     }
 
     // this test first triggers an alarm, then opens a
@@ -62,7 +67,11 @@ protected void releaseResources() throws IOException {
         Connection connection = connection(latch);
         publish(connection);
 
-        assertTrue(latch.await(10, TimeUnit.SECONDS));
+        try {
+            assertTrue(latch.await(10, TimeUnit.SECONDS));
+        } finally {
+            TestUtils.abort(connection);
+        }
     }
 
     private Connection connection(final CountDownLatch latch) throws IOException, TimeoutException {

src/test/java/com/rabbitmq/client/test/server/ChannelLimitNegotiation.java

@@ -42,6 +42,7 @@
 import com.rabbitmq.tools.Host;
 
 public class ChannelLimitNegotiation extends BrokerTestCase {
+
     class SpecialConnection extends AMQConnection {
         private final int channelMax;
 
@@ -68,8 +69,9 @@ protected int negotiateChannelMax(int requestedChannelMax, int serverMax) {
         ConnectionFactory cf = TestUtils.connectionFactory();
         cf.setRequestedChannelMax(n);
 
-        Connection conn = cf.newConnection();
-        assertEquals(n, conn.getChannelMax());
+        try (Connection conn = cf.newConnection()) {
+            assertEquals(n, conn.getChannelMax());
+        }
     }
 
     @Test public void channelMaxGreaterThanServerValue() throws Exception {
@@ -91,10 +93,11 @@ protected int negotiateChannelMax(int requestedChannelMax, int serverMax) {
     @Test public void openingTooManyChannels() throws Exception {
         int n = 48;
 
+        Connection conn = null;
         try {
             Host.rabbitmqctl("eval 'application:set_env(rabbit, channel_max, " + n + ").'");
             ConnectionFactory cf = TestUtils.connectionFactory();
-            Connection conn = cf.newConnection();
+            conn = cf.newConnection();
             assertEquals(n, conn.getChannelMax());
 
             for (int i = 1; i <= n; i++) {
@@ -118,6 +121,7 @@ public void shutdownCompleted(ShutdownSignalException cause) {
         } catch (IOException e) {
             checkShutdownSignal(530, e);
         } finally {
+            TestUtils.abort(conn);
             Host.rabbitmqctl("eval 'application:set_env(rabbit, channel_max, 0).'");
         }
     }