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

INT-1992 moved REDIS module to the main branch

Browse Source
This commit is contained in:
Oleg Zhurakousky
2011-07-26 10:41:06 -04:00
parent c26a1c6c6b
commit 80fa07f356
26 changed files with 2054 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
build.gradle
View File

@@ -125,6 +125,7 @@ configure(javaprojects) {
springAmqpVersion = '1.0.0.RC2'
springDataMongoVersion = '1.0.0.BUILD-SNAPSHOT'
springDataCommonsVersion = '1.2.0.BUILD-SNAPSHOT'
springDataRedisVersion = '1.0.0.BUILD-SNAPSHOT'
springSecurityVersion = '3.0.5.RELEASE'
springWsVersion = '2.0.2.RELEASE'
@@ -333,6 +334,16 @@ project('spring-integration-mongodb') {
}
}
project('spring-integration-redis') {
description = 'Spring Integration Redis Support'
dependencies {
compile project(":spring-integration-core")
compile "org.springframework:spring-context:$springVersion"
compile "org.springframework.data:spring-data-redis:$springDataRedisVersion"
compile "org.springframework.data:spring-data-commons-core:$springDataCommonsVersion"
}
}
project('spring-integration-rmi') {
description = 'Spring Integration RMI Support'
dependencies {

1
settings.gradle
View File

@@ -31,6 +31,7 @@ include 'spring-integration-jms'
include 'spring-integration-jmx'
include 'spring-integration-mail'
include 'spring-integration-mongodb'
include 'spring-integration-redis'
include 'spring-integration-rmi'
include 'spring-integration-security'
include 'spring-integration-sftp'

556
spring-integration-redis/jredis.log Normal file
View File

File diff suppressed because one or more lines are too long

179
spring-integration-redis/pom.xml Normal file
View File

@@ -0,0 +1,179 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-redis</artifactId>
<version>2.1.0.BUILD-SNAPSHOT</version>
<name>Spring Integration Redis Support</name>
<licenses>
<license>
<name>The Apache Software License, Version 2.0</name>
<url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
<distribution>repo</distribution>
</license>
</licenses>
<build>
<resources>
<resource>
<directory>src/main/java</directory>
<includes>
<include>**/*</include>
</includes>
<excludes>
<exclude>**/*.java</exclude>
</excludes>
</resource>
<resource>
<directory>src/main/resources</directory>
<includes>
<include>**/*</include>
</includes>
</resource>
</resources>
<testResources>
<testResource>
<directory>src/test/java</directory>
<includes>
<include>**/*</include>
</includes>
<excludes>
<exclude>**/*.java</exclude>
</excludes>
</testResource>
<testResource>
<directory>src/test/resources</directory>
<includes>
<include>**/*</include>
</includes>
</testResource>
</testResources>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>1.5</source>
<target>1.5</target>
</configuration>
</plugin>
<plugin>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<includes>
<include>**/*Tests.java</include>
</includes>
<excludes>
<exclude>**/*Abstract*.java</exclude>
</excludes>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<id>http://download.java.net/maven/2</id>
<url>http://download.java.net/maven/2</url>
</repository>
<repository>
<id>http://maven.springframework.org/milestone</id>
<url>http://maven.springframework.org/milestone</url>
</repository>
<repository>
<id>http://maven.springframework.org/release</id>
<url>http://maven.springframework.org/release</url>
</repository>
<repository>
<id>http://maven.springframework.org/snapshot</id>
<url>http://maven.springframework.org/snapshot</url>
</repository>
<repository>
<id>http://repository.springsource.com/maven/bundles/external</id>
<url>http://repository.springsource.com/maven/bundles/external</url>
</repository>
<repository>
<id>http://repository.springsource.com/maven/bundles/milestone</id>
<url>http://repository.springsource.com/maven/bundles/milestone</url>
</repository>
<repository>
<id>http://repository.springsource.com/maven/bundles/release</id>
<url>http://repository.springsource.com/maven/bundles/release</url>
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.easymock</groupId>
<artifactId>easymock</artifactId>
<version>2.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.easymock</groupId>
<artifactId>easymockclassextension</artifactId>
<version>2.3</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-commons-core</artifactId>
<version>1.2.0.BUILD-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>3.1.0.M2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.data</groupId>
<artifactId>spring-data-redis</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>3.1.0.M2</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.hamcrest</groupId>
<artifactId>hamcrest-all</artifactId>
<version>1.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>
<version>1.8.4</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.12</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>cglib</groupId>
<artifactId>cglib-nodep</artifactId>
<version>2.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit-dep</artifactId>
<version>4.8.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>2.1.0.BUILD-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
</dependencies>
<properties>
<project.build.sourceEncoding>UTF8</project.build.sourceEncoding>
</properties>
</project>

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

@@ -0,0 +1,169 @@
/*
* Copyright 2002-2011 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.channel;
import java.util.concurrent.Executor;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
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.channel.AbstractMessageChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.support.converter.MessageConverter;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.util.ErrorHandler;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
@SuppressWarnings("rawtypes")
public class SubscribableRedisChannel extends AbstractMessageChannel implements SubscribableChannel, SmartLifecycle, DisposableBean {
private final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
private final JedisConnectionFactory connectionFactory;
private final RedisTemplate redisTemplate;
private final String topicName;
private volatile MessageDispatcher dispatcher;
// defaults
private volatile Executor taskExecutor = new SimpleAsyncTaskExecutor();
private volatile RedisSerializer<?> serializer = new StringRedisSerializer();
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
public SubscribableRedisChannel(JedisConnectionFactory connectionFactory, String topicName) {
this.connectionFactory = connectionFactory;
this.redisTemplate = new StringRedisTemplate(connectionFactory);
this.topicName = topicName;
}
public void setTaskExecutor(Executor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public void setMessageConverter(MessageConverter messageConverter) {
this.messageConverter = messageConverter;
}
public void setSerializer(RedisSerializer<?> serializer) {
this.serializer = serializer;
}
public boolean subscribe(MessageHandler handler) {
return this.dispatcher.addHandler(handler);
}
public boolean unsubscribe(MessageHandler handler) {
return this.dispatcher.removeHandler(handler);
}
@Override
protected boolean doSend(Message<?> message, long arg1) {
this.redisTemplate.convertAndSend(this.topicName, this.messageConverter.fromMessage(message));
return true;
}
@Override
public void onInit() throws Exception {
super.onInit();
this.configureDispatcher();
if (this.messageConverter == null){
this.messageConverter = new SimpleMessageConverter();
}
this.container.setConnectionFactory(connectionFactory);
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
ErrorHandler errorHandler = new MessagePublishingErrorHandler(
new BeanFactoryChannelResolver(this.getBeanFactory()));
this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, errorHandler);
}
this.container.setTaskExecutor(this.taskExecutor);
MessageListenerAdapter adapter = new MessageListenerAdapter(new MessageListenerDelegate());
adapter.setSerializer(serializer);
this.container.addMessageListener(adapter, new ChannelTopic(topicName));
this.container.afterPropertiesSet();
}
private void configureDispatcher() {
this.dispatcher = new BroadcastingDispatcher();
}
/*
* SmartLifecycle implementation (delegates to the MessageListener container)
*/
public boolean isAutoStartup() {
return (this.container != null) ? this.container.isAutoStartup() : false;
}
public int getPhase() {
return 0;
}
public boolean isRunning() {
return (this.container != null) ? this.container.isRunning() : false;
}
public void start() {
if (this.container != null) {
this.container.start();
}
}
public void stop() {
this.connectionFactory.getConnection().discard();
if (this.container != null) {
this.container.stop();
}
}
public void stop(Runnable callback) {
if (this.container != null) {
this.container.stop(callback);
}
}
public void destroy() throws Exception {
if (this.container != null) {
this.container.destroy();
}
}
private class MessageListenerDelegate {
@SuppressWarnings("unused")
public void handleMessage(String s) {
Message<?> siMessage = messageConverter.toMessage(s);
dispatcher.dispatch(siMessage);
}
}
}

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

@@ -0,0 +1,55 @@
/*
* Copyright 2002-2010 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.AbstractChannelParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
/**
* Parser for the 'channel' and 'publish-subscribe-channel' elements of the
* Spring Integration Redis namespace.
*
* @author Oleg Zhurakusky
* @since 2.1
*/
public class RedisChannelParser extends AbstractChannelParser {
@Override
protected BeanDefinitionBuilder buildBeanDefinition(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.redis.channel.SubscribableRedisChannel");
String connectionFactory = element.getAttribute("connection-factory");
if (!StringUtils.hasText(connectionFactory)) {
connectionFactory = "redisConnectionFactory";
}
builder.addConstructorArgReference(connectionFactory);
String topicName = element.getAttribute("topic-name");
builder.addConstructorArgValue(topicName);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "task-executor");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");
// IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "serializer");
return builder;
}
}

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

