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

INT-2638 added support for Redis inbound channel adapters

Browse Source 
Currenly there is support for:
- list-inbound-channel-adapter - http://redis.io/commands#list
- zset-inbound-channel-adapter - http://redis.io/commands#sorted_set

INT-2638 polishing

INT-2638 polished schema docs

INT-2638 polishing

INT-2638 polishing

INT-2638 refactored Redis collection inbound adapters to use RedisStore (view)

INT-2638 polishing, exposed redis-template via the namespace

INT-2638 polished based on PR comments

INT-2638 polishing

INT-2638 Polishing
This commit is contained in:
Oleg Zhurakousky
2012-07-25 07:59:30 -04:00
committed by Gary Russell
parent 9423c21e7a
commit bfb5cbdb2a
12 changed files with 875 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -17,6 +17,10 @@ import static org.springframework.beans.factory.xml.AbstractBeanDefinitionParser
import java.util.List;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.config.RuntimeBeanReference;
@@ -24,9 +28,12 @@ import org.springframework.beans.factory.config.TypedStringValue;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.core.Conventions;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.endpoint.AbstractPollingEndpoint;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.MatchAlwaysTransactionAttributeSource;
@@ -34,9 +41,6 @@ import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
/**
* Shared utility methods for integration namespace parsers.
@@ -379,4 +383,36 @@ public abstract class IntegrationNamespaceUtils {
return adviceChain;
}
public static RootBeanDefinition createExpressionDefinitionFromValueOrExpression(String valueElementName,
String expressionElementName, ParserContext parserContext, Element element, boolean oneRequired) {
Assert.hasText(valueElementName, "'valueElementName' must not be empty");
Assert.hasText(expressionElementName, "'expressionElementName' must no be empty");
String valueElementValue = element.getAttribute(valueElementName);
String expressionElementValue = element.getAttribute(expressionElementName);
boolean hasAttributeValue = StringUtils.hasText(valueElementValue);
boolean hasAttributeExpression = StringUtils.hasText(expressionElementValue);
if (hasAttributeValue && hasAttributeExpression){
parserContext.getReaderContext().error("Only one of '" + valueElementName + "' or '"
+ expressionElementName + "' is allowed", element);
}
if (oneRequired && (!hasAttributeValue && !hasAttributeExpression)){
parserContext.getReaderContext().error("One of '" + valueElementName + "' or '"
+ expressionElementName + "' is required", element);
}
RootBeanDefinition expressionDef = null;
if (hasAttributeValue) {
expressionDef = new RootBeanDefinition(LiteralExpression.class);
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(valueElementValue);
}
else if (hasAttributeExpression){
expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(expressionElementValue);
}
return expressionDef;
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -152,11 +152,15 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme
Object resource = NO_TX_RESOURCE;
if (this.isPseudoTxMessageSource) {
messageSource = (PseudoTransactionalMessageSource<?,Object>) this.source;
resource = messageSource.getResource();
}
Message<?> message;
try {
message = this.source.receive();
if (this.isPseudoTxMessageSource) {
Object actualResource = messageSource.getResource();
resource = actualResource != null ? actualResource : resource;
}
if (TransactionSynchronizationManager.isActualTransactionActive()) {
TransactionSynchronizationManager.bindResource(this, resource);
TransactionSynchronizationManager.registerSynchronization(

20
spring-integration-core/src/main/java/org/springframework/integration/util/ExpressionUtils.java
View File

@@ -15,17 +15,22 @@
*/
package org.springframework.integration.util;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.expression.BeanFactoryResolver;
import org.springframework.context.expression.MapAccessor;
import org.springframework.core.convert.ConversionService;
import org.springframework.expression.BeanResolver;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.expression.spel.support.StandardTypeConverter;
import org.springframework.integration.context.IntegrationContextUtils;
/**
* Utility class with static methods for helping with establishing environments for
* SpEL expressions.
*
* @author Gary Russell
* @author Oleg Zhurakousky
* @since 2.2
*
*/
@@ -83,4 +88,19 @@ public class ExpressionUtils {
}
return evaluationContext;
}
/**
* Creates {@link BeanFactoryResolver}, extracts {@link ConversionService} and delegates to
* {@link #createStandardEvaluationContext(BeanResolver, ConversionService)}
*
* @param beanFactory
* @return
*/
public static StandardEvaluationContext createStandardEvaluationContext(BeanFactory beanFactory) {
ConversionService conversionService = IntegrationContextUtils.getConversionService(beanFactory);
if (conversionService == null && beanFactory instanceof ConfigurableListableBeanFactory){
conversionService = ((ConfigurableListableBeanFactory)beanFactory).getConversionService();
}
return createStandardEvaluationContext(new BeanFactoryResolver(beanFactory), conversionService);
}
}

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

