Finalize lambdas for *Listener objects

Add functional interfaces for confirm, blocked, and return
listeners.

Fixes #246

Author: acogoluegnes

src/main/java/com/rabbitmq/client/BlockedCallback.java

@@ -18,8 +18,12 @@
 import java.io.IOException;
 
 /**
- *
+ * Implement this interface in order to be notified of connection block events.
+ * Prefer it over {@link BlockedListener} for a lambda-oriented syntax.
+ * @see BlockedListener
+ * @see UnblockedCallback
  */
+@FunctionalInterface
 public interface BlockedCallback {
 
     void handle(String reason) throws IOException;

src/main/java/com/rabbitmq/client/BlockedListener.java

@@ -21,6 +21,8 @@
 /**
  * Implement this interface in order to be notified of connection block and
  * unblock events.
+ * For a lambda-oriented syntax, use {@link BlockedCallback} and
+ * {@link UnblockedCallback}.
  */
 public interface BlockedListener {
     void handleBlocked(String reason) throws IOException;

src/main/java/com/rabbitmq/client/Channel.java

@@ -107,6 +107,14 @@ public interface Channel extends ShutdownNotifier {
      */
     void addReturnListener(ReturnListener listener);
 
+    /**
+     * Add a lambda-based {@link ReturnListener}.
+     * @see ReturnListener
+     * @see ReturnCallback
+     * @see Return
+     * @param returnCallback the callback when the message is returned
+     * @return the listener that wraps the callback
+     */
     ReturnListener addReturnListener(ReturnCallback returnCallback);
 
     /**
@@ -129,10 +137,12 @@ public interface Channel extends ShutdownNotifier {
     void addConfirmListener(ConfirmListener listener);
 
     /**
-     *
-     * @param ackCallback
-     * @param nackCallback
-     * @return
+     * Add a lambda-based {@link ConfirmListener}.
+     * @see ConfirmListener
+     * @see ConfirmCallback
+     * @param ackCallback callback on ack
+     * @param nackCallback call on ack
+     * @return the listener that wraps the callbacks
      */
     ConfirmListener addConfirmListener(ConfirmCallback ackCallback, ConfirmCallback nackCallback);
 

src/main/java/com/rabbitmq/client/ConfirmCallback.java

@@ -19,8 +19,16 @@
 import java.io.IOException;
 
 /**
- *
+ * Implement this interface in order to be notified of Confirm events.
+ * Acks represent messages handled successfully; Nacks represent
+ * messages lost by the broker.  Note, the lost messages could still
+ * have been delivered to consumers, but the broker cannot guarantee
+ * this.
+ * Prefer this interface over {@link ConfirmListener} for
+ * a lambda-oriented syntax.
+ * @see ConfirmListener
  */
+@FunctionalInterface
 public interface ConfirmCallback {
 
     void handle(long deliveryTag, boolean multiple) throws IOException;

src/main/java/com/rabbitmq/client/ConfirmListener.java

@@ -24,6 +24,7 @@
  * messages lost by the broker.  Note, the lost messages could still
  * have been delivered to consumers, but the broker cannot guarantee
  * this.
+ * For a lambda-oriented syntax, use {@link ConfirmCallback}.
  */
 public interface ConfirmListener {
     void handleAck(long deliveryTag, boolean multiple)

src/main/java/com/rabbitmq/client/Connection.java

@@ -239,6 +239,15 @@ public interface Connection extends ShutdownNotifier, Closeable { // rename to A
      */
     void addBlockedListener(BlockedListener listener);
 
+    /**
+     * Add a lambda-based {@link BlockedListener}.
+     * @see BlockedListener
+     * @see BlockedCallback
+     * @see UnblockedCallback
+     * @param blockedCallback the callback when the connection is blocked
+     * @param unblockedCallback the callback when the connection is unblocked
+     * @return the listener that wraps the callback
+     */
     BlockedListener addBlockedListener(BlockedCallback blockedCallback, UnblockedCallback unblockedCallback);
 
     /**

src/main/java/com/rabbitmq/client/ReturnCallback.java

@@ -16,8 +16,16 @@
 package com.rabbitmq.client;
 
 /**
- *
+ * Implement this interface in order to be notified of failed
+ * deliveries when basicPublish is called with "mandatory" or
+ * "immediate" flags set.
+ * Prefer this interface over {@link ReturnListener} for
+ * a simpler, lambda-oriented syntax.
+ * @see Channel#basicPublish
+ * @see ReturnListener
+ * @see Return
  */
+@FunctionalInterface
 public interface ReturnCallback {
 
     void handle(Return returnMessage);

src/main/java/com/rabbitmq/client/ReturnListener.java

@@ -22,6 +22,7 @@
  * Implement this interface in order to be notified of failed
  * deliveries when basicPublish is called with "mandatory" or
  * "immediate" flags set.
+ * For a lambda-oriented syntax, use {@link ReturnCallback}.
  * @see Channel#basicPublish
  */
 public interface ReturnListener {

src/main/java/com/rabbitmq/client/ShutdownListener.java

@@ -29,6 +29,7 @@
  * @see ShutdownNotifier
  * @see ShutdownSignalException
  */
+@FunctionalInterface
 public interface ShutdownListener extends EventListener {
     void shutdownCompleted(ShutdownSignalException cause);
 }

src/main/java/com/rabbitmq/client/UnblockedCallback.java

@@ -18,8 +18,12 @@
 import java.io.IOException;
 
 /**
- *
+ * Implement this interface in order to be notified of connection unblock events.
+ * Prefer it over {@link BlockedListener} for a lambda-oriented syntax.
+ * @see BlockedListener
+ * @see BlockedCallback
  */
+@FunctionalInterface
 public interface UnblockedCallback {
 
     void handle() throws IOException;

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

@@ -46,7 +46,8 @@
     StandardMetricsCollectorTest.class,
     DnsSrvRecordAddressResolverTest.class,
     JavaNioTest.class,
-    RpcTest.class
+    RpcTest.class,
+    LambdaListenerCallbackTest.class
 })
 public class ClientTests {
 

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

@@ -0,0 +1,86 @@
+// Copyright (c) 2017 Pivotal Software, Inc.  All rights reserved.
+//
+// This software, the RabbitMQ Java client library, is triple-licensed under the
+// Mozilla Public License 1.1 ("MPL"), the GNU General Public License version 2
+// ("GPL") and the Apache License version 2 ("ASL"). For the MPL, please see
+// LICENSE-MPL-RabbitMQ. For the GPL, please see LICENSE-GPL2.  For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// [email protected].
+
+package com.rabbitmq.client.test;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertTrue;
+
+public class LambdaListenerCallbackTest extends BrokerTestCase {
+
+    protected void releaseResources() throws IOException {
+        try {
+            unblock();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    @Test public void shutdownListener() throws Exception {
+        CountDownLatch latch = new CountDownLatch(2);
+        try(Connection connection = TestUtils.connectionFactory().newConnection()) {
+            connection.addShutdownListener(cause -> latch.countDown());
+            Channel channel = connection.createChannel();
+            channel.addShutdownListener(cause -> latch.countDown());
+        }
+        assertTrue("Connection closed, shutdown listeners should have been called", latch.await(1, TimeUnit.SECONDS));
+    }
+
+    @Test public void confirmListener() throws Exception {
+        channel.confirmSelect();
+        CountDownLatch latch = new CountDownLatch(1);
+        channel.addConfirmListener(
+            (deliveryTag, multiple) -> latch.countDown(),
+            (deliveryTag, multiple) -> {}
+        );
+        channel.basicPublish("", "whatever", null, "dummy".getBytes());
+        assertTrue("Should have received publisher confirm", latch.await(1, TimeUnit.SECONDS));
+    }
+
+    @Test public void returnListener() throws Exception {
+        CountDownLatch latch = new CountDownLatch(1);
+        channel.addReturnListener(returnMessage -> latch.countDown());
+        channel.basicPublish("", "notlikelytoexist", true, null, "dummy".getBytes());
+        assertTrue("Should have received returned message", latch.await(1, TimeUnit.SECONDS));
+    }
+
+    @Test public void blockedListener() throws Exception {
+        final CountDownLatch latch = new CountDownLatch(1);
+        try(Connection connection = TestUtils.connectionFactory().newConnection()) {
+            connection.addBlockedListener(
+                reason -> {
+                    try {
+                        unblock();
+                    } catch (InterruptedException e) {
+                        e.printStackTrace();
+                    }
+            },
+                () -> latch.countDown()
+            );
+            block();
+            Channel ch = connection.createChannel();
+            ch.basicPublish("", "", null, "dummy".getBytes());
+            assertTrue("Should have been blocked and unblocked", latch.await(10, TimeUnit.SECONDS));
+        }
+    }
+
+}