From 3e6eb40b6dfa3b69720bb0329b744649068fc87d Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 6 Sep 2012 09:22:47 -0400 Subject: [PATCH] INT-2733/4/6 Configure Redis Serializers Add support for serializers to the RedisStore Inbound adapters. Fix the serializer support for RedisStore outbound adapters to ensure that serializers cannot be set if adapter is bootstrapped with a RedisTemplate. Add support for serializers to Redis Inbound/Outbound adapters. NOTE: unlike the RedisStore adapters Redis in/out adapters only use one serializer, so the name of the attribute is 'serializer' and, on the Redis Outbound adapter, it corresponds to the RedisTemplate.valueSerializer. INT-2733/4/6 Fix RedisCollectionPopulatingMessageHandler to ensure that the checks for setting serializers also exist at the class level in case an instance is created outside of the namespace parser's control. INT-2733/4/6 polishing INT-2733/4/6 polishing INT-2733/4/6 polishing INT-2733/4/6 Polishing --- ...ollectionsInboundChannelAdapterParser.java | 2 + ...llectionsOutboundChannelAdapterParser.java | 5 +- .../RedisInboundChannelAdapterParser.java | 6 +- .../RedisOutboundChannelAdapterParser.java | 6 +- .../redis/config/RedisParserUtils.java | 47 +++ .../inbound/RedisInboundChannelAdapter.java | 19 +- .../inbound/RedisStoreMessageSource.java | 56 +++- ...disCollectionPopulatingMessageHandler.java | 35 ++- .../RedisPublishingMessageHandler.java | 30 +- .../config/spring-integration-redis-2.2.xsd | 80 +++++ ...InboundChannelAdapterIntegrationTests.java | 237 +++++++++++++++ ...tionsInboundChannelAdapterParserTests.java | 286 ++---------------- ...ionsOutboundChannelAdapterParserTests.java | 14 +- ...boundChannelAdapterParserTests-context.xml | 5 +- ...RedisInboundChannelAdapterParserTests.java | 45 +++ ...boundChannelAdapterParserTests-context.xml | 5 +- ...edisOutboundChannelAdapterParserTests.java | 13 +- .../inbound-store-adapter-parser-fail.xml | 33 ++ .../config/inbound-store-adapter-parser.xml | 38 +++ .../redis/config/list-inbound-adapter.xml | 6 - .../store-outbound-adapter-parser-fail.xml | 37 +++ .../config/store-outbound-adapter-parser.xml | 4 +- .../redis/config/zset-inbound-adapter.xml | 2 +- .../redis/rules/RedisAvailableTests.java | 9 +- 24 files changed, 683 insertions(+), 337 deletions(-) create mode 100644 spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisParserUtils.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterIntegrationTests.java create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-store-adapter-parser-fail.xml create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-store-adapter-parser.xml create mode 100644 spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter-parser-fail.xml 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 index aaa821a292..3a7dbef971 100644 --- 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 @@ -64,6 +64,8 @@ public class RedisCollectionsInboundChannelAdapterParser extends AbstractPolling IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "collection-type"); + RedisParserUtils.setRedisSerializers(StringUtils.hasText(redisTemplate), element, parserContext, builder); + 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/RedisCollectionsOutboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParser.java index b3f54fc650..cb8a131be8 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParser.java @@ -62,10 +62,7 @@ public class RedisCollectionsOutboundChannelAdapterParser extends AbstractOutbou IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "collection-type"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "extract-payload-elements"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "key-serializer"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "value-serializer"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "hash-key-serializer"); - IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "hash-value-serializer"); + RedisParserUtils.setRedisSerializers(StringUtils.hasText(redisTemplateRef), element, parserContext, builder); if (expressionDef != null){ builder.addConstructorArgValue(expressionDef); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java index 8e84e01fec..6a4add53cd 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParser.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. @@ -16,13 +16,14 @@ package org.springframework.integration.redis.config; +import org.w3c.dom.Element; + 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.AbstractChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.util.StringUtils; -import org.w3c.dom.Element; /** * @author Oleg Zhurakousky @@ -46,6 +47,7 @@ public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterPars IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer"); return builder.getBeanDefinition(); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java index 934e2d381f..b97e31572d 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParser.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. @@ -16,13 +16,14 @@ package org.springframework.integration.redis.config; +import org.w3c.dom.Element; + 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.util.StringUtils; -import org.w3c.dom.Element; /** * @author Oleg Zhurakousky @@ -42,6 +43,7 @@ public class RedisOutboundChannelAdapterParser extends AbstractOutboundChannelAd builder.addConstructorArgReference(connectionFactory); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topic", "defaultTopic"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer"); return builder.getBeanDefinition(); } diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisParserUtils.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisParserUtils.java new file mode 100644 index 0000000000..da398a8bc5 --- /dev/null +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/config/RedisParserUtils.java @@ -0,0 +1,47 @@ +/* + * Copyright 2007-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.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; + +/** + * + * @author Oleg Zhurakousky + * @since 2.2 + */ +public class RedisParserUtils { + + public static void setRedisSerializers(boolean redisTemplateSet, Element element, ParserContext parserContext, BeanDefinitionBuilder builder){ + if (redisTemplateSet){ + if (element.hasAttribute("key-serializer") || + element.hasAttribute("value-serializer") || + element.hasAttribute("hash-key-serializer") || + element.hasAttribute("hash-value-serializer")){ + + parserContext.getReaderContext().error("Serializers can not be provided if this adapter is initialized " + + "with an external RedisTemplate. You may set serializers directly on the RedisTemplate", element); + } + } + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "key-serializer"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "value-serializer"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "hash-key-serializer"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "hash-value-serializer"); + } +} diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java index 83f3c3955a..0fc7eb78be 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java @@ -23,6 +23,7 @@ import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.listener.ChannelTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer; import org.springframework.data.redis.listener.adapter.MessageListenerAdapter; +import org.springframework.data.redis.serializer.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.Message; import org.springframework.integration.endpoint.MessageProducerSupport; @@ -38,27 +39,33 @@ import org.springframework.util.Assert; public class RedisInboundChannelAdapter extends MessageProducerSupport { private final RedisMessageListenerContainer container = new RedisMessageListenerContainer(); - + private volatile MessageConverter messageConverter = new SimpleMessageConverter(); private volatile String[] topics; + private volatile RedisSerializer serializer = new StringRedisSerializer(); public RedisInboundChannelAdapter(RedisConnectionFactory connectionFactory) { Assert.notNull(connectionFactory, "connectionFactory must not be null"); this.container.setConnectionFactory(connectionFactory); } + public void setSerializer(RedisSerializer serializer) { + Assert.notNull(serializer, "'serializer' must not be null"); + this.serializer = serializer; + } public void setTopics(String... topics) { this.topics = topics; } - + public void setMessageConverter(MessageConverter messageConverter) { Assert.notNull(messageConverter, "messageConverter must not be null"); this.messageConverter = messageConverter; } + @Override public String getComponentType() { return "redis:inbound-channel-adapter"; } @@ -69,23 +76,23 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport { Assert.notEmpty(this.topics, "at least one topis is required for subscription"); MessageListenerDelegate delegate = new MessageListenerDelegate(); MessageListenerAdapter adapter = new MessageListenerAdapter(delegate); - adapter.setSerializer(new StringRedisSerializer()); + adapter.setSerializer(this.serializer); List topicList = new ArrayList(); for (String topic : this.topics) { topicList.add(new ChannelTopic(topic)); - } + } adapter.afterPropertiesSet(); this.container.addMessageListener(adapter, topicList); this.container.afterPropertiesSet(); } - + @Override protected void doStart() { super.doStart(); this.container.start(); } - + @Override protected void doStop() { super.doStop(); 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 index daee63d903..0aade29bac 100644 --- 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 @@ -21,6 +21,7 @@ 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.RedisSerializer; import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean; import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType; @@ -57,6 +58,16 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport private final RedisTemplate redisTemplate; + private volatile RedisSerializer keySerializer; + + private volatile RedisSerializer valueSerializer; + + private volatile RedisSerializer hashKeySerializer; + + private volatile RedisSerializer hashValueSerializer; + + private volatile boolean usingDefaultTemplate = true; + /** * Creates this instance with provided {@link RedisTemplate} and SpEL expression * which should resolve to a 'key' name of the collection to be used. @@ -73,6 +84,7 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport Assert.notNull(redisTemplate, "'redisTemplate' must not be null"); this.redisTemplate = redisTemplate; + this.usingDefaultTemplate = false; this.keyExpression = keyExpression; } @@ -96,15 +108,35 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport 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 setKeySerializer(RedisSerializer keySerializer) { + Assert.state(this.usingDefaultTemplate, "'keySerializer' can not be set if a RedisTemplate was explicitly provided"); + Assert.notNull(keySerializer, "'keySerializer' must not be null"); + this.keySerializer = keySerializer; + } + + public void setValueSerializer(RedisSerializer valueSerializer) { + Assert.state(this.usingDefaultTemplate, "'valueSerializer' can not be set if a RedisTemplate was explicitly provided"); + Assert.notNull(valueSerializer, "'valueSerializer' must not be null"); + this.valueSerializer = valueSerializer; + } + + public void setHashKeySerializer(RedisSerializer hashKeySerializer) { + Assert.state(this.usingDefaultTemplate, "'hashKeySerializer' can not be set if a RedisTemplate was explicitly provided"); + Assert.notNull(hashKeySerializer, "'hashKeySerializer' must not be null"); + this.hashKeySerializer = hashKeySerializer; + } + + public void setHashValueSerializer(RedisSerializer hashValueSerializer) { + Assert.state(this.usingDefaultTemplate, "'hashValueSerializer' can not be set if a RedisTemplate was explicitly provided"); + Assert.notNull(hashValueSerializer, "'hashValueSerializer' must not be null"); + this.hashValueSerializer = hashValueSerializer; + } + public void setCollectionType(CollectionType collectionType) { this.collectionType = collectionType; } @@ -152,6 +184,22 @@ public class RedisStoreMessageSource extends IntegrationObjectSupport else { this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(); } + + if (this.usingDefaultTemplate){ + if (this.keySerializer != null){ + redisTemplate.setKeySerializer(this.keySerializer); + } + if (this.valueSerializer != null){ + redisTemplate.setValueSerializer(this.valueSerializer); + } + if (this.hashKeySerializer != null){ + redisTemplate.setHashKeySerializer(this.hashKeySerializer); + } + if (this.hashValueSerializer != null){ + redisTemplate.setHashValueSerializer(this.hashValueSerializer); + } + this.redisTemplate.afterPropertiesSet(); + } } public RedisStore getResource() { diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisCollectionPopulatingMessageHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisCollectionPopulatingMessageHandler.java index 8133ced0c7..7491043fe7 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisCollectionPopulatingMessageHandler.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisCollectionPopulatingMessageHandler.java @@ -21,7 +21,6 @@ import java.util.Properties; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.data.redis.connection.RedisConnection; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.BoundSetOperations; @@ -95,6 +94,8 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand private volatile RedisSerializer hashValueSerializer; + private volatile boolean usingDefaultTemplate = true; + /** * Will construct this instance using fully created and initialized instance of * provided {@link RedisTemplate} @@ -120,6 +121,7 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand public RedisCollectionPopulatingMessageHandler(RedisTemplate redisTemplate, Expression keyExpression) { Assert.notNull(redisTemplate, "'redisTemplate' must not be null"); this.redisTemplate = redisTemplate; + this.usingDefaultTemplate = false; if (keyExpression != null) { this.keyExpression = keyExpression; } @@ -162,21 +164,25 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand } public void setKeySerializer(RedisSerializer keySerializer) { + Assert.state(this.usingDefaultTemplate, "'keySerializer' can not be set if a RedisTemplate was explicitly provided"); Assert.notNull(keySerializer, "'keySerializer' must not be null"); this.keySerializer = keySerializer; } public void setValueSerializer(RedisSerializer valueSerializer) { + Assert.state(this.usingDefaultTemplate, "'valueSerializer' can not be set if a RedisTemplate was explicitly provided"); Assert.notNull(valueSerializer, "'valueSerializer' must not be null"); this.valueSerializer = valueSerializer; } public void setHashKeySerializer(RedisSerializer hashKeySerializer) { + Assert.state(this.usingDefaultTemplate, "'hashKeySerializer' can not be set if a RedisTemplate was explicitly provided"); Assert.notNull(hashKeySerializer, "'hashKeySerializer' must not be null"); this.hashKeySerializer = hashKeySerializer; } public void setHashValueSerializer(RedisSerializer hashValueSerializer) { + Assert.state(this.usingDefaultTemplate, "'hashValueSerializer' can not be set if a RedisTemplate was explicitly provided"); Assert.notNull(hashValueSerializer, "'hashValueSerializer' must not be null"); this.hashValueSerializer = hashValueSerializer; } @@ -234,19 +240,22 @@ public class RedisCollectionPopulatingMessageHandler extends AbstractMessageHand Assert.state(!this.mapKeyExpressionExplicitlySet || (this.collectionType == CollectionType.MAP || this.collectionType == CollectionType.PROPERTIES), "'mapKeyExpression' can only be set for CollectionType.MAP or CollectionType.PROPERTIES"); - if (this.keySerializer != null){ - redisTemplate.setKeySerializer(this.keySerializer); + + if (this.usingDefaultTemplate){ + if (this.keySerializer != null){ + redisTemplate.setKeySerializer(this.keySerializer); + } + if (this.valueSerializer != null){ + redisTemplate.setValueSerializer(this.valueSerializer); + } + if (this.hashKeySerializer != null){ + redisTemplate.setHashKeySerializer(this.hashKeySerializer); + } + if (this.hashValueSerializer != null){ + redisTemplate.setHashValueSerializer(this.hashValueSerializer); + } + this.redisTemplate.afterPropertiesSet(); } - if (this.valueSerializer != null){ - redisTemplate.setValueSerializer(this.valueSerializer); - } - if (this.hashKeySerializer != null){ - redisTemplate.setHashKeySerializer(this.hashKeySerializer); - } - if (this.hashValueSerializer != null){ - redisTemplate.setHashValueSerializer(this.hashValueSerializer); - } - this.redisTemplate.afterPropertiesSet(); } /** diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java index c6c7e67f90..731e54265b 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java @@ -18,9 +18,10 @@ package org.springframework.integration.redis.outbound; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.serializer.RedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; import org.springframework.integration.Message; -import org.springframework.integration.MessagingException; -import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.support.converter.MessageConverter; import org.springframework.integration.support.converter.SimpleMessageConverter; import org.springframework.util.Assert; @@ -29,7 +30,7 @@ import org.springframework.util.Assert; * @author Mark Fisher * @since 2.1 */ -public class RedisPublishingMessageHandler implements MessageHandler { +public class RedisPublishingMessageHandler extends AbstractMessageHandler { private final StringRedisTemplate template; @@ -37,12 +38,17 @@ public class RedisPublishingMessageHandler implements MessageHandler { private volatile String defaultTopic; + private volatile RedisSerializer serializer = new StringRedisSerializer(); public RedisPublishingMessageHandler(RedisConnectionFactory connectionFactory) { Assert.notNull(connectionFactory, "connectionFactory must not be null"); this.template = new StringRedisTemplate(connectionFactory); } + public void setSerializer(RedisSerializer serializer) { + Assert.notNull(serializer, "'serializer' must not be null"); + this.serializer = serializer; + } public void setMessageConverter(MessageConverter messageConverter) { Assert.notNull(messageConverter, "messageConverter must not be null"); @@ -53,12 +59,6 @@ public class RedisPublishingMessageHandler implements MessageHandler { this.defaultTopic = defaultTopic; } - public void handleMessage(Message message) throws MessagingException { - String topic = this.determineTopic(message); - Object value = this.messageConverter.fromMessage(message); - this.template.convertAndSend(topic, value.toString()); - } - private String determineTopic(Message message) { // TODO: add support for determining topic by evaluating SpEL against the Message Assert.hasText(this.defaultTopic, "Failed to determine Redis topic " + @@ -66,4 +66,16 @@ public class RedisPublishingMessageHandler implements MessageHandler { return this.defaultTopic; } + @Override + protected void handleMessageInternal(Message message) throws Exception { + String topic = this.determineTopic(message); + Object value = this.messageConverter.fromMessage(message); + this.template.convertAndSend(topic, value.toString()); + } + + @Override + protected void onInit() throws Exception { + this.template.setValueSerializer(this.serializer); + this.template.afterPropertiesSet(); + } } 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 2b3b7799b0..fe33d3f5ce 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 @@ -191,6 +191,18 @@ ]]> + + + + + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer + + + + + + + @@ -237,6 +249,18 @@ ]]> + + + + + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer + + + + + + + @@ -394,6 +418,62 @@ + + + + + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer + to serialize the 'key' value for this collection. Please refer to the JavaDoc of the RedisTemplate + for more detail. + + + + + + + + + + + + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer + to serialize the 'value' entered into the collection. Please refer to the JavaDoc of the RedisTemplate + for more detail. + + + + + + + + + + + + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer + to be used when serializing hash keys (only relevant for hash-typed collections). + Please refer to the JavaDoc of the RedisTemplate for more detail. + + + + + + + + + + + + Reference to an instance of org.springframework.data.redis.serializer.RedisSerializer + to be used when serializing hash values (only relevant for hash-typed collections). + Please refer to the JavaDoc of the RedisTemplate for more detail. + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterIntegrationTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterIntegrationTests.java new file mode 100644 index 0000000000..ba3ecdee03 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterIntegrationTests.java @@ -0,0 +1,237 @@ +/* + * 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 RedisCollectionsInboundChannelAdapterIntegrationTests 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/RedisCollectionsInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsInboundChannelAdapterParserTests.java index e871581f98..ff242dd74d 100644 --- 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 @@ -16,288 +16,36 @@ 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 static org.junit.Assert.assertTrue; 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.core.BoundListOperations; -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.RedisList; -import org.springframework.data.redis.support.collections.RedisZSet; -import org.springframework.integration.Message; -import org.springframework.integration.MessagingException; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.core.MessageHandler; -import org.springframework.integration.core.SubscribableChannel; -import org.springframework.integration.endpoint.SourcePollingChannelAdapter; -import org.springframework.integration.redis.rules.RedisAvailable; -import org.springframework.integration.redis.rules.RedisAvailableTests; +import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType; +import org.springframework.expression.spel.standard.SpelExpression; +import org.springframework.integration.redis.inbound.RedisStoreMessageSource; +import org.springframework.integration.test.util.TestUtils; /** * @author Oleg Zhurakousky * @since 2.2 */ -public class RedisCollectionsInboundChannelAdapterParserTests extends RedisAvailableTests{ - +public class RedisCollectionsInboundChannelAdapterParserTests { @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 - public void testListInboundConfigurationWithSynchronizationAndRollback() throws Exception{ - JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.prepareList(jcf); - - RedisTemplate redisTemplate = new RedisTemplate(); - redisTemplate.setConnectionFactory(jcf); - redisTemplate.setKeySerializer(new StringRedisSerializer()); - redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); - - BoundListOperations ops = redisTemplate.boundListOps("baz"); - - assertTrue(ops.size() == 0); - - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass()); - SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronizationAndRollback", SourcePollingChannelAdapter.class); - SubscribableChannel redisFailChannel = context.getBean("redisFailChannel", SubscribableChannel.class); - redisFailChannel.subscribe(new MessageHandler() { - - public void handleMessage(Message message) throws MessagingException { - throw new MessagingException("intentional"); - } - }); - spca.start(); - - Thread.sleep(1000); - ops = redisTemplate.boundListOps("baz"); - assertTrue(ops.size() == 13); - ops = redisTemplate.boundListOps("bar"); - assertTrue(ops.size() == 0); - - 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 testListInboundConfigurationWithBeforeCommit() throws Exception{ - JedisConnectionFactory jcf = this.getConnectionFactoryForTest(); - this.prepareList(jcf); - ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("list-inbound-adapter.xml", this.getClass()); - SourcePollingChannelAdapter spca = context.getBean("listAdapterWithSynchronizationBeforeCommit", SourcePollingChannelAdapter.class); - spca.start(); - QueueChannel redisChannel = context.getBean("redisChannel", QueueChannel.class); - - QueueChannel adapterErrors = context.getBean("adapterErrors", QueueChannel.class); - - Message> message = (Message>) redisChannel.receive(1000); - assertNotNull(message); - assertNotNull(adapterErrors.receive(2000)); - 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(); + public void validateFullConfiguration(){ + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("inbound-store-adapter-parser.xml", this.getClass()); + RedisStoreMessageSource source = + TestUtils.getPropertyValue(context.getBean("adapterWithConnectionFactory"), "source", RedisStoreMessageSource.class); + assertEquals(context.getBean("keySerializer"), TestUtils.getPropertyValue(source, "keySerializer")); + assertEquals(context.getBean("valueSerializer"), TestUtils.getPropertyValue(source, "valueSerializer")); + assertEquals(context.getBean("hashKeySerializer"), TestUtils.getPropertyValue(source, "hashKeySerializer")); + assertEquals(context.getBean("hashValueSerializer"), TestUtils.getPropertyValue(source, "hashValueSerializer")); + assertEquals("'presidents'", ((SpelExpression)TestUtils.getPropertyValue(source, "keyExpression")).getExpressionString()); + assertEquals("LIST", ((CollectionType)TestUtils.getPropertyValue(source, "collectionType")).toString()); } @Test(expected=BeanDefinitionParsingException.class) - public void testTemplateAndCfMutualExclusivity(){ - new ClassPathXmlApplicationContext("inbound-template-cf-fail.xml", this.getClass()); + public void validateFailureIfTemplateAndSerializers(){ + new ClassPathXmlApplicationContext("inbound-store-adapter-parser-fail.xml", this.getClass()); } } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParserTests.java index 867c816088..2fd68d12bb 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisCollectionsOutboundChannelAdapterParserTests.java @@ -18,6 +18,7 @@ import static org.junit.Assert.assertEquals; import org.junit.Test; +import org.springframework.beans.factory.parsing.BeanDefinitionParsingException; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.data.redis.support.collections.RedisCollectionFactoryBean.CollectionType; import org.springframework.expression.common.LiteralExpression; @@ -34,11 +35,16 @@ public class RedisCollectionsOutboundChannelAdapterParserTests { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("store-outbound-adapter-parser.xml", this.getClass()); RedisCollectionPopulatingMessageHandler handler = TestUtils.getPropertyValue(context.getBean("storeAdapter.adapter"), "handler", RedisCollectionPopulatingMessageHandler.class); - assertEquals(TestUtils.getPropertyValue(handler, "keySerializer"), context.getBean("keySerializer")); - assertEquals(TestUtils.getPropertyValue(handler, "valueSerializer"), context.getBean("valueSerializer")); - assertEquals(TestUtils.getPropertyValue(handler, "hashKeySerializer"), context.getBean("hashKeySerializer")); - assertEquals(TestUtils.getPropertyValue(handler, "hashValueSerializer"), context.getBean("hashValueSerializer")); + assertEquals(context.getBean("keySerializer"), TestUtils.getPropertyValue(handler, "keySerializer")); + assertEquals(context.getBean("valueSerializer"), TestUtils.getPropertyValue(handler, "valueSerializer")); + assertEquals(context.getBean("hashKeySerializer"), TestUtils.getPropertyValue(handler, "hashKeySerializer")); + assertEquals(context.getBean("hashValueSerializer"), TestUtils.getPropertyValue(handler, "hashValueSerializer")); assertEquals("pepboys", ((LiteralExpression)TestUtils.getPropertyValue(handler, "keyExpression")).getExpressionString()); assertEquals("PROPERTIES", ((CollectionType)TestUtils.getPropertyValue(handler, "collectionType")).toString()); } + + @Test(expected=BeanDefinitionParsingException.class) + public void validateFailureIfTemplateAndSerializers(){ + new ClassPathXmlApplicationContext("store-outbound-adapter-parser-fail.xml", this.getClass()); + } } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml index 31a60b9c33..8e6c72c56a 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests-context.xml @@ -8,7 +8,8 @@ + message-converter="testConverter" + serializer="serializer"/> @@ -26,6 +27,8 @@ + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java index 0a6e39811a..13592fa843 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java @@ -26,9 +26,13 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; import org.springframework.data.redis.connection.jedis.JedisConnectionFactory; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.data.redis.serializer.StringRedisSerializer; +import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter; +import org.springframework.integration.redis.inbound.RedisStoreMessageSource; import org.springframework.integration.redis.rules.RedisAvailable; import org.springframework.integration.redis.rules.RedisAvailableTests; import org.springframework.integration.support.converter.SimpleMessageConverter; @@ -65,6 +69,7 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests{ assertEquals(errorChannelBean, accessor.getPropertyValue("errorChannel")); Object converterBean = context.getBean("testConverter"); assertEquals(converterBean, accessor.getPropertyValue("messageConverter")); + assertEquals(context.getBean("serializer"), accessor.getPropertyValue("serializer")); } @Test @@ -87,6 +92,46 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests{ assertSame(autoChannel, TestUtils.getPropertyValue(autoChannelAdapter, "outputChannel")); } + @Test(expected=IllegalStateException.class) + public void cantSetKeySerializer() { + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(new JedisConnectionFactory()); + RedisStoreMessageSource source = new RedisStoreMessageSource(redisTemplate, + new LiteralExpression("foo")); + source.setKeySerializer(new StringRedisSerializer()); + source.afterPropertiesSet(); + } + + @Test(expected=IllegalStateException.class) + public void cantSetValueSerializer() { + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(new JedisConnectionFactory()); + RedisStoreMessageSource source = new RedisStoreMessageSource(redisTemplate, + new LiteralExpression("foo")); + source.setValueSerializer(new StringRedisSerializer()); + source.afterPropertiesSet(); + } + + @Test(expected=IllegalStateException.class) + public void cantSetHashKeySerializer() { + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(new JedisConnectionFactory()); + RedisStoreMessageSource source = new RedisStoreMessageSource(redisTemplate, + new LiteralExpression("foo")); + source.setHashKeySerializer(new StringRedisSerializer()); + source.afterPropertiesSet(); + } + + @Test(expected=IllegalStateException.class) + public void cantSetHashValueSerializer() { + RedisTemplate redisTemplate = new RedisTemplate(); + redisTemplate.setConnectionFactory(new JedisConnectionFactory()); + RedisStoreMessageSource source = new RedisStoreMessageSource(redisTemplate, + new LiteralExpression("foo")); + source.setHashValueSerializer(new StringRedisSerializer()); + source.afterPropertiesSet(); + } + @SuppressWarnings("unused") private static class TestMessageConverter extends SimpleMessageConverter { } diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml index 620a792ee6..dcac318cea 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests-context.xml @@ -12,7 +12,8 @@ + message-converter="testConverter" + serializer="serializer"/> @@ -29,5 +30,7 @@ + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java index ada5daccf9..85406c4765 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java @@ -16,6 +16,8 @@ package org.springframework.integration.redis.config; +import static junit.framework.Assert.assertEquals; + import org.junit.Test; import org.junit.runner.RunWith; @@ -33,8 +35,6 @@ import org.springframework.integration.support.converter.SimpleMessageConverter; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import static junit.framework.Assert.assertEquals; - /** * @author Oleg Zhurakousky * @author Mark Fisher @@ -46,8 +46,8 @@ public class RedisOutboundChannelAdapterParserTests extends RedisAvailableTests{ @Autowired private ApplicationContext context; - - @Test + + @Test @RedisAvailable public void validateConfiguration() { EventDrivenConsumer adapter = context.getBean("outboundAdapter", EventDrivenConsumer.class); @@ -58,9 +58,10 @@ public class RedisOutboundChannelAdapterParserTests extends RedisAvailableTests{ assertEquals("foo", accessor.getPropertyValue("defaultTopic")); Object converterBean = context.getBean("testConverter"); assertEquals(converterBean, accessor.getPropertyValue("messageConverter")); + assertEquals(context.getBean("serializer"), accessor.getPropertyValue("serializer")); } - - @Test + + @Test @RedisAvailable public void testOutboundChannelAdapterMessaging() throws Exception{ MessageChannel sendChannel = context.getBean("sendChannel", MessageChannel.class); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-store-adapter-parser-fail.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-store-adapter-parser-fail.xml new file mode 100644 index 0000000000..3b451e9802 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-store-adapter-parser-fail.xml @@ -0,0 +1,33 @@ + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-store-adapter-parser.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-store-adapter-parser.xml new file mode 100644 index 0000000000..cfe4572d5b --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/inbound-store-adapter-parser.xml @@ -0,0 +1,38 @@ + + + + + + + + + + + + + + + + + + + + + + + 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 index 68a4577233..0d345eddf5 100644 --- 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 @@ -72,12 +72,6 @@ - - - - - - diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter-parser-fail.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter-parser-fail.xml new file mode 100644 index 0000000000..41bfd769a4 --- /dev/null +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter-parser-fail.xml @@ -0,0 +1,37 @@ + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter-parser.xml b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter-parser.xml index 4e030fd510..713d528b94 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter-parser.xml +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/config/store-outbound-adapter-parser.xml @@ -25,8 +25,8 @@ - - + + 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 index ecdc30682a..f410d05033 100644 --- 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 @@ -46,7 +46,7 @@ - + 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 8c0f898417..3900b833dc 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 @@ -26,8 +26,6 @@ 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 @@ -59,9 +57,7 @@ public class RedisAvailableTests { RedisTemplate redisTemplate = new RedisTemplate(); redisTemplate.setConnectionFactory(jcf); - redisTemplate.setKeySerializer(new StringRedisSerializer()); - redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); - + redisTemplate.afterPropertiesSet(); BoundListOperations ops = redisTemplate.boundListOps("presidents"); ops.rightPush("John Adams"); @@ -85,8 +81,7 @@ public class RedisAvailableTests { RedisTemplate redisTemplate = new RedisTemplate(); redisTemplate.setConnectionFactory(jcf); - redisTemplate.setKeySerializer(new StringRedisSerializer()); - redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer()); + redisTemplate.afterPropertiesSet(); BoundZSetOperations ops = redisTemplate.boundZSetOps("presidents");