SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Move calling afterProperties on consumer and producer endpoints from binder implementations to core

Browse Source
This commit is contained in:
Soby Chacko
2016-07-22 11:35:44 -04:00
committed by Marius Bogoevici
parent 08d65a92bb
commit 8dd22ebca0
1 changed files with 16 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
View File

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.binder;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
@@ -75,7 +76,10 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
* Binds an outbound channel to a given destination. The implementation delegates to
* {@link #createProducerDestinationIfNecessary(String, ProducerProperties)}
* and {@link #createProducerMessageHandler(String, ProducerProperties)} for
* handling the middleware specific logic.
* handling the middleware specific logic. If the returned producer message handler is an
* {@link InitializingBean} then {@link InitializingBean#afterPropertiesSet()} will be
* called on it. Similarly, if the returned producer message handler endpoint is a
* {@link Lifecycle}, then {@link Lifecycle#start()} will be called on it.
* @param destination the name of the destination
* @param outputChannel the channel to be bound
* @param producerProperties the {@link ProducerProperties} of the binding
@@ -91,6 +95,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
final MessageHandler producerMessageHandler;
try {
producerMessageHandler = createProducerMessageHandler(destination, producerProperties);
if (producerMessageHandler instanceof InitializingBean) {
((InitializingBean) producerMessageHandler).afterPropertiesSet();
}
}
catch (Exception e) {
if (e instanceof BinderException) {
@@ -158,7 +165,10 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
* Binds an inbound channel to a given destination. The implementation delegates to
* {@link #createConsumerDestinationIfNecessary(String, String, ConsumerProperties)}
* and {@link #createConsumerEndpoint(String, String, Object, ConsumerProperties)}
* for handling middleware-specific logic.
* for handling middleware-specific logic. If the returned consumer endpoint is an
* {@link InitializingBean} then {@link InitializingBean#afterPropertiesSet()} will be
* called on it. Similarly, if the returned consumer endpoint is a {@link Lifecycle},
* then {@link Lifecycle#start()} will be called on it.
* @param name the name of the destination
* @param group the consumer group
* @param inputChannel the channel to be bound
@@ -180,6 +190,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
bridge.setBeanName("bridge." + name);
consumerEndpoint = createConsumerEndpoint(name, group, destination, properties);
consumerEndpoint.setOutputChannel(bridge);
if (consumerEndpoint instanceof InitializingBean) {
((InitializingBean) consumerEndpoint).afterPropertiesSet();
}
if (consumerEndpoint instanceof Lifecycle) {
((Lifecycle) consumerEndpoint).start();
}
@@ -201,7 +214,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
((Lifecycle) consumerEndpoint).stop();
}
if (e instanceof BinderException) {
throw e;
throw (BinderException) e;
}
else {
throw new BinderException("Exception thrown while starting consumer: ", e);