Improve Redis module nullability

Author: artembilan

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java

@@ -19,6 +19,7 @@
 import java.time.Duration;
 import java.util.function.Function;
 
+import org.jspecify.annotations.Nullable;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -39,7 +40,6 @@
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.redis.support.RedisHeaders;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
-import org.jspecify.annotations.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.converter.MessageConversionException;
@@ -136,7 +136,7 @@ public void setAutoAck(boolean autoAck) {
 	 * {@link #createConsumerGroup}. If not set, the defined bean name {@link #getBeanName()} is used.
 	 * @param consumerGroup the Consumer Group on which this adapter should register to listen messages.
 	 */
-	public void setConsumerGroup(@Nullable String consumerGroup) {
+	public void setConsumerGroup(String consumerGroup) {
 		this.consumerGroup = consumerGroup;
 	}
 
@@ -295,8 +295,8 @@ protected void doStart() {
 			Mono<?> consumerGroupMono = Mono.empty();
 			if (this.createConsumerGroup) {
 				consumerGroupMono =
-						this.reactiveStreamOperations.createGroup(this.streamKey, this.consumerGroup) // NOSONAR
-						.onErrorReturn(this.consumerGroup);
+						this.reactiveStreamOperations.createGroup(this.streamKey, this.consumerGroup)
+								.onErrorReturn(this.consumerGroup);
 			}
 
 			Consumer consumer = Consumer.from(this.consumerGroup, this.consumerName);
@@ -348,8 +348,8 @@ private <T> Publisher<T> handleReceiverError(Throwable error) {
 			}
 		}
 		MessagingException conversionException = (failedMessage != null)
-					? new MessageConversionException(failedMessage, "Cannot deserialize Redis Stream Record", error)
-					: new MessageConversionException("Cannot deserialize Redis Stream Record", error);
+				? new MessageConversionException(failedMessage, "Cannot deserialize Redis Stream Record", error)
+				: new MessageConversionException("Cannot deserialize Redis Stream Record", error);
 		if (!sendErrorMessageIfNecessary(null, conversionException)) {
 			logger.getLog().error(conversionException);
 		}

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java

@@ -19,6 +19,8 @@
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
@@ -37,7 +39,6 @@
 import org.springframework.jmx.export.annotation.ManagedMetric;
 import org.springframework.jmx.export.annotation.ManagedOperation;
 import org.springframework.jmx.export.annotation.ManagedResource;
-import org.jspecify.annotations.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessagingException;
 import org.springframework.scheduling.SchedulingAwareRunnable;
@@ -282,7 +283,7 @@ private Message<Object> prepareRequestMessage(byte[] value) {
 	}
 
 	@SuppressWarnings("unchecked")
-	private @Nullable byte[] extractReplyPayload(Message<?> replyMessage) {
+	private byte @Nullable [] extractReplyPayload(Message<?> replyMessage) {
 		byte[] value = null;
 		if (!(replyMessage.getPayload() instanceof byte[])) {
 			if (replyMessage.getPayload() instanceof String && !this.serializerExplicitlySet) {
@@ -324,12 +325,7 @@ private void sleepBeforeRecoveryAttempt() {
 	}
 
 	private void publishException(Exception e) {
-		if (this.applicationEventPublisher != null) {
-			this.applicationEventPublisher.publishEvent(new RedisExceptionEvent(this, e));
-		}
-		else {
-			logger.debug(() -> "No application event publisher for exception: " + e.getMessage());
-		}
+		this.applicationEventPublisher.publishEvent(new RedisExceptionEvent(this, e));
 	}
 
 	private void restart() {
@@ -355,7 +351,7 @@ public boolean isListening() {
 	/**
 	 * Returns the size of the Queue specified by {@link #boundListOperations}. The queue is
 	 * represented by a Redis list. If the queue does not exist <code>0</code>
-	 * is returned. See also https://redis.io/commands/llen
+	 * is returned. See also <a href="https://redis.io/commands/llen">LLEN</a>
 	 * @return Size of the queue. Never negative.
 	 */
 	@ManagedMetric

spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java

@@ -21,6 +21,7 @@
 import java.util.concurrent.TimeUnit;
 
 import org.jspecify.annotations.Nullable;
+
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.context.ApplicationEventPublisher;
@@ -195,15 +196,15 @@ public String getComponentType() {
 
 	@SuppressWarnings("unchecked")
 	private void popMessageAndSend() {
-		@Nullable byte[] value = popForValue();
+		byte[] value = popForValue();
 
 		Message<Object> message = null;
 
 		if (value != null) {
 			if (this.expectMessage) {
 				try {
 					if (this.serializer != null) {
-						message = (Message<Object>) serializer.deserialize(value);
+						message = (Message<Object>) this.serializer.deserialize(value);
 					}
 				}
 				catch (Exception e) {
@@ -236,8 +237,8 @@ private void popMessageAndSend() {
 		}
 	}
 
-	private @Nullable byte[] popForValue() {
-		@Nullable byte[] value = null;
+	private byte @Nullable [] popForValue() {
+		byte[] value = null;
 		try {
 			if (this.rightPop) {
 				value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);

spring-integration-redis/src/main/java/org/springframework/integration/redis/metadata/RedisMetadataStore.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.redis.metadata;
 
 import org.jspecify.annotations.Nullable;
+
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.BoundHashOperations;
 import org.springframework.data.redis.core.RedisOperations;

spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ExpressionArgumentsStrategy.java

@@ -35,6 +35,7 @@
 /**
  * @author Artem Bilan
  * @author Gary Russell
+ *
  * @since 4.0
  */
 public class ExpressionArgumentsStrategy implements ArgumentsStrategy, BeanFactoryAware, InitializingBean {
@@ -43,11 +44,11 @@ public class ExpressionArgumentsStrategy implements ArgumentsStrategy, BeanFacto
 
 	private final Expression[] argumentExpressions;
 
+	private final boolean useCommandVariable;
+
 	@SuppressWarnings("NullAway.Init")
 	private EvaluationContext evaluationContext;
 
-	private final boolean useCommandVariable;
-
 	@SuppressWarnings("NullAway.Init")
 	private BeanFactory beanFactory;
 
@@ -58,28 +59,22 @@ public ExpressionArgumentsStrategy(String[] argumentExpressions) {
 	public ExpressionArgumentsStrategy(String[] argumentExpressions, boolean useCommandVariable) {
 		Assert.notNull(argumentExpressions, "'argumentExpressions' must not be null");
 		Assert.noNullElements(argumentExpressions, "'argumentExpressions' cannot have null values.");
-		List<Expression> expressions = new LinkedList<Expression>();
+		List<Expression> expressions = new LinkedList<>();
 		for (String argumentExpression : argumentExpressions) {
 			expressions.add(PARSER.parseExpression(argumentExpression));
 		}
-		this.argumentExpressions = expressions.toArray(new Expression[expressions.size()]);
+		this.argumentExpressions = expressions.toArray(new Expression[0]);
 		this.useCommandVariable = useCommandVariable;
 	}
 
-	public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
-		this.evaluationContext = evaluationContext;
-	}
-
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
 	}
 
 	@Override
 	public void afterPropertiesSet() {
-		if (this.evaluationContext == null) {
-			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
-		}
+		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
 	}
 
 	@Override
@@ -90,7 +85,7 @@ public Object[] resolve(String command, Message<?> message) {
 			evaluationContextToUse = IntegrationContextUtils.getEvaluationContext(this.beanFactory);
 			evaluationContextToUse.setVariable("cmd", command);
 		}
-		List<Object> arguments = new ArrayList<Object>();
+		List<Object> arguments = new ArrayList<>();
 		for (Expression argumentExpression : this.argumentExpressions) {
 			Object argument = argumentExpression.getValue(evaluationContextToUse, message);
 			if (argument != null) {

spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/ReactiveRedisStreamMessageHandler.java

@@ -18,6 +18,7 @@
 
 import java.util.function.Function;
 
+import org.jspecify.annotations.Nullable;
 import reactor.core.publisher.Mono;
 
 import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
@@ -33,7 +34,6 @@
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.handler.AbstractReactiveMessageHandler;
-import org.jspecify.annotations.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 

spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisOutboundGateway.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.redis.outbound;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.RedisTemplate;
@@ -25,9 +27,9 @@
 import org.springframework.expression.Expression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.expression.ExpressionUtils;
+import org.springframework.integration.expression.FunctionExpression;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.redis.support.RedisHeaders;
-import org.jspecify.annotations.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
@@ -50,11 +52,13 @@ public class RedisOutboundGateway extends AbstractReplyProducingMessageHandler {
 	@SuppressWarnings("NullAway.Init")
 	private EvaluationContext evaluationContext;
 
-	private volatile RedisSerializer<Object> argumentsSerializer = new GenericToStringSerializer<>(Object.class);
+	private RedisSerializer<Object> argumentsSerializer = new GenericToStringSerializer<>(Object.class);
 
-	private volatile Expression commandExpression = PARSER.parseExpression("headers[" + RedisHeaders.COMMAND + "]");
+	@SuppressWarnings("NullAway") // Cannot see nullability on the generics
+	private Expression commandExpression =
+			new FunctionExpression<Message<?>>(message -> message.getHeaders().get(RedisHeaders.COMMAND));
 
-	private volatile ArgumentsStrategy argumentsStrategy = new PayloadArgumentsStrategy();
+	private @Nullable ArgumentsStrategy argumentsStrategy = new PayloadArgumentsStrategy();
 
 	public RedisOutboundGateway(RedisTemplate<?, ?> redisTemplate) {
 		Assert.notNull(redisTemplate, "'redisTemplate' must not be null");
@@ -63,7 +67,7 @@ public RedisOutboundGateway(RedisTemplate<?, ?> redisTemplate) {
 
 	public RedisOutboundGateway(RedisConnectionFactory connectionFactory) {
 		Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
-		this.redisTemplate = new RedisTemplate<Object, Object>();
+		this.redisTemplate = new RedisTemplate<>();
 		this.redisTemplate.setConnectionFactory(connectionFactory);
 		this.redisTemplate.afterPropertiesSet();
 	}
@@ -91,14 +95,10 @@ public void setCommandExpressionString(String commandExpression) {
 		this.commandExpression = EXPRESSION_PARSER.parseExpression(commandExpression);
 	}
 
-	public void setArgumentsStrategy(ArgumentsStrategy argumentsStrategy) {
+	public void setArgumentsStrategy(@Nullable ArgumentsStrategy argumentsStrategy) {
 		this.argumentsStrategy = argumentsStrategy;
 	}
 
-	public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
-		this.evaluationContext = evaluationContext;
-	}
-
 	@Override
 	public String getComponentType() {
 		return "redis:outbound-gateway";
@@ -107,9 +107,7 @@ public String getComponentType() {
 	@Override
 	protected void doInit() {
 		super.doInit();
-		if (this.evaluationContext == null) {
-			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
-		}
+		this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
 	}
 
 	@Override
@@ -124,22 +122,15 @@ protected void doInit() {
 
 				for (int i = 0; i < arguments.length; i++) {
 					Object argument = arguments[i];
-					byte[] arg = null;
-					if (argument instanceof byte[]) {
-						arg = (byte[]) argument;
-					}
-					else {
-						arg = this.argumentsSerializer.serialize(argument);
-					}
-					args[i] = arg;
+					args[i] = argument instanceof byte[] bytes ? bytes : this.argumentsSerializer.serialize(argument);
 				}
 			}
 		}
 
 		final byte[][] actualArgs = args != null ? args : EMPTY_ARGS;
 
 		return this.redisTemplate.execute(
-				(RedisCallback<Object>) connection -> connection.execute(command, actualArgs));
+				(RedisCallback<@Nullable Object>) connection -> connection.execute(command, actualArgs));
 	}
 
 	private static class PayloadArgumentsStrategy implements ArgumentsStrategy {

spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java

@@ -18,14 +18,15 @@
 
 import java.util.concurrent.TimeUnit;
 
+import org.jspecify.annotations.Nullable;
+
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.BoundListOperations;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
 import org.springframework.data.redis.serializer.RedisSerializer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
-import org.jspecify.annotations.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.AlternativeJdkIdGenerator;
 import org.springframework.util.Assert;