@@ -0,0 +1,72 @@
/*
* Copyright 2002-2012 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.BeanMetadataElement;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.redis.inbound.RedisStoreMessageSource;
import org.springframework.util.StringUtils;
/**
* Parser for Redis store inbound adapters
*
* @author Oleg Zhurakousky
* @since 2.2
*/
public class RedisCollectionsInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(RedisStoreMessageSource.class);
String redisTemplate = element.getAttribute("redis-template");
String connectionFactory = element.getAttribute("connection-factory");
if (StringUtils.hasText(redisTemplate) && StringUtils.hasText(connectionFactory)){
parserContext.getReaderContext().error("Only one of '" + redisTemplate + "' or '"
+ connectionFactory + "' is allowed", element);
}
if (StringUtils.hasText(redisTemplate)){
builder.addConstructorArgReference(redisTemplate);
}
else {
if (!StringUtils.hasText(connectionFactory)) {
connectionFactory = "redisConnectionFactory";
}
builder.addConstructorArgReference(connectionFactory);
}
boolean atLeastOneRequired = true;
RootBeanDefinition expressionDef =
IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("key", "key-expression",
parserContext, element, atLeastOneRequired);
builder.addConstructorArgValue(expressionDef);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "collection-type");
String beanName = BeanDefinitionReaderUtils.registerWithGeneratedName(
builder.getBeanDefinition(), parserContext.getRegistry());
return new RuntimeBeanReference(beanName);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2012 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.
@@ -19,7 +19,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
/**
* Namespace handler for Spring Integration's 'redis' namespace.
*
*
* @author Oleg Zhurakousky
* @since 2.1
*/
@@ -28,7 +28,7 @@ public class RedisNamespaceHandler extends AbstractIntegrationNamespaceHandler {
public void init() {
registerBeanDefinitionParser("publish-subscribe-channel", new RedisChannelParser());
registerBeanDefinitionParser("inbound-channel-adapter", new RedisInboundChannelAdapterParser());
registerBeanDefinitionParser("store-inbound-channel-adapter", new RedisCollectionsInboundChannelAdapterParser());
registerBeanDefinitionParser("outbound-channel-adapter", new RedisOutboundChannelAdapterParser());
}
}
}

169
spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java Normal file
View File