@@ -0,0 +1,52 @@
/*
* Copyright 2002-2011 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.springframework.beans.factory.config.RuntimeBeanReference;
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
* @author Mark Fisher
* @since 2.1
*/
public class RedisInboundChannelAdapterParser extends AbstractChannelAdapterParser {
@Override
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.redis.inbound.RedisInboundChannelAdapter");
String connectionFactory = element.getAttribute("connection-factory");
if (!StringUtils.hasText(connectionFactory)) {
connectionFactory = "redisConnectionFactory";
}
builder.addConstructorArgReference(connectionFactory);
builder.addPropertyValue("outputChannel", new RuntimeBeanReference(channelName));
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topics");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
return builder.getBeanDefinition();
}
}

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

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

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

@@ -0,0 +1,48 @@
/*
* Copyright 2002-2011 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.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
* @author Mark Fisher
* @since 2.1
*/
public class RedisOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
"org.springframework.integration.redis.outbound.RedisPublishingMessageHandler");
String connectionFactory = element.getAttribute("connection-factory");
if (!StringUtils.hasText(connectionFactory)) {
connectionFactory = "redisConnectionFactory";
}
builder.addConstructorArgReference(connectionFactory);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "topic", "defaultTopic");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
return builder.getBeanDefinition();
}
}

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

