From e8bd31cc372457bc7ec4db29269cd9bb1a99c7d1 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 19 Dec 2018 09:47:57 -0500 Subject: [PATCH] Sonar fixes Critical smells `0.s.i.r*` * Polishing - PR Comments. --- .../context/IntegrationObjectSupport.java | 11 ++++- .../router/AbstractMappingMessageRouter.java | 5 ++- .../router/AbstractMessageRouter.java | 43 +++++++++++------- .../channel/SubscribableRedisChannel.java | 6 ++- .../inbound/RedisQueueInboundGateway.java | 21 +++++++-- .../outbound/RedisQueueOutboundGateway.java | 39 +++++++++++----- .../RedisStoreWritingMessageHandler.java | 8 ++-- .../redis/store/RedisChannelMessageStore.java | 12 +++-- .../RedisChannelPriorityMessageStore.java | 45 +++++++++++-------- 9 files changed, 130 insertions(+), 60 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java index c2e23f8e5b..d36c5952f1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java @@ -17,6 +17,7 @@ package org.springframework.integration.context; import java.util.Properties; +import java.util.UUID; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -44,7 +45,9 @@ import org.springframework.integration.support.utils.IntegrationUtils; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.core.DestinationResolver; import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.AlternativeJdkIdGenerator; import org.springframework.util.Assert; +import org.springframework.util.IdGenerator; import org.springframework.util.StringUtils; /** @@ -68,10 +71,12 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo protected static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser(); + private static final IdGenerator idGenerator = new AlternativeJdkIdGenerator(); // NOSONAR lower case + /** * Logger that is available to subclasses */ - protected final Log logger = LogFactory.getLog(getClass()); + protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR protected private final ConversionService defaultConversionService = DefaultConversionService.getSharedInstance(); @@ -312,4 +317,8 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo return (this.beanName != null) ? this.beanName : super.toString(); } + public static UUID generateId() { + return idGenerator.generateId(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java index 6545620b04..ac3ec4bde4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMappingMessageRouter.java @@ -293,7 +293,10 @@ public abstract class AbstractMappingMessageRouter extends AbstractMessageRouter addToCollection(channels, (Collection) channelKey, message); } else if (getRequiredConversionService().canConvert(channelKey.getClass(), String.class)) { - addChannelFromString(channels, getConversionService().convert(channelKey, String.class), message); + String converted = getConversionService().convert(channelKey, String.class); + if (converted != null) { + addChannelFromString(channels, converted, message); + } } else { throw new MessagingException("unsupported return type for router [" + channelKey.getClass() + "]"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java index 2387d7ccf8..4acbcf1271 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java @@ -17,6 +17,7 @@ package org.springframework.integration.router; import java.util.Collection; +import java.util.UUID; import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.support.DefaultConversionService; @@ -171,24 +172,20 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler imple int sequenceSize = results.size(); int sequenceNumber = 1; for (MessageChannel channel : results) { - final Message messageToSend = - !this.applySequence ? message : (this.getMessageBuilderFactory() + final Message messageToSend; + if (!this.applySequence) { + messageToSend = message; + } + else { + UUID id = message.getHeaders().getId(); + messageToSend = getMessageBuilderFactory() .fromMessage(message) - .pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize) - .build()); + .pushSequenceDetails(id == null ? generateId() : id, + sequenceNumber++, sequenceSize) + .build(); + } if (channel != null) { - try { - this.messagingTemplate.send(channel, messageToSend); - sent = true; - } - catch (MessagingException e) { - if (!this.ignoreSendFailures) { - throw e; - } - else if (this.logger.isDebugEnabled()) { - this.logger.debug(e); - } - } + sent |= doSend(channel, messageToSend); } } } @@ -204,4 +201,18 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler imple } } + private boolean doSend(MessageChannel channel, final Message messageToSend) { + try { + this.messagingTemplate.send(channel, messageToSend); + return true; + } + catch (MessagingException e) { + if (!this.ignoreSendFailures) { + throw e; + } + this.logger.debug("Send failure ignored", e); + return false; + } + } + } 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 f9140a4764..a80cc8794c 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 @@ -218,10 +218,12 @@ public class SubscribableRedisChannel extends AbstractMessageChannel StringUtils.hasText(SubscribableRedisChannel.this.topicName) ? SubscribableRedisChannel.this.topicName : "unknown"; - throw new MessageDeliveryException(siMessage, e.getMessage() + String exceptionMessage = e.getMessage(); + throw new MessageDeliveryException(siMessage, + (exceptionMessage == null ? e.getClass().getSimpleName() : exceptionMessage) + " for redis-channel '" + topicName - + "' (" + SubscribableRedisChannel.this.getFullChannelName() + ").", e); + + "' (" + getFullChannelName() + ").", e); // NOSONAR false - never null } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java index 94b5d92fc4..227d49f614 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisQueueInboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-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. @@ -194,6 +194,9 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements return; } uuid = stringSerializer.deserialize(value); + if (uuid == null) { + return; + } try { value = this.template.boundListOps(uuid).rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS); } @@ -212,18 +215,24 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements Object payload = value; if (this.serializer != null) { payload = this.serializer.deserialize(value); + if (payload == null) { + return; + } } - requestMessage = this.getMessageBuilderFactory().withPayload(payload).build(); + requestMessage = getMessageBuilderFactory().withPayload(payload).build(); } else { try { requestMessage = (Message) this.serializer.deserialize(value); + if (requestMessage == null) { + return; + } } catch (Exception e) { throw new MessagingException("Deserialization of Message failed.", e); } } - Message replyMessage = this.sendAndReceiveMessage(requestMessage); + Message replyMessage = sendAndReceiveMessage(requestMessage); if (replyMessage != null) { if (this.extractPayload) { value = extractReplyPayload(replyMessage); @@ -231,6 +240,9 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements else { if (this.serializer != null) { value = ((RedisSerializer) this.serializer).serialize(replyMessage); + if (value == null) { + return; + } } } this.template.boundListOps(uuid + QUEUE_NAME_SUFFIX).leftPush(value); @@ -320,7 +332,8 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements */ @ManagedMetric public long getQueueSize() { - return this.boundListOperations.size(); + Long size = this.boundListOperations.size(); + return size == null ? 0 : size; } /** diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java index 78346991a5..f1056641ad 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundGateway.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2017 the original author or authors. + * Copyright 2014-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. @@ -25,6 +25,7 @@ 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.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.AlternativeJdkIdGenerator; import org.springframework.util.Assert; @@ -33,6 +34,7 @@ import org.springframework.util.IdGenerator; /** * @author David Liu * @author Artem Bilan + * @author Gary Russell * * @since 4.1 */ @@ -90,12 +92,14 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand @Override @SuppressWarnings("unchecked") + @Nullable protected Object handleRequestMessage(Message message) { Object value = message; if (this.extractPayload) { value = message.getPayload(); } + Object beforeSerialization = value; if (!(value instanceof byte[])) { if (value instanceof String && !this.serializerExplicitlySet) { value = stringSerializer.serialize((String) value); @@ -104,6 +108,12 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand value = ((RedisSerializer) this.serializer).serialize(value); } } + if (value == null) { + if (this.logger.isDebugEnabled()) { + this.logger.debug("Serializer produced null for " + beforeSerialization); + } + return null; + } String uuid = defaultIdGenerator.generateId().toString(); byte[] uuidByte = uuid.getBytes(); @@ -113,19 +123,24 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand BoundListOperations boundListOperations = this.template.boundListOps(uuid + QUEUE_NAME_SUFFIX); byte[] reply = (byte[]) boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS); if (reply != null && reply.length > 0) { - Object replyMessage = this.serializer.deserialize(reply); - if (replyMessage == null) { - return null; - } - if (this.extractPayload) { - return getMessageBuilderFactory() - .withPayload(replyMessage); - } - else { - return replyMessage; - } + return createReply(reply); } return null; } + @Nullable + private Object createReply(byte[] reply) { + Object replyMessage = this.serializer.deserialize(reply); + if (replyMessage == null) { + return null; + } + if (this.extractPayload) { + return getMessageBuilderFactory() + .withPayload(replyMessage); + } + else { + return replyMessage; + } + } + } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandler.java index 9894383e48..6f48cc3b81 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandler.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisStoreWritingMessageHandler.java @@ -1,5 +1,5 @@ /* - * Copyright 2007-2017 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. @@ -416,15 +416,17 @@ public class RedisStoreWritingMessageHandler extends AbstractMessageHandler { } private void processInPipeline(PipelineCallback callback) { + RedisConnectionFactory connectionFactoryForPipeline = this.redisTemplate.getConnectionFactory(); + Assert.state(connectionFactoryForPipeline != null, "RedisTemplate returned no connection factory"); RedisConnection connection = - RedisConnectionUtils.bindConnection(this.redisTemplate.getConnectionFactory()); + RedisConnectionUtils.bindConnection(connectionFactoryForPipeline); try { connection.openPipeline(); callback.process(); } finally { connection.closePipeline(); - RedisConnectionUtils.unbindConnection(this.redisTemplate.getConnectionFactory()); + RedisConnectionUtils.unbindConnection(connectionFactoryForPipeline); } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java index 54ad88d8be..0859bc44df 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-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. @@ -114,7 +114,8 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw @Override @ManagedAttribute public int messageGroupSize(Object groupId) { - return (int) this.redisTemplate.boundListOps(groupId).size().longValue(); + Long size = this.redisTemplate.boundListOps(groupId).size(); + return size == null ? 0 : (int) size.longValue(); } @Override @@ -129,6 +130,7 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw return null; } + @Override public void removeMessageGroup(Object groupId) { this.redisTemplate.boundListOps(groupId).trim(1, 0); } @@ -141,6 +143,9 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw @ManagedAttribute public int getMessageCountForAllMessageGroups() { Set keys = this.redisTemplate.keys(this.beanName + ":*"); + if (keys == null) { + return 0; + } int count = 0; for (Object key : keys) { count += this.messageGroupSize(key); @@ -150,7 +155,8 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw @ManagedAttribute public int getMessageGroupCount() { - return this.redisTemplate.keys(this.beanName + ":*").size(); + Set keys = this.redisTemplate.keys(this.beanName + ":*"); + return keys == null ? 0 : keys.size(); } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java index 4d19bed227..5659554971 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelPriorityMessageStore.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-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. @@ -65,7 +65,10 @@ public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore List list = sortedKeys((String) groupId); int count = 0; for (String key : list) { - count += this.getRedisTemplate().boundListOps(key).size(); + Long size = getRedisTemplate().boundListOps(key).size(); + if (size != null) { + count += size; + } } return count; } @@ -77,7 +80,9 @@ public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore List list = sortedKeys((String) groupId); for (String key : list) { List> messages = this.getRedisTemplate().boundListOps(key).range(0, -1); - allMessages.addAll(messages); + if (messages != null) { + allMessages.addAll(messages); + } } return getMessageGroupFactory().create(allMessages, groupId); } @@ -110,12 +115,14 @@ public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore private List sortedKeys(String groupId) { Set keys = this.getRedisTemplate().keys(groupId == null ? (this.getBeanName() + ":*") : (groupId + "*")); - List list = new LinkedList(); - for (Object key : keys) { - Assert.isInstanceOf(String.class, key); - list.add((String) key); + List list = new LinkedList<>(); + if (keys != null) { + for (Object key : keys) { + Assert.isInstanceOf(String.class, key); + list.add((String) key); + } + Collections.sort(list, this.keysComparator); } - Collections.sort(list, this.keysComparator); return list; } @@ -129,16 +136,18 @@ public class RedisChannelPriorityMessageStore extends RedisChannelMessageStore private Set narrowedKeys() { Set keys = this.getRedisTemplate().keys(this.getBeanName() + ":*"); - Set narrowedKeys = new HashSet(); - for (Object key : keys) { - Assert.isInstanceOf(String.class, key); - String keyString = (String) key; - int lastIndexOfColon = keyString.lastIndexOf(":"); - if (keyString.indexOf(":") != lastIndexOfColon) { - narrowedKeys.add(keyString.substring(0, lastIndexOfColon)); - } - else { - narrowedKeys.add(key); + Set narrowedKeys = new HashSet<>(); + if (keys != null) { + for (Object key : keys) { + Assert.isInstanceOf(String.class, key); + String keyString = (String) key; + int lastIndexOfColon = keyString.lastIndexOf(":"); + if (keyString.indexOf(":") != lastIndexOfColon) { + narrowedKeys.add(keyString.substring(0, lastIndexOfColon)); + } + else { + narrowedKeys.add(key); + } } } return narrowedKeys;