diff --git a/spring-bus-core/pom.xml b/spring-bus-core/pom.xml index 2d6071a6b..a32884eb4 100644 --- a/spring-bus-core/pom.xml +++ b/spring-bus-core/pom.xml @@ -27,6 +27,11 @@ org.springframework.boot spring-boot-starter-actuator + + org.springframework.cloud + spring-cloud-starter + true + org.springframework.boot spring-boot-starter-web @@ -57,6 +62,11 @@ spring-boot-starter-test test + + org.scala-lang + scala-library + 2.10.4 + diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/ChannelLocator.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/ChannelLocator.java new file mode 100644 index 000000000..b235c1ef5 --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/ChannelLocator.java @@ -0,0 +1,28 @@ +/* + * 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; + +/** + * @author Dave Syer + * + */ +// TODO: Use DestinationResolver? +public interface ChannelLocator { + + String locate(String name); + +} diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/ChannelsMetadata.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/ChannelsMetadata.java new file mode 100644 index 000000000..c3a32a4c2 --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/ChannelsMetadata.java @@ -0,0 +1,58 @@ +/* + * 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; + +import java.util.Collection; +import java.util.Collections; + +import org.springframework.bus.runner.config.MessageBusProperties; + +/** + * @author Dave Syer + * + */ +public class ChannelsMetadata { + + private Collection outputChannels = Collections.emptySet(); + private Collection inputChannels = Collections.emptySet(); + private MessageBusProperties module; + + public MessageBusProperties getModule() { + return module; + } + + public void setModule(MessageBusProperties module) { + this.module = module; + } + + public Collection getOutputChannels() { + return outputChannels; + } + + public void setOutputChannels(Collection outputChannels) { + this.outputChannels = outputChannels; + } + + public Collection getInputChannels() { + return inputChannels; + } + + public void setInputChannels(Collection inputChannels) { + this.inputChannels = inputChannels; + } + +} 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 new file mode 100644 index 000000000..5f17d0b23 --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DefaultChannelLocator.java @@ -0,0 +1,85 @@ +/* + * 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; + +import org.springframework.bus.runner.config.MessageBusProperties; + +import reactor.util.StringUtils; + +/** + * @author Dave Syer + * + */ +public class DefaultChannelLocator implements ChannelLocator { + + private MessageBusProperties module; + + public DefaultChannelLocator(MessageBusProperties module) { + this.module = module; + } + + @Override + public String locate(String name) { + String channelName = extractChannelName("input", name, this.module.getInputChannelName()); + if (channelName!=null) { + return channelName; + } + channelName = extractChannelName("output", name, this.module.getOutputChannelName()); + if (channelName!=null) { + return channelName; + } + return null; + } + + + private String extractChannelName(String start, String name, + String externalChannelName) { + if (name.equals(start)) { + return externalChannelName; + } + else if (name.startsWith(start + ".") || name.startsWith(start + "_")) { + String prefix = ""; + String channelName = name.substring(start.length() + 1); + if (channelName.contains(":")) { + String[] tokens = channelName.split(":", 2); + String type = tokens[0]; + if ("queue".equals(type)) { + // omit the type for a queue + if (StringUtils.hasText(tokens[1])) { + prefix = tokens[1] + "."; + } + } + else { + prefix = channelName + (channelName.endsWith(":") ? "" : "."); + } + } + else { + prefix = channelName + "."; + } + return prefix + getPlainChannelName(externalChannelName); + } + return null; + } + + private String getPlainChannelName(String name) { + if (name.contains(":")) { + name = name.substring(name.indexOf(":") + 1); + } + return name; + } + +} diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DiscoveryChannelLocator.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DiscoveryChannelLocator.java new file mode 100644 index 000000000..57e9b78ef --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/DiscoveryChannelLocator.java @@ -0,0 +1,73 @@ +/* + * 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; + +import java.net.URI; +import java.util.List; +import java.util.Random; + +import org.springframework.cloud.client.ServiceInstance; +import org.springframework.cloud.client.discovery.DiscoveryClient; +import org.springframework.web.client.RestTemplate; + +/** + * @author Dave Syer + * + */ +public class DiscoveryChannelLocator implements ChannelLocator { + + private DiscoveryClient discovery; + + private RestTemplate restTemplate = new RestTemplate(); + + private String serviceId; + + public DiscoveryChannelLocator(DiscoveryClient discovery, String serviceId) { + this.discovery = discovery; + this.serviceId = serviceId; + } + + @Override + public String locate(String name) { + List instances = discovery.getInstances(serviceId); + if (instances==null || instances.isEmpty()) { + return null; + } + URI uri = pickUrl(instances); + try { + ChannelsMetadata channels = restTemplate.getForObject(uri, ChannelsMetadata.class); + for (OutputChannelSpec spec : channels.getOutputChannels()) { + if (name.equals(spec.getLocalName())) { + return spec.getName(); + } + } + for (InputChannelSpec spec : channels.getInputChannels()) { + if (name.equals(spec.getLocalName())) { + return spec.getName(); + } + } + } catch (Exception e) { + return null; + } + return null; + } + + private URI pickUrl(List instances) { + return instances.get(new Random().nextInt(instances.size())).getUri(); + } + +} diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java index 949a77129..6e164b498 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java @@ -16,11 +16,6 @@ package org.springframework.bus.runner.adapter; -import org.springframework.integration.support.context.NamedComponent; -import org.springframework.messaging.MessageChannel; - -import com.fasterxml.jackson.annotation.JsonIgnore; - /** * @author Dave Syer * @@ -28,25 +23,22 @@ import com.fasterxml.jackson.annotation.JsonIgnore; public class InputChannelSpec { private String name; - private MessageChannel channel; + private String localName; - public InputChannelSpec(String name, MessageChannel channel) { - this.name = name; - this.channel = channel; + public InputChannelSpec(String localName) { + this.localName = localName; } public String getName() { - return name; + return this.name; + } + + public void setName(String name) { + this.name = name; } public String getLocalName() { - return (channel instanceof NamedComponent) ? ((NamedComponent) channel) - .getComponentName() : channel.toString(); - } - - @JsonIgnore - public MessageChannel getMessageChannel() { - return channel; + return this.localName; } } 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 ab6ec1664..9c0a49262 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 @@ -19,6 +19,7 @@ package org.springframework.bus.runner.adapter; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.Map; @@ -38,11 +39,13 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.interceptor.WireTap; import org.springframework.integration.support.DefaultMessageBuilderFactory; import org.springframework.integration.support.MessageBuilderFactory; +import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.core.DestinationResolver; import org.springframework.messaging.support.ChannelInterceptorAdapter; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -74,16 +77,38 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { private ConfigurableApplicationContext applicationContext; + private ChannelLocator inputChannelLocator; + + private ChannelLocator outputChannelLocator; + + private DestinationResolver channelResolver; + + private Map bindings = new HashMap(); + public MessageBusAdapter(MessageBusProperties module, MessageBus messageBus) { this.module = module; this.messageBus = messageBus; + this.inputChannelLocator = new DefaultChannelLocator(module); + this.outputChannelLocator = new DefaultChannelLocator(module); + } + + public void setInputChannelLocator(ChannelLocator channelLocator) { + this.inputChannelLocator = channelLocator; + } + + public void setOutputChannelLocator(ChannelLocator channelLocator) { + this.outputChannelLocator = channelLocator; } @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = (ConfigurableApplicationContext) applicationContext; + this.channelResolver = new BeanFactoryChannelResolver(applicationContext); + } + public void setChannelResolver(DestinationResolver channelResolver) { + this.channelResolver = channelResolver; } public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) { @@ -94,20 +119,6 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { this.trackHistory = trackHistory; } - public void setOutputChannel(MessageChannel outputChannel) { - if (outputChannel != null) { - String name = module.getOutputChannelName(); - this.outputChannels.add(new OutputChannelSpec(name, outputChannel)); - } - } - - public void setInputChannel(MessageChannel inputChannel) { - if (inputChannel != null) { - String name = module.getInputChannelName(); - this.inputChannels.add(new InputChannelSpec(name, inputChannel)); - } - } - public void setOutputChannels(Collection outputChannels) { this.outputChannels = new LinkedHashSet(outputChannels); } @@ -116,20 +127,25 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { this.inputChannels = new LinkedHashSet(inputChannels); } - public Collection getOutputChannels() { - return outputChannels; + public ChannelsMetadata getChannelsMetadata() { + ChannelsMetadata channels = new ChannelsMetadata(); + channels.setModule(this.module); + channels.setInputChannels(new LinkedHashSet(this.inputChannels)); + channels.setOutputChannels(new LinkedHashSet( + this.outputChannels)); + return channels; } public OutputChannelSpec getOutputChannel(String name) { - if (name==null) { + if (name == null) { return null; } - for (OutputChannelSpec spec : outputChannels) { + for (OutputChannelSpec spec : this.outputChannels) { if (name.equals(spec.getName())) { return spec; } } - for (OutputChannelSpec spec : outputChannels) { + for (OutputChannelSpec spec : this.outputChannels) { if (name.equals(spec.getLocalName())) { return spec; } @@ -138,15 +154,15 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { } public InputChannelSpec getInputChannel(String name) { - if (name==null) { + if (name == null) { return null; } - for (InputChannelSpec spec : inputChannels) { + for (InputChannelSpec spec : this.inputChannels) { if (name.equals(spec.getName())) { return spec; } } - for (InputChannelSpec spec : inputChannels) { + for (InputChannelSpec spec : this.inputChannels) { if (name.equals(spec.getLocalName())) { return spec; } @@ -154,111 +170,153 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { return null; } - public Collection getInputChannels() { - return inputChannels; - } - public void tap(String outputChannel) { OutputChannelSpec channel = getOutputChannel(outputChannel); - if (channel==null || channel.isTapped()) { + if (channel == null || channel.isTapped()) { return; } - createAndBindTapChannel(channel.getTapChannelName(), channel.getMessageChannel()); + createAndBindTapChannel(channel.getTapChannelName(), channel.getLocalName()); channel.setTapped(true); } public void untap(String outputChannel) { OutputChannelSpec channel = getOutputChannel(outputChannel); - if (channel==null || !channel.isTapped()) { + if (channel == null || !channel.isTapped()) { return; } String tapChannelName = channel.getTapChannelName(); - messageBus.unbindProducers(tapChannelName); + this.messageBus.unbindProducers(tapChannelName); channel.setTapped(false); } @Override @ManagedOperation public void start() { - if (!running) { + if (!this.running) { // Start everything, but don't call ourselves - if (!active.get()) { - if (active.compareAndSet(false, true)) { + if (!this.active.get()) { + if (this.active.compareAndSet(false, true)) { bindChannels(); - applicationContext.start(); - active.set(false); + this.applicationContext.start(); + this.active.set(false); } } } - running = true; + this.running = true; } @Override @ManagedOperation public void stop() { - if (running) { - if (!active.get()) { - if (active.compareAndSet(false, true)) { + if (this.running) { + if (!this.active.get()) { + if (this.active.compareAndSet(false, true)) { unbindChannels(); - applicationContext.stop(); - active.set(false); + this.applicationContext.stop(); + this.active.set(false); } } } - running = false; + this.running = false; } @Override @ManagedAttribute public boolean isRunning() { - return running && applicationContext.isRunning(); + return this.running && this.applicationContext.isRunning(); } protected final void unbindChannels() { - for (InputChannelSpec spec : inputChannels) { - messageBus.unbindConsumers(spec.getName()); + for (InputChannelSpec spec : this.inputChannels) { + String name = this.bindings.get(spec.getName()); + if (name == null) { + continue; + } + this.messageBus.unbindConsumers(name); } - for (OutputChannelSpec spec : outputChannels) { - messageBus.unbindProducers(spec.getName()); + for (OutputChannelSpec spec : this.outputChannels) { + String name = this.bindings.get(spec.getName()); + if (name == null) { + continue; + } + this.messageBus.unbindProducers(name); if (spec.isTapped()) { String tapChannelName = spec.getTapChannelName(); - messageBus.unbindProducers(tapChannelName); + this.messageBus.unbindProducers(tapChannelName); } } } protected final void bindChannels() { Map historyProperties = new LinkedHashMap(); - if (trackHistory) { + if (this.trackHistory) { // TODO: addHistoryTag(); } - for (OutputChannelSpec spec : outputChannels) { - String name = spec.getName(); - MessageChannel outputChannel = spec.getMessageChannel(); - bindMessageProducer(outputChannel, name, module.getProducerProperties()); + 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); + MessageChannel outputChannel = this.channelResolver.resolveDestination(spec + .getLocalName()); + bindMessageProducer(outputChannel, name, this.module.getProducerProperties()); if (spec.isTapped()) { - String tapChannelName = spec.getTapChannelName(); + String tapChannelName = getTapChannelName(name); + spec.setTapChannelName(tapChannelName); // tappableChannels.put(tapChannelName, outputChannel); // if (isTapActive(tapChannelName)) { - createAndBindTapChannel(tapChannelName, outputChannel); + createAndBindTapChannel(tapChannelName, name); // } } - if (trackHistory) { + if (this.trackHistory) { historyProperties.put("outputChannel", name); track(outputChannel, historyProperties); } } - for (InputChannelSpec spec : inputChannels) { - String name = spec.getName(); - MessageChannel inputChannel = spec.getMessageChannel(); - bindMessageConsumer(inputChannel, name, module.getConsumerProperties()); - if (trackHistory && outputChannels.size() != 1) { + 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()); + bindMessageConsumer(inputChannel, name, this.module.getConsumerProperties()); + if (this.trackHistory && this.outputChannels.size() != 1) { historyProperties.put("inputChannel", name); track(inputChannel, historyProperties); } } } + // TODO: move this to ChannelLocator? + private String getTapChannelName(String name) { + return !isDefaultOuputChannel(name) ? this.module + .getTapChannelName(getPlainChannelName(name)) : this.module + .getTapChannelName(); + } + + // TODO: move this to ChannelLocator? + private String getPlainChannelName(String name) { + if (name.contains(":")) { + name = name.substring(name.indexOf(":") + 1); + } + return name; + } + + // TODO: move this to ChannelLocator? + private boolean isDefaultOuputChannel(String channelName) { + if (channelName.contains(":")) { + String[] tokens = channelName.split(":", 2); + channelName = tokens[1]; + } + return channelName.equals(this.module.getOutputChannelName()); + } + /* * Following methods copied from parent to support the bindChannels() method above */ @@ -266,22 +324,24 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { private void bindMessageConsumer(MessageChannel inputChannel, String inputChannelName, Properties consumerProperties) { if (isChannelPubSub(inputChannelName)) { - messageBus.bindPubSubConsumer(inputChannelName, inputChannel, + this.messageBus.bindPubSubConsumer(inputChannelName, inputChannel, consumerProperties); } else { - messageBus.bindConsumer(inputChannelName, inputChannel, consumerProperties); + this.messageBus.bindConsumer(inputChannelName, inputChannel, + consumerProperties); } } private void bindMessageProducer(MessageChannel outputChannel, String outputChannelName, Properties producerProperties) { if (isChannelPubSub(outputChannelName)) { - messageBus.bindPubSubProducer(outputChannelName, outputChannel, + this.messageBus.bindPubSubProducer(outputChannelName, outputChannel, producerProperties); } else { - messageBus.bindProducer(outputChannelName, outputChannel, producerProperties); + this.messageBus.bindProducer(outputChannelName, outputChannel, + producerProperties); } } @@ -296,18 +356,19 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { * {@link MessageBus}'s message target. * * @param tapChannelName the name of the tap channel - * @param outputChannel the channel to tap + * @param localName the channel to tap */ - private void createAndBindTapChannel(String tapChannelName, - MessageChannel outputChannel) { + private void createAndBindTapChannel(String tapChannelName, String localName) { logger.info("creating and binding tap channel for {}", tapChannelName); - if (outputChannel instanceof ChannelInterceptorAware) { + MessageChannel channel = this.channelResolver.resolveDestination(localName); + if (channel instanceof ChannelInterceptorAware) { DirectChannel tapChannel = new DirectChannel(); tapChannel.setBeanName(tapChannelName + ".tap.bridge"); - messageBus.bindPubSubProducer(tapChannelName, tapChannel, null); // TODO tap - // producer - // props - tapOutputChannel(tapChannel, (ChannelInterceptorAware) outputChannel); + this.messageBus.bindPubSubProducer(tapChannelName, tapChannel, null); // TODO + // tap + // producer + // props + tapOutputChannel(tapChannel, (ChannelInterceptorAware) channel); } else { if (logger.isDebugEnabled()) { @@ -325,29 +386,30 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { private void track(MessageChannel channel, final Map historyProps) { if (channel instanceof ChannelInterceptorAware) { ((ChannelInterceptorAware) channel) - .addInterceptor(new ChannelInterceptorAdapter() { + .addInterceptor(new ChannelInterceptorAdapter() { - @Override - public Message preSend(Message message, - MessageChannel channel) { - @SuppressWarnings("unchecked") - Collection> history = (Collection>) message - .getHeaders().get(XdHeaders.XD_HISTORY); - if (history == null) { - history = new ArrayList>(1); - } - else { - history = new ArrayList>(history); - } - Map map = new LinkedHashMap(); - map.putAll(historyProps); - map.put("thread", Thread.currentThread().getName()); - history.add(map); - Message out = messageBuilderFactory.fromMessage(message) - .setHeader(XdHeaders.XD_HISTORY, history).build(); - return out; - } - }); + @Override + public Message preSend(Message message, + MessageChannel channel) { + @SuppressWarnings("unchecked") + Collection> history = (Collection>) message + .getHeaders().get(XdHeaders.XD_HISTORY); + if (history == null) { + history = new ArrayList>(1); + } + else { + history = new ArrayList>(history); + } + Map map = new LinkedHashMap(); + map.putAll(historyProps); + map.put("thread", Thread.currentThread().getName()); + history.add(map); + Message out = MessageBusAdapter.this.messageBuilderFactory + .fromMessage(message) + .setHeader(XdHeaders.XD_HISTORY, history).build(); + return out; + } + }); } } diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java index 710f839f9..0ebb9cdad 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java @@ -16,7 +16,6 @@ package org.springframework.bus.runner.adapter; -import org.springframework.messaging.MessageChannel; /** * @author Dave Syer @@ -27,24 +26,24 @@ public class OutputChannelSpec extends InputChannelSpec { private boolean tapped = false; private String tapChannelName; - public OutputChannelSpec(String name, MessageChannel channel) { - super(name, channel); + public OutputChannelSpec(String localName) { + super(localName); } public boolean isTapped() { - return tapped; + return this.tapped; } - + public void setTapped(boolean tapped) { this.tapped = tapped; } - + public void setTapChannelName(String tapChannelName) { this.tapChannelName = tapChannelName; } public String getTapChannelName() { - return tapChannelName==null ? "" : tapChannelName; + return this.tapChannelName==null ? "" : this.tapChannelName; } } 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 cdd69a87c..e620d6554 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 @@ -32,6 +32,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.bus.runner.adapter.InputChannelSpec; import org.springframework.bus.runner.adapter.MessageBusAdapter; import org.springframework.bus.runner.adapter.OutputChannelSpec; +import org.springframework.bus.runner.endpoint.ChannelsEndpoint; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.ImportResource; @@ -40,8 +41,6 @@ import org.springframework.util.Assert; import org.springframework.xd.dirt.integration.bus.MessageBus; import org.springframework.xd.dirt.integration.bus.MessageBusAwareRouterBeanPostProcessor; -import reactor.util.StringUtils; - /** * @author Dave Syer * @@ -68,84 +67,28 @@ public class MessageBusAdapterConfiguration { @Bean public ChannelsEndpoint channelsEndpoint(MessageBusAdapter adapter) { - return new ChannelsEndpoint(module, adapter); + return new ChannelsEndpoint(adapter); } protected Collection getOutputChannels() { Set channels = new LinkedHashSet(); - String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(beanFactory, - MessageChannel.class); + String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors( + this.beanFactory, MessageChannel.class); for (String name : names) { - String channelName = extractChannelName("output", name, - module.getOutputChannelName()); - if (channelName != null) { - OutputChannelSpec channel = new OutputChannelSpec(channelName, - beanFactory.getBean(name, MessageChannel.class)); - String tapChannelName = !isDefaultOuputChannel(channelName) ? module - .getTapChannelName(getPlainChannelName(channel.getName())) - : module.getTapChannelName(); - channel.setTapChannelName(tapChannelName); - channel.setTapped(false); - channels.add(channel); + if (name.startsWith("output")) { + channels.add(new OutputChannelSpec(name)); } } return channels; } - private boolean isDefaultOuputChannel(String channelName) { - if (channelName.contains(":")) { - String[] tokens = channelName.split(":", 2); - channelName = tokens[1]; - } - return channelName.equals(module.getOutputChannelName()); - } - - private String extractChannelName(String start, String name, - String externalChannelName) { - if (name.equals(start)) { - return externalChannelName; - } - else if (name.startsWith(start + ".") || name.startsWith(start + "_")) { - String prefix = ""; - String channelName = name.substring(start.length() + 1); - if (channelName.contains(":")) { - String[] tokens = channelName.split(":", 2); - String type = tokens[0]; - if ("queue".equals(type)) { - // omit the type for a queue - if (StringUtils.hasText(tokens[1])) { - prefix = tokens[1] + "."; - } - } - else { - prefix = channelName + (channelName.endsWith(":") ? "" : "."); - } - } - else { - prefix = channelName + "."; - } - return prefix + getPlainChannelName(externalChannelName); - } - return null; - } - - private String getPlainChannelName(String name) { - if (name.contains(":")) { - name = name.substring(name.indexOf(":") + 1); - } - return name; - } - protected Collection getInputChannels() { Set channels = new LinkedHashSet(); - String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(beanFactory, - MessageChannel.class); + String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors( + this.beanFactory, MessageChannel.class); for (String name : names) { - String channelName = extractChannelName("input", name, - module.getInputChannelName()); - if (channelName != null) { - channels.add(new InputChannelSpec(channelName, beanFactory.getBean(name, - MessageChannel.class))); + if (name.startsWith("input")) { + channels.add(new InputChannelSpec(name)); } } return channels; @@ -161,11 +104,11 @@ public class MessageBusAdapterConfiguration { @Bean public MessageBusAwareRouterBeanPostProcessor messageBusAwareRouterBeanPostProcessor() { - return new MessageBusAwareRouterBeanPostProcessor( - createLazyProxy(beanFactory, MessageBus.class), new Properties()); + return new MessageBusAwareRouterBeanPostProcessor(createLazyProxy( + this.beanFactory, MessageBus.class), new Properties()); } - private T createLazyProxy(ListableBeanFactory beanFactory, Class type) { + private T createLazyProxy(ListableBeanFactory beanFactory, Class type) { ProxyFactory factory = new ProxyFactory(); LazyInitTargetSource source = new LazyInitTargetSource(); source.setTargetClass(type); @@ -180,9 +123,10 @@ public class MessageBusAdapterConfiguration { } private String getBeanNameFor(ListableBeanFactory beanFactory, Class type) { - String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(beanFactory, type, false, false); - Assert.state(names.length==1, "No unique MessageBus (found " + names.length + ": " - + Arrays.asList(names) + ")"); + String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors( + beanFactory, type, false, false); + Assert.state(names.length == 1, "No unique MessageBus (found " + names.length + + ": " + Arrays.asList(names) + ")"); return names[0]; } diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java index 204bda274..f846acea4 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java @@ -48,9 +48,11 @@ public class MessageBusProperties { private Properties consumerProperties = new Properties(); private Properties producerProperties = new Properties(); - + private Tap tap; + private Discovery discovery = new Discovery(); + private boolean autoStartup = true; public String getName() { @@ -79,10 +81,11 @@ public class MessageBusProperties { public String getInputChannelName() { if (isTap()) { - return String.format("%s.%s.%s", BusUtils.constructTapPrefix(tap.getGroup()), tap.getName(), tap.getIndex()); + return String.format("%s.%s.%s", BusUtils.constructTapPrefix(tap.getGroup()), + tap.getName(), tap.getIndex()); } return (inputChannelName != null) ? inputChannelName : BusUtils - .constructPipeName(group, index>0 ? index - 1 : index); + .constructPipeName(group, index > 0 ? index - 1 : index); } public String getOutputChannelName() { @@ -97,7 +100,8 @@ public class MessageBusProperties { public String getTapChannelName(String prefix) { Assert.isTrue(!type.equals("job"), "Job module type not supported."); // for Stream return channel name with indexed elements - return String.format("%s.%s.%s", BusUtils.constructTapPrefix(prefix), name, index); + return String + .format("%s.%s.%s", BusUtils.constructTapPrefix(prefix), name, index); } public void setOutputChannelName(String outputChannelName) { @@ -131,11 +135,11 @@ public class MessageBusProperties { public void setProducerProperties(Properties producerProperties) { this.producerProperties = producerProperties; } - + public boolean isAutoStartup() { return autoStartup; } - + public void setAutoStartup(boolean autoStartup) { this.autoStartup = autoStartup; } @@ -147,19 +151,58 @@ public class MessageBusProperties { public void setTap(Tap tap) { this.tap = tap; } - + private boolean isTap() { - if (tap!=null) { - Assert.state(tap.getName()!=null, "Tap name not provided"); - Assert.state(!tap.getGroup().equals(group), "Tap group cannot be the same as module group"); + if (tap != null) { + Assert.state(tap.getName() != null, "Tap name not provided"); + Assert.state(!tap.getGroup().equals(group), + "Tap group cannot be the same as module group"); } - return tap!=null; + return tap != null; + } + + public Discovery getDiscovery() { + return discovery; + } + + public static class Discovery { + + private boolean enabled = false; + + private String inputServiceId; + + private String outputServiceId; + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public String getInputServiceId() { + return inputServiceId; + } + + public void setInputServiceId(String inputServiceId) { + this.inputServiceId = inputServiceId; + } + + public String getOutputServiceId() { + return outputServiceId; + } + + public void setOutputServiceId(String outputServiceId) { + this.outputServiceId = outputServiceId; + } + } public static class Tap { private String group = "group"; - + private String name; public String getName() { diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/ChannelsEndpoint.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/endpoint/ChannelsEndpoint.java similarity index 81% rename from spring-bus-core/src/main/java/org/springframework/bus/runner/config/ChannelsEndpoint.java rename to spring-bus-core/src/main/java/org/springframework/bus/runner/endpoint/ChannelsEndpoint.java index 5919d2588..9686a6ffb 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/config/ChannelsEndpoint.java +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/endpoint/ChannelsEndpoint.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.bus.runner.config; +package org.springframework.bus.runner.endpoint; import java.util.ArrayList; import java.util.LinkedHashMap; @@ -22,6 +22,7 @@ import java.util.List; import java.util.Map; import org.springframework.boot.actuate.endpoint.AbstractEndpoint; +import org.springframework.bus.runner.adapter.ChannelsMetadata; import org.springframework.bus.runner.adapter.MessageBusAdapter; import org.springframework.bus.runner.adapter.OutputChannelSpec; import org.springframework.web.bind.annotation.RequestMapping; @@ -31,20 +32,18 @@ import org.springframework.web.bind.annotation.RestController; @RestController public class ChannelsEndpoint extends AbstractEndpoint> { - - private MessageBusProperties module; + private MessageBusAdapter adapter; - public ChannelsEndpoint(MessageBusProperties module, MessageBusAdapter adapter) { + public ChannelsEndpoint(MessageBusAdapter adapter) { super("channels"); - this.module = module; this.adapter = adapter; } @RequestMapping(value="/channels/taps") public List taps() { List list = new ArrayList(); - for (OutputChannelSpec spec : adapter.getOutputChannels()) { + for (OutputChannelSpec spec : adapter.getChannelsMetadata().getOutputChannels()) { if (spec.isTapped()) { list.add(spec); } @@ -67,9 +66,10 @@ public class ChannelsEndpoint extends AbstractEndpoint> { @Override public Map invoke() { LinkedHashMap map = new LinkedHashMap(); - map.put("inputChannels", adapter.getInputChannels()); - map.put("outputChannels", adapter.getOutputChannels()); - map.put("module", module); + ChannelsMetadata channels = adapter.getChannelsMetadata(); + map.put("inputChannels", channels.getInputChannels()); + map.put("outputChannels", channels.getOutputChannels()); + map.put("module", channels.getModule()); return map; } diff --git a/spring-bus-core/src/test/java/org/springframework/bus/runner/adapter/DefaultChannelLocatorTests.java b/spring-bus-core/src/test/java/org/springframework/bus/runner/adapter/DefaultChannelLocatorTests.java new file mode 100644 index 000000000..38f9b8f21 --- /dev/null +++ b/spring-bus-core/src/test/java/org/springframework/bus/runner/adapter/DefaultChannelLocatorTests.java @@ -0,0 +1,87 @@ +/* + * 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; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; +import org.springframework.bus.runner.config.MessageBusProperties; + +/** + * @author Dave Syer + * + */ +public class DefaultChannelLocatorTests { + + private MessageBusProperties module = new MessageBusProperties(); + + private DefaultChannelLocator locator = new DefaultChannelLocator(this.module); + + @Test + public void oneOutput() throws Exception { + assertEquals("group.0", this.locator.locate("output")); + } + + @Test + public void oneOutputTopic() throws Exception { + assertEquals("topic:group.0", this.locator.locate("output.topic:")); + } + + @Test + public void outputWithNamedTopic() throws Exception { + assertEquals("topic:foo.group.0", this.locator.locate("output.topic:foo")); + } + + @Test + public void outputWithNamedQueue() throws Exception { + assertEquals("foo.group.0", this.locator.locate("output.queue:foo")); + } + + @Test + public void overrideNaturalOutputChannelName() throws Exception { + this.module.setOutputChannelName("bar"); + assertEquals("foo.bar", this.locator.locate("output.queue:foo")); + } + + @Test + public void noQueueQualifier() throws Exception { + assertEquals("foo.group.0", this.locator.locate("output.foo")); + } + + @Test + public void underscoreSeparatorForChannelName() throws Exception { + assertEquals("foo.group.0", this.locator.locate("output_foo")); + } + + @Test + public void overrideNaturalOutputChannelNamedQueue() throws Exception { + this.module.setOutputChannelName("queue:bar"); + assertEquals("foo.bar", this.locator.locate("output.foo")); + } + + @Test + public void overrideNaturalOutputChannelNamedQueueWithTopic() throws Exception { + this.module.setOutputChannelName("queue:bar"); + assertEquals("topic:foo.bar", this.locator.locate("output.topic:foo")); + } + + @Test + public void overrideNaturalOutputChannelNamedTopic() throws Exception { + this.module.setOutputChannelName("topic:bar"); + assertEquals("foo.bar", this.locator.locate("output.queue:foo")); + } + +} diff --git a/spring-bus-core/src/test/java/org/springframework/bus/runner/config/MessageBusAdapterConfigurationTests.java b/spring-bus-core/src/test/java/org/springframework/bus/runner/config/MessageBusAdapterConfigurationTests.java index 2e41abcf2..80dc286bc 100644 --- a/spring-bus-core/src/test/java/org/springframework/bus/runner/config/MessageBusAdapterConfigurationTests.java +++ b/spring-bus-core/src/test/java/org/springframework/bus/runner/config/MessageBusAdapterConfigurationTests.java @@ -22,12 +22,14 @@ import java.util.ArrayList; import java.util.Collection; import java.util.List; +import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.boot.test.SpringApplicationConfiguration; import org.springframework.bus.runner.adapter.InputChannelSpec; +import org.springframework.bus.runner.adapter.MessageBusAdapter; import org.springframework.bus.runner.adapter.OutputChannelSpec; import org.springframework.bus.runner.config.MessageBusAdapterConfigurationTests.Empty; import org.springframework.context.annotation.Bean; @@ -45,52 +47,65 @@ import org.springframework.xd.dirt.integration.bus.local.LocalMessageBus; */ @RunWith(SpringJUnit4ClassRunner.class) @SpringApplicationConfiguration(classes = Empty.class) -@DirtiesContext(classMode=ClassMode.AFTER_EACH_TEST_METHOD) +@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD) public class MessageBusAdapterConfigurationTests { @Autowired private DefaultListableBeanFactory context; + @Autowired + private MessageBusAdapter adapter; + @Autowired private MessageBusAdapterConfiguration configuration; @Autowired private MessageBusProperties module; + @Before + public void init() { + } + @Test public void oneOutput() throws Exception { - context.registerSingleton("output", new DirectChannel()); - Collection channels = configuration.getOutputChannels(); + this.context.registerSingleton("output", new DirectChannel()); + refresh(); + Collection channels = this.adapter.getChannelsMetadata() + .getOutputChannels(); assertEquals(1, channels.size()); assertEquals("group.0", channels.iterator().next().getName()); - assertEquals("tap:stream:group.module.0", channels.iterator().next().getTapChannelName()); + assertEquals("tap:stream:group.module.0", channels.iterator().next() + .getTapChannelName()); + } + + private void refresh() { + Collection channels = this.configuration.getOutputChannels(); + for (OutputChannelSpec channel : channels) { + channel.setTapped(true); + } + this.adapter.setOutputChannels(channels); + this.adapter.start(); } @Test public void oneOutputTopic() throws Exception { - context.registerSingleton("output.topic:", new DirectChannel()); - Collection channels = configuration.getOutputChannels(); + this.context.registerSingleton("output.topic:", new DirectChannel()); + refresh(); + Collection channels = this.adapter.getChannelsMetadata() + .getOutputChannels(); assertEquals(1, channels.size()); assertEquals("topic:group.0", channels.iterator().next().getName()); - assertEquals("tap:stream:group.module.0", channels.iterator().next().getTapChannelName()); - } - - @Test - public void twoOutputsWithTopic() throws Exception { - context.registerSingleton("output", new DirectChannel()); - context.registerSingleton("output.topic:foo", new DirectChannel()); - Collection channels = configuration.getOutputChannels(); - List names = getChannelNames(channels); - assertEquals(2, channels.size()); - assertTrue(names.contains("group.0")); - assertTrue(names.contains("topic:foo.group.0")); + assertEquals("tap:stream:group.module.0", channels.iterator().next() + .getTapChannelName()); } @Test public void twoOutputsWithQueue() throws Exception { - context.registerSingleton("output", new DirectChannel()); - context.registerSingleton("output.queue:foo", new DirectChannel()); - Collection channels = configuration.getOutputChannels(); + this.context.registerSingleton("output", new DirectChannel()); + this.context.registerSingleton("output.queue:foo", new DirectChannel()); + refresh(); + Collection channels = this.adapter.getChannelsMetadata() + .getOutputChannels(); List names = getChannelNames(channels); assertEquals(2, channels.size()); assertTrue(names.contains("group.0")); @@ -102,62 +117,34 @@ public class MessageBusAdapterConfigurationTests { for (InputChannelSpec spec : channels) { list.add(spec.getName()); } - return list ; + return list; } @Test public void overrideNaturalOutputChannelName() throws Exception { - module.setOutputChannelName("bar"); - context.registerSingleton("output.queue:foo", new DirectChannel()); - Collection channels = configuration.getOutputChannels(); + this.module.setOutputChannelName("bar"); + this.context.registerSingleton("output.queue:foo", new DirectChannel()); + refresh(); + Collection channels = this.adapter.getChannelsMetadata() + .getOutputChannels(); assertEquals(1, channels.size()); assertEquals("foo.bar", channels.iterator().next().getName()); // TODO: fix this. What should it be? - assertEquals("tap:stream:foo.bar.module.0", channels.iterator().next().getTapChannelName()); - } - - @Test - public void noQueueQualifier() throws Exception { - context.registerSingleton("output.foo", new DirectChannel()); - Collection channels = configuration.getOutputChannels(); - assertEquals(1, channels.size()); - assertEquals("foo.group.0", channels.iterator().next().getName()); - } - - @Test - public void underscoreSeparatorForChannelName() throws Exception { - context.registerSingleton("output_foo", new DirectChannel()); - Collection channels = configuration.getOutputChannels(); - assertEquals(1, channels.size()); - assertEquals("foo.group.0", channels.iterator().next().getName()); - } - - @Test - public void overrideNaturalOutputChannelNamedQueue() throws Exception { - module.setOutputChannelName("queue:bar"); - context.registerSingleton("output.queue:foo", new DirectChannel()); - Collection channels = configuration.getOutputChannels(); - assertEquals(1, channels.size()); - assertEquals("foo.bar", channels.iterator().next().getName()); + assertEquals("tap:stream:foo.bar.module.0", channels.iterator().next() + .getTapChannelName()); } @Test public void overrideNaturalOutputChannelNamedQueueWithTopic() throws Exception { - module.setOutputChannelName("queue:bar"); - context.registerSingleton("output.topic:foo", new DirectChannel()); - Collection channels = configuration.getOutputChannels(); + this.module.setOutputChannelName("queue:bar"); + this.context.registerSingleton("output.topic:foo", new DirectChannel()); + refresh(); + Collection channels = this.adapter.getChannelsMetadata() + .getOutputChannels(); assertEquals(1, channels.size()); assertEquals("topic:foo.bar", channels.iterator().next().getName()); - assertEquals("tap:stream:foo.bar.module.0", channels.iterator().next().getTapChannelName()); - } - - @Test - public void overrideNaturalOutputChannelNamedTopic() throws Exception { - module.setOutputChannelName("topic:bar"); - context.registerSingleton("output.queue:foo", new DirectChannel()); - Collection channels = configuration.getOutputChannels(); - assertEquals(1, channels.size()); - assertEquals("foo.bar", channels.iterator().next().getName()); + assertEquals("tap:stream:foo.bar.module.0", channels.iterator().next() + .getTapChannelName()); } @Configuration