@@ -0,0 +1,94 @@
/*
* Copyright 2007-2011 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.ArrayList;
import java.util.List;
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.StringRedisSerializer;
import org.springframework.integration.Message;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.converter.MessageConverter;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.1
*/
public class RedisInboundChannelAdapter extends MessageProducerSupport {
private final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
private volatile String[] topics;
public RedisInboundChannelAdapter(RedisConnectionFactory connectionFactory) {
Assert.notNull(connectionFactory, "connectionFactory must not be null");
this.container.setConnectionFactory(connectionFactory);
}
public void setTopics(String... topics) {
this.topics = topics;
}
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "messageConverter must not be null");
this.messageConverter = messageConverter;
}
public String getComponentType() {
return "redis:inbound-channel-adapter";
}
@Override
protected void onInit() {
super.onInit();
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());
List<ChannelTopic> topicList = new ArrayList<ChannelTopic>();
for (String topic : this.topics) {
topicList.add(new ChannelTopic(topic));
}
this.container.addMessageListener(adapter, topicList);
this.container.afterPropertiesSet();
}
private Message<?> convertMessage(String s) {
return this.messageConverter.toMessage(s);
}
private class MessageListenerDelegate {
@SuppressWarnings("unused")
public void handleMessage(String s) {
sendMessage(convertMessage(s));
}
}
}

69
spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java Normal file
View File

@@ -0,0 +1,69 @@
/*
* Copyright 2007-2011 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.outbound;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.support.converter.MessageConverter;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
* @since 2.1
*/
public class RedisPublishingMessageHandler implements MessageHandler {
private final StringRedisTemplate template;
private volatile MessageConverter messageConverter = new SimpleMessageConverter();
private volatile String defaultTopic;
public RedisPublishingMessageHandler(RedisConnectionFactory connectionFactory) {
Assert.notNull(connectionFactory, "connectionFactory must not be null");
this.template = new StringRedisTemplate(connectionFactory);
}
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "messageConverter must not be null");
this.messageConverter = messageConverter;
}
public void setDefaultTopic(String defaultTopic) {
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 " +
"from Message, and no defaultTopic has been provided.");
return this.defaultTopic;
}
}

1
spring-integration-redis/src/main/resources/META-INF/spring.handlers Normal file
View File

