From ec0966875cf20687c6e1e53159e53eefb5e71b08 Mon Sep 17 00:00:00 2001
From: Oleg Zhurakousky <ozhurakousky@pivotal.io>
Date: Wed, 28 Feb 2018 13:00:05 -0500
Subject: [PATCH] GH-1267: Created BindingCreatedEvent

Fixes spring-cloud/spring-cloud-stream#1267

- updated AbstractChannelBinder to fire this event whenever bindings are created

* Simple code style polishing and fix JavaDocs
---
 .../binder/AbstractMessageChannelBinder.java  | 45 ++++++++++++++-----
 .../stream/binder/BindingCreatedEvent.java    | 37 +++++++++++++++
 2 files changed, 70 insertions(+), 12 deletions(-)
 create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BindingCreatedEvent.java

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 91f6a94d9..7ed4061a0 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
@@ -25,12 +25,15 @@ import org.apache.commons.logging.Log;
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.cloud.stream.provisioning.ProvisioningException;
 import org.springframework.cloud.stream.provisioning.ProvisioningProvider;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.PublishSubscribeChannel;
@@ -76,7 +79,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 
 	private final EmbeddedHeadersChannelInterceptor embeddedHeadersChannelInterceptor =
 			new EmbeddedHeadersChannelInterceptor(this.logger);
-	
+
 	private final ObjectMapper objectMapper = new ObjectMapper();
 
 	/**
@@ -90,8 +93,12 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	 */
 	private final String[] headersToEmbed;
 
+	@Autowired
+	private ApplicationEventPublisher applicationEventPublisher;
+
 	public AbstractMessageChannelBinder(String[] headersToEmbed,
 			PP provisioningProvider) {
+
 		this.headersToEmbed = headersToEmbed == null ? new String[0] : headersToEmbed;
 		this.provisioningProvider = provisioningProvider;
 	}
@@ -102,6 +109,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	@Deprecated
 	protected AbstractMessageChannelBinder(boolean supportsHeadersNatively, String[] headersToEmbed,
 			PP provisioningProvider) {
+
 		this(headersToEmbed, provisioningProvider);
 	}
 
@@ -159,14 +167,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 						.equals(producerProperties.getHeaderMode()), this.headersToEmbed,
 						producerProperties.isUseNativeEncoding()));
 
-		return new DefaultBinding<MessageChannel>(destination, null, outputChannel,
+		Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(destination, null, outputChannel,
 				producerMessageHandler instanceof Lifecycle ? (Lifecycle) producerMessageHandler : null) {
 
 			@Override
-			public Map<String, Object> getExtendedInfo() {	
+			public Map<String, Object> getExtendedInfo() {
 				return doGetExtendedInfo(destination, producerProperties);
 			}
-			
+
 			@Override
 			public void afterUnbind() {
 				try {
@@ -182,6 +190,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				afterUnbindProducer(producerDestination, producerProperties);
 			}
 		};
+
+		doPublishEvent(new BindingCreatedEvent(binding));
+		return binding;
 	}
 
 	/**
@@ -265,14 +276,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				((Lifecycle) consumerEndpoint).start();
 			}
 
-			return new DefaultBinding<MessageChannel>(name, group, inputChannel,
+			Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(name, group, inputChannel,
 					consumerEndpoint instanceof Lifecycle ? (Lifecycle) consumerEndpoint : null) {
-				
+
 				@Override
 				public Map<String, Object> getExtendedInfo() {
 					return doGetExtendedInfo(destination, properties);
 				}
-				
+
 				@Override
 				protected void afterUnbind() {
 					try {
@@ -289,6 +300,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 				}
 
 			};
+			doPublishEvent(new BindingCreatedEvent(binding));
+			return binding;
 		}
 		catch (Exception e) {
 			if (consumerEndpoint instanceof Lifecycle) {
@@ -336,9 +349,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		if (resources.getSource() instanceof Lifecycle) {
 				((Lifecycle) resources.getSource()).start();
 		}
-		return new DefaultBinding<PollableSource<MessageHandler>>(name, group, inboundBindTarget,
+		Binding<PollableSource<MessageHandler>> binding = new DefaultBinding<PollableSource<MessageHandler>>(name, group, inboundBindTarget,
 				resources.getSource() instanceof Lifecycle ? (Lifecycle) resources.getSource() : null) {
-			
+
 			@Override
 			public Map<String, Object> getExtendedInfo() {
 				return doGetExtendedInfo(destination, properties);
@@ -351,6 +364,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 			}
 
 		};
+
+		doPublishEvent(new BindingCreatedEvent(binding));
+		return binding;
 	}
 
 	protected void postProcessPollableSource(DefaultPollableMessageSource bindingTarget) {
@@ -459,7 +475,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	 * @param destination the destination.
 	 * @param group the group.
 	 * @param consumerProperties the properties.
-	 * @param true if this is for a polled consumer.
+	 * @param polled true if this is for a polled consumer.
 	 * @return the ErrorInfrastructure which is a holder for the error channel, the recoverer and the
 	 * message handler that is subscribed to the channel.
 	 */
@@ -659,7 +675,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	protected String errorsBaseName(ProducerDestination destination) {
 		return destination.getName() + ".errors";
 	}
-	
+
 	private Map<String, Object> doGetExtendedInfo(Object destination, Object properties) {
 		Map<String, Object> extendedInfo = new LinkedHashMap<>();
 		extendedInfo.put("bindingDestination", destination.toString());
@@ -667,6 +683,12 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		return extendedInfo;
 	}
 
+	private void doPublishEvent(ApplicationEvent event) {
+		if (this.applicationEventPublisher != null) {
+			this.applicationEventPublisher.publishEvent(event);
+		}
+	}
+
 	private final class SendingHandler extends AbstractMessageHandler implements Lifecycle {
 
 		private final boolean embedHeaders;
@@ -824,5 +846,4 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 
 	}
-
 }
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BindingCreatedEvent.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BindingCreatedEvent.java
new file mode 100644
index 000000000..03f9183e8
--- /dev/null
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BindingCreatedEvent.java
@@ -0,0 +1,37 @@
+/*
+ * Copyright 2018 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.cloud.stream.binder;
+
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * ApplicationEvent fired whenever the the Binding is created.
+ *
+ * @author Oleg Zhurakousky
+ *
+ * @since 2.0
+ *
+ * #see AbstractMessageChannelBinder
+ */
+@SuppressWarnings("serial")
+public class BindingCreatedEvent extends ApplicationEvent {
+
+	public BindingCreatedEvent(Binding<?> source) {
+		super(source);
+	}
+
+}