@@ -0,0 +1,169 @@
/*
* Copyright 2002-2012 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.inbound;
import java.util.Collection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean;
import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType;
import org.springframework.data.redis.support.collections.RedisStore;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.Message;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.PseudoTransactionalMessageSource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.ExpressionUtils;
import org.springframework.util.Assert;
/**
* Inbound channel adapter which returns a Message representing a view into
* a Redis store. The type of store depends on the {@link #collectionType} attribute.
* Default is LIST. This adapter supports 5 types of collections identified by
* {@link CollectionType}
*
* @author Oleg Zhurakousky
* @since 2.2
*/
public class RedisStoreMessageSource extends IntegrationObjectSupport
implements PseudoTransactionalMessageSource<RedisStore, RedisStore> {
private final ThreadLocal<RedisStore> resourceHolder = new ThreadLocal<RedisStore>();
private volatile StandardEvaluationContext evaluationContext;
private volatile Expression keyExpression;
private volatile CollectionType collectionType = CollectionType.LIST;
private final RedisTemplate<String, ?> redisTemplate;
/**
* Creates this instance with provided {@link RedisTemplate} and SpEL expression
* which should resolve to a 'key' name of the collection to be used.
* It assumes that {@link RedisTemplate} is fully initialized and ready to be used.
* The 'keyExpression' will be evaluated on every call to the {@link #receive()} method.
*
* @param redisTemplate
* @param keyExpression
*/
public RedisStoreMessageSource(RedisTemplate<String, ?> redisTemplate,
Expression keyExpression) {
Assert.notNull(keyExpression, "'keyExpression' must not be null");
Assert.notNull(redisTemplate, "'redisTemplate' must not be null");
this.redisTemplate = redisTemplate;
this.keyExpression = keyExpression;
}
/**
* Creates this instance with provided {@link RedisConnectionFactory} and SpEL expression
* which should resolve to a 'key' name of the collection to be used.
* It will create and initialize an instance of the {@link RedisTemplate} using
* {@link StringRedisSerializer} as key serializer and {@link JdkSerializationRedisSerializer} for
* value, hashKey and hashValue serializer.
*
* The 'keyExpression' will be evaluated on every call to the {@link #receive()} method.
*
* @param connectionFactory
* @param keyExpression
*/
public RedisStoreMessageSource(RedisConnectionFactory connectionFactory,
Expression keyExpression) {
Assert.notNull(keyExpression, "'keyExpression' must not be null");
Assert.notNull(connectionFactory, "'connectionFactory' must not be null");
RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
redisTemplate.setConnectionFactory(connectionFactory);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
redisTemplate.setHashKeySerializer(new JdkSerializationRedisSerializer());
redisTemplate.setHashValueSerializer(new JdkSerializationRedisSerializer());
this.redisTemplate = redisTemplate;
this.keyExpression = keyExpression;
}
public void setCollectionType(CollectionType collectionType) {
this.collectionType = collectionType;
}
/**
* Returns a Message with the view into a {@link RedisStore} identified
* by {@link #keyExpression}
*/
@SuppressWarnings("unchecked")
public Message<RedisStore> receive() {
String key = this.keyExpression.getValue(this.evaluationContext, String.class);
Assert.hasText(key, "Failed to determine the key for the collection");
RedisStore store = this.createStoreView(key);
this.resourceHolder.set(store);
if (store instanceof Collection<?> && ((Collection<Object>)store).size() < 1){
return null;
}
else {
return MessageBuilder.withPayload(store).build();
}
}
private RedisStore createStoreView(String key){
RedisCollectionFactoryBean fb = new RedisCollectionFactoryBean();
fb.setKey(key);
fb.setTemplate(this.redisTemplate);
fb.setType(this.collectionType);
fb.afterPropertiesSet();
return fb.getObject();
}
@Override
protected void onInit() throws Exception {
if (this.getBeanFactory() != null){
this.evaluationContext =
ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
}
else {
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext();
}
}
public RedisStore getResource() {
return resourceHolder.get();
}
public void afterCommit(Object object) {
this.resourceHolder.remove();
}
public void afterRollback(Object object) {
this.resourceHolder.remove();
}
public void afterReceiveNoTx(RedisStore resource) {
this.resourceHolder.remove();
}
public void afterSendNoTx(RedisStore resource) {
this.resourceHolder.remove();
}
}

148
spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd
View File

