issues/111 Added @nullable to Nullable params and return types

Author: vanseverk

src/main/java/reactor/rabbitmq/BindingSpecification.java

@@ -16,6 +16,8 @@
 
 package reactor.rabbitmq;
 
+import reactor.util.annotation.Nullable;
+
 import java.util.Map;
 
 /**
@@ -49,7 +51,7 @@ public BindingSpecification routingKey(String routingKey) {
         return this;
     }
 
-    public BindingSpecification arguments(Map<String, Object> arguments) {
+    public BindingSpecification arguments(@Nullable Map<String, Object> arguments) {
         this.arguments = arguments;
         return this;
     }
@@ -66,6 +68,7 @@ public String getRoutingKey() {
         return routingKey;
     }
 
+    @Nullable
     public Map<String, Object> getArguments() {
         return arguments;
     }

src/main/java/reactor/rabbitmq/ChannelPoolOptions.java

@@ -17,6 +17,7 @@
 package reactor.rabbitmq;
 
 import reactor.core.scheduler.Scheduler;
+import reactor.util.annotation.Nullable;
 
 /**
  * Options to when creating a {@link ChannelPool}.
@@ -44,6 +45,7 @@ public ChannelPoolOptions maxCacheSize(int maxCacheSize) {
         return this;
     }
 
+    @Nullable
     public Integer getMaxCacheSize() {
         return maxCacheSize;
     }
@@ -54,11 +56,12 @@ public Integer getMaxCacheSize() {
      * @param subscriptionScheduler
      * @return this {@link ChannelPoolOptions} instance
      */
