Detect changes in service registry and rebind channels if possible

pom.xml

@@ -12,8 +12,8 @@
 	</organization>
     <properties>
       <java.version>1.7</java.version>
-      <spring-boot.version>1.2.3.RELEASE</spring-boot.version>
-      <spring-cloud.version>1.0.2.RELEASE</spring-cloud.version>
+      <spring-boot.version>1.3.0.BUILD-SNAPSHOT</spring-boot.version>
+      <spring-cloud.version>1.0.3.BUILD-SNAPSHOT</spring-cloud.version>
       <spring-xd.version>1.2.0.BUILD-SNAPSHOT</spring-xd.version>
     </properties>
 	<modules>

spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DefaultChannelLocator.java

@@ -17,8 +17,7 @@
 package org.springframework.bus.runner.adapter;
 
 import org.springframework.bus.runner.config.MessageBusProperties;
-
-import reactor.util.StringUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * @author Dave Syer

spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/Downstream.java

@@ -36,6 +36,6 @@ import org.springframework.beans.factory.annotation.Qualifier;
 @Retention(RetentionPolicy.RUNTIME)
 @Inherited
 @Documented
-public @interface InputChannel {
+public @interface Downstream {
 
 }

spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java

@@ -189,6 +189,17 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware {
 		channel.setTapped(false);
 	}
 
+	@ManagedOperation
+	public void rebind() {
+		boolean runnable = locateChannels();
+		if (runnable && !this.running) {
+			start();
+		}
+		if (!runnable && this.running) {
+			stop();
+		}
+	}
+
 	@Override
 	@ManagedOperation
 	public void start() {
@@ -196,13 +207,15 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware {
 			// Start everything, but don't call ourselves
 			if (!this.active.get()) {
 				if (this.active.compareAndSet(false, true)) {
-					bindChannels();
-					this.applicationContext.start();
+					boolean ready = bindChannels();
+					if (ready) {
+						this.running = true;
+						this.applicationContext.start();
+					}
 					this.active.set(false);
 				}
 			}
 		}
-		this.running = true;
 	}
 
 	@Override
@@ -247,19 +260,16 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware {
 		}
 	}
 