@@ -28,7 +28,7 @@
<xsd:attribute name="topic-name" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>
Name of the Redis topic that backs this channel.
Name of the Redis topic that backs this channel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
@@ -119,24 +119,6 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<!-- The following two attributes should be added once configurable on the RedisMessageListenerContainer -->
<!-- <xsd:attribute name="auto-startup" type="xsd:string"> -->
<!-- <xsd:annotation> -->
<!-- <xsd:documentation><![CDATA[ -->
<!-- Boolean value indicating whether this channel's listener container should start automatically. -->
<!-- ]]></xsd:documentation> -->
<!-- </xsd:annotation> -->
<!-- </xsd:attribute> -->
<!-- <xsd:attribute name="phase" type="xsd:string"> -->
<!-- <xsd:annotation> -->
<!-- <xsd:documentation><![CDATA[ -->
<!-- The lifecycle phase within which this channel's listener container should start and stop. -->
<!-- The lower the value the earlier this container will start and the later it will stop. The -->
<!-- default is Integer.MAX_VALUE meaning the container will start as late as possible -->
<!-- and stop as soon as possible. -->
<!-- ]]></xsd:documentation> -->
<!-- </xsd:annotation> -->
<!-- </xsd:attribute> -->
<xsd:attributeGroup ref="integration:subscribersAttributeGroup" />
</xsd:complexType>
@@ -177,7 +159,7 @@
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Channel to which Messages will be sent.
Channel to which Messages will be sent.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -262,6 +244,68 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="store-inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines Redis inbound channel adapter that creates a Message which contains
a view into a redis store. THe view could be one of the subclasses of
org.springframework.data.redis.support.collections.RedisStore
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="redisCollectionInboundAdapterType">
<xsd:attribute name="collection-type">
<xsd:annotation>
<xsd:documentation>
Collection type supported by this adapter
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:restriction base="xsd:string">
<xsd:enumeration value="LIST">
<xsd:annotation>
<xsd:documentation>
[DEFAULT] Redis List
</xsd:documentation>
</xsd:annotation>
</xsd:enumeration>
<xsd:enumeration value="SET">
<xsd:annotation>
<xsd:documentation>
Redis Set
</xsd:documentation>
</xsd:annotation>
</xsd:enumeration>
<xsd:enumeration value="ZSET">
<xsd:annotation>
<xsd:documentation>
Redis Sorted Set
</xsd:documentation>
</xsd:annotation>
</xsd:enumeration>
<xsd:enumeration value="MAP">
<xsd:annotation>
<xsd:documentation>
Redis Map
</xsd:documentation>
</xsd:annotation>
</xsd:enumeration>
<xsd:enumeration value="PROPERTIES">
<xsd:annotation>
<xsd:documentation>
Redis Properties
</xsd:documentation>
</xsd:annotation>
</xsd:enumeration>
</xsd:restriction>
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="redisAdapterType">
<xsd:annotation>
<xsd:documentation>
@@ -281,4 +325,68 @@
<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
</xsd:complexType>
<xsd:complexType name="redisCollectionInboundAdapterType">
<xsd:complexContent>
<xsd:extension base="redisAdapterType">
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="redis-template" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
RedisTemplate to be used with this adapter
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.data.redis.core.RedisTemplate"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Channel to which Messages will be sent.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key-expression" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
SpEL expression which returns the name of the key for the collection being used. If you want to provide a
literal string value use the 'key' attribute.
This attribute is mutually exclusive with the 'key' attribute.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="key" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The name of the key for the collection being used. If you require a key to be dynamically
determined per each poll use the 'key-expression' attribute.
This attribute is mutually exclusive with the 'key-expression' attribute.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Channel to which Error Messages will be sent.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:schema>

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

