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

INT-3017: Redis Queue Adapters: Namespace Support

Browse Source 
JIRA: https://jira.springsource.org/browse/INT-3017

* Add parsers tests and integration tests
* Polishing Redis XSD
* Fix `JmsMessageDrivenEndpointParser` `LifeCycle` attributes
* Catch `RedisSystemException` after `this.boundListOperations.rightPop`.
It maybe an exception about 'connection closed'.
If the `RedisQueueMessageDrivenEndpoint` is 'active' this exception is rethrown,
otherwise just logged under error category.
* Add Redis queue components docs

JIRA: https://jira.springsource.org/browse/INT-3019

INT-3017 Doc Polishing
This commit is contained in:
Artem Bilan
2013-11-04 20:42:56 +02:00
committed by Gary Russell
parent 5be8ef3fd8
commit ce1f467793
19 changed files with 993 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java
View File

@@ -172,9 +172,19 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName, String propertyName) {
String attributeValue = element.getAttribute(attributeName);
if (StringUtils.hasText(attributeValue)) {
builder.addPropertyReference(propertyName, attributeValue);
setReferenceIfAttributeDefined(builder, element, attributeName, propertyName, false);
}
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName, String propertyName, boolean emptyStringAllowed) {
if (element.hasAttribute(attributeName)) {
String attributeValue = element.getAttribute(attributeName);
if (StringUtils.hasText(attributeValue)) {
builder.addPropertyReference(propertyName, attributeValue);
}
else if (emptyStringAllowed) {
builder.addPropertyValue(propertyName, null);
}
}
}
@@ -198,8 +208,13 @@ public abstract class IntegrationNamespaceUtils {
*/
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName) {
setReferenceIfAttributeDefined(builder, element, attributeName, false);
}
public static void setReferenceIfAttributeDefined(BeanDefinitionBuilder builder, Element element,
String attributeName, boolean emptyStringAllowed) {
setReferenceIfAttributeDefined(builder, element, attributeName,
Conventions.attributeNameToPropertyName(attributeName));
Conventions.attributeNameToPropertyName(attributeName), emptyStringAllowed);
}
/**

3
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsMessageDrivenEndpointParser.java
View File

@@ -95,7 +95,8 @@ public class JmsMessageDrivenEndpointParser extends AbstractSingleBeanDefinition
String listenerBeanName = this.parseMessageListener(element, parserContext);
builder.addConstructorArgReference(containerBeanName);
builder.addConstructorArgReference(listenerBeanName);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
}
private String parseMessageListenerContainer(Element element, ParserContext parserContext) {

7
spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -21,6 +21,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
* Namespace handler for Spring Integration's 'redis' namespace.
*
* @author Oleg Zhurakousky
* @author Artem Bilan
* @since 2.1
*/
public class RedisNamespaceHandler extends AbstractIntegrationNamespaceHandler {
@@ -31,5 +32,7 @@ public class RedisNamespaceHandler extends AbstractIntegrationNamespaceHandler {
registerBeanDefinitionParser("store-inbound-channel-adapter", new RedisStoreInboundChannelAdapterParser());
registerBeanDefinitionParser("store-outbound-channel-adapter", new RedisStoreOutboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-channel-adapter", new RedisOutboundChannelAdapterParser());
registerBeanDefinitionParser("queue-inbound-channel-adapter", new RedisQueueInboundChannelAdapterParser());
registerBeanDefinitionParser("queue-outbound-channel-adapter", new RedisQueueOutboundChannelAdapterParser());
}
}
}

83
spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueInboundChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,83 @@
/*
* Copyright 2013 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.redis.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
import org.springframework.util.StringUtils;
/**
* Parser for the <queue-inbound-channel-adapter> element of the 'redis' namespace.
*
* @author Artem Bilan
* @since 3.0
*/
public class RedisQueueInboundChannelAdapterParser extends AbstractSingleBeanDefinitionParser {
@Override
protected Class<?> getBeanClass(Element element) {
return RedisQueueMessageDrivenEndpoint.class;
}
@Override
protected final String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
String id = element.getAttribute("id");
if (!element.hasAttribute("channel")) {
// the created channel will get the 'id', so the adapter's bean name includes a suffix
id = id + ".adapter";
}
else if (!StringUtils.hasText(id)) {
id = parserContext.getReaderContext().generateBeanName(definition);
}
return id;
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
builder.addConstructorArgValue(element.getAttribute("queue"));
String connectionFactory = element.getAttribute("connection-factory");
if (!StringUtils.hasText(connectionFactory)) {
connectionFactory = "redisConnectionFactory";
}
builder.addConstructorArgReference(connectionFactory);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer", true);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "expect-message");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "receive-timeout");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.AUTO_STARTUP);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IntegrationNamespaceUtils.PHASE);
String channelName = element.getAttribute("channel");
if (!StringUtils.hasText(channelName)) {
channelName = IntegrationNamespaceUtils.createDirectChannel(element, parserContext);
}
builder.addPropertyReference("outputChannel", channelName);
}
}

