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

INT-3939: Consistency for Redis Queue Gateways

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3939

Previously `RedisQueueInboundGateway` and `RedisQueueOutboundGateway` used different `RedisSerializer`s
for non-String objects.

* Change the default to the `JdkSerializationRedisSerializer` for the consistency in case of client/server scenarios.

The previous behavior with the `StringRedisSerializer` can be reinstated with `serializer` injection.

Doc Polishing.
This commit is contained in:
Artem Bilan
2016-01-27 17:32:22 -05:00
committed by Gary Russell
parent f762f2c3f0
commit d2eba0927d
6 changed files with 22 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,6 +25,7 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor;
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.channel.MessagePublishingErrorHandler;
@@ -69,7 +70,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
private volatile Executor taskExecutor;
private volatile RedisSerializer<?> serializer = new StringRedisSerializer();
private volatile RedisSerializer<?> serializer = new JdkSerializationRedisSerializer();
private volatile long receiveTimeout = DEFAULT_RECEIVE_TIMEOUT;
@@ -84,7 +85,7 @@ public class RedisQueueInboundGateway extends MessagingGatewaySupport implements
private volatile Runnable stopCallback;
/**
* @param queueName Must not be an empty String
* @param queueName Must not be an empty String
* @param connectionFactory Must not be null
*/
public RedisQueueInboundGateway(String queueName, RedisConnectionFactory connectionFactory) {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors
* Copyright 2014-2016 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.
@@ -84,7 +84,7 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand
@Override
public String getComponentType() {
return "redis:queue-outbound-gatewway";
return "redis:queue-outbound-gateway";
}
@Override
@@ -111,7 +111,7 @@ public class RedisQueueOutboundGateway extends AbstractReplyProducingMessageHand
BoundListOperations<String, Object> boundListOperations = template.boundListOps(uuid + QUEUE_NAME_SUFFIX);
byte[] reply = (byte[]) boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
if(reply != null && reply.length > 0) {
if (reply != null && reply.length > 0) {
Object replyMessage = this.serializer.deserialize(reply);
if (replyMessage == null) {
return null;

9
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests-context.xml
View File

@@ -28,20 +28,15 @@
queue="#{redisQueue.toString()}"
reply-timeout="1000"
requires-reply="true"
reply-channel="outputChannel"
serializer="serializer"/>
<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
reply-channel="outputChannel"/>
<int-redis:queue-inbound-gateway id="inboundGateway"
queue="#{redisQueue.toString()}"
request-channel="requestChannel"
serializer="serializer"
reply-timeout="20001"
receive-timeout="100"
request-timeout="20000"/>
<int:service-activator input-channel="requestChannel" expression="payload.toUpperCase()"/>
<int:service-activator input-channel="requestChannel" expression="payload + 1"/>
</beans>

8
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueGatewayIntegrationTests.java
View File

@@ -78,10 +78,10 @@ public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests {
@Test
@RedisAvailable
public void testRequestWithReply() throws Exception {
this.sendChannel.send(new GenericMessage<String>("test1"));
this.sendChannel.send(new GenericMessage<Integer>(1));
Message<?> receive = this.outputChannel.receive(10000);
assertNotNull(receive);
assertEquals("test1".toUpperCase(), receive.getPayload());
assertEquals(2, receive.getPayload());
}
@Test
@@ -124,10 +124,10 @@ public class RedisQueueGatewayIntegrationTests extends RedisAvailableTests {
this.inboundGateway.setExtractPayload(false);
this.outboundGateway.setSerializer(new JdkSerializationRedisSerializer());
this.outboundGateway.setExtractPayload(false);
this.sendChannel.send(new GenericMessage<String>("test1"));
this.sendChannel.send(new GenericMessage<Integer>(2));
Message<?> receive = this.outputChannel.receive(10000);
assertNotNull(receive);
assertEquals("test1".toUpperCase(), receive.getPayload());
assertEquals(3, receive.getPayload());
this.inboundGateway.setSerializer(new StringRedisSerializer());
this.inboundGateway.setExtractPayload(true);
this.outboundGateway.setSerializer(new StringRedisSerializer());