@@ -0,0 +1,238 @@
/*
* Copyright 2002-2012 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.assertNotNull;
import static org.junit.Assert.assertNull;
import org.junit.Test;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.support.collections.RedisList;
import org.springframework.data.redis.support.collections.RedisZSet;
import org.springframework.integration.Message;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
/**
* @author Oleg Zhurakousky
* @since 2.2
*/
public class RedisCollectionsInboundChannelAdapterParserTests extends RedisAvailableTests{
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
public void testListInboundConfiguration() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
this.prepareList(jcf);
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("listAdapter", SourcePollingChannelAdapter.class);
spca.start();
QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(13, message.getPayload().size());
//poll again, should get the same stuff
message = (Message<RedisList<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(13, message.getPayload().size());
context.close();
}
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
public void testListInboundConfigurationWithSynchronization() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
this.prepareList(jcf);
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronization", SourcePollingChannelAdapter.class);
spca.start();
QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(13, message.getPayload().size());
//poll again, should get nothing since the collection was removed during synchronization
message = (Message<RedisList<Object>>) redisChannel.receive(1000);
assertNull(message);
spca.stop();
context.close();
}
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
public void testListInboundConfigurationWithSynchronizationAndTemplate() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
this.prepareList(jcf);
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass());
SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronizationAndRedisTemplate", SourcePollingChannelAdapter.class);
spca.start();
QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
Message<RedisList<Object>> message = (Message<RedisList<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(13, message.getPayload().size());
//poll again, should get nothing since the collection was removed during synchronization
message = (Message<RedisList<Object>>) redisChannel.receive(1000);
assertNull(message);
spca.stop();
context.close();
}
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
public void testZsetInboundConfiguration(){
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
this.prepareZset(jcf);
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
SourcePollingChannelAdapter zsetAdapterNoScore =
context.getBean("zsetAdapterNoScore", SourcePollingChannelAdapter.class);
zsetAdapterNoScore.start();
QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(13, message.getPayload().size());
//poll again, should get the same stuff
message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(13, message.getPayload().size());
zsetAdapterNoScore.stop();
context.close();
}
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
public void testZsetInboundConfigurationWithScoreRange(){
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
this.prepareZset(jcf);
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
SourcePollingChannelAdapter zsetAdapterWithScoreRange =
context.getBean("zsetAdapterWithScoreRange", SourcePollingChannelAdapter.class);
zsetAdapterWithScoreRange.start();
QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(11, message.getPayload().rangeByScore(18, 20).size());
//poll again, should get the same stuff
message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(11, message.getPayload().rangeByScore(18, 20).size());
zsetAdapterWithScoreRange.stop();
context.close();
}
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
public void testZsetInboundConfigurationWithSingleScore(){
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
this.prepareZset(jcf);
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
SourcePollingChannelAdapter zsetAdapterWithSingleScore =
context.getBean("zsetAdapterWithSingleScore", SourcePollingChannelAdapter.class);
zsetAdapterWithSingleScore.start();
QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(2, message.getPayload().rangeByScore(18, 18).size());
//poll again, should get the same stuff
message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(2, message.getPayload().rangeByScore(18, 18).size());
zsetAdapterWithSingleScore.stop();
context.close();
}
@Test
@RedisAvailable
@SuppressWarnings("unchecked")
public void testZsetInboundConfigurationWithSingleScoreAndSynchronization() throws Exception{
JedisConnectionFactory jcf = this.getConnectionFactoryForTest();
this.prepareZset(jcf);
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("zset-inbound-adapter.xml", this.getClass());
SourcePollingChannelAdapter zsetAdapterWithSingleScoreAndSynchronization =
context.getBean("zsetAdapterWithSingleScoreAndSynchronization", SourcePollingChannelAdapter.class);
SourcePollingChannelAdapter zsetAdapterNoScore =
context.getBean("zsetAdapterNoScore", SourcePollingChannelAdapter.class);
QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class);
QueueChannel otherRedisChannel = context.getBean("otherRedisChannel", QueueChannel.class);
// get all 13 presidents
zsetAdapterNoScore.start();
Message<RedisZSet<Object>> message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(13, message.getPayload().size());
zsetAdapterNoScore.stop();
Thread.sleep(1000);
// get only presidents for 18th century
zsetAdapterWithSingleScoreAndSynchronization.start();
message = (Message<RedisZSet<Object>>) otherRedisChannel.receive(1000);
assertNotNull(message);
assertEquals(2, message.getPayload().rangeByScore(18, 18).size());
//message.getPayload().remo
zsetAdapterWithSingleScoreAndSynchronization.stop();
Thread.sleep(1000);
// ... however other elements are still available 13-2=11
zsetAdapterNoScore.start();
message = (Message<RedisZSet<Object>>) redisChannel.receive(1000);
assertNotNull(message);
assertEquals(11, message.getPayload().size());
zsetAdapterNoScore.stop();
context.close();
}
@Test(expected=BeanDefinitionParsingException.class)
public void testTemplateAndCfMutualExclusivity(){
new ClassPathXmlApplicationContext("inbound-template-cf-fail.xml", this.getClass());
}
}

