From 8dd22ebca0bd6439d3f4d75ce8674f07d09eca70 Mon Sep 17 00:00:00 2001
From: Soby Chacko <schacko@pivotal.io>
Date: Fri, 22 Jul 2016 11:35:44 -0400
Subject: [PATCH] Move calling afterProperties on consumer and producer
 endpoints from binder implementations to core

---
 .../binder/AbstractMessageChannelBinder.java  | 19 ++++++++++++++++---
 1 file changed, 16 insertions(+), 3 deletions(-)

diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
index 096616eab..52bb8e434 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
@@ -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);