@@ -0,0 +1 @@
http\://www.springframework.org/schema/integration/redis=org.springframework.integration.redis.config.RedisNamespaceHandler

2
spring-integration-redis/src/main/resources/META-INF/spring.schemas Normal file
View File

@@ -0,0 +1,2 @@
http\://www.springframework.org/schema/integration/redis/spring-integration-redis-2.0.xsd=org/springframework/integration/redis/config/spring-integration-redis-2.0.xsd
http\://www.springframework.org/schema/integration/redis/spring-integration-redis.xsd=org/springframework/integration/redis/config/spring-integration-redis-2.0.xsd

4
spring-integration-redis/src/main/resources/META-INF/spring.tooling Normal file
View File

@@ -0,0 +1,4 @@
# Tooling related information for the integration redis namespace
http\://www.springframework.org/schema/integration/redis@name=integration redis Namespace
http\://www.springframework.org/schema/integration/redis@prefix=int-redis
http\://www.springframework.org/schema/integration/redis@icon=org/springframework/integration/redis/config/spring-integration-redis.gif

282
spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis-2.0.xsd Normal file
View File

@@ -0,0 +1,282 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/redis"
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/redis"
elementFormDefault="qualified" attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans" />
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-2.0.xsd" />
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for Spring Integration Redis Adapters and Channels.
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="publish-subscribe-channel">
<xsd:annotation>
<xsd:documentation>
Defines a publish-subscribe Message Channel that is backed by a Redis topic.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="channelType">
<xsd:attribute name="topic-name" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>
Name of the Redis topic that backs this channel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="channelType">
<xsd:annotation>
<xsd:documentation><![CDATA[
Base type for Redis backed Message Channels (either 'channel' for a
list-backed channel or 'publish-subscribe-channel' for a topic-backed channel).
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:exports type="org.springframework.integration.channel.AbstractMessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
<xsd:sequence>
<xsd:element name="interceptors" type="integration:channelInterceptorsType" minOccurs="0" maxOccurs="1">
<xsd:annotation>
<xsd:documentation><![CDATA[
A list of ChannelInterceptor instances to be applied to this channel.
]]></xsd:documentation>
</xsd:annotation>
</xsd:element>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID" use="required">
<xsd:annotation>
<xsd:documentation>
ID for this channel. Required.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to a RedisConnectionFactory. If none is provided, the default
bean name for the reference will be "redisConnectionFactory".
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.data.redis.connection.RedisConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="task-executor" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to a Spring TaskExecutor (or standard JDK 1.5+ Executor) for executing
JMS listener invokers. Default is a SimpleAsyncTaskExecutor in case of a
DefaultMessageListenerContainer, using internally managed threads. For a
SimpleMessageListenerContainer, listeners will always get invoked within the
JMS provider's receive thread by default.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="java.util.concurrent.Executor"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-converter" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to the MessageConverter strategy for converting between Redis Messages
and the Spring Integration Message payloads. Default is a SimpleMessageConverter.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.jms.support.converter.MessageConverter"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="serializer" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to the RedisSerializer strategy
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.data.redis.serializer.RedisSerializer"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<!-- <xsd:attribute name="auto-startup" type="xsd:string"> -->
<!-- <xsd:annotation> -->
<!-- <xsd:documentation><![CDATA[ -->
<!-- Boolean value indicating whether this channel's listener container should start automatically. -->
<!-- ]]></xsd:documentation> -->
<!-- </xsd:annotation> -->
<!-- </xsd:attribute> -->
<!-- <xsd:attribute name="phase" type="xsd:string"> -->
<!-- <xsd:annotation> -->
<!-- <xsd:documentation><![CDATA[ -->
<!-- The lifecycle phase within which this channel's listener container should start and stop. -->
<!-- The lower the value the earlier this container will start and the later it will stop. The -->
<!-- default is Integer.MAX_VALUE meaning the container will start as late as possible -->
<!-- and stop as soon as possible. -->
<!-- ]]></xsd:documentation> -->
<!-- </xsd:annotation> -->
<!-- </xsd:attribute> -->
</xsd:complexType>
<xsd:element name="inbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines an inbound Channel Adapter for subscribing to a Redis topic.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
The ID for this Channel Adapter.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Reference to a RedisConnectionFactory. If none is provided, the default
bean name for the reference will be "redisConnectionFactory".
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.data.redis.connection.RedisConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="topics" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Redis topic names as a comma-delimited list of Strings.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string" use="required">
<xsd:annotation>
<xsd:documentation>
Channel to which Messages will be sent.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="error-channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
Channel to which Error Messages will be sent.
</xsd:documentation>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-converter" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to the MessageConverter strategy for converting between Redis Messages
and the Spring Integration Message payloads. Default is a SimpleMessageConverter.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.support.converter.MessageConverter"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines an outbound Redis Message-sending Channel Adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="redisAdapterType">
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="topic" type="xsd:string"/>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-converter" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
A reference to the MessageConverter strategy for converting between Redis Messages
and the Spring Integration Message payloads. Default is a SimpleMessageConverter.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.support.converter.MessageConverter"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="order" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Specifies the order for invocation when this adapter is connected as a
subscriber to a SubscribableChannel.
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="redisAdapterType">
<xsd:annotation>
<xsd:documentation>
Common configuration for Redis adapters.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:ID"/>
<xsd:attribute name="connection-factory" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.data.redis.connection.RedisConnectionFactory"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
</xsd:complexType>
</xsd:schema>

BIN
spring-integration-redis/src/main/resources/org/springframework/integration/redis/config/spring-integration-redis.gif Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 552 B

54
spring-integration-redis/src/test/java/org/springframework/integration/redis/channel/SubscribableRedisChannelTests.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* Copyright 2002-2011 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.channel;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import org.junit.Ignore;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.integration.Message;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.message.GenericMessage;
/**
* @author Oleg Zhurakousky
* @since 2.0
*/
public class SubscribableRedisChannelTests {
@Test @Ignore // requires instance of redis-server
public void pubSubChanneTest() throws Exception{
JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
connectionFactory.afterPropertiesSet();
SubscribableRedisChannel channel = new SubscribableRedisChannel(connectionFactory, "si.test.channel");
channel.setBeanFactory(mock(BeanFactory.class));
channel.afterPropertiesSet();
channel.start();
MessageHandler handler = mock(MessageHandler.class);
channel.subscribe(handler);
channel.send(new GenericMessage<String>("1"));
channel.send(new GenericMessage<String>("2"));
channel.send(new GenericMessage<String>("3"));
Thread.sleep(100);
verify(handler, times(3)).handleMessage(Mockito.any(Message.class));
}
}

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

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
xmlns:redis="http://www.springframework.org/schema/redis"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis-2.0.xsd
http://www.springframework.org/schema/redis http://www.springframework.org/schema/redis/spring-redis-1.0.xsd">
<int-redis:publish-subscribe-channel id="redisChannel" topic-name="si.test.topic"/>
<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory"/>
</beans>

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

@@ -0,0 +1,62 @@
/*
* Copyright 2002-2011 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 junit.framework.Assert.assertEquals;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.test.util.TestUtils;
/**
* @author Oleg Zhurakousky
*
*/
public class RedisChannelParserTests {
@Test @Ignore // requires redis-server instance
public void testPubSubChannelConfig(){
ApplicationContext context = new ClassPathXmlApplicationContext("RedisChannelParserTests-context.xml", this.getClass());
SubscribableChannel redisChannel = context.getBean("redisChannel", SubscribableChannel.class);
JedisConnectionFactory connectionFactory =
TestUtils.getPropertyValue(redisChannel, "connectionFactory", JedisConnectionFactory.class);
assertEquals(connectionFactory, context.getBean("redisConnectionFactory"));
assertEquals("si.test.topic", TestUtils.getPropertyValue(redisChannel, "topicName"));
}
@Test
@Ignore
public void testPubSubChannelUsage(){
ApplicationContext context = new ClassPathXmlApplicationContext("RedisChannelParserTests-context.xml", this.getClass());
SubscribableChannel redisChannel = context.getBean("redisChannel", SubscribableChannel.class);
redisChannel.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) throws MessagingException {
System.out.println("Message: " + message);
}
});
redisChannel.send(new GenericMessage<String>("Hello Redis"));
}
}

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