57
spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParser.java Normal file
View File

@@ -0,0 +1,57 @@
/*
* Copyright 2013 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.redis.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.redis.outbound.RedisQueueOutboundChannelAdapter;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;int-redis:queue-outbound-channel-adapter&gt; element.
*
* @author Artem Bilan
* @since 3.0
*/
public class RedisQueueOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(RedisQueueOutboundChannelAdapter.class);
BeanDefinition queueExpression = IntegrationNamespaceUtils
.createExpressionDefinitionFromValueOrExpression("queue", "queue-expression", parserContext, element, true);
builder.addConstructorArgValue(queueExpression);
String connectionFactory = element.getAttribute("connection-factory");
if (!StringUtils.hasText(connectionFactory)) {
connectionFactory = "redisConnectionFactory";
}
builder.addConstructorArgReference(connectionFactory);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer");
return builder.getBeanDefinition();
}
}

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

@@ -15,10 +15,11 @@
*/
package org.springframework.integration.redis.inbound;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.RedisTemplate;
@@ -31,6 +32,7 @@ import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
@@ -52,7 +54,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
private MessageChannel errorChannel;
private volatile TaskExecutor taskExecutor;
private volatile Executor taskExecutor;
private volatile RedisSerializer<?> serializer = new JdkSerializationRedisSerializer();
@@ -117,7 +119,7 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
this.receiveTimeout = receiveTimeout;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
@@ -138,7 +140,8 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
this.taskExecutor = new SimpleAsyncTaskExecutor((beanName == null ? "" : beanName + "-") + this.getComponentType());
}
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
MessagePublishingErrorHandler errorHandler =
new MessagePublishingErrorHandler(new BeanFactoryChannelResolver(this.getBeanFactory()));
errorHandler.setDefaultErrorChannel(this.errorChannel);
this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler);
}
@@ -146,14 +149,25 @@ public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport {
@Override
public String getComponentType() {
return "int-redis:message-driven-channel-adapter";
return "redis:queue-inbound-channel-adapter";
}
@SuppressWarnings("unchecked")
private void popMessageAndSend() {
Message<Object> message = null;
byte[] value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
byte[] value = null;
try {
value = this.boundListOperations.rightPop(this.receiveTimeout, TimeUnit.MILLISECONDS);
}
catch (RedisSystemException e) {
if (this.active) {
throw e;
}
else {
logger.error(e);
}
}
if (value != null) {
if (this.expectMessage) {

2
spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapter.java
View File

@@ -84,7 +84,7 @@ public class RedisQueueOutboundChannelAdapter extends AbstractMessageHandler imp
@Override
public String getComponentType() {
return "int-redis:outbound-channel-adapter";
return "redis:outbound-channel-adapter";
}
@Override

145
spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-3.0.xsd
View File

@@ -82,10 +82,7 @@
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) for executing
JMS listener invokers. Default is a SimpleAsyncTaskExecutor in case of a
DefaultMessageListenerContainer, using internally managed threads. For a
SimpleMessageListenerContainer, listeners will always get invoked within the
JMS provider's receive thread by default.
Redis listener invokers. Default is a SimpleAsyncTaskExecutor.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -329,6 +326,146 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="queue-inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines a Message Driven Endpoint for listening a Redis queue.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="redisAdapterType">
<xsd:attribute name="queue" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>
Redis queue name.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Identifies the channel to which error messages will be sent if a failure occurs in this
Endpoint's process. If no "error-channel" reference is provided, this Endpoint will
propagate Exceptions to the caller. To completely suppress Exceptions, provide a
reference to the "nullChannel" here.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="serializer" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer.
It can be specified as an empty String value, which means the Endpoint's 'serializer' property is
set to 'null', in which case the Message will contain the raw byte[] payload.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.data.redis.serializer.RedisSerializer"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="receive-timeout" type="xsd:string" default="1000">
<xsd:annotation>
<xsd:documentation>
Specify the timeout in milliseconds to wait for the result of the
'rightPop' operation on Redis queue.
Default is 1 second.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="expect-message" type="xsd:string" default="false">
<xsd:annotation>
<xsd:documentation>
When true, specifies that the 'byte[]' from a Redis message should be deserialized
as an entire Spring Integration Message. Otherwise the data becomes just the
payload of the message (deserialized or not).
If this attribute is 'true', the 'serializer' must not be an empty String.
Default is 'false'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) for executing
the listening task on the Redis queue. Default is a SimpleAsyncTaskExecutor.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.concurrent.Executor"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:element name="queue-outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines an outbound Redis Queue Message-sending Channel Adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="redisAdapterType">
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="queue" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the name of the Redis queue.
This attribute is mutually exclusive with 'queue-expression' attribute.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="queue-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the expression to determine the name of the Redis queue
against the Message at runtime.
This attribute is mutually exclusive with 'queue' attribute.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the order for invocation when this adapter is connected as a
subscriber to a SubscribableChannel.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="serializer" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.data.redis.serializer.RedisSerializer"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="extract-payload" type="xsd:string" default="true">
<xsd:annotation>
<xsd:documentation>
Specifies if the Message payload or the entire (serialized) Message will be send to the Redis queue.
Default is 'true'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="redisAdapterType">
<xsd:annotation>
<xsd:documentation>

44
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests-context.xml Normal file
View File

@@ -0,0 +1,44 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
<bean id="redisConnectionFactory"
class="org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory">
<property name="port" value="#{T(org.springframework.integration.redis.rules.RedisAvailableRule).REDIS_PORT}"/>
</bean>
<bean id="customRedisConnectionFactory" parent="redisConnectionFactory"/>
<int-redis:queue-inbound-channel-adapter id="defaultAdapter" queue="si.test.Int3017.Inbound1"/>
<int:channel id="sendChannel"/>
<int-redis:queue-inbound-channel-adapter id="customAdapter"
queue="si.test.Int3017.Inbound2"
channel="sendChannel"
connection-factory="customRedisConnectionFactory"
expect-message="true"
serializer="serializer"
error-channel="errorChannel"
receive-timeout="2000"
task-executor="executor"
auto-startup="false"
phase="100"/>
<bean id="executor" class="org.springframework.integration.util.ErrorHandlingTaskExecutor">
<constructor-arg ref="threadPoolTaskExecutor"/>
<constructor-arg value="#{T(org.springframework.scheduling.support.TaskUtils).LOG_AND_SUPPRESS_ERROR_HANDLER}"/>
</bean>
<task:executor id="threadPoolTaskExecutor" pool-size="5"/>
<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
</beans>

114
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisMessageDrivenEndpointParserTests.java Normal file
View File

@@ -0,0 +1,114 @@
/*
* Copyright 2013 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.redis.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.task.TaskExecutor;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.redis.inbound.RedisQueueMessageDrivenEndpoint;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Artem Bilan
* @since 3.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class RedisMessageDrivenEndpointParserTests {
@Autowired
@Qualifier("redisConnectionFactory")
private RedisConnectionFactory connectionFactory;
@Autowired
@Qualifier("customRedisConnectionFactory")
private RedisConnectionFactory customRedisConnectionFactory;
@Autowired
@Qualifier("defaultAdapter.adapter")
private RedisQueueMessageDrivenEndpoint defaultAdapter;
@Autowired
@Qualifier("defaultAdapter")
private MessageChannel defaultAdapterChannel;
@Autowired
@Qualifier("customAdapter")
private RedisQueueMessageDrivenEndpoint customAdapter;
@Autowired
@Qualifier("errorChannel")
private MessageChannel errorChannel;
@Autowired
@Qualifier("sendChannel")
private MessageChannel sendChannel;
@Autowired
@Qualifier("executor")
private TaskExecutor taskExecutor;
@Autowired
private RedisSerializer<?> serializer;
@Test
public void testInt3017DefaultConfig() {
assertSame(this.connectionFactory, TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.ops.template.connectionFactory"));
assertEquals("si.test.Int3017.Inbound1", TestUtils.getPropertyValue(this.defaultAdapter, "boundListOperations.key"));
assertFalse(TestUtils.getPropertyValue(this.defaultAdapter, "expectMessage", Boolean.class));
assertEquals(new Long(1000), TestUtils.getPropertyValue(this.defaultAdapter, "receiveTimeout", Long.class));
assertNull(TestUtils.getPropertyValue(this.defaultAdapter, "errorChannel"));
assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "taskExecutor"), Matchers.instanceOf(ErrorHandlingTaskExecutor.class));
assertThat(TestUtils.getPropertyValue(this.defaultAdapter, "serializer"), Matchers.instanceOf(JdkSerializationRedisSerializer.class));
assertTrue(TestUtils.getPropertyValue(this.defaultAdapter, "autoStartup", Boolean.class));
assertSame(this.defaultAdapterChannel, TestUtils.getPropertyValue(this.defaultAdapter, "outputChannel"));
}
@Test
public void testInt3017CustomConfig() {
assertSame(this.customRedisConnectionFactory, TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.ops.template.connectionFactory"));
assertEquals("si.test.Int3017.Inbound2", TestUtils.getPropertyValue(this.customAdapter, "boundListOperations.key"));
assertTrue(TestUtils.getPropertyValue(this.customAdapter, "expectMessage", Boolean.class));
assertEquals(new Long(2000), TestUtils.getPropertyValue(this.customAdapter, "receiveTimeout", Long.class));
assertSame(this.errorChannel, TestUtils.getPropertyValue(this.customAdapter, "errorChannel"));
assertSame(this.taskExecutor, TestUtils.getPropertyValue(this.customAdapter, "taskExecutor"));
assertSame(this.serializer, TestUtils.getPropertyValue(this.customAdapter, "serializer"));
assertFalse(TestUtils.getPropertyValue(this.customAdapter, "autoStartup", Boolean.class));
assertEquals(new Integer(100), TestUtils.getPropertyValue(this.customAdapter, "phase", Integer.class));
assertSame(this.sendChannel, TestUtils.getPropertyValue(this.customAdapter, "outputChannel"));
}
}

29
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParserTests-context.xml Normal file
View File

@@ -0,0 +1,29 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd">
<bean id="redisConnectionFactory"
class="org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory">
<property name="port" value="#{T(org.springframework.integration.redis.rules.RedisAvailableRule).REDIS_PORT}"/>
</bean>
<bean id="customRedisConnectionFactory" parent="redisConnectionFactory"/>
<int:channel id="sendChannel"/>
<int-redis:queue-outbound-channel-adapter id="defaultAdapter" channel="sendChannel" queue="foo"/>
<int-redis:queue-outbound-channel-adapter id="customAdapter" channel="sendChannel"
queue-expression="headers['redis_queue']"
extract-payload="false"
serializer="serializer"
connection-factory="customRedisConnectionFactory"/>
<bean id="serializer" class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
</beans>

83
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisQueueOutboundChannelAdapterParserTests.java Normal file
View File

@@ -0,0 +1,83 @@
/*
* Copyright 2013 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.redis.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.expression.Expression;
import org.springframework.integration.redis.outbound.RedisQueueOutboundChannelAdapter;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Artem Bilan
* @since 3.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class RedisQueueOutboundChannelAdapterParserTests {
@Autowired
@Qualifier("redisConnectionFactory")
private RedisConnectionFactory connectionFactory;
@Autowired
@Qualifier("customRedisConnectionFactory")
private RedisConnectionFactory customRedisConnectionFactory;
@Autowired
@Qualifier("defaultAdapter.handler")
private RedisQueueOutboundChannelAdapter defaultAdapter;
@Autowired
@Qualifier("customAdapter.handler")
private RedisQueueOutboundChannelAdapter customAdapter;
@Autowired
private RedisSerializer<?> serializer;
@Test
public void testInt3017DefaultConfig() {
assertSame(this.connectionFactory, TestUtils.getPropertyValue(this.defaultAdapter, "template.connectionFactory"));
assertEquals("foo", TestUtils.getPropertyValue(this.defaultAdapter, "queueNameExpression", Expression.class).getExpressionString());
assertTrue(TestUtils.getPropertyValue(this.defaultAdapter, "extractPayload", Boolean.class));
assertFalse(TestUtils.getPropertyValue(this.defaultAdapter, "serializerExplicitlySet", Boolean.class));
}
@Test
public void testInt3017CustomConfig() {
assertSame(this.customRedisConnectionFactory, TestUtils.getPropertyValue(this.customAdapter, "template.connectionFactory"));
assertEquals("headers['redis_queue']", TestUtils.getPropertyValue(this.customAdapter, "queueNameExpression", Expression.class).getExpressionString());
assertFalse(TestUtils.getPropertyValue(this.customAdapter, "extractPayload", Boolean.class));
assertTrue(TestUtils.getPropertyValue(this.customAdapter, "serializerExplicitlySet", Boolean.class));
assertSame(this.serializer, TestUtils.getPropertyValue(this.customAdapter, "serializer"));
}
}

44
spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisQueueMessageDrivenEndpointTests-context.xml Normal file
View File

@@ -0,0 +1,44 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">
<bean id="redisConnectionFactory"
class="org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory">
<property name="port" value="#{T(org.springframework.integration.redis.rules.RedisAvailableRule).REDIS_PORT}"/>
</bean>
<int:channel id="fromChannel">
<int:queue/>
</int:channel>
<int-redis:queue-inbound-channel-adapter queue="si.test.Int3017IntegrationInbound"
channel="fromChannel"
expect-message="true"
serializer="testSerializer"/>
<bean id="testSerializer" class="org.springframework.integration.redis.util.CustomJsonSerializer"/>
<int:chain input-channel="symmetricalInputChannel">
<int:payload-serializing-transformer/>
<int-redis:queue-outbound-channel-adapter queue-expression="headers.redis_queue"/>
</int:chain>
<int-redis:queue-inbound-channel-adapter queue="si.test.Int3017IntegrationSymmetrical"
channel="symmetricalRedisChannel"
serializer=""/>
<int:payload-deserializing-transformer input-channel="symmetricalRedisChannel" output-channel="symmetricalOutputChannel"/>
<int:channel id="symmetricalOutputChannel">
<int:queue/>
</int:channel>
</beans>

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

@@ -21,15 +21,22 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import java.util.Date;
import java.util.UUID;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.PollableChannel;
@@ -37,14 +44,31 @@ import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gunnar Hillert
* @author Artem Bilan
* @since 3.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
@Autowired
private RedisConnectionFactory connectionFactory;
@Autowired
private PollableChannel fromChannel;
@Autowired
private MessageChannel symmetricalInputChannel;
@Autowired
private PollableChannel symmetricalOutputChannel;
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
@@ -52,10 +76,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
String queueName = "si.test.redisQueueInboundChannelAdapterTests";
RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
redisTemplate.setConnectionFactory(connectionFactory);
redisTemplate.setConnectionFactory(this.connectionFactory);
redisTemplate.setEnableDefaultSerializer(false);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
@@ -71,7 +93,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
PollableChannel channel = new QueueChannel();
RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, connectionFactory);
RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
endpoint.setOutputChannel(channel);
endpoint.setReceiveTimeout(1000);
endpoint.afterPropertiesSet();
@@ -86,7 +109,6 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
assertEquals(payload2, receive.getPayload());
endpoint.stop();
this.waitUntilListening(endpoint);
}
@Test
@@ -96,10 +118,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
final String queueName = "si.test.redisQueueInboundChannelAdapterTests2";
RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
redisTemplate.setConnectionFactory(connectionFactory);
redisTemplate.setConnectionFactory(this.connectionFactory);
redisTemplate.setEnableDefaultSerializer(false);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
@@ -115,7 +135,8 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
PollableChannel errorChannel = new QueueChannel();
RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, connectionFactory);
RedisQueueMessageDrivenEndpoint endpoint = new RedisQueueMessageDrivenEndpoint(queueName, this.connectionFactory);
endpoint.setBeanFactory(Mockito.mock(BeanFactory.class));
endpoint.setExpectMessage(true);
endpoint.setOutputChannel(channel);
endpoint.setErrorChannel(errorChannel);
@@ -139,19 +160,38 @@ public class RedisQueueMessageDrivenEndpointTests extends RedisAvailableTests {
endpoint.stop();
this.waitUntilListening(endpoint);
}
@Test
@RedisAvailable
public void testInt3017IntegrationInbound() throws Exception {
public void waitUntilListening(RedisQueueMessageDrivenEndpoint endpoint) throws Exception {
int n = 0;
while (endpoint.isListening()) {
Thread.sleep(100);
if (n++ > 100) {
throw new Exception("RedisQueueMessageDrivenEndpoint failed to stop.");
}
}
String payload = new Date().toString();
RedisTemplate<String, String> redisTemplate = new StringRedisTemplate();
redisTemplate.setConnectionFactory(this.connectionFactory);
redisTemplate.afterPropertiesSet();
redisTemplate.boundListOps("si.test.Int3017IntegrationInbound").leftPush("{\"payload\":\"" + payload + "\",\"headers\":{}}");
Message<?> receive = this.fromChannel.receive(2000);
assertNotNull(receive);
assertEquals(payload, receive.getPayload());
}
@Test
@RedisAvailable
public void testInt3017IntegrationSymmetrical() throws Exception {
UUID payload = UUID.randomUUID();
Message<UUID> message = MessageBuilder.withPayload(payload)
.setHeader("redis_queue", "si.test.Int3017IntegrationSymmetrical")
.build();
this.symmetricalInputChannel.send(message);
Message<?> receive = this.symmetricalOutputChannel.receive(2000);
assertNotNull(receive);
assertEquals(payload, receive.getPayload());
}
}

23
spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests-context.xml Normal file
View File

@@ -0,0 +1,23 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd">
<bean id="redisConnectionFactory"
class="org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory">
<property name="port" value="#{T(org.springframework.integration.redis.rules.RedisAvailableRule).REDIS_PORT}"/>
</bean>
<int:chain input-channel="toRedisQueueChannel">
<int-redis:queue-outbound-channel-adapter queue-expression="payload"
extract-payload="false"
serializer="testSerializer"/>
</int:chain>
<bean id="testSerializer" class="org.springframework.integration.redis.util.CustomJsonSerializer"/>
</beans>

59
spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisQueueOutboundChannelAdapterTests.java
View File

@@ -24,7 +24,10 @@ import java.util.Date;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
@@ -32,33 +35,47 @@ import org.springframework.data.redis.serializer.JacksonJsonRedisSerializer;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.json.Jackson2JsonMessageParser;
import org.springframework.integration.support.json.JsonInboundMessageMapper;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gunnar Hillert
* @author Artem Bilan
* @since 3.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests {
@Autowired
private RedisConnectionFactory connectionFactory;
@Autowired
@Qualifier("toRedisQueueChannel")
private MessageChannel sendChannel;
@Test
@RedisAvailable
public void testInt3015Default() throws Exception {
final String queueName = "si.test.testRedisQueueOutboundChannelAdapter";
RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory);
final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, this.connectionFactory);
String payload = "testing";
handler.handleMessage(MessageBuilder.withPayload(payload).build());
RedisTemplate<String, ?> redisTemplate = new StringRedisTemplate();
redisTemplate.setConnectionFactory(connectionFactory);
redisTemplate.setConnectionFactory(this.connectionFactory);
redisTemplate.afterPropertiesSet();
Object result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS);
@@ -70,7 +87,7 @@ public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests {
handler.handleMessage(MessageBuilder.withPayload(payload2).build());
RedisTemplate<String, ?> redisTemplate2 = new RedisTemplate<String, Object>();
redisTemplate2.setConnectionFactory(connectionFactory);
redisTemplate2.setConnectionFactory(this.connectionFactory);
redisTemplate2.setEnableDefaultSerializer(false);
redisTemplate2.setKeySerializer(new StringRedisSerializer());
redisTemplate2.setValueSerializer(new JdkSerializationRedisSerializer());
@@ -88,16 +105,14 @@ public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests {
final String queueName = "si.test.testRedisQueueOutboundChannelAdapter2";
RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory);
final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, this.connectionFactory);
handler.setExtractPayload(false);
Message<String> message = MessageBuilder.withPayload("testing").build();
handler.handleMessage(message);
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
redisTemplate.setConnectionFactory(connectionFactory);
redisTemplate.setConnectionFactory(this.connectionFactory);
redisTemplate.setEnableDefaultSerializer(false);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
@@ -116,13 +131,11 @@ public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests {
final String queueName = "si.test.testRedisQueueOutboundChannelAdapter2";
RedisConnectionFactory connectionFactory = this.getConnectionFactoryForTest();
final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, connectionFactory);
final RedisQueueOutboundChannelAdapter handler = new RedisQueueOutboundChannelAdapter(queueName, this.connectionFactory);
handler.setSerializer(new JacksonJsonRedisSerializer<Object>(Object.class));
RedisTemplate<String, ?> redisTemplate = new StringRedisTemplate();
redisTemplate.setConnectionFactory(connectionFactory);
redisTemplate.setConnectionFactory(this.connectionFactory);
redisTemplate.afterPropertiesSet();
handler.handleMessage(new GenericMessage<Object>(Arrays.asList("foo", "bar", "baz")));
@@ -140,4 +153,24 @@ public class RedisQueueOutboundChannelAdapterTests extends RedisAvailableTests {
assertEquals("\"test\"", result);
}
@Test
@RedisAvailable
public void testInt3017IntegrationOutbound() throws Exception {
final String queueName = "si.test.Int3017IntegrationOutbound";
GenericMessage<Object> message = new GenericMessage<Object>(queueName);
this.sendChannel.send(message);
RedisTemplate<String, String> redisTemplate = new StringRedisTemplate();
redisTemplate.setConnectionFactory(this.connectionFactory);
redisTemplate.afterPropertiesSet();
String result = redisTemplate.boundListOps(queueName).rightPop(5000, TimeUnit.MILLISECONDS);
assertNotNull(result);
InboundMessageMapper<String> mapper = new JsonInboundMessageMapper(String.class, new Jackson2JsonMessageParser());
Message<?> resultMessage = mapper.toMessage(result);
assertEquals(message.getPayload(), resultMessage.getPayload());
}
}

60
spring-integration-redis/src/test/java/org/springframework/integration/redis/util/CustomJsonSerializer.java Normal file
View File

@@ -0,0 +1,60 @@
/*
* Copyright 2013 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.redis.util;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;
import org.springframework.integration.Message;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.support.json.Jackson2JsonMessageParser;
import org.springframework.integration.support.json.JsonInboundMessageMapper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
/**
* @author Artem Bilan
* @since 3.0
*/
public class CustomJsonSerializer implements RedisSerializer<Message<?>> {
private final ObjectMapper objectMapper = new ObjectMapper();
private final InboundMessageMapper<String> mapper =
new JsonInboundMessageMapper(String.class, new Jackson2JsonMessageParser());
@Override
public byte[] serialize(Message<?> message) throws SerializationException {
try {
return this.objectMapper.writeValueAsBytes(message);
}
catch (JsonProcessingException e) {
throw new SerializationException("Fail to serialize 'message' to json.", e);
}
}
@Override
public Message<?> deserialize(byte[] bytes) throws SerializationException {
try {
return mapper.toMessage(new String(bytes));
}
catch (Exception e) {
throw new SerializationException("Fail to deserialize 'message' from json.", e);
}
}
}

156
src/reference/docbook/redis.xml
View File

@@ -179,6 +179,160 @@ rt.setConnectionFactory(redisConnectionFactory);]]></programlisting>
This example also includes the optional, custom <classname>MessageConverter</classname> (the '<code>testConverter</code>' bean).
</para>
</section>
<section id="redis-queue-inbound-channel-adapter">
<title>Redis Queue Inbound Channel Adapter</title>
<para>
Since <emphasis>Spring Integration 3.0</emphasis>, a Queue Inbound Channel Adapter
is available to 'right pop' messages from a Redis List.
The adapter is message-driven using an internal listener thread and does not use a poller.
<programlisting language="xml"><![CDATA[<int-redis:queue-inbound-channel-adapter id="" ]]><co id="redis-m-d-c-a-id"/><![CDATA[
channel="" ]]><co id="redis-m-d-c-a-channel"/><![CDATA[
auto-startup="" ]]><co id="redis-m-d-c-a-autoStartup"/><![CDATA[
phase="" ]]><co id="redis-m-d-c-a-phase"/><![CDATA[
connection-factory="" ]]><co id="redis-m-d-c-a-connectionFactory"/><![CDATA[
queue="" ]]><co id="redis-m-d-c-a-queue"/><![CDATA[
error-channel="" ]]><co id="redis-m-d-c-a-errorChannel"/><![CDATA[
serializer="" ]]><co id="redis-m-d-c-a-serializer"/><![CDATA[
receive-timeout="" ]]><co id="redis-m-d-c-a-receiveTimeout"/><![CDATA[
expect-message="" ]]><co id="redis-m-d-c-a-expectMessage"/><![CDATA[
task-executor=""/> ]]><co id="redis-m-d-c-a-task-executor"/>
</programlisting>
<calloutlist>
<callout arearefs="redis-m-d-c-a-id">
<para>
The component bean name. If the <code>channel</code> attribute isn't provided a <classname>DirectChannel</classname>
is created and registered with application context with this <code>id</code> attribute as the bean name.
In this case, the endpoint itself is registered with the bean name <code>id + '.adapter'</code>.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-channel">
<para>
The <interfacename>MessageChannel</interfacename> to which to send <interfacename>Message</interfacename>s from this Endpoint.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-autoStartup">
<para>
A <interfacename>SmartLifecycle</interfacename> attribute to specify whether this Endpoint should start automatically after
the application context start or not. Default is <code>true</code>.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-phase">
<para>
A <interfacename>SmartLifecycle</interfacename> attribute to specify the <emphasis>phase</emphasis> in which
this Endpoint will be started. Default is <code>0</code>.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-connectionFactory">
<para>
A reference to a <interfacename>RedisConnectionFactory</interfacename> bean. Defaults to
<code>redisConnectionFactory</code>.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-queue">
<para>
The name of the Redis List on which the queue-based 'right pop' operation is performed to get Redis messages.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-errorChannel">
<para>
The <interfacename>MessageChannel</interfacename> to which to send <interfacename>ErrorMessage</interfacename>s with
<interfacename>Exception</interfacename>s from the listening task of the Endpoint.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-serializer">
<para>
The <interfacename>RedisSerializer</interfacename> bean reference. Can be an empty string, which means 'no serializer'.
In this case the raw <code>byte[]</code> from the inbound Redis message is sent to the <code>channel</code> as the
<interfacename>Message</interfacename> payload. By default it is a <classname>JdkSerializationRedisSerializer</classname>.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-receiveTimeout">
<para>
The timeout in milliseconds for 'right pop' operation to wait for a Redis message from the queue. Default is 1 second.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-expectMessage">
<para>
Specify if this Endpoint expects data from the Redis queue to contain entire <interfacename>Message</interfacename>s.
If this attribute is set to <code>true</code>, the <code>serializer</code> can't be an empty string because messages
require some form of deserialization (JDK serialization by default).
Default is <code>false</code>.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-task-executor">
<para>
A reference to a Spring <interfacename>TaskExecutor</interfacename> (or standard JDK 1.5+ <interfacename>Executor</interfacename>)
bean. It is used for the underlying listening task. By default a <classname>SimpleAsyncTaskExecutor</classname>
is used.
</para>
</callout>
</calloutlist>
</para>
</section>
<section id="redis-queue-outbound-channel-adapter">
<title>Redis Queue Outbound Channel Adapter</title>
<para>
Since <emphasis>Spring Integration 3.0</emphasis>, a Queue Outbound Channel Adapter
is available to 'left push' to a Redis List from Spring Integration messages:
<programlisting language="xml"><![CDATA[<int-redis:queue-outbound-channel-adapter id="" ]]><co id="redis-q-u-c-a-id"/><![CDATA[
channel="" ]]><co id="redis-q-u-c-a-channel"/><![CDATA[
connection-factory="" ]]><co id="redis-q-u-c-a-connectionFactory"/><![CDATA[
queue="" ]]><co id="redis-q-u-c-a-queue"/><![CDATA[
queue-expression="" ]]><co id="redis-q-u-c-a-queueExpression"/><![CDATA[
serializer="" ]]><co id="redis-q-u-c-a-serializer"/><![CDATA[
extract-payload="" />]]><co id="redis-q-u-c-a-extractPayload"/>
</programlisting>
<calloutlist>
<callout arearefs="redis-q-u-c-a-id">
<para>
The component bean name. If the <code>channel</code> attribute isn't provided, a <classname>DirectChannel</classname>
is created and registered with the application context with this <code>id</code> attribute as the bean name.
In this case, the endpoint is registered with the bean name <code>id + '.adapter'</code>.
</para>
</callout>
<callout arearefs="redis-q-u-c-a-channel">
<para>
The <interfacename>MessageChannel</interfacename> from which this Endpoint receives <interfacename>Message</interfacename>s.
</para>
</callout>
<callout arearefs="redis-q-u-c-a-connectionFactory">
<para>
A reference to a <interfacename>RedisConnectionFactory</interfacename> bean. Defaults to
<code>redisConnectionFactory</code>.
</para>
</callout>
<callout arearefs="redis-q-u-c-a-queue">
<para>
The name of the Redis List on which the queue-based 'left push' operation is performed to send Redis messages.
This attribute is mutually exclusive with <code>queue-expression</code>.
</para>
</callout>
<callout arearefs="redis-q-u-c-a-queueExpression">
<para>
A SpEL <interfacename>Expression</interfacename> to determine the name of the Redis List
using the incoming <interfacename>Message</interfacename> at runtime as the <code>#root</code> variable.
This attribute is mutually exclusive with <code>queue</code>.
</para>
</callout>
<callout arearefs="redis-m-d-c-a-serializer">
<para>
A <interfacename>RedisSerializer</interfacename> bean reference.
By default it is a <classname>JdkSerializationRedisSerializer</classname>.
However, for <classname>String</classname> payloads, a <classname>StringRedisSerializer</classname>
is used, if a <code>serializer</code> reference isn't provided.
</para>
</callout>
<callout arearefs="redis-q-u-c-a-extractPayload">
<para>
Specify if this Endpoint should send just the <emphasis>payload</emphasis> to the Redis queue,
or the entire <interfacename>Message</interfacename>.
Default is <code>true
</code>.
</para>
</callout>
</calloutlist>
</para>
</section>
</section>
<section id="redis-message-store">
@@ -423,4 +577,4 @@ the serialization of values, you may want to consider providing your own
</para>
</section>
</chapter>
</chapter>

