Merge pull request #251 from rabbitmq/rabbitmq-java-client-246

Use lambdas for *Listener objects

Author: michaelklishin

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

@@ -0,0 +1,31 @@
+// 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;
+
+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,16 @@ 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);
+
     /**
      * Remove a {@link ReturnListener}.
      * @param listener the listener to remove
@@ -126,6 +136,16 @@ public interface Channel extends ShutdownNotifier {
      */
     void addConfirmListener(ConfirmListener listener);
 
+    /**
+     * Add a lambda-based {@link ConfirmListener}.
+     * @see ConfirmListener
+     * @see ConfirmCallback
+     * @param ackCallback callback on ack
+     * @param nackCallback call on nack (negative ack)
+     * @return the listener that wraps the callbacks
+     */
+    ConfirmListener addConfirmListener(ConfirmCallback ackCallback, ConfirmCallback nackCallback);
+
     /**
      * Remove a {@link ConfirmListener}.
      * @param listener the listener to remove

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

@@ -0,0 +1,36 @@
+// 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;
+
+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,17 @@ 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);
+
     /**
      * Remove a {@link BlockedListener}.
      * @param listener the listener to remove

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

@@ -0,0 +1,62 @@
+// Copyright (c) 2007-Present 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;
+
+/**
+ *
+ */
+public class Return {
+
+    private final int replyCode;
+    private final String replyText;
+    private final String exchange;
+    private final String routingKey;
+    private final AMQP.BasicProperties properties;
+    private final byte[] body;
+
+    public Return(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) {
+        this.replyCode = replyCode;
+        this.replyText = replyText;
+        this.exchange = exchange;
+        this.routingKey = routingKey;
+        this.properties = properties;
+        this.body = body;
+    }
+
+    public int getReplyCode() {
+        return replyCode;
+    }
+
+    public String getReplyText() {
+        return replyText;
+    }
+
+    public String getExchange() {
+        return exchange;
+    }
+
+    public String getRoutingKey() {
+        return routingKey;
+    }
+
+    public AMQP.BasicProperties getProperties() {
+        return properties;
+    }
+
+    public byte[] getBody() {
+        return body;
+    }
+}

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

@@ -0,0 +1,33 @@
+// 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;
+
+/**
+ * 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 {
-    public void shutdownCompleted(ShutdownSignalException cause);
+    void shutdownCompleted(ShutdownSignalException cause);
 }

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

@@ -30,26 +30,26 @@ public interface ShutdownNotifier {
      *
      * @param listener {@link ShutdownListener} to the component
      */
-    public void addShutdownListener(ShutdownListener listener);
+    void addShutdownListener(ShutdownListener listener);
 
     /**
      * Remove shutdown listener for the component.
      *
      * @param listener {@link ShutdownListener} to be removed
      */
-    public void removeShutdownListener(ShutdownListener listener);
+    void removeShutdownListener(ShutdownListener listener);
 
     /**
      * Get the shutdown reason object
      * @return ShutdownSignalException if component is closed, null otherwise
      */
-    public ShutdownSignalException getCloseReason();
+    ShutdownSignalException getCloseReason();
 
     /**
      * Protected API - notify the listeners attached to the component
      * @see com.rabbitmq.client.ShutdownListener
      */
-    public void notifyListeners();
+    void notifyListeners();
 
     /**
      * Determine whether the component is currently open.

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

@@ -0,0 +1,31 @@
+// 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;
+
+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/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -1031,6 +1031,24 @@ public void addBlockedListener(BlockedListener listener) {
         blockedListeners.add(listener);
     }
 
+    @Override
+    public BlockedListener addBlockedListener(BlockedCallback blockedCallback, UnblockedCallback unblockedCallback) {
+        BlockedListener blockedListener = new BlockedListener() {
+
+            @Override
+            public void handleBlocked(String reason) throws IOException {
+                blockedCallback.handle(reason);
+            }
+
+            @Override
+            public void handleUnblocked() throws IOException {
+                unblockedCallback.handle();
+            }
+        };
+        this.addBlockedListener(blockedListener);
+        return blockedListener;
+    }
+
     @Override
     public boolean removeBlockedListener(BlockedListener listener) {
         return blockedListeners.remove(listener);

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

@@ -26,6 +26,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeoutException;
 
+import com.rabbitmq.client.ConfirmCallback;
 import com.rabbitmq.client.*;
 import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.Method;
@@ -135,6 +136,15 @@ public void addReturnListener(ReturnListener listener) {
         returnListeners.add(listener);
     }
 
+    @Override
+    public ReturnListener addReturnListener(ReturnCallback returnCallback) {
+        ReturnListener returnListener = (replyCode, replyText, exchange, routingKey, properties, body) -> returnCallback.handle(new Return(
+            replyCode, replyText, exchange, routingKey, properties, body
+        ));
+        this.addReturnListener(returnListener);
+        return returnListener;
+    }
+
     @Override
     public boolean removeReturnListener(ReturnListener listener) {
         return returnListeners.remove(listener);
@@ -150,6 +160,24 @@ public void addConfirmListener(ConfirmListener listener) {
         confirmListeners.add(listener);
     }
 
+    @Override
+    public ConfirmListener addConfirmListener(ConfirmCallback ackCallback, ConfirmCallback nackCallback) {
+        ConfirmListener confirmListener = new ConfirmListener() {
+
+            @Override
+            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
+                ackCallback.handle(deliveryTag, multiple);
+            }
+
+            @Override
+            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
+                nackCallback.handle(deliveryTag, multiple);
+            }
+        };
+        this.addConfirmListener(confirmListener);
+        return confirmListener;
+    }
+
     @Override
     public boolean removeConfirmListener(ConfirmListener listener) {
         return confirmListeners.remove(listener);