@@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis-2.0.xsd">
<int-redis:inbound-channel-adapter id="adapter"
topics="foo, bar"
channel="receiveChannel"
error-channel="testErrorChannel"
message-converter="testConverter"/>
<int:channel id="receiveChannel">
<int:queue/>
</int:channel>
<int:channel id="testErrorChannel"/>
<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory"/>
<bean id="testConverter" class="org.springframework.integration.redis.config.RedisInboundChannelAdapterParserTests$TestMessageConverter"/>
</beans>

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

@@ -0,0 +1,73 @@
/*
* Copyright 2002-2011 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 junit.framework.Assert.assertEquals;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Oleg Zhurakousky
* @author Mark Fisher
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class RedisInboundChannelAdapterParserTests {
@Autowired
private ApplicationContext context;
@Test
public void validateConfiguration() {
RedisInboundChannelAdapter adapter = context.getBean("adapter", RedisInboundChannelAdapter.class);
assertEquals("adapter", adapter.getComponentName());
assertEquals("redis:inbound-channel-adapter", adapter.getComponentType());
DirectFieldAccessor accessor = new DirectFieldAccessor(adapter);
Object errorChannelBean = context.getBean("testErrorChannel");
assertEquals(errorChannelBean, accessor.getPropertyValue("errorChannel"));
Object converterBean = context.getBean("testConverter");
assertEquals(converterBean, accessor.getPropertyValue("messageConverter"));
}
@Test @Ignore
public void testInboundChannelAdapterMessaging() {
JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
connectionFactory.afterPropertiesSet();
connectionFactory.getConnection().publish("foo".getBytes(), "Hello Redis from foo".getBytes());
QueueChannel receiveChannel = context.getBean("receiveChannel", QueueChannel.class);
assertEquals("Hello Redis from foo", receiveChannel.receive(1000).getPayload());
connectionFactory.getConnection().publish("bar".getBytes(), "Hello Redis from bar".getBytes());
assertEquals("Hello Redis from bar", receiveChannel.receive(1000).getPayload());
}
@SuppressWarnings("unused")
private static class TestMessageConverter extends SimpleMessageConverter {
}
}

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

@@ -0,0 +1,27 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-redis="http://www.springframework.org/schema/integration/redis"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/integration/redis http://www.springframework.org/schema/integration/redis/spring-integration-redis-2.0.xsd">
<int:channel id="sendChannel"/>
<int-redis:outbound-channel-adapter id="outboundAdapter"
channel="sendChannel"
topic="foo"
message-converter="testConverter"/>
<int-redis:inbound-channel-adapter channel="receiveChannel" topics="foo"/>
<int:channel id="receiveChannel">
<int:queue/>
</int:channel>
<bean id="redisConnectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory"/>
<bean id="testConverter" class="org.springframework.integration.redis.config.RedisOutboundChannelAdapterParserTests$TestMessageConverter"/>
</beans>

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

@@ -0,0 +1,72 @@
/*
* Copyright 2002-2011 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 junit.framework.Assert.assertEquals;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.redis.outbound.RedisPublishingMessageHandler;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Oleg Zhurakousky
* @author Mark Fisher
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class RedisOutboundChannelAdapterParserTests {
@Autowired
private ApplicationContext context;
@Test
public void validateConfiguration() {
EventDrivenConsumer adapter = context.getBean("outboundAdapter", EventDrivenConsumer.class);
RedisPublishingMessageHandler handler = (RedisPublishingMessageHandler)
new DirectFieldAccessor(adapter).getPropertyValue("handler");
assertEquals("outboundAdapter", adapter.getComponentName());
DirectFieldAccessor accessor = new DirectFieldAccessor(handler);
assertEquals("foo", accessor.getPropertyValue("defaultTopic"));
Object converterBean = context.getBean("testConverter");
assertEquals(converterBean, accessor.getPropertyValue("messageConverter"));
}
@Test @Ignore // requires redis-server instance
public void testOutboundChannelAdapterMessaging(){
MessageChannel sendChannel = context.getBean("sendChannel", MessageChannel.class);
sendChannel.send(new GenericMessage<String>("Hello Redis"));
QueueChannel receiveChannel = context.getBean("receiveChannel", QueueChannel.class);
assertEquals("Hello Redis", receiveChannel.receive(1000).getPayload());
}
@SuppressWarnings("unused")
private static class TestMessageConverter extends SimpleMessageConverter {
}
}

66
spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapterTests.java Normal file
View File

@@ -0,0 +1,66 @@
/*
* Copyright 2007-2011 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 static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.integration.Message;
import org.springframework.integration.channel.QueueChannel;
/**
* @author Mark Fisher
* @since 2.1
*/
public class RedisInboundChannelAdapterTests {
@Test @Ignore
public void testRedisInboundChannelAdapter() throws Exception {
int numToTest = 100;
String redisChannelName = "si.test.channel";
QueueChannel channel = new QueueChannel();
JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
connectionFactory.afterPropertiesSet();
RedisInboundChannelAdapter adapter = new RedisInboundChannelAdapter(connectionFactory);
adapter.setTopics("si.test.channel");
adapter.setOutputChannel(channel);
adapter.afterPropertiesSet();
adapter.start();
StringRedisTemplate redisTemplate = new StringRedisTemplate(connectionFactory);
redisTemplate.afterPropertiesSet();
for (int i = 0; i < numToTest; i++) {
redisTemplate.convertAndSend(redisChannelName, "test-" + i);
}
int counter = 0;
for (int i = 0; i < numToTest; i++) {
Message<?> message = channel.receive(1000);
assertNotNull(message);
assertTrue(message.getPayload().toString().startsWith("test-"));
counter++;
}
assertEquals(numToTest, counter);
}
}

