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

INT-2328-v2

Browse Source 
SubscribableRedisChannel as well as other components to initialize MessageListenerAdapter as it is now neccessery after Redis upgrade

INT-2328-v2 polishing

INT-2328-v2 polishing
This commit is contained in:
Oleg Zhurakousky
2011-12-21 12:10:49 -05:00
committed by Mark Fisher
parent 0137a3023f
commit 70f8177a6d
3 changed files with 25 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java
View File

@@ -55,7 +55,9 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
private final RedisTemplate redisTemplate;
private final String topicName;
private volatile MessageDispatcher dispatcher;
private final MessageDispatcher dispatcher = new BroadcastingDispatcher();
private volatile boolean initialized;
// defaults
private volatile Executor taskExecutor = new SimpleAsyncTaskExecutor();
@@ -86,12 +88,10 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
}
public boolean subscribe(MessageHandler handler) {
Assert.state(this.dispatcher != null, "'MessageDispatcher' must not be null. This channel might not have been initialized");
return this.dispatcher.addHandler(handler);
}
public boolean unsubscribe(MessageHandler handler) {
Assert.state(this.dispatcher != null, "'MessageDispatcher' must not be null. This channel might not have been initialized");
return this.dispatcher.removeHandler(handler);
}
@@ -103,30 +103,26 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
@Override
public void onInit() throws Exception {
synchronized (this.dispatcher) {
if (this.dispatcher == null){
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();
}
if (this.initialized){
return;
}
}
private void configureDispatcher() {
this.dispatcher = new BroadcastingDispatcher();
super.onInit();
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);
adapter.afterPropertiesSet();
this.container.addMessageListener(adapter, new ChannelTopic(topicName));
this.container.afterPropertiesSet();
this.initialized = true;
}
/*

3
spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java
View File

@@ -73,7 +73,8 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
List<ChannelTopic> topicList = new ArrayList<ChannelTopic>();
for (String topic : this.topics) {
topicList.add(new ChannelTopic(topic));
}
}
adapter.afterPropertiesSet();
this.container.addMessageListener(adapter, topicList);
this.container.afterPropertiesSet();
}

1
spring-integration-redis/src/test/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandlerTests.java
View File

@@ -53,6 +53,7 @@ public class RedisPublishingMessageHandlerTests extends RedisAvailableTests{
MessageListenerAdapter listener = new MessageListenerAdapter();
listener.setDelegate(new Listener(latch));
listener.setSerializer(new StringRedisSerializer());
listener.afterPropertiesSet();
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(connectionFactory);