From 1e66b6b55d2734a3ebd60bd57eb927af6d9ae8ee Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 26 Oct 2018 14:16:26 -0400 Subject: [PATCH] Redis Sonar fixes --- .../channel/SubscribableRedisChannel.java | 20 +++--- .../RedisQueueMessageDrivenEndpoint.java | 62 +++++++++++-------- .../inbound/RedisStoreMessageSource.java | 7 +-- .../RedisPublishingMessageHandler.java | 9 +-- 4 files changed, 54 insertions(+), 44 deletions(-) diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java index 9bb149294c..f9140a4764 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java +++ b/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 diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java index ae5be9fb65..6619e30373 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpoint.java +++ b/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 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 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 0 * 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); } /** diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java index d1c798283c..afa516831f 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java @@ -88,11 +88,10 @@ public class RedisStoreMessageSource extends AbstractMessageSource { 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; } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java index feced7b238..70e355f66b 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java +++ b/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);