85
spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandlerTests.java Normal file
View File

@@ -0,0 +1,85 @@
/*
* Copyright 2007-2011 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.outbound;
import static org.junit.Assert.assertEquals;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.integration.support.MessageBuilder;
/**
* @author Mark Fisher
* @since 2.1
*/
public class RedisPublishingMessageHandlerTests {
@Test @Ignore
public void testRedisPublishingMessageHandler() throws Exception {
int numToTest = 100;
String topic = "si.test.channel";
final CountDownLatch latch = new CountDownLatch(numToTest);
JedisConnectionFactory connectionFactory = new JedisConnectionFactory();
connectionFactory.afterPropertiesSet();
MessageListenerAdapter listener = new MessageListenerAdapter();
listener.setDelegate(new Listener(latch));
listener.setSerializer(new StringRedisSerializer());
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.afterPropertiesSet();
container.addMessageListener(listener, Collections.<Topic>singletonList(new ChannelTopic(topic)));
container.start();
Thread.sleep(500);
final RedisPublishingMessageHandler handler = new RedisPublishingMessageHandler(connectionFactory);
handler.setDefaultTopic(topic);
for (int i = 0; i < numToTest; i++) {
handler.handleMessage(MessageBuilder.withPayload("test-" + i).build());
}
latch.await(3, TimeUnit.SECONDS);
assertEquals(0, latch.getCount());
}
private static class Listener {
private final CountDownLatch latch;
private Listener(CountDownLatch latch) {
this.latch = latch;
}
@SuppressWarnings("unused")
public void handleMessage(String s) {
this.latch.countDown();
}
}
}

16
spring-integration-redis/template.mf Normal file
View File

@@ -0,0 +1,16 @@
Bundle-SymbolicName: org.springframework.integration.redis
Bundle-Name: Spring Integration Redis Support
Bundle-Vendor: SpringSource
Bundle-ManifestVersion: 2
Import-Template:
org.apache.commons.logging;version="[1.1.1, 2.0.0)",
org.springframework.integration.*;version="[2.0.0, 2.0.1)",
org.springframework.beans.*;version="[3.0.0, 4.0.0)",
org.springframework.context;version="[3.0.0, 4.0.0)",
org.springframework.core.*;version="[3.0.0, 4.0.0)",
org.springframework.expression.*;version="[3.0.0, 4.0.0)",
org.springframework.util;version="[3.0.0, 4.0.0)",
org.springframework.expression.*;version="[3.0.0, 4.0.0)",
org.springframework.data.keyvalue.redis.*;version="[1.0.0, 2.0.0)",
javax.*;version="0",
org.w3c.dom.*;version="0"