diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java index 5c5b81f00a..28469b516f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/IntegrationNamespaceUtils.java @@ -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; + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 598ef95d23..801945db81 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -152,11 +152,15 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint impleme Object resource = NO_TX_RESOURCE; if (this.isPseudoTxMessageSource) { messageSource = (PseudoTransactionalMessageSource) 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( diff --git a/spring-integration-core/src/main/java/org/springframework/integration/util/ExpressionUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/util/ExpressionUtils.java index eb37830969..05cb5d5d5b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/util/ExpressionUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/util/ExpressionUtils.java @@ -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); + } } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParser.java new file mode 100644 index 0000000000..acd69becf7 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParser.java @@ -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); + } +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java index 86a9fc9aa5..4c67c2e7e3 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisNamespaceHandler.java @@ -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()); } - -} +} \ No newline at end of file diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java new file mode 100644 index 0000000000..a299ed6b2c --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisStoreMessageSource.java @@ -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 { + + private final ThreadLocal resourceHolder = new ThreadLocal(); + + private volatile StandardEvaluationContext evaluationContext; + + private volatile Expression keyExpression; + + private volatile CollectionType collectionType = CollectionType.LIST; + + private final RedisTemplate 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 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 redisTemplate = new RedisTemplate(); + 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 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)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(); + } +} diff --git a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd index c7677020a7..a045d2505c 100644 --- a/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd +++ b/spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.2.xsd @@ -28,7 +28,7 @@ - Name of the Redis topic that backs this channel. + Name of the Redis topic that backs this channel. @@ -119,24 +119,6 @@ - - - - - - - - - - - - - - - - - - @@ -177,7 +159,7 @@ - Channel to which Messages will be sent. + Channel to which Messages will be sent. @@ -262,6 +244,68 @@ + + + + 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 + + + + + + + + + Collection type supported by this adapter + + + + + + + + [DEFAULT] Redis List + + + + + + + Redis Set + + + + + + + Redis Sorted Set + + + + + + + Redis Map + + + + + + + Redis Properties + + + + + + + + + + + @@ -281,4 +325,68 @@ + + + + + + + + + + RedisTemplate to be used with this adapter + + + + + + + + + + + + Channel to which Messages will be sent. + + + + + + + + + + + + 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. + + + + + + + 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. + + + + + + + + Channel to which Error Messages will be sent. + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java new file mode 100644 index 0000000000..a5af6a4e07 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java @@ -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> message = (Message>) redisChannel.receive(1000); + assertNotNull(message); + assertEquals(13, message.getPayload().size()); + + //poll again, should get the same stuff + message = (Message>) 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> message = (Message>) redisChannel.receive(1000); + assertNotNull(message); + assertEquals(13, message.getPayload().size()); + + //poll again, should get nothing since the collection was removed during synchronization + message = (Message>) 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> message = (Message>) redisChannel.receive(1000); + assertNotNull(message); + assertEquals(13, message.getPayload().size()); + + //poll again, should get nothing since the collection was removed during synchronization + message = (Message>) 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> message = (Message>) redisChannel.receive(1000); + assertNotNull(message); + assertEquals(13, message.getPayload().size()); + + //poll again, should get the same stuff + message = (Message>) 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> message = (Message>) redisChannel.receive(1000); + assertNotNull(message); + assertEquals(11, message.getPayload().rangeByScore(18, 20).size()); + + //poll again, should get the same stuff + message = (Message>) 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> message = (Message>) redisChannel.receive(1000); + assertNotNull(message); + assertEquals(2, message.getPayload().rangeByScore(18, 18).size()); + + //poll again, should get the same stuff + message = (Message>) 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> message = (Message>) 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>) 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>) 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()); + } +} diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-template-cf-fail.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-template-cf-fail.xml new file mode 100644 index 0000000000..04592a7bfc --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-template-cf-fail.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/list-inbound-adapter.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/list-inbound-adapter.xml new file mode 100644 index 0000000000..020849b97a --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/list-inbound-adapter.xml @@ -0,0 +1,57 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml new file mode 100644 index 0000000000..cf15ba25d4 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/zset-inbound-adapter.xml @@ -0,0 +1,61 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java index e898c92986..8c0f898417 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/rules/RedisAvailableTests.java @@ -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 redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(jcf); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + + BoundListOperations 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 redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(jcf); + redisTemplate.setKeySerializer(new StringRedisSerializer()); + redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + + BoundZSetOperations 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); + } }