Add deliver/cancel/shutdown lambda consume methods

References #247

Author: acogoluegnes

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

@@ -26,6 +26,7 @@
  * @see ConsumerShutdownSignalCallback
  * @see Channel#basicConsume(String, boolean, String, boolean, boolean, Map, DeliverCallback, CancelCallback)
  * @see Channel#basicConsume(String, boolean, String, boolean, boolean, Map, DeliverCallback, ConsumerShutdownSignalCallback)
+ * @see Channel#basicConsume(String, boolean, String, boolean, boolean, Map, DeliverCallback, CancelCallback, ConsumerShutdownSignalCallback)
  * @since 5.0
  */
 @FunctionalInterface

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

@@ -836,6 +836,27 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      */
     String basicConsume(String queue, DeliverCallback deliverCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;
 
+    /**
+     * Start a non-nolocal, non-exclusive consumer, with
+     * explicit acknowledgement and a server-generated consumerTag.
+     * Provide access to <code>basic.deliver</code>, <code>basic.cancel</code>
+     * and shutdown signal callbacks (which is sufficient
+     * for most cases). See methods with a {@link Consumer} argument
+     * to have access to all the application callbacks.
+     * @param queue the name of the queue
+     * @param deliverCallback callback when a message is delivered
+     * @param cancelCallback callback when the consumer is cancelled
+     * @param shutdownSignalCallback callback when the channel/connection is shut down
+     * @return the consumerTag generated by the server
+     * @throws IOException if an error is encountered
+     * @see com.rabbitmq.client.AMQP.Basic.Consume
+     * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
+     * @see #basicAck
+     * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
+     * @since 5.0
+     */
+    String basicConsume(String queue, DeliverCallback deliverCallback, CancelCallback cancelCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;
+
     /**
      * Start a non-nolocal, non-exclusive consumer, with
      * a server-generated consumerTag.
@@ -898,6 +919,30 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      */
     String basicConsume(String queue, boolean autoAck, DeliverCallback deliverCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;
 
+    /**
+     * Start a non-nolocal, non-exclusive consumer, with
+     * a server-generated consumerTag.
+     * Provide access to <code>basic.deliver</code>, <code>basic.cancel</code>
+     * and shutdown signal callbacks (which is sufficient
+     * for most cases). See methods with a {@link Consumer} argument
+     * to have access to all the application callbacks.
+     * @param queue the name of the queue
+     * @param autoAck true if the server should consider messages
+     * acknowledged once delivered; false if the server should expect
+     * explicit acknowledgements
+     * @param deliverCallback callback when a message is delivered
+     * @param cancelCallback callback when the consumer is cancelled
+     * @param shutdownSignalCallback callback when the channel/connection is shut down
+     * @return the consumerTag generated by the server
+     * @throws IOException if an error is encountered
+     * @see com.rabbitmq.client.AMQP.Basic.Consume
+     * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
+     * @see #basicAck
+     * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
+     * @since 5.0
+     */
+    String basicConsume(String queue, boolean autoAck, DeliverCallback deliverCallback, CancelCallback cancelCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;
+
     /**
      * Start a non-nolocal, non-exclusive consumer, with
      * a server-generated consumerTag and specified arguments.
@@ -963,6 +1008,31 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      */
     String basicConsume(String queue, boolean autoAck, Map<String, Object> arguments, DeliverCallback deliverCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;
 
+    /**
+     * Start a non-nolocal, non-exclusive consumer, with
+     * a server-generated consumerTag and specified arguments.
+     * Provide access to <code>basic.deliver</code>, <code>basic.cancel</code>
+     * and shutdown signal callbacks (which is sufficient
+     * for most cases). See methods with a {@link Consumer} argument
+     * to have access to all the application callbacks.
+     * @param queue the name of the queue
+     * @param autoAck true if the server should consider messages
+     * acknowledged once delivered; false if the server should expect
+     * explicit acknowledgements
+     * @param arguments a set of arguments for the consume
+     * @param deliverCallback callback when a message is delivered
+     * @param cancelCallback callback when the consumer is cancelled
+     * @param shutdownSignalCallback callback when the channel/connection is shut down
+     * @return the consumerTag generated by the server
+     * @throws IOException if an error is encountered
+     * @see com.rabbitmq.client.AMQP.Basic.Consume
+     * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
+     * @see #basicAck
+     * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
+     * @since 5.0
+     */
+    String basicConsume(String queue, boolean autoAck, Map<String, Object> arguments, DeliverCallback deliverCallback, CancelCallback cancelCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;
+
     /**
      * Start a non-nolocal, non-exclusive consumer.
      * @param queue the name of the queue
@@ -1023,6 +1093,29 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      */
     String basicConsume(String queue, boolean autoAck, String consumerTag, DeliverCallback deliverCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;
 
+    /**
+     * Start a non-nolocal, non-exclusive consumer.
+     * Provide access to <code>basic.deliver</code>, <code>basic.cancel</code>
+     * and shutdown signal callbacks (which is sufficient
+     * for most cases). See methods with a {@link Consumer} argument
+     * to have access to all the application callbacks.
+     * @param queue the name of the queue
+     * @param autoAck true if the server should consider messages
+     * acknowledged once delivered; false if the server should expect
+     * explicit acknowledgements
+     * @param consumerTag a client-generated consumer tag to establish context
+     * @param deliverCallback callback when a message is delivered
+     * @param cancelCallback callback when the consumer is cancelled
+     * @param shutdownSignalCallback callback when the channel/connection is shut down
+     * @return the consumerTag associated with the new consumer
+     * @throws java.io.IOException if an error is encountered
+     * @see com.rabbitmq.client.AMQP.Basic.Consume
+     * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
+     * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
+     * @since 5.0
+     */
+    String basicConsume(String queue, boolean autoAck, String consumerTag, DeliverCallback deliverCallback, CancelCallback cancelCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;
+
     /**
      * Start a consumer. Calls the consumer's {@link Consumer#handleConsumeOk}
      * method.
@@ -1095,6 +1188,33 @@ void basicNack(long deliveryTag, boolean multiple, boolean requeue)
      */
     String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments, DeliverCallback deliverCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;
 
+    /**
+     * Start a consumer. Calls the consumer's {@link Consumer#handleConsumeOk}
+     * method.
+     * Provide access to <code>basic.deliver</code>, <code>basic.cancel</code>
+     * and shutdown signal callbacks (which is sufficient
+     * for most cases). See methods with a {@link Consumer} argument
+     * to have access to all the application callbacks.
+     * @param queue the name of the queue
+     * @param autoAck true if the server should consider messages
+     * acknowledged once delivered; false if the server should expect
+     * explicit acknowledgements
+     * @param consumerTag a client-generated consumer tag to establish context
+     * @param noLocal true if the server should not deliver to this consumer
+     * messages published on this channel's connection
+     * @param exclusive true if this is an exclusive consumer
+     * @param arguments a set of arguments for the consume
+     * @param deliverCallback callback when a message is delivered
+     * @param cancelCallback callback when the consumer is cancelled
+     * @param shutdownSignalCallback callback when the channel/connection is shut down
+     * @return the consumerTag associated with the new consumer
+     * @throws java.io.IOException if an error is encountered
+     * @see com.rabbitmq.client.AMQP.Basic.Consume
+     * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
+     * @since 5.0
+     */
+    String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments, DeliverCallback deliverCallback, CancelCallback cancelCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException;
+
     /**
      * Cancel a consumer. Calls the consumer's {@link Consumer#handleCancelOk}
      * method.

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

@@ -26,6 +26,7 @@
  * @see DeliverCallback
  * @see Channel#basicConsume(String, boolean, String, boolean, boolean, Map, DeliverCallback, CancelCallback)
  * @see Channel#basicConsume(String, boolean, String, boolean, boolean, Map, DeliverCallback, ConsumerShutdownSignalCallback)
+ * @see Channel#basicConsume(String, boolean, String, boolean, boolean, Map, DeliverCallback, CancelCallback, ConsumerShutdownSignalCallback)
  * @since 5.0
  */
 @FunctionalInterface

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

@@ -26,6 +26,7 @@
  * @see ConsumerShutdownSignalCallback
  * @see Channel#basicConsume(String, boolean, String, boolean, boolean, Map, DeliverCallback, CancelCallback)
  * @see Channel#basicConsume(String, boolean, String, boolean, boolean, Map, DeliverCallback, ConsumerShutdownSignalCallback)
+ * @see Channel#basicConsume(String, boolean, String, boolean, boolean, Map, DeliverCallback, CancelCallback, ConsumerShutdownSignalCallback)
  * @since 5.0
  */
 @FunctionalInterface

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

@@ -1201,6 +1201,13 @@ public String basicConsume(String queue, DeliverCallback deliverCallback, Consum
         return basicConsume(queue, consumerFromDeliverShutdownCallbacks(deliverCallback, shutdownSignalCallback));
     }
 
+    /** Public API - {@inheritDoc} */
+    @Override
+    public String basicConsume(String queue, DeliverCallback deliverCallback, CancelCallback cancelCallback,
+        ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException {
+        return basicConsume(queue, consumerFromDeliverCancelShutdownCallbacks(deliverCallback, cancelCallback, shutdownSignalCallback));
+    }
+
     /** Public API - {@inheritDoc} */
     @Override
     public String basicConsume(String queue, boolean autoAck, Consumer callback)
@@ -1222,6 +1229,12 @@ public String basicConsume(String queue, boolean autoAck, DeliverCallback delive
         return basicConsume(queue, autoAck, "", consumerFromDeliverCancelCallbacks(deliverCallback, cancelCallback));
     }
 
+    @Override
+    public String basicConsume(String queue, boolean autoAck, DeliverCallback deliverCallback, CancelCallback cancelCallback,
+        ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException {
+        return basicConsume(queue, autoAck, "", consumerFromDeliverCancelShutdownCallbacks(deliverCallback, cancelCallback, shutdownSignalCallback));
+    }
+
     /** Public API - {@inheritDoc} */
     @Override
     public String basicConsume(String queue, boolean autoAck, Map<String, Object> arguments,
@@ -1245,6 +1258,13 @@ public String basicConsume(String queue, boolean autoAck, Map<String, Object> ar
         return basicConsume(queue, autoAck, "", false, false, arguments, consumerFromDeliverShutdownCallbacks(deliverCallback, shutdownSignalCallback));
     }
 
+    /** Public API - {@inheritDoc} */
+    @Override
+    public String basicConsume(String queue, boolean autoAck, Map<String, Object> arguments, DeliverCallback deliverCallback, CancelCallback cancelCallback,
+        ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException {
+        return basicConsume(queue, autoAck, "", false, false, arguments, consumerFromDeliverCancelShutdownCallbacks(deliverCallback, cancelCallback, shutdownSignalCallback));
+    }
+
     /** Public API - {@inheritDoc} */
     @Override
     public String basicConsume(String queue, boolean autoAck, String consumerTag,
@@ -1265,7 +1285,14 @@ public String basicConsume(String queue, boolean autoAck, String consumerTag, De
     @Override
     public String basicConsume(String queue, boolean autoAck, String consumerTag, DeliverCallback deliverCallback,
         ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException {
-        return basicConsume(queue, autoAck, consumerTag, consumerFromDeliverShutdownCallbacks(deliverCallback, shutdownSignalCallback));
+        return basicConsume(queue, autoAck, consumerTag, false, false, null, consumerFromDeliverShutdownCallbacks(deliverCallback, shutdownSignalCallback));
+    }
+
+    /** Public API - {@inheritDoc} */
+    @Override
+    public String basicConsume(String queue, boolean autoAck, String consumerTag, DeliverCallback deliverCallback, CancelCallback cancelCallback,
+        ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException {
+        return basicConsume(queue, autoAck, consumerTag , false, false, null, consumerFromDeliverCancelShutdownCallbacks(deliverCallback, cancelCallback, shutdownSignalCallback));
     }
 
     /** Public API - {@inheritDoc} */
@@ -1282,6 +1309,12 @@ public String basicConsume(String queue, boolean autoAck, String consumerTag, bo
         return basicConsume(queue, autoAck, consumerTag, noLocal, exclusive, arguments, consumerFromDeliverShutdownCallbacks(deliverCallback, shutdownSignalCallback));
     }
 
+    @Override
+    public String basicConsume(String queue, boolean autoAck, String consumerTag, boolean noLocal, boolean exclusive, Map<String, Object> arguments,
+        DeliverCallback deliverCallback, CancelCallback cancelCallback, ConsumerShutdownSignalCallback shutdownSignalCallback) throws IOException {
+        return basicConsume(queue, autoAck, consumerTag, noLocal, exclusive, arguments, consumerFromDeliverCancelShutdownCallbacks(deliverCallback, cancelCallback, shutdownSignalCallback));
+    }
+
     /** Public API - {@inheritDoc} */
     @Override
     public String basicConsume(String queue, final boolean autoAck, String consumerTag,
@@ -1373,6 +1406,34 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
         };
     }
 
+    private Consumer consumerFromDeliverCancelShutdownCallbacks(final DeliverCallback deliverCallback, final CancelCallback cancelCallback, final ConsumerShutdownSignalCallback shutdownSignalCallback) {
+        return new Consumer() {
+            @Override
+            public void handleConsumeOk(String consumerTag) { }
+
+            @Override
+            public void handleCancelOk(String consumerTag) { }
+
+            @Override
+            public void handleCancel(String consumerTag) throws IOException {
+                cancelCallback.handle(consumerTag);
+            }
+
+            @Override
+            public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
+                shutdownSignalCallback.handleShutdownSignal(consumerTag, sig);
+            }
+
+            @Override
+            public void handleRecoverOk(String consumerTag) { }
+
+            @Override
+            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {
+                deliverCallback.handle(consumerTag, new Delivery(envelope, properties, body));
+            }
+        };
+    }
+
     /** Public API - {@inheritDoc} */
     @Override
     public void basicCancel(final String consumerTag)