Redis Sonar fixes

spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java

@@ -73,11 +73,11 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
 	private volatile boolean initialized;
 
 	// defaults
-	private volatile Executor taskExecutor = new SimpleAsyncTaskExecutor();
+	private Executor taskExecutor = new SimpleAsyncTaskExecutor();
 
-	private volatile RedisSerializer<?> serializer = new StringRedisSerializer();
+	private RedisSerializer<?> serializer = new StringRedisSerializer();
 
-	private volatile MessageConverter messageConverter = new SimpleMessageConverter();
+	private MessageConverter messageConverter = new SimpleMessageConverter();
 
 	public SubscribableRedisChannel(RedisConnectionFactory connectionFactory, String topicName) {
 		Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
@@ -125,7 +125,8 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
 
 	@Override
 	protected boolean doSend(Message<?> message, long arg1) {
-		this.redisTemplate.convertAndSend(this.topicName, this.messageConverter.fromMessage(message, Object.class));
+		Object value = this.messageConverter.fromMessage(message, Object.class);
+		this.redisTemplate.convertAndSend(this.topicName, value); // NOSONAR - null can be sent
 		return true;
 	}
 
@@ -136,9 +137,8 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
 		}
 		super.onInit();
 		if (this.maxSubscribers == null) {
-			Integer maxSubscribers =
-					getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class);
-			this.setMaxSubscribers(maxSubscribers);
+			setMaxSubscribers(
+					getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class));
 		}
 		if (this.messageConverter == null) {
 			this.messageConverter = new SimpleMessageConverter();
@@ -214,8 +214,10 @@ public class SubscribableRedisChannel extends AbstractMessageChannel
 				SubscribableRedisChannel.this.dispatcher.dispatch(siMessage);
 			}
 			catch (MessageDispatchingException e) {
-				String topicName = SubscribableRedisChannel.this.topicName;
-				topicName = StringUtils.hasText(topicName) ? topicName : "unknown";
+				String topicName =
+						StringUtils.hasText(SubscribableRedisChannel.this.topicName)
+								? SubscribableRedisChannel.this.topicName
+								: "unknown";
 				throw new MessageDeliveryException(siMessage, e.getMessage()
 						+ " for redis-channel '"
 						+ topicName

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.redis.inbound;
 
+import java.util.Optional;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
@@ -187,30 +188,9 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
 
 	@SuppressWarnings("unchecked")
 	private void popMessageAndSend() {
-		Message<Object> message = null;
+		byte[] value = popForValue();
 
-		byte[] value = null;
-		try {
-			if (this.rightPop) {
-				value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
-			}
-			else {
-				value = this.boundListOperations.leftPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
-			}
-		}
-		catch (Exception e) {
-			this.listening = false;
-			if (this.active) {
-				logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval
-						+ " milliseconds.", e);
-				this.publishException(e);
-				this.sleepBeforeRecoveryAttempt();
-			}
-			else {
-				logger.debug("Failed to execute listening task. " + e.getClass() + ": " + e.getMessage());
-			}
-			return;
-		}
+		Message<Object> message = null;
 
 		if (value != null) {
 			if (this.expectMessage) {
@@ -226,7 +206,9 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
 				if (this.serializer != null) {
 					payload = this.serializer.deserialize(value);
 				}
-				message = this.getMessageBuilderFactory().withPayload(payload).build();
+				if (payload != null) {
+					message = getMessageBuilderFactory().withPayload(payload).build();
+				}
 			}
 		}
 
@@ -245,6 +227,31 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
 		}
 	}
 
+	private byte[] popForValue() {
+		byte[] value = null;
+		try {
+			if (this.rightPop) {
+				value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
+			}
+			else {
+				value = this.boundListOperations.leftPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
+			}
+		}
+		catch (Exception e) {
+			this.listening = false;
+			if (this.active) {
+				logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval
+						+ " milliseconds.", e);
+				publishException(e);
+				sleepBeforeRecoveryAttempt();
+			}
+			else {
+				logger.debug("Failed to execute listening task. " + e.getClass() + ": " + e.getMessage());
+			}
+		}
+		return value;
+	}
+
 	@Override
 	protected void doStart() {
 		if (!this.active) {
@@ -293,7 +300,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
 	@Override
 	protected void doStop() {
 		super.doStop();
-		this.active = this.listening = false;
+		this.active = false;
+		this.listening = false;
 	}
 
 	public boolean isListening() {
@@ -304,12 +312,12 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
 	 * 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 http://redis.io/commands/llen
-	 *
 	 * @return Size of the queue. Never negative.
 	 */
 	@ManagedMetric
 	public long getQueueSize() {
-		return this.boundListOperations.size();
+		return Optional.ofNullable(this.boundListOperations.size())
+				.orElse(0L);
 	}
 
 	/**

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

@@ -88,11 +88,10 @@ public class RedisStoreMessageSource extends AbstractMessageSource<RedisStore> {
 		Assert.notNull(keyExpression, "'keyExpression' must not be null");
 		Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
 
-		StringRedisTemplate redisTemplate = new StringRedisTemplate();
-		redisTemplate.setConnectionFactory(connectionFactory);
-		redisTemplate.afterPropertiesSet();
+		this.redisTemplate = new StringRedisTemplate();
+		this.redisTemplate.setConnectionFactory(connectionFactory);
+		this.redisTemplate.afterPropertiesSet();
 
-		this.redisTemplate = redisTemplate;
 		this.keyExpression = keyExpression;
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2007-2016 the original author or authors.
+ * Copyright 2007-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -34,6 +34,7 @@ import org.springframework.util.Assert;
 /**
  * @author Mark Fisher
  * @author Artem Bilan
+ *
  * @since 2.1
  */
 public class RedisPublishingMessageHandler extends AbstractMessageHandler {
@@ -86,7 +87,7 @@ public class RedisPublishingMessageHandler extends AbstractMessageHandler {
 	}
 
 	@Override
-	protected void onInit() throws Exception {
+	protected void onInit() {
 		Assert.notNull(this.topicExpression, "'topicExpression' must not be null.");
 		if (this.messageConverter instanceof BeanFactoryAware) {
 			((BeanFactoryAware) this.messageConverter).setBeanFactory(getBeanFactory());
@@ -98,9 +99,9 @@ public class RedisPublishingMessageHandler extends AbstractMessageHandler {
 
 	@Override
 	@SuppressWarnings("unchecked")
-	protected void handleMessageInternal(Message<?> message) throws Exception {
+	protected void handleMessageInternal(Message<?> message) {
 		String topic = this.topicExpression.getValue(this.evaluationContext, message, String.class);
-		Object value = this.messageConverter.fromMessage(message, null);
+		Object value = this.messageConverter.fromMessage(message, Object.class);
 
 		if (value instanceof byte[]) {
 			this.template.convertAndSend(topic, value);