-	protected final void bindChannels() {
+	protected final boolean bindChannels() {
+		if (!locateChannels()) {
+			return false;
+		}
 		Map<String, Object> historyProperties = new LinkedHashMap<String, Object>();
 		if (this.trackHistory) {
 			// TODO: addHistoryTag();
 		}
 		for (OutputChannelSpec spec : this.outputChannels) {
-			String name = this.outputChannelLocator.locate(spec.getLocalName());
-			if (name == null) {
-				logger.info("No channel found for: " + spec.getLocalName());
-				continue;
-			}
-			spec.setName(name);
-			this.bindings.put(spec.getName(), name);
+			String name = spec.getName();
 			MessageChannel outputChannel = this.channelResolver.resolveDestination(spec
 					.getLocalName());
 			bindMessageProducer(outputChannel, name, this.module.getProducerProperties());
@@ -277,20 +287,40 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware {
 			}
 		}
 		for (InputChannelSpec spec : this.inputChannels) {
-			String name = this.inputChannelLocator.locate(spec.getLocalName());
-			if (name == null) {
-				logger.info("No channel found for: " + spec.getLocalName());
-				continue;
-			}
-			spec.setName(name);
-			this.bindings.put(spec.getName(), name);
-			MessageChannel inputChannel = this.channelResolver.resolveDestination(spec.getLocalName());
+			String name = spec.getName();
+			MessageChannel inputChannel = this.channelResolver.resolveDestination(spec
+					.getLocalName());
 			bindMessageConsumer(inputChannel, name, this.module.getConsumerProperties());
 			if (this.trackHistory && this.outputChannels.size() != 1) {
 				historyProperties.put("inputChannel", name);
 				track(inputChannel, historyProperties);
 			}
 		}
+		return true;
+	}
+
+	private boolean locateChannels() {
+		logger.info("Locating channels");
+		boolean located = true;
+		for (OutputChannelSpec spec : this.outputChannels) {
+			String name = this.outputChannelLocator.locate(spec.getLocalName());
+			if (name == null) {
+				logger.info("No channel found for: " + spec.getLocalName());
+				located = false;
+			}
+			spec.setName(name);
+			this.bindings.put(spec.getName(), name);
+		}
+		for (InputChannelSpec spec : this.inputChannels) {
+			String name = this.inputChannelLocator.locate(spec.getLocalName());
+			if (name == null) {
+				logger.info("No channel found for: " + spec.getLocalName());
+				located = false;
+			}
+			spec.setName(name);
+			this.bindings.put(spec.getName(), name);
+		}
+		return located;
 	}
 
 	// TODO: move this to ChannelLocator?

spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/Upstream.java

@@ -36,6 +36,6 @@ import org.springframework.beans.factory.annotation.Qualifier;
 @Retention(RetentionPolicy.RUNTIME)
 @Inherited
 @Documented
-public @interface OutputChannel {
+public @interface Upstream {
 
 }

spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientAutoConfiguration.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2015 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.bus.runner.adapter.discovery;
+
+import javax.annotation.PostConstruct;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.bus.runner.adapter.ChannelLocator;
+import org.springframework.bus.runner.adapter.Downstream;
+import org.springframework.bus.runner.adapter.MessageBusAdapter;
+import org.springframework.bus.runner.adapter.Upstream;
+import org.springframework.cloud.client.discovery.DiscoveryClient;
+import org.springframework.cloud.client.discovery.event.HeartbeatEvent;
+import org.springframework.cloud.client.discovery.event.HeartbeatMonitor;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.event.EventListener;
+
+/**
+ * Autoconfiguration for use when user has provided a
+ * {@link DiscoveryClientChannelLocator}. Listens for changes in the service registry and
+ * rebinds the external channels as needed.
+ *
+ * @author Dave Syer
+ *
+ */
+@Configuration
+@ConditionalOnClass(DiscoveryClient.class)
+public class DiscoveryClientAutoConfiguration {
+
+	private HeartbeatMonitor monitor = new HeartbeatMonitor();
+
+	@Autowired
+	private MessageBusAdapter adapter;
+
+	@Autowired(required = false)
+	@Upstream
+	private ChannelLocator inputChannelLocator;
+
+	@Autowired(required = false)
+	@Downstream
+	private ChannelLocator outputChannelLocator;
+
+	private boolean enabled = false;
+
+	@PostConstruct
+	public void init() {
+		if (this.inputChannelLocator instanceof DiscoveryClientChannelLocator
+				|| this.outputChannelLocator instanceof DiscoveryClientChannelLocator) {
+			this.enabled = true;
+		}
+	}
+
+	@EventListener
+	public void discoveryHeartbeat(HeartbeatEvent event) {
+		if (this.enabled && this.monitor.update(event.getValue())) {
+			this.adapter.rebind();
+		}
+	}
+
+}

spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientChannelLocator.java

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-package org.springframework.bus.runner.adapter;
+package org.springframework.bus.runner.adapter.discovery;
 
 import java.net.URI;
 import java.util.Collection;
@@ -24,6 +24,9 @@ import java.util.Random;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.bus.runner.adapter.ChannelLocator;
+import org.springframework.bus.runner.adapter.ChannelsMetadata;
+import org.springframework.bus.runner.adapter.InputChannelSpec;
 import org.springframework.cloud.client.ServiceInstance;
 import org.springframework.cloud.client.discovery.DiscoveryClient;
 import org.springframework.web.client.RestOperations;

spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java

@@ -30,10 +30,10 @@ import org.springframework.beans.factory.ListableBeanFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.bus.runner.adapter.ChannelLocator;
-import org.springframework.bus.runner.adapter.InputChannel;
+import org.springframework.bus.runner.adapter.Upstream;
 import org.springframework.bus.runner.adapter.InputChannelSpec;
 import org.springframework.bus.runner.adapter.MessageBusAdapter;
-import org.springframework.bus.runner.adapter.OutputChannel;
+import org.springframework.bus.runner.adapter.Downstream;
 import org.springframework.bus.runner.adapter.OutputChannelSpec;
 import org.springframework.bus.runner.endpoint.ChannelsEndpoint;
 import org.springframework.context.annotation.Bean;
@@ -60,11 +60,11 @@ public class MessageBusAdapterConfiguration {
 	private ListableBeanFactory beanFactory;
 
 	@Autowired(required=false)
-	@InputChannel
+	@Upstream
 	private ChannelLocator inputChannelLocator;
 
 	@Autowired(required=false)
-	@OutputChannel
+	@Downstream
 	private ChannelLocator outputChannelLocator;
 
 	@Bean

spring-bus-core/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,3 @@
+# AutoConfiguration
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+org.springframework.bus.runner.adapter.discovery.DiscoveryClientAutoConfiguration

spring-bus-core/src/test/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocatorTests.java

@@ -26,6 +26,7 @@ import java.util.HashSet;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
+import org.springframework.bus.runner.adapter.discovery.DiscoveryClientChannelLocator;
 import org.springframework.bus.runner.config.MessageBusProperties;
 import org.springframework.cloud.client.DefaultServiceInstance;
 import org.springframework.cloud.client.discovery.DiscoveryClient;