24
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-template-cf-fail.xml Normal file
View File

@@ -0,0 +1,24 @@
<?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-redis="http://www.springframework.org/schema/integration/redis"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis-2.2.xsd">
<int-redis:store-inbound-channel-adapter id="listAdapterWithSynchronizationAndRedisTemplate"
redis-template="redisTemplate"
connection-factory="redisConnectionFactory"
key-expression="'presidents'"
auto-startup="false">
<int:poller fixed-rate="2000" max-messages-per-poll="10"/>
</int-redis:store-inbound-channel-adapter>
<bean id="redisConnectionFactory"
class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
<property name="port" value="7379" />
</bean>
</beans>

57
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/list-inbound-adapter.xml Normal file
View File

@@ -0,0 +1,57 @@
<?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-2.2.xsd
http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis-2.2.xsd">
<int-redis:store-inbound-channel-adapter id="listAdapter"
connection-factory="redisConnectionFactory"
key-expression="'presidents'"
channel="redisChannel"
auto-startup="false"
collection-type="LIST">
<int:poller fixed-rate="2000" max-messages-per-poll="10"/>
</int-redis:store-inbound-channel-adapter>
<int-redis:store-inbound-channel-adapter id="listAdapterWithSynchronization"
connection-factory="redisConnectionFactory"
key-expression="'presidents'"
channel="redisChannel"
auto-startup="false">
<int:poller fixed-rate="2000" max-messages-per-poll="10">
<int:pseudo-transactional on-success-expression="#resource.rename('bar')"/>
</int:poller>
</int-redis:store-inbound-channel-adapter>
<int-redis:store-inbound-channel-adapter id="listAdapterWithSynchronizationAndRedisTemplate"
redis-template="redisTemplate"
key-expression="'presidents'"
channel="redisChannel"
auto-startup="false">
<int:poller fixed-rate="2000" max-messages-per-poll="10">
<int:pseudo-transactional on-success-expression="#resource.rename('bar')"/>
</int:poller>
</int-redis:store-inbound-channel-adapter>
<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
<property name="keySerializer">
<bean class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
</property>
<property name="valueSerializer">
<bean class="org.springframework.data.redis.serializer.JdkSerializationRedisSerializer"/>
</property>
<property name="connectionFactory" ref="redisConnectionFactory"/>
</bean>
<int:channel id="redisChannel">
<int:queue/>
</int:channel>
<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
<property name="port" value="7379"/>
</bean>
</beans>

61
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml Normal file
View File

