Polish previous code with Java 8 features

[#154263515]

References #341

Author: acogoluegnes

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -220,7 +220,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler, Metrics
         this._virtualHost = params.getVirtualHost();
         this._exceptionHandler = params.getExceptionHandler();
 
-        this._clientProperties = new HashMap<String, Object>(params.getClientProperties());
+        this._clientProperties = new HashMap<>(params.getClientProperties());
         this.requestedFrameMax = params.getRequestedFrameMax();
         this.requestedChannelMax = params.getRequestedChannelMax();
         this.requestedHeartbeat = params.getRequestedHeartbeat();
@@ -252,10 +252,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler, Metrics
         this.metricsCollector = metricsCollector;
 
         this.errorOnWriteListener = params.getErrorOnWriteListener() != null ? params.getErrorOnWriteListener() :
-            new ErrorOnWriteListener() {
-                @Override
-                public void handle(Connection connection, IOException exception) { }
-        };
+            (connection, exception) -> { };
         this.workPoolTimeout = params.getWorkPoolTimeout();
     }
 

src/main/java/com/rabbitmq/client/impl/ChannelN.java

@@ -453,10 +453,10 @@ protected void processDelivery(Command command, Basic.Deliver method) {
             // in case a manual ack in the callback, the stats will be able to record the ack
             metricsCollector.consumedMessage(this, m.getDeliveryTag(), m.getConsumerTag());
             this.dispatcher.handleDelivery(callback,
-                m.getConsumerTag(),
-                envelope,
-                (BasicProperties) command.getContentHeader(),
-                command.getContentBody());
+                                           m.getConsumerTag(),
+                                           envelope,
+                                           (BasicProperties) command.getContentHeader(),
+                                           command.getContentBody());
         } catch (WorkPoolFullException e) {
             // couldn't enqueue in work pool, propagating
             throw e;

src/main/java/com/rabbitmq/client/impl/ConsumerWorkService.java

@@ -34,8 +34,8 @@ final public class ConsumerWorkService {
     public ConsumerWorkService(ExecutorService executor, ThreadFactory threadFactory, int queueingTimeout, int shutdownTimeout) {
         this.privateExecutor = (executor == null);
         this.executor = (executor == null) ? Executors.newFixedThreadPool(DEFAULT_NUM_THREADS, threadFactory)
-            : executor;
-        this.workPool = new WorkPool<Channel, Runnable>(queueingTimeout);
+                                           : executor;
+        this.workPool = new WorkPool<>(queueingTimeout);
         this.shutdownTimeout = shutdownTimeout;
     }
 

src/main/java/com/rabbitmq/client/impl/WorkPool.java

@@ -21,9 +21,8 @@
 import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
+import java.util.function.BiConsumer;
 
 /**
  * <p>This is a generic implementation of the channels specification
@@ -64,33 +63,26 @@ public class WorkPool<K, W> {
     private final Map<K, VariableLinkedBlockingQueue<W>> pool = new HashMap<K, VariableLinkedBlockingQueue<W>>();
     /** Those keys which want limits to be removed. We do not limit queue size if this is non-empty. */
     private final Set<K> unlimited = new HashSet<K>();
-    private final EnqueueingCallback<W> enqueueingCallback;
+    private final BiConsumer<VariableLinkedBlockingQueue<W>, W> enqueueingCallback;
 
     public WorkPool(final int queueingTimeout) {
         if (queueingTimeout > 0) {
-            this.enqueueingCallback = new EnqueueingCallback<W>() {
-                @Override
-                public void enqueue(BlockingQueue<W> queue, W item) {
-                    try {
-                        boolean offered = queue.offer(item, queueingTimeout, TimeUnit.MILLISECONDS);
-                        if (!offered) {
-                            throw new WorkPoolFullException("Could not enqueue in work pool after " + queueingTimeout + " ms.");
-                        }
-                    } catch (InterruptedException e) {
-                        Thread.currentThread();
+            this.enqueueingCallback = (queue, item) -> {
+                try {
+                    boolean offered = queue.offer(item, queueingTimeout, TimeUnit.MILLISECONDS);
+                    if (!offered) {
+                        throw new WorkPoolFullException("Could not enqueue in work pool after " + queueingTimeout + " ms.");
                     }
+                } catch (InterruptedException e) {
+                    Thread.currentThread();
                 }
             };
         } else {
-            this.enqueueingCallback = new EnqueueingCallback<W>() {
-
-                @Override
-                public void enqueue(BlockingQueue<W> queue, W item) {
-                    try {
-                        queue.put(item);
-                    } catch (InterruptedException e) {
-                        Thread.currentThread().interrupt();
-                    }
+            this.enqueueingCallback = (queue, item) -> {
+                try {
+                    queue.put(item);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
                 }
             };
         }
@@ -212,7 +204,7 @@ public boolean addWorkItem(K key, W item) {
         }
         // The put operation may block. We need to make sure we are not holding the lock while that happens.
         if (queue != null) {
-            enqueueingCallback.enqueue(queue, item);
+            enqueueingCallback.accept(queue, item);
 
             synchronized (this) {
                 if (isDormant(key)) {
@@ -274,9 +266,4 @@ private K readyToInProgress() {
         return key;
     }
 
-    private interface EnqueueingCallback<W> {
-
-        void enqueue(BlockingQueue<W> queue, W item);
-
-    }
 }

src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringConnection.java

@@ -18,7 +18,6 @@
 import com.rabbitmq.client.*;
 import com.rabbitmq.client.impl.AMQConnection;
 import com.rabbitmq.client.impl.ConnectionParams;
-import com.rabbitmq.client.impl.ErrorOnWriteListener;
 import com.rabbitmq.client.impl.FrameHandlerFactory;
 import com.rabbitmq.client.impl.NetworkConnection;
 import com.rabbitmq.utility.Utility;
@@ -104,28 +103,25 @@ public AutorecoveringConnection(ConnectionParams params, FrameHandlerFactory f,
     private void setupErrorOnWriteListenerForPotentialRecovery() {
         final ThreadFactory threadFactory = this.params.getThreadFactory();
         final Lock errorOnWriteLock = new ReentrantLock();
-        this.params.setErrorOnWriteListener(new ErrorOnWriteListener() {
-            @Override
-            public void handle(final Connection connection, final IOException exception) throws IOException {
-                // this is called for any write error
-                // we should trigger the error handling and the recovery only once
-                if (errorOnWriteLock.tryLock()) {
-                    try {
-                        Thread recoveryThread = threadFactory.newThread(new Runnable() {
-                            @Override
-                            public void run() {
-                                AMQConnection c = (AMQConnection) connection;
-                                c.handleIoError(exception);
-                            }
-                        });
-                        recoveryThread.setName("RabbitMQ Error On Write Thread");
-                        recoveryThread.start();
-                    } finally {
-                        errorOnWriteLock.unlock();
-                    }
+        this.params.setErrorOnWriteListener((connection, exception) -> {
+            // this is called for any write error
+            // we should trigger the error handling and the recovery only once
+            if (errorOnWriteLock.tryLock()) {
+                try {
+                    Thread recoveryThread = threadFactory.newThread(new Runnable() {
+                        @Override
+                        public void run() {
+                            AMQConnection c = (AMQConnection) connection;
+                            c.handleIoError(exception);
+                        }
+                    });
+                    recoveryThread.setName("RabbitMQ Error On Write Thread");
+                    recoveryThread.start();
+                } finally {
+                    errorOnWriteLock.unlock();
                 }
-                throw exception;
             }
+            throw exception;
         });
     }
 

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

@@ -23,6 +23,7 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 
+import com.rabbitmq.client.Connection;
 import org.junit.Test;
 
 import com.rabbitmq.client.ConnectionFactory;
@@ -32,34 +33,50 @@ public class SharedThreadPoolTest extends BrokerTestCase {
     @Test public void willShutDownExecutor() throws IOException, TimeoutException {
         ExecutorService executor1 = null;
         ExecutorService executor2 = null;
+        AMQConnection conn1 = null;
+        AMQConnection conn2 = null;
+        AMQConnection conn3 = null;
+        AMQConnection conn4 = null;
         try {
             ConnectionFactory cf = TestUtils.connectionFactory();
             cf.setAutomaticRecoveryEnabled(false);
             executor1 = Executors.newFixedThreadPool(8);
             cf.setSharedExecutor(executor1);
 
-            AMQConnection conn1 = (AMQConnection)cf.newConnection();
+            conn1 = (AMQConnection)cf.newConnection();
             assertFalse(conn1.willShutDownConsumerExecutor());
 
             executor2 = Executors.newSingleThreadExecutor();
-            AMQConnection conn2 = (AMQConnection)cf.newConnection(executor2);
+            conn2 = (AMQConnection)cf.newConnection(executor2);
             assertFalse(conn2.willShutDownConsumerExecutor());
 
-            AMQConnection conn3 = (AMQConnection)cf.newConnection((ExecutorService)null);
+            conn3 = (AMQConnection)cf.newConnection((ExecutorService)null);
             assertTrue(conn3.willShutDownConsumerExecutor());
 
             cf.setSharedExecutor(null);
 
-            AMQConnection conn4 = (AMQConnection)cf.newConnection();
+            conn4 = (AMQConnection)cf.newConnection();
             assertTrue(conn4.willShutDownConsumerExecutor());
         } finally {
-            if (executor1 != null) {
-                executor1.shutdownNow();
-            }
-            if (executor2 != null) {
-                executor2.shutdownNow();
-            }
+            close(conn1);
+            close(conn2);
+            close(conn3);
+            close(conn4);
+            close(executor1);
+            close(executor2);
         }
 
     }
+
+    void close(ExecutorService executor) {
+        if (executor != null) {
+            executor.shutdownNow();
+        }
+    }
+
+    void close(Connection connection) throws IOException {
+        if (connection != null) {
+            connection.close();
+        }
+    }
 }