13
src/reference/docbook/whats-new.xml
View File

@@ -143,12 +143,12 @@
For more information see <xref linkend="http-namespace"/>.
</para>
</section>
<section id="3.0-redis-meta-data-store">
<title>Redis Metadata Store</title>
<section id="3.0-redis-new-components">
<title>Redis: New Components</title>
<para>
A new Redis-based
<interfacename><ulink url="http://docs.spring.io/spring-integration/docs/latest-ga/api/org/springframework/integration/store/MetadataStore.html">MetadataStore</ulink></interfacename>
implementation was added. The <classname>RedisMetadataStore</classname> can
implementation has been added. The <classname>RedisMetadataStore</classname> can
be used to maintain state of a <interfacename>MetadataStore</interfacename>
across application restarts. This new <interfacename>MetadataStore</interfacename>
implementation can be used with adapters such as:
@@ -158,7 +158,12 @@
<listitem>Feed Inbound Channel Adapter</listitem>
</itemizedlist>
<para>
For more information see <xref linkend="redis-metadata-store" />.
New queue-based components has been added. The <code>&lt;int-redis:queue-inbound-channel-adapter/&gt;</code>
and the <code>&lt;int-redis:queue-outbound-channel-adapter/&gt;</code> components are provided
to perform 'right pop' and 'left push' operations on a Redis List, respectively.
</para>
<para>
For more information see <xref linkend="redis" />.
</para>
</section>
</section>