-    public ChannelPoolOptions subscriptionScheduler(Scheduler subscriptionScheduler) {
+    public ChannelPoolOptions subscriptionScheduler(@Nullable Scheduler subscriptionScheduler) {
         this.subscriptionScheduler = subscriptionScheduler;
         return this;
     }
 
+    @Nullable
     public Scheduler getSubscriptionScheduler() {
         return subscriptionScheduler;
     }

src/main/java/reactor/rabbitmq/ExchangeSpecification.java

@@ -16,6 +16,8 @@
 
 package reactor.rabbitmq;
 
+import reactor.util.annotation.Nullable;
+
 import java.util.Map;
 
 /**
@@ -66,7 +68,7 @@ public ExchangeSpecification passive(boolean passive) {
         return this;
     }
 
-    public ExchangeSpecification arguments(Map<String, Object> arguments) {
+    public ExchangeSpecification arguments(@Nullable Map<String, Object> arguments) {
         this.arguments = arguments;
         return this;
     }
@@ -95,6 +97,7 @@ public boolean isPassive() {
         return passive;
     }
 
+    @Nullable
     public Map<String, Object> getArguments() {
         return arguments;
     }

src/main/java/reactor/rabbitmq/IdempotentClosedConnection.java

@@ -24,6 +24,7 @@
 import com.rabbitmq.client.ShutdownListener;
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.UnblockedCallback;
+import reactor.util.annotation.Nullable;
 
 import java.io.IOException;
 import java.net.InetAddress;
@@ -81,6 +82,7 @@ public Map<String, Object> getClientProperties() {
     }
 
     @Override
+    @Nullable
     public String getClientProvidedName() {
         return this.delegate.getClientProvidedName();
     }
@@ -91,11 +93,13 @@ public Map<String, Object> getServerProperties() {
     }
 
     @Override
+    @Nullable
     public Channel createChannel() throws IOException {
         return this.delegate.createChannel();
     }
 
     @Override
+    @Nullable
     public Channel createChannel(int channelNumber) throws IOException {
         return this.delegate.createChannel(channelNumber);
     }
@@ -206,6 +210,7 @@ public void removeShutdownListener(ShutdownListener listener) {
     }
 
     @Override
+    @Nullable
     public ShutdownSignalException getCloseReason() {
         return this.delegate.getCloseReason();
     }

src/main/java/reactor/rabbitmq/OutboundMessage.java

@@ -17,6 +17,7 @@
 package reactor.rabbitmq;
 
 import com.rabbitmq.client.AMQP.BasicProperties;
+import reactor.util.annotation.Nullable;
 
 import java.util.Arrays;
 
@@ -55,7 +56,7 @@ public OutboundMessage(String exchange, String routingKey, byte[] body) {
      * @param properties AMQP compatible properties that will be used during the publishing of the message.
      * @param body       The main body of the message.
      */
-    public OutboundMessage(String exchange, String routingKey, BasicProperties properties, byte[] body) {
+    public OutboundMessage(String exchange, String routingKey, @Nullable BasicProperties properties, byte[] body) {
         this.exchange = exchange;
         this.routingKey = routingKey;
         this.properties = properties;
@@ -85,6 +86,7 @@ public String getRoutingKey() {
      *
      * @return All the properties that have been set. Null if no property is set.
      */
+    @Nullable
     public BasicProperties getProperties() {
         return properties;
     }

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

@@ -16,8 +16,9 @@
 
 package reactor.rabbitmq;
 
+import reactor.util.annotation.Nullable;
+
 import java.util.Map;
-import java.util.Queue;
 
 /**
  * Fluent API to specify the creation of a queue.
@@ -40,11 +41,11 @@ public static QueueSpecification queue() {
         return new NullNameQueueSpecification();
     }
 
-    public static QueueSpecification queue(String name) {
+    public static QueueSpecification queue(@Nullable String name) {
         return new QueueSpecification().name(name);
     }
 
-    public QueueSpecification name(String queue) {
+    public QueueSpecification name(@Nullable String queue) {
         if (queue == null) {
             return new NullNameQueueSpecification().arguments(this.arguments);
         }
@@ -68,7 +69,7 @@ public QueueSpecification autoDelete(boolean autoDelete) {
         return this;
     }
 
-    public QueueSpecification arguments(Map<String, Object> arguments) {
+    public QueueSpecification arguments(@Nullable Map<String, Object> arguments) {
         this.arguments = arguments;
         return this;
     }
@@ -78,6 +79,7 @@ public QueueSpecification passive(boolean passive) {
         return this;
     }
 
+    @Nullable
     public String getName() {
         return name;
     }
@@ -98,6 +100,7 @@ public boolean isPassive() {
         return passive;
     }
 
+    @Nullable
     public Map<String, Object> getArguments() {
         return arguments;
     }
@@ -118,7 +121,7 @@ private static class NullNameQueueSpecification extends QueueSpecification {
         }
 
         @Override
-        public QueueSpecification name(String name) {
+        public QueueSpecification name(@Nullable String name) {
             if (name == null) {
                 return this;
             }

src/main/java/reactor/rabbitmq/ReceiverOptions.java

@@ -20,6 +20,7 @@
 import com.rabbitmq.client.ConnectionFactory;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
+import reactor.util.annotation.Nullable;
 
 import java.time.Duration;
 import java.util.function.Function;
@@ -62,6 +63,7 @@ public ReceiverOptions connectionFactory(ConnectionFactory connectionFactory) {
         return this;
     }
 
+    @Nullable
     public Scheduler getConnectionSubscriptionScheduler() {
         return connectionSubscriptionScheduler;
     }
@@ -73,7 +75,7 @@ public Scheduler getConnectionSubscriptionScheduler() {
      * @param connectionSubscriptionScheduler
      * @return the current {@link ReceiverOptions} instance
      */
-    public ReceiverOptions connectionSubscriptionScheduler(Scheduler connectionSubscriptionScheduler) {
+    public ReceiverOptions connectionSubscriptionScheduler(@Nullable Scheduler connectionSubscriptionScheduler) {
         this.connectionSubscriptionScheduler = connectionSubscriptionScheduler;
         return this;
     }
@@ -118,7 +120,7 @@ public ReceiverOptions connectionSupplier(ConnectionFactory connectionFactory, U
      * @param connectionMono
      * @return this current {@link ReceiverOptions}
      */
-    public ReceiverOptions connectionMono(Mono<? extends Connection> connectionMono) {
+    public ReceiverOptions connectionMono(@Nullable Mono<? extends Connection> connectionMono) {
         this.connectionMono = connectionMono;
         return this;
     }
@@ -139,10 +141,12 @@ public ReceiverOptions connectionMonoConfigurator(Function<Mono<? extends Connec
         return this;
     }
 
+    @Nullable
     public Mono<? extends Connection> getConnectionMono() {
         return connectionMono;
     }
 
+    @Nullable
     public Utils.ExceptionFunction<ConnectionFactory, ? extends Connection> getConnectionSupplier() {
         return connectionSupplier;
     }
@@ -160,11 +164,12 @@ public Function<Mono<? extends Connection>, Mono<? extends Connection>> getConne
      * @return this {@link ReceiverOptions} instance
      * @since 1.3.0
      */
-    public ReceiverOptions connectionClosingTimeout(Duration connectionClosingTimeout) {
+    public ReceiverOptions connectionClosingTimeout(@Nullable Duration connectionClosingTimeout) {
         this.connectionClosingTimeout = connectionClosingTimeout;
         return this;
     }
 
+    @Nullable
     public Duration getConnectionClosingTimeout() {
         return connectionClosingTimeout;
     }

src/main/java/reactor/rabbitmq/ResourceManagementOptions.java

@@ -18,6 +18,7 @@
 
 import com.rabbitmq.client.Channel;
 import reactor.core.publisher.Mono;
+import reactor.util.annotation.Nullable;
 
 /**
  * Options for resource management in {@link Sender}.
@@ -26,11 +27,12 @@ public class ResourceManagementOptions {
 
     private Mono<Channel> channelMono;
 
-    public ResourceManagementOptions channelMono(Mono<Channel> channelMono) {
+    public ResourceManagementOptions channelMono(@Nullable Mono<Channel> channelMono) {
         this.channelMono = channelMono;
         return this;
     }
 
+    @Nullable
     public Mono<Channel> getChannelMono() {
         return channelMono;
     }

src/main/java/reactor/rabbitmq/ResourcesSpecification.java

@@ -16,6 +16,8 @@
 
 package reactor.rabbitmq;
 
+import reactor.util.annotation.Nullable;
+
 /**
  * API that combines {@link ExchangeSpecification}, {@link QueueSpecification}, and {@link BindingSpecification}.
  */
@@ -25,7 +27,7 @@ public static ExchangeSpecification exchange(String name) {
         return ExchangeSpecification.exchange(name);
     }
 
-    public static QueueSpecification queue(String name) {
+    public static QueueSpecification queue(@Nullable String name) {
         return QueueSpecification.queue(name);
     }
 

src/main/java/reactor/rabbitmq/RpcClient.java

@@ -29,6 +29,7 @@
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoOperator;
 import reactor.core.publisher.Operators;
+import reactor.util.annotation.Nullable;
 
 import java.io.IOException;
 import java.util.concurrent.ConcurrentHashMap;
@@ -121,7 +122,7 @@ public static class RpcRequest {
         private final AMQP.BasicProperties properties;
         private final byte[] body;
 
-        public RpcRequest(AMQP.BasicProperties properties, byte[] body) {
+        public RpcRequest(@Nullable AMQP.BasicProperties properties, byte[] body) {
             this.properties = properties;
             this.body = body;
         }