Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix Redis components for JDK deserialization

Browse Source 
It turns out that `JdkSerializationRedisSerializer` by default is
based on the default Java class loader which may lead into
`ClassCastException` downstream after deserialization

* Make all the `JdkSerializationRedisSerializer` usage (default)
in Redis components based on the BeanFactory `ClassLoader`
* Fix tests to call `setBeanClassLoader()`
* Fix Mark Fisher's name in the `MultipartFileReader` :-)

**Cherry-pick to 5.1.x, 5.0.x & 4.3.x after restoring diamonds**
This commit is contained in:
Artem Bilan
2019-06-28 15:51:59 -04:00
committed by Gary Russell
parent 890cd1feb9
commit f89e8aafa5
6 changed files with 91 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-http/src/main/java/org/springframework/integration/http/multipart/MultipartFileReader.java
View File

@@ -23,7 +23,8 @@ import org.springframework.web.multipart.MultipartFile;
/**
* Strategy for reading {@link MultipartFile} content.
*
* @author mark Fisher
* @author Mark Fisher
*
* @since 2.0
*/
public interface MultipartFileReader<T> {

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

@@ -19,6 +19,7 @@ package org.springframework.integration.redis.inbound;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
@@ -51,7 +52,8 @@ import org.springframework.util.Assert;
*/
@ManagedResource
@IntegrationManagedResource
public class RedisQueueInboundGateway extends MessagingGatewaySupport implements ApplicationEventPublisherAware {
public class RedisQueueInboundGateway extends MessagingGatewaySupport
implements ApplicationEventPublisherAware, BeanClassLoaderAware {
private static final String QUEUE_NAME_SUFFIX = ".reply";
@@ -65,24 +67,24 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
private final BoundListOperations<String, byte[]> boundListOperations;
private volatile ApplicationEventPublisher applicationEventPublisher;
private ApplicationEventPublisher applicationEventPublisher;
private volatile boolean serializerExplicitlySet;
private boolean serializerExplicitlySet;
private volatile Executor taskExecutor;
private Executor taskExecutor;
private volatile RedisSerializer<?> serializer = new JdkSerializationRedisSerializer();
private RedisSerializer<?> serializer;
private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
private long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
private boolean extractPayload = true;
private volatile boolean active;
private volatile boolean listening;
private volatile boolean extractPayload = true;
private volatile Runnable stopCallback;
/**
@@ -92,7 +94,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
public RedisQueueInboundGateway(String queueName, RedisConnectionFactory connectionFactory) {
Assert.hasText(queueName, "'queueName' is required");
Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
this.template = new RedisTemplate<String, byte[]>();
this.template = new RedisTemplate<>();
this.template.setConnectionFactory(connectionFactory);
this.template.setEnableDefaultSerializer(false);
this.template.setKeySerializer(new StringRedisSerializer());
@@ -109,12 +111,18 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
this.applicationEventPublisher = applicationEventPublisher;
}
@Override
public void setBeanClassLoader(ClassLoader beanClassLoader) {
if (!this.serializerExplicitlySet) {
this.serializer = new JdkSerializationRedisSerializer(beanClassLoader);
}
}
public void setSerializer(RedisSerializer<?> serializer) {
this.serializer = serializer;
this.serializerExplicitlySet = true;
}
/**
* This timeout (milliseconds) is used when retrieving elements from the queue
* specified by {@link #boundListOperations}.
@@ -146,11 +154,11 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
Assert.notNull(this.serializer, "'serializer' has to be provided where 'extractPayload == false'.");
}
if (this.taskExecutor == null) {
String beanName = this.getComponentName();
String beanName = getComponentName();
this.taskExecutor = new SimpleAsyncTaskExecutor((beanName == null ? "" : beanName + "-")
+ this.getComponentType());
+ getComponentType());
}
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor) && this.getBeanFactory() != null) {
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor) && getBeanFactory() != null) {
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
errorHandler.setBeanFactory(getBeanFactory());
errorHandler.setDefaultErrorChannel(getErrorChannel());
@@ -168,8 +176,8 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
if (this.active) {
logger.error("Failed to execute listening task. Will attempt to resubmit in " + this.recoveryInterval
+ " milliseconds.", e);
this.publishException(e);
this.sleepBeforeRecoveryAttempt();
publishException(e);
sleepBeforeRecoveryAttempt();
}
else {
logger.debug("Failed to execute listening task. " + e.getClass() + ": " + e.getMessage());
@@ -178,7 +186,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
@SuppressWarnings("unchecked")
private void receiveAndReply() {
byte[] value = null;
byte[] value;
try {
value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
}

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

@@ -20,6 +20,7 @@ import java.util.Optional;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
@@ -40,7 +41,6 @@ import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.scheduling.SchedulingAwareRunnable;
import org.springframework.util.Assert;
@@ -57,7 +57,8 @@ import org.springframework.util.Assert;
*/
@ManagedResource
@IntegrationManagedResource
public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport implements ApplicationEventPublisherAware {
public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport
implements ApplicationEventPublisherAware, BeanClassLoaderAware {
public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;
@@ -65,19 +66,21 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
private final BoundListOperations<String, byte[]> boundListOperations;
private volatile ApplicationEventPublisher applicationEventPublisher;
private ApplicationEventPublisher applicationEventPublisher;
private volatile MessageChannel errorChannel;
private Executor taskExecutor;
private volatile Executor taskExecutor;
private RedisSerializer<?> serializer;
private volatile RedisSerializer<?> serializer = new JdkSerializationRedisSerializer();
private boolean serializerExplicitlySet;
private volatile boolean expectMessage = false;
private boolean expectMessage = false;
private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
private long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
private volatile long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
private long recoveryInterval = DEFAULT_RECOVERY_INTERVAL;
private boolean rightPop = true;
private volatile boolean active;
@@ -85,8 +88,6 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
private volatile Runnable stopCallback;
private volatile boolean rightPop = true;
/**
* @param queueName Must not be an empty String
* @param connectionFactory Must not be null
@@ -94,7 +95,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
public RedisQueueMessageDrivenEndpoint(String queueName, RedisConnectionFactory connectionFactory) {
Assert.hasText(queueName, "'queueName' is required");
Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
RedisTemplate<String, byte[]> template = new RedisTemplate<String, byte[]>();
RedisTemplate<String, byte[]> template = new RedisTemplate<>();
template.setConnectionFactory(connectionFactory);
template.setEnableDefaultSerializer(false);
template.setKeySerializer(new StringRedisSerializer());
@@ -107,8 +108,16 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
this.applicationEventPublisher = applicationEventPublisher;
}
@Override
public void setBeanClassLoader(ClassLoader beanClassLoader) {
if (!this.serializerExplicitlySet) {
this.serializer = new JdkSerializationRedisSerializer(beanClassLoader);
}
}
public void setSerializer(RedisSerializer<?> serializer) {
this.serializer = serializer;
this.serializerExplicitlySet = true;
}
/**
@@ -116,8 +125,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
* just the payload for a Message, or does the data represent a serialized
* {@link Message}?. {@code expectMessage} defaults to false. This means
* the retrieved data will be used as the payload for a new Spring Integration
* Message. Otherwise, the data is deserialized as Spring Integration
* Message.
* Message. Otherwise, the data is deserialized as Spring Integration Message.
* @param expectMessage Defaults to false
*/
public void setExpectMessage(boolean expectMessage) {
@@ -144,12 +152,6 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
this.taskExecutor = taskExecutor;
}
@Override
public void setErrorChannel(MessageChannel errorChannel) {
super.setErrorChannel(errorChannel);
this.errorChannel = errorChannel;
}
public void setRecoveryInterval(long recoveryInterval) {
this.recoveryInterval = recoveryInterval;
}
@@ -178,7 +180,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport impl
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor) && beanFactory != null) {
MessagePublishingErrorHandler errorHandler =
new MessagePublishingErrorHandler(ChannelResolverUtils.getChannelResolver(beanFactory));
errorHandler.setDefaultErrorChannel(this.errorChannel);
errorHandler.setDefaultErrorChannel(getErrorChannel());
this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler);
}
}

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

@@ -46,19 +46,19 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand
private static final IdGenerator defaultIdGenerator = new AlternativeJdkIdGenerator();
private static final RedisSerializer<String> stringSerializer = new StringRedisSerializer();
private static final RedisSerializer<String> stringSerializer = new StringRedisSerializer();
private final RedisTemplate<String, Object> template = new RedisTemplate<>();
private final BoundListOperations<String, Object> boundListOps;
private volatile boolean extractPayload = true;
private boolean extractPayload = true;
private volatile RedisSerializer<?> serializer = new JdkSerializationRedisSerializer();
private RedisSerializer<?> serializer;
private volatile boolean serializerExplicitlySet;
private boolean serializerExplicitlySet;
private volatile int receiveTimeout = TIMEOUT;
private int receiveTimeout = TIMEOUT;
public RedisQueueOutboundGateway(String queueName, RedisConnectionFactory connectionFactory) {
Assert.hasText(queueName, "'queueName' is required");
@@ -70,6 +70,14 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand
this.boundListOps = this.template.boundListOps(queueName);
}
@Override
public void setBeanClassLoader(ClassLoader beanClassLoader) {
super.setBeanClassLoader(beanClassLoader);
if (!this.serializerExplicitlySet) {
this.serializer = new JdkSerializationRedisSerializer(beanClassLoader);
}
}
public void setReceiveTimeout(int timeout) {
this.receiveTimeout = timeout;
}

21
spring-integration-redis/src/main/java/org/springframework/integration/redis/store/RedisChannelMessageStore.java
View File

@@ -19,6 +19,7 @@ package org.springframework.integration.redis.store;
import java.util.List;
import java.util.Set;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.data.redis.connection.RedisConnectionFactory;
@@ -41,30 +42,41 @@ import org.springframework.util.Assert;
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 4.0
*
*/
public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAware, InitializingBean {
public class RedisChannelMessageStore
implements ChannelMessageStore, BeanNameAware, InitializingBean, BeanClassLoaderAware {
private final RedisTemplate<Object, Message<?>> redisTemplate;
private volatile MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory();
private String beanName;
private MessageGroupFactory messageGroupFactory = new SimpleMessageGroupFactory();
private boolean valueSerializerExplicitlySet;
/**
* Construct a message store that uses Java Serialization for messages.
*
* @param connectionFactory The redis connection factory.
*/
public RedisChannelMessageStore(RedisConnectionFactory connectionFactory) {
this.redisTemplate = new RedisTemplate<Object, Message<?>>();
this.redisTemplate = new RedisTemplate<>();
this.redisTemplate.setConnectionFactory(connectionFactory);
this.redisTemplate.setKeySerializer(new StringRedisSerializer());
this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
this.redisTemplate.afterPropertiesSet();
}
@Override
public void setBeanClassLoader(ClassLoader classLoader) {
if (!this.valueSerializerExplicitlySet) {
this.redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer(classLoader));
}
}
/**
* Use a different serializer (default {@link JdkSerializationRedisSerializer} for
* the {@link Message}.
@@ -74,6 +86,7 @@ public class RedisChannelMessageStore implements ChannelMessageStore, BeanNameAw
public void setValueSerializer(RedisSerializer<?> valueSerializer) {
Assert.notNull(valueSerializer, "'valueSerializer' must not be null");
this.redisTemplate.setValueSerializer(valueSerializer);
this.valueSerializerExplicitlySet = true;
}
/**

25
spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests.java
View File

@@ -63,8 +63,8 @@ import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.ClassUtils;
/**
* @author Gunnar Hillert
@@ -74,8 +74,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*
* @since 3.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@RunWith(SpringRunner.class)
@DirtiesContext
public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
@@ -95,7 +94,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
@RedisAvailable
@SuppressWarnings("unchecked")
public void testInt3014Default() {
String queueName = "si.test.redisQueueInboundChannelAdapterTests";
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
@@ -118,6 +116,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
RedisQueueMessageDrivenEndpoint endpoint =
new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
endpoint.setBeanClassLoader(ClassUtils.getDefaultClassLoader());
endpoint.setOutputChannel(channel);
endpoint.setReceiveTimeout(10);
endpoint.afterPropertiesSet();
@@ -138,8 +137,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
@RedisAvailable
@SuppressWarnings("unchecked")
public void testInt3014ExpectMessageTrue() {
final String queueName = "si.test.redisQueueInboundChannelAdapterTests2";
String queueName = "si.test.redisQueueInboundChannelAdapterTests2";
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(this.connectionFactory);
@@ -188,7 +186,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testInt3017IntegrationInbound() {
String payload = new Date().toString();
RedisTemplate<String, String> redisTemplate = new StringRedisTemplate();
@@ -222,9 +219,9 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
@RedisAvailable
@SuppressWarnings("unchecked")
public void testInt3442ProperlyStop() throws Exception {
final String queueName = "si.test.testInt3442ProperlyStopTest";
String queueName = "si.test.testInt3442ProperlyStopTest";
final RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(this.connectionFactory);
redisTemplate.setEnableDefaultSerializer(false);
redisTemplate.setKeySerializer(new StringRedisSerializer());
@@ -257,9 +254,9 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
redisTemplate.boundListOps(queueName).leftPush("foo");
final CountDownLatch stopLatch = new CountDownLatch(1);
CountDownLatch stopLatch = new CountDownLatch(1);
endpoint.stop(() -> stopLatch.countDown());
endpoint.stop(stopLatch::countDown);
executorService.shutdown();
assertThat(executorService.awaitTermination(20, TimeUnit.SECONDS)).isTrue();
@@ -309,7 +306,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
((InitializingBean) this.connectionFactory).afterPropertiesSet();
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(this.getConnectionFactoryForTest());
redisTemplate.setEnableDefaultSerializer(false);
redisTemplate.setKeySerializer(new StringRedisSerializer());
@@ -331,7 +328,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
@RedisAvailable
@SuppressWarnings("unchecked")
public void testInt3932ReadFromLeft() {
String queueName = "si.test.redisQueueInboundChannelAdapterTests3932";
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
@@ -354,6 +350,7 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
RedisQueueMessageDrivenEndpoint endpoint =
new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
endpoint.setBeanClassLoader(ClassUtils.getDefaultClassLoader());
endpoint.setOutputChannel(channel);
endpoint.setReceiveTimeout(10);
endpoint.setRightPop(false);