diff --git a/pom.xml b/pom.xml index df9b8a7a7..ae662e34c 100644 --- a/pom.xml +++ b/pom.xml @@ -12,8 +12,8 @@ 1.7 - 1.2.3.RELEASE - 1.0.2.RELEASE + 1.3.0.BUILD-SNAPSHOT + 1.0.3.BUILD-SNAPSHOT 1.2.0.BUILD-SNAPSHOT diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DefaultChannelLocator.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DefaultChannelLocator.java index 5f17d0b23..94f0eac45 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DefaultChannelLocator.java +++ b/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 diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannel.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/Downstream.java similarity index 97% rename from spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannel.java rename to spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/Downstream.java index ff3ba2b9c..87a5c0992 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannel.java +++ b/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 { } diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java index 9c0a49262..df5bfd6e5 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/MessageBusAdapter.java +++ b/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 historyProperties = new LinkedHashMap(); 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? diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannel.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/Upstream.java similarity index 97% rename from spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannel.java rename to spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/Upstream.java index 459a68749..8ec070d9a 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannel.java +++ b/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 { } diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientAutoConfiguration.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientAutoConfiguration.java new file mode 100644 index 000000000..f93ed6936 --- /dev/null +++ b/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(); + } + } + +} diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocator.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientChannelLocator.java similarity index 92% rename from spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocator.java rename to spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientChannelLocator.java index d598ae547..8877a2ea3 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocator.java +++ b/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; diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java index 5a5aaa727..c31e609db 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusAdapterConfiguration.java +++ b/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 diff --git a/spring-bus-core/src/main/resources/META-INF/spring.factories b/spring-bus-core/src/main/resources/META-INF/spring.factories new file mode 100644 index 000000000..f30e7d9e3 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/spring-bus-core/src/test/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocatorTests.java b/spring-bus-core/src/test/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocatorTests.java index 46e16de66..83de2d051 100644 --- a/spring-bus-core/src/test/java/org/springframework/bus/runner/adapter/DiscoveryClientChannelLocatorTests.java +++ b/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;