@@ -0,0 +1,61 @@
<?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-2.2.xsd
http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis-2.2.xsd">
<int-redis:store-inbound-channel-adapter id="zsetAdapterNoScore"
connection-factory="redisConnectionFactory"
key-expression="'presidents'"
channel="redisChannel"
auto-startup="false"
collection-type="ZSET">
<int:poller fixed-rate="1000" max-messages-per-poll="2"/>
</int-redis:store-inbound-channel-adapter>
<int-redis:store-inbound-channel-adapter id="zsetAdapterWithScoreRange"
connection-factory="redisConnectionFactory"
key-expression="'presidents'"
channel="redisChannel"
auto-startup="false"
collection-type="ZSET">
<int:poller fixed-rate="1000" max-messages-per-poll="2"/>
</int-redis:store-inbound-channel-adapter>
<int-redis:store-inbound-channel-adapter id="zsetAdapterWithSingleScore"
connection-factory="redisConnectionFactory"
key-expression="'presidents'"
channel="redisChannel"
auto-startup="false"
collection-type="ZSET">
<int:poller fixed-rate="1000" max-messages-per-poll="2"/>
</int-redis:store-inbound-channel-adapter>
<int-redis:store-inbound-channel-adapter id="zsetAdapterWithSingleScoreAndSynchronization"
connection-factory="redisConnectionFactory"
key-expression="'presidents'"
channel="otherRedisChannel"
auto-startup="false"
collection-type="ZSET">
<int:poller fixed-rate="1000" max-messages-per-poll="2">
<int:pseudo-transactional on-success-expression="#resource.removeByScore(18, 18)" on-success-result-channel="nullChannel"/>
</int:poller>
</int-redis:store-inbound-channel-adapter>
<int:channel id="redisChannel">
<int:queue/>
</int:channel>
<int:channel id="otherRedisChannel">
<int:queue/>
</int:channel>
<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
<property name="port" value="7379"/>
</bean>
</beans>

59
spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java
View File

@@ -18,11 +18,16 @@ package org.springframework.integration.redis.rules;
import java.util.UUID;
import org.junit.Rule;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.BoundListOperations;
import org.springframework.data.redis.core.BoundZSetOperations;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
/**
* @author Oleg Zhurakousky
@@ -31,7 +36,7 @@ import org.springframework.data.redis.core.RedisTemplate;
public class RedisAvailableTests {
@Rule
public RedisAvailableRule redisAvailableRule = new RedisAvailableRule();
@SuppressWarnings({ "rawtypes", "unchecked" })
public JedisConnectionFactory getConnectionFactoryForTest(){
JedisConnectionFactory jcf = new JedisConnectionFactory();
@@ -49,4 +54,56 @@ public class RedisAvailableTests {
});
return jcf;
}
protected void prepareList(JedisConnectionFactory jcf){
RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<Object, Object>();
redisTemplate.setConnectionFactory(jcf);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
BoundListOperations<Object, Object> ops = redisTemplate.boundListOps("presidents");
ops.rightPush("John Adams");
ops.rightPush("Barack Obama");
ops.rightPush("Thomas Jefferson");
ops.rightPush("John Quincy Adams");
ops.rightPush("Zachary Taylor");
ops.rightPush("Theodore Roosevelt");
ops.rightPush("Woodrow Wilson");
ops.rightPush("George W. Bush");
ops.rightPush("Franklin D. Roosevelt");
ops.rightPush("Ronald Reagan");
ops.rightPush("William J. Clinton");
ops.rightPush("Abraham Lincoln");
ops.rightPush("George Washington");
}
protected void prepareZset(JedisConnectionFactory jcf){
RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<Object, Object>();
redisTemplate.setConnectionFactory(jcf);
redisTemplate.setKeySerializer(new StringRedisSerializer());
redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
BoundZSetOperations<Object, Object> ops = redisTemplate.boundZSetOps("presidents");
ops.add("John Adams", 18);
ops.add("Barack Obama", 21);
ops.add("Thomas Jefferson", 19);
ops.add("John Quincy Adams", 19);
ops.add("Zachary Taylor", 19);
ops.add("Theodore Roosevelt", 20);
ops.add("Woodrow Wilson", 20);
ops.add("George W. Bush", 21);
ops.add("Franklin D. Roosevelt", 20);
ops.add("Ronald Reagan", 20);
ops.add("William J. Clinton", 20);
ops.add("Abraham Lincoln", 19);
ops.add("George Washington", 18);
}
}