Polish queue specification

Closes #44

Author: acogoluegnes

build.gradle

@@ -44,6 +44,7 @@ ext {
 	mockitoVersion = '2.23.4'
 	junitPlatformVersion = '1.3.2'
 	junitJupiterVersion  = '5.3.2'
+	assertjVersion = '3.11.1'
 
 	javadocLinks = ["http://docs.oracle.com/javase/7/docs/api/",
 					"http://docs.oracle.com/javaee/6/api/",
@@ -127,6 +128,7 @@ configure(allprojects) { project ->
 
 		testCompile "io.projectreactor:reactor-test:$reactorCoreVersion"
 		testCompile "org.mockito:mockito-core:$mockitoVersion"
+		testCompile "org.assertj:assertj-core:${assertjVersion}"
 		testRuntime "org.slf4j:jcl-over-slf4j:$slf4jVersion"
 		testRuntime "org.slf4j:slf4j-api:$slf4jVersion"
 		testRuntime "ch.qos.logback:logback-classic:$logbackVersion"

src/docs/asciidoc/api-guide.adoc

@@ -112,6 +112,18 @@ include::{test-examples}/ApiGuideSender.java[tag=resource-declaration]
 Note the `Sender#declare*` methods return their respective AMQP results
 wrapped into a `Mono`.
 
+[NOTE]
+====
+For queue creation, note that if a queue specification has a
+null name, the queue to be created will have a server-generated name
+and will be non-durable, exclusive, and auto-delete. If you want
+a queue to have a server-generated name but other parameters,
+specify an empty name `""` and set the parameters accordingly on
+the `QueueSpecification` instance. For more information about queues,
+see the https://www.rabbitmq.com/queues.html[official documentation].
+====
+
+
 One can also use the `ResourcesSpecification` factory class
 with a static import to reduce boilerplate code. Combined with
 `Mono` chaining and `Sender#declare` shortcuts, it allows for condensed syntax:

src/docs/asciidoc/new.adoc

@@ -19,3 +19,5 @@
 * Handle error signal of `connectionMono` subscription to enable proper error handling
 * Use Reactor 3.2.3.RELEASE
 * Use Java client 5.5.1
+
+

src/main/java/reactor/rabbitmq/QueueSpecification.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017 Pivotal Software Inc, All Rights Reserved.
+ * Copyright (c) 2017-2019 Pivotal Software Inc, All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,59 +20,19 @@
 
 /**
  * Fluent API to specify the creation of a queue.
+ * <p>
+ * If a queue specification has a null name, the queue to be created
+ * will have a server-generated name and will be non-durable, exclusive, and
+ * auto-delete. To have more control over the properties of a queue with
+ * a server-generated name, specify a non-null, empty string name, <code>""</code>.
  */
 public class QueueSpecification {
 
-    static class NullNameQueueSpecification extends QueueSpecification {
-
-        NullNameQueueSpecification() {
-            this.name = null;
-            this.durable = false;
-            this.exclusive = true;
-            this.autoDelete = true;
-        }
-
-        @Override
-        public QueueSpecification name(String name) {
-            if (name == null) {
-                return this;
-            }
-            return QueueSpecification.queue(name)
-                .durable(durable)
-                .exclusive(exclusive)
-                .autoDelete(autoDelete);
-        }
-
-        @Override
-        public QueueSpecification durable(boolean durable) {
-            if (this.durable != durable) {
-                throw new IllegalArgumentException("once a queue has null name, durable is always false");
-            }
-            return this;
-        }
-
-        @Override
-        public QueueSpecification exclusive(boolean exclusive) {
-            if (this.exclusive != exclusive) {
-                throw new IllegalArgumentException("once a queue has null name, exclusive is always true");
-            }
-            return this;
-        }
-
-        @Override
-        public QueueSpecification autoDelete(boolean autoDelete) {
-            if (this.autoDelete != autoDelete) {
-                throw new IllegalArgumentException("once a queue has null name, autoDelete is always true");
-            }
-            return this;
-        }
-    }
-
-    String name;
-    boolean durable = false;
-    boolean exclusive = false;
-    boolean autoDelete = false;
-    Map<String, Object> arguments;
+    protected String name;
+    protected boolean durable = false;
+    protected boolean exclusive = false;
+    protected boolean autoDelete = false;
+    protected Map<String, Object> arguments;
 
     public static QueueSpecification queue() {
         return new NullNameQueueSpecification();
@@ -84,7 +44,7 @@ public static QueueSpecification queue(String name) {
 
     public QueueSpecification name(String queue) {
         if (queue == null) {
-            return new NullNameQueueSpecification();
+            return new NullNameQueueSpecification().arguments(this.arguments);
         }
 
         this.name = queue;
@@ -130,4 +90,54 @@ public boolean isAutoDelete() {
     public Map<String, Object> getArguments() {
         return arguments;
     }
+
+    /**
+     * Internal class to handle queues with a null name.
+     * Those queues always have a server-generated name are non-durable,
+     * exclusive, and auto-delete.
+     */
+    private static class NullNameQueueSpecification extends QueueSpecification {
+
+        NullNameQueueSpecification() {
+            this.name = null;
+            this.durable = false;
+            this.exclusive = true;
+            this.autoDelete = true;
+        }
+
+        @Override
+        public QueueSpecification name(String name) {
+            if (name == null) {
+                return this;
+            }
+            return QueueSpecification.queue(name)
+                    .durable(durable)
+                    .exclusive(exclusive)
+                    .autoDelete(autoDelete);
+        }
+
+        @Override
+        public QueueSpecification durable(boolean durable) {
+            if (this.durable != durable) {
+                throw new IllegalArgumentException("Once a queue has a null name, durable is always false");
+            }
+            return this;
+        }
+
+        @Override
+        public QueueSpecification exclusive(boolean exclusive) {
+            if (this.exclusive != exclusive) {
+                throw new IllegalArgumentException("Once a queue has a null name, exclusive is always true");
+            }
+            return this;
+        }
+
+        @Override
+        public QueueSpecification autoDelete(boolean autoDelete) {
+            if (this.autoDelete != autoDelete) {
+                throw new IllegalArgumentException("Once a queue has a null name, autoDelete is always true");
+            }
+            return this;
+        }
+    }
 }

src/main/java/reactor/rabbitmq/Sender.java

@@ -27,11 +27,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.CoreSubscriber;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.FluxOperator;
-import reactor.core.publisher.Mono;
-import reactor.core.publisher.SignalType;
-import reactor.core.publisher.Operators;
+import reactor.core.publisher.*;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
@@ -188,18 +184,50 @@ public RpcClient rpcClient(String exchange, String routingKey, Supplier<String>
         return new RpcClient(connectionMono.map(CHANNEL_CREATION_FUNCTION).cache(), exchange, routingKey, correlationIdProvider);
     }
 
+    /**
+     * Declare a queue following the specification.
+     *
+     * @param specification the specification of the queue
+     * @return a mono wrapping the result of the declaration
+     * @see QueueSpecification
+     */
     public Mono<AMQP.Queue.DeclareOk> declare(QueueSpecification specification) {
         return this.declareQueue(specification, null);
     }
 
+    /**
+     * Declare a queue following the specification and the resource management options.
+     *
+     * @param specification the specification of the queue
+     * @param options       options for resource management
+     * @return a mono wrapping the result of the declaration
+     * @see QueueSpecification
+     * @see ResourceManagementOptions
+     */
     public Mono<AMQP.Queue.DeclareOk> declare(QueueSpecification specification, ResourceManagementOptions options) {
         return this.declareQueue(specification, options);
     }
 
+    /**
+     * Declare a queue following the specification.
+     *
+     * @param specification the specification of the queue
+     * @return a mono wrapping the result of the declaration
+     * @see QueueSpecification
+     */
     public Mono<AMQP.Queue.DeclareOk> declareQueue(QueueSpecification specification) {
         return this.declareQueue(specification, null);
     }
 
+    /**
+     * Declare a queue following the specification and the resource management options.
+     *
+     * @param specification the specification of the queue
+     * @param options       options for resource management
+     * @return a mono wrapping the result of the declaration
+     * @see QueueSpecification
+     * @see ResourceManagementOptions
+     */
     public Mono<AMQP.Queue.DeclareOk> declareQueue(QueueSpecification specification, ResourceManagementOptions options) {
         Mono<? extends Channel> channelMono = getChannelMonoForResourceManagement(options);