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/ChannelBinding.java similarity index 70% rename from spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java rename to spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/ChannelBinding.java index 01deb70a9..2c9cc4600 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/ChannelBinding.java @@ -17,32 +17,34 @@ package org.springframework.bus.runner.adapter; /** - * @author Dave Syer + * Represents a binding between a local and remote message channel. * + * @author Dave Syer + * @author Mark Fisher */ -public class InputChannelSpec { +public abstract class ChannelBinding { - private String name; private String localName; + private String remoteName; - protected InputChannelSpec() { + protected ChannelBinding() { this(null); } - public InputChannelSpec(String localName) { + protected ChannelBinding(String localName) { this.localName = localName; } - public String getName() { - return this.name; - } - - public void setName(String name) { - this.name = name; - } - public String getLocalName() { return this.localName; } + public String getRemoteName() { + return this.remoteName; + } + + public void setRemoteName(String name) { + this.remoteName = 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 index c3a32a4c2..90fc03a33 100644 --- 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 @@ -23,12 +23,11 @@ import org.springframework.bus.runner.config.MessageBusProperties; /** * @author Dave Syer - * */ public class ChannelsMetadata { - private Collection outputChannels = Collections.emptySet(); - private Collection inputChannels = Collections.emptySet(); + private Collection outputChannels = Collections.emptySet(); + private Collection inputChannels = Collections.emptySet(); private MessageBusProperties module; public MessageBusProperties getModule() { @@ -39,19 +38,19 @@ public class ChannelsMetadata { this.module = module; } - public Collection getOutputChannels() { + public Collection getOutputChannels() { return outputChannels; } - public void setOutputChannels(Collection outputChannels) { + public void setOutputChannels(Collection outputChannels) { this.outputChannels = outputChannels; } - public Collection getInputChannels() { + public Collection getInputChannels() { return inputChannels; } - public void setInputChannels(Collection 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 index 94f0eac45..b17c90374 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 @@ -21,7 +21,6 @@ import org.springframework.util.StringUtils; /** * @author Dave Syer - * */ public class DefaultChannelLocator implements ChannelLocator { @@ -45,8 +44,7 @@ public class DefaultChannelLocator implements ChannelLocator { } - private String extractChannelName(String start, String name, - String externalChannelName) { + private String extractChannelName(String start, String name, String externalChannelName) { if (name.equals(start)) { return externalChannelName; } diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelBinding.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelBinding.java new file mode 100644 index 000000000..3871293bc --- /dev/null +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelBinding.java @@ -0,0 +1,32 @@ +/* + * 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 + */ +public class InputChannelBinding extends ChannelBinding { + + protected InputChannelBinding() { + super(null); + } + + public InputChannelBinding(String localName) { + super(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 df5bfd6e5..64d21ad59 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 @@ -64,8 +64,8 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { private MessageBus messageBus; private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); - private Collection outputChannels = Collections.emptySet(); - private Collection inputChannels = Collections.emptySet(); + private Collection outputChannels = Collections.emptySet(); + private Collection inputChannels = Collections.emptySet(); private boolean running = false; @@ -101,8 +101,7 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { } @Override - public void setApplicationContext(ApplicationContext applicationContext) - throws BeansException { + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = (ConfigurableApplicationContext) applicationContext; this.channelResolver = new BeanFactoryChannelResolver(applicationContext); } @@ -119,59 +118,58 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { this.trackHistory = trackHistory; } - public void setOutputChannels(Collection outputChannels) { - this.outputChannels = new LinkedHashSet(outputChannels); + public void setOutputChannels(Collection outputChannels) { + this.outputChannels = new LinkedHashSet(outputChannels); } - public void setInputChannels(Collection inputChannels) { - this.inputChannels = new LinkedHashSet(inputChannels); + public void setInputChannels(Collection inputChannels) { + this.inputChannels = new LinkedHashSet(inputChannels); } public ChannelsMetadata getChannelsMetadata() { ChannelsMetadata channels = new ChannelsMetadata(); channels.setModule(this.module); - channels.setInputChannels(new LinkedHashSet(this.inputChannels)); - channels.setOutputChannels(new LinkedHashSet( - this.outputChannels)); + channels.setInputChannels(new LinkedHashSet(this.inputChannels)); + channels.setOutputChannels(new LinkedHashSet(this.outputChannels)); return channels; } - public OutputChannelSpec getOutputChannel(String name) { + public OutputChannelBinding getOutputChannel(String name) { if (name == null) { return null; } - for (OutputChannelSpec spec : this.outputChannels) { - if (name.equals(spec.getName())) { - return spec; + for (OutputChannelBinding binding : this.outputChannels) { + if (name.equals(binding.getRemoteName())) { + return binding; } } - for (OutputChannelSpec spec : this.outputChannels) { - if (name.equals(spec.getLocalName())) { - return spec; + for (OutputChannelBinding binding : this.outputChannels) { + if (name.equals(binding.getLocalName())) { + return binding; } } return null; } - public InputChannelSpec getInputChannel(String name) { + public InputChannelBinding getInputChannel(String name) { if (name == null) { return null; } - for (InputChannelSpec spec : this.inputChannels) { - if (name.equals(spec.getName())) { - return spec; + for (InputChannelBinding binding : this.inputChannels) { + if (name.equals(binding.getRemoteName())) { + return binding; } } - for (InputChannelSpec spec : this.inputChannels) { - if (name.equals(spec.getLocalName())) { - return spec; + for (InputChannelBinding binding : this.inputChannels) { + if (name.equals(binding.getLocalName())) { + return binding; } } return null; } public void tap(String outputChannel) { - OutputChannelSpec channel = getOutputChannel(outputChannel); + OutputChannelBinding channel = getOutputChannel(outputChannel); if (channel == null || channel.isTapped()) { return; } @@ -180,7 +178,7 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { } public void untap(String outputChannel) { - OutputChannelSpec channel = getOutputChannel(outputChannel); + OutputChannelBinding channel = getOutputChannel(outputChannel); if (channel == null || !channel.isTapped()) { return; } @@ -240,21 +238,21 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { } protected final void unbindChannels() { - for (InputChannelSpec spec : this.inputChannels) { - String name = this.bindings.get(spec.getName()); + for (InputChannelBinding binding : this.inputChannels) { + String name = this.bindings.get(binding.getRemoteName()); if (name == null) { continue; } this.messageBus.unbindConsumers(name); } - for (OutputChannelSpec spec : this.outputChannels) { - String name = this.bindings.get(spec.getName()); + for (OutputChannelBinding binding : this.outputChannels) { + String name = this.bindings.get(binding.getRemoteName()); if (name == null) { continue; } this.messageBus.unbindProducers(name); - if (spec.isTapped()) { - String tapChannelName = spec.getTapChannelName(); + if (binding.isTapped()) { + String tapChannelName = binding.getTapChannelName(); this.messageBus.unbindProducers(tapChannelName); } } @@ -268,14 +266,13 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { if (this.trackHistory) { // TODO: addHistoryTag(); } - for (OutputChannelSpec spec : this.outputChannels) { - String name = spec.getName(); - MessageChannel outputChannel = this.channelResolver.resolveDestination(spec - .getLocalName()); + for (OutputChannelBinding binding : this.outputChannels) { + String name = binding.getRemoteName(); + MessageChannel outputChannel = this.channelResolver.resolveDestination(binding.getLocalName()); bindMessageProducer(outputChannel, name, this.module.getProducerProperties()); - if (spec.isTapped()) { + if (binding.isTapped()) { String tapChannelName = getTapChannelName(name); - spec.setTapChannelName(tapChannelName); + binding.setTapChannelName(tapChannelName); // tappableChannels.put(tapChannelName, outputChannel); // if (isTapActive(tapChannelName)) { createAndBindTapChannel(tapChannelName, name); @@ -286,10 +283,9 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { track(outputChannel, historyProperties); } } - for (InputChannelSpec spec : this.inputChannels) { - String name = spec.getName(); - MessageChannel inputChannel = this.channelResolver.resolveDestination(spec - .getLocalName()); + for (InputChannelBinding binding : this.inputChannels) { + String name = binding.getRemoteName(); + MessageChannel inputChannel = this.channelResolver.resolveDestination(binding.getLocalName()); bindMessageConsumer(inputChannel, name, this.module.getConsumerProperties()); if (this.trackHistory && this.outputChannels.size() != 1) { historyProperties.put("inputChannel", name); @@ -302,32 +298,31 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { private boolean locateChannels() { logger.info("Locating channels"); boolean located = true; - for (OutputChannelSpec spec : this.outputChannels) { - String name = this.outputChannelLocator.locate(spec.getLocalName()); + for (OutputChannelBinding binding : this.outputChannels) { + String name = this.outputChannelLocator.locate(binding.getLocalName()); if (name == null) { - logger.info("No channel found for: " + spec.getLocalName()); + logger.info("No channel found for: " + binding.getLocalName()); located = false; } - spec.setName(name); - this.bindings.put(spec.getName(), name); + binding.setRemoteName(name); + this.bindings.put(binding.getRemoteName(), name); } - for (InputChannelSpec spec : this.inputChannels) { - String name = this.inputChannelLocator.locate(spec.getLocalName()); + for (InputChannelBinding binding : this.inputChannels) { + String name = this.inputChannelLocator.locate(binding.getLocalName()); if (name == null) { - logger.info("No channel found for: " + spec.getLocalName()); + logger.info("No channel found for: " + binding.getLocalName()); located = false; } - spec.setName(name); - this.bindings.put(spec.getName(), name); + binding.setRemoteName(name); + this.bindings.put(binding.getRemoteName(), name); } return located; } // TODO: move this to ChannelLocator? private String getTapChannelName(String name) { - return !isDefaultOuputChannel(name) ? this.module - .getTapChannelName(getPlainChannelName(name)) : this.module - .getTapChannelName(); + return !isDefaultOuputChannel(name) ? this.module.getTapChannelName(getPlainChannelName(name)) + : this.module.getTapChannelName(); } // TODO: move this to ChannelLocator? @@ -354,30 +349,25 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { private void bindMessageConsumer(MessageChannel inputChannel, String inputChannelName, Properties consumerProperties) { if (isChannelPubSub(inputChannelName)) { - this.messageBus.bindPubSubConsumer(inputChannelName, inputChannel, - consumerProperties); + this.messageBus.bindPubSubConsumer(inputChannelName, inputChannel, consumerProperties); } else { - this.messageBus.bindConsumer(inputChannelName, inputChannel, - consumerProperties); + this.messageBus.bindConsumer(inputChannelName, inputChannel, consumerProperties); } } private void bindMessageProducer(MessageChannel outputChannel, String outputChannelName, Properties producerProperties) { if (isChannelPubSub(outputChannelName)) { - this.messageBus.bindPubSubProducer(outputChannelName, outputChannel, - producerProperties); + this.messageBus.bindPubSubProducer(outputChannelName, outputChannel, producerProperties); } else { - this.messageBus.bindProducer(outputChannelName, outputChannel, - producerProperties); + this.messageBus.bindProducer(outputChannelName, outputChannel, producerProperties); } } private boolean isChannelPubSub(String channelName) { - Assert.isTrue(StringUtils.hasText(channelName), - "Channel name should not be empty/null."); + Assert.isTrue(StringUtils.hasText(channelName), "Channel name should not be empty/null."); return (channelName.startsWith("tap:") || channelName.startsWith("topic:")); } @@ -407,8 +397,7 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { } } - private MessageChannel tapOutputChannel(MessageChannel tapChannel, - ChannelInterceptorAware outputChannel) { + private MessageChannel tapOutputChannel(MessageChannel tapChannel, ChannelInterceptorAware outputChannel) { outputChannel.addInterceptor(new WireTap(tapChannel)); return tapChannel; } @@ -419,8 +408,7 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { .addInterceptor(new ChannelInterceptorAdapter() { @Override - public Message preSend(Message message, - MessageChannel channel) { + public Message preSend(Message message, MessageChannel channel) { @SuppressWarnings("unchecked") Collection> history = (Collection>) message .getHeaders().get(XdHeaders.XD_HISTORY); @@ -434,8 +422,7 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { map.putAll(historyProps); map.put("thread", Thread.currentThread().getName()); history.add(map); - Message out = MessageBusAdapter.this.messageBuilderFactory - .fromMessage(message) + 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/OutputChannelBinding.java similarity index 88% rename from spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelSpec.java rename to spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/OutputChannelBinding.java index 4fed56fc1..a43075653 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/OutputChannelBinding.java @@ -19,18 +19,17 @@ package org.springframework.bus.runner.adapter; /** * @author Dave Syer - * */ -public class OutputChannelSpec extends InputChannelSpec { +public class OutputChannelBinding extends ChannelBinding { private boolean tapped = false; private String tapChannelName; - protected OutputChannelSpec() { + protected OutputChannelBinding() { this(null); } - public OutputChannelSpec(String localName) { + public OutputChannelBinding(String localName) { super(localName); } diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientChannelLocator.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientChannelLocator.java index 8877a2ea3..b81588f6d 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientChannelLocator.java +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/discovery/DiscoveryClientChannelLocator.java @@ -24,9 +24,9 @@ import java.util.Random; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.bus.runner.adapter.ChannelBinding; 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; @@ -35,7 +35,6 @@ import org.springframework.web.util.UriComponentsBuilder; /** * @author Dave Syer - * */ public class DiscoveryClientChannelLocator implements ChannelLocator { @@ -64,22 +63,21 @@ public class DiscoveryClientChannelLocator implements ChannelLocator { } URI uri = pickUrl(instances); try { - ChannelsMetadata channels = this.restTemplate.getForObject(uri, - ChannelsMetadata.class); - Collection specs = Collections.emptySet(); + ChannelsMetadata channels = this.restTemplate.getForObject(uri, ChannelsMetadata.class); + Collection bindings = Collections.emptySet(); if (name.startsWith("input")) { name = name.replace("input", "output"); - specs = channels.getOutputChannels(); + bindings = channels.getOutputChannels(); } else if (name.startsWith("output")) { name = name.replace("output", "input"); - specs = channels.getInputChannels(); + bindings = channels.getInputChannels(); } - for (InputChannelSpec spec : specs) { - if (name.equals(spec.getLocalName())) { + for (ChannelBinding binding : bindings) { + if (name.equals(binding.getLocalName())) { this.logger.debug("Discovered channel for '" + this.serviceId + "' (" - + name + "=" + spec.getName() + ")"); - return spec.getName(); + + name + "=" + binding.getRemoteName() + ")"); + return binding.getRemoteName(); } } } 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 c31e609db..996905ce6 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 @@ -31,10 +31,10 @@ 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.Upstream; -import org.springframework.bus.runner.adapter.InputChannelSpec; +import org.springframework.bus.runner.adapter.InputChannelBinding; import org.springframework.bus.runner.adapter.MessageBusAdapter; import org.springframework.bus.runner.adapter.Downstream; -import org.springframework.bus.runner.adapter.OutputChannelSpec; +import org.springframework.bus.runner.adapter.OutputChannelBinding; import org.springframework.bus.runner.endpoint.ChannelsEndpoint; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -87,25 +87,25 @@ public class MessageBusAdapterConfiguration { return new ChannelsEndpoint(adapter); } - protected Collection getOutputChannels() { - Set channels = new LinkedHashSet(); + protected Collection getOutputChannels() { + Set channels = new LinkedHashSet(); String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors( this.beanFactory, MessageChannel.class); for (String name : names) { if (name.startsWith("output")) { - channels.add(new OutputChannelSpec(name)); + channels.add(new OutputChannelBinding(name)); } } return channels; } - protected Collection getInputChannels() { - Set channels = new LinkedHashSet(); + protected Collection getInputChannels() { + Set channels = new LinkedHashSet(); String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors( this.beanFactory, MessageChannel.class); for (String name : names) { if (name.startsWith("input")) { - channels.add(new InputChannelSpec(name)); + channels.add(new InputChannelBinding(name)); } } return channels; diff --git a/spring-bus-core/src/main/java/org/springframework/bus/runner/endpoint/ChannelsEndpoint.java b/spring-bus-core/src/main/java/org/springframework/bus/runner/endpoint/ChannelsEndpoint.java index 9686a6ffb..df3e832bf 100644 --- a/spring-bus-core/src/main/java/org/springframework/bus/runner/endpoint/ChannelsEndpoint.java +++ b/spring-bus-core/src/main/java/org/springframework/bus/runner/endpoint/ChannelsEndpoint.java @@ -24,12 +24,15 @@ 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.bus.runner.adapter.OutputChannelBinding; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; +/** + * @author Dave Syer + */ @RestController public class ChannelsEndpoint extends AbstractEndpoint> { @@ -41,24 +44,24 @@ public class ChannelsEndpoint extends AbstractEndpoint> { } @RequestMapping(value="/channels/taps") - public List taps() { - List list = new ArrayList(); - for (OutputChannelSpec spec : adapter.getChannelsMetadata().getOutputChannels()) { - if (spec.isTapped()) { - list.add(spec); + public List taps() { + List list = new ArrayList(); + for (OutputChannelBinding binding : adapter.getChannelsMetadata().getOutputChannels()) { + if (binding.isTapped()) { + list.add(binding); } } return list ; } @RequestMapping(value="/channels/taps", method=RequestMethod.POST) - public OutputChannelSpec tap(@RequestParam String channel) { + public OutputChannelBinding tap(@RequestParam String channel) { adapter.tap(channel); return adapter.getOutputChannel(channel); } @RequestMapping(value="/channels/taps", method=RequestMethod.DELETE) - public OutputChannelSpec untap(@RequestParam String channel) { + public OutputChannelBinding untap(@RequestParam String channel) { adapter.untap(channel); return adapter.getOutputChannel(channel); } 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 83de2d051..a0f983f05 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 @@ -34,7 +34,6 @@ import org.springframework.web.client.RestOperations; /** * @author Dave Syer - * */ public class DiscoveryClientChannelLocatorTests { @@ -42,8 +41,7 @@ public class DiscoveryClientChannelLocatorTests { private RestOperations restTemplate = Mockito.mock(RestOperations.class); - private DiscoveryClientChannelLocator locator = new DiscoveryClientChannelLocator( - this.client, "service"); + private DiscoveryClientChannelLocator locator = new DiscoveryClientChannelLocator(this.client, "service"); private ChannelsMetadata metadata = new ChannelsMetadata(); @@ -51,28 +49,27 @@ public class DiscoveryClientChannelLocatorTests { public void init() { this.locator.setRestTemplate(this.restTemplate); this.metadata.setModule(new MessageBusProperties()); - this.metadata.setInputChannels(new HashSet()); - this.metadata.setOutputChannels(new HashSet()); + this.metadata.setInputChannels(new HashSet()); + this.metadata.setOutputChannels(new HashSet()); Mockito.when( this.restTemplate.getForObject(Mockito.any(URI.class), anyChannels())) .thenReturn(this.metadata); Mockito.when(this.client.getInstances(Mockito.anyString())).thenReturn( - Arrays.asList(new DefaultServiceInstance("service", "example.com", 888, - false))); + Arrays.asList(new DefaultServiceInstance("service", "example.com", 888, false))); } @Test public void locateInputFromOutput() { - OutputChannelSpec output = new OutputChannelSpec("output"); - output.setName("foo.0"); + OutputChannelBinding output = new OutputChannelBinding("output"); + output.setRemoteName("foo.0"); this.metadata.getOutputChannels().add(output); assertEquals("foo.0", this.locator.locate("input")); } @Test public void locateOutputFromInput() { - InputChannelSpec input = new InputChannelSpec("input"); - input.setName("foo.0"); + InputChannelBinding input = new InputChannelBinding("input"); + input.setRemoteName("foo.0"); this.metadata.getInputChannels().add(input); assertEquals("foo.0", this.locator.locate("output")); } 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 80dc286bc..834e33239 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 @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.bus.runner.config; import static org.junit.Assert.assertEquals; @@ -28,9 +29,9 @@ 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.ChannelBinding; import org.springframework.bus.runner.adapter.MessageBusAdapter; -import org.springframework.bus.runner.adapter.OutputChannelSpec; +import org.springframework.bus.runner.adapter.OutputChannelBinding; import org.springframework.bus.runner.config.MessageBusAdapterConfigurationTests.Empty; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -43,7 +44,6 @@ import org.springframework.xd.dirt.integration.bus.local.LocalMessageBus; /** * @author Dave Syer - * */ @RunWith(SpringJUnit4ClassRunner.class) @SpringApplicationConfiguration(classes = Empty.class) @@ -70,17 +70,15 @@ public class MessageBusAdapterConfigurationTests { public void oneOutput() throws Exception { this.context.registerSingleton("output", new DirectChannel()); refresh(); - Collection channels = this.adapter.getChannelsMetadata() - .getOutputChannels(); + 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("group.0", channels.iterator().next().getRemoteName()); + assertEquals("tap:stream:group.module.0", channels.iterator().next().getTapChannelName()); } private void refresh() { - Collection channels = this.configuration.getOutputChannels(); - for (OutputChannelSpec channel : channels) { + Collection channels = this.configuration.getOutputChannels(); + for (OutputChannelBinding channel : channels) { channel.setTapped(true); } this.adapter.setOutputChannels(channels); @@ -91,12 +89,10 @@ public class MessageBusAdapterConfigurationTests { public void oneOutputTopic() throws Exception { this.context.registerSingleton("output.topic:", new DirectChannel()); refresh(); - Collection channels = this.adapter.getChannelsMetadata() - .getOutputChannels(); + 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()); + assertEquals("topic:group.0", channels.iterator().next().getRemoteName()); + assertEquals("tap:stream:group.module.0", channels.iterator().next().getTapChannelName()); } @Test @@ -104,18 +100,17 @@ public class MessageBusAdapterConfigurationTests { this.context.registerSingleton("output", new DirectChannel()); this.context.registerSingleton("output.queue:foo", new DirectChannel()); refresh(); - Collection channels = this.adapter.getChannelsMetadata() - .getOutputChannels(); + Collection channels = this.adapter.getChannelsMetadata().getOutputChannels(); List names = getChannelNames(channels); assertEquals(2, channels.size()); assertTrue(names.contains("group.0")); assertTrue(names.contains("foo.group.0")); } - private List getChannelNames(Collection channels) { + private List getChannelNames(Collection channels) { List list = new ArrayList(); - for (InputChannelSpec spec : channels) { - list.add(spec.getName()); + for (ChannelBinding binding : channels) { + list.add(binding.getRemoteName()); } return list; } @@ -125,13 +120,11 @@ public class MessageBusAdapterConfigurationTests { this.module.setOutputChannelName("bar"); this.context.registerSingleton("output.queue:foo", new DirectChannel()); refresh(); - Collection channels = this.adapter.getChannelsMetadata() - .getOutputChannels(); + Collection channels = this.adapter.getChannelsMetadata().getOutputChannels(); assertEquals(1, channels.size()); - assertEquals("foo.bar", channels.iterator().next().getName()); + assertEquals("foo.bar", channels.iterator().next().getRemoteName()); // TODO: fix this. What should it be? - assertEquals("tap:stream:foo.bar.module.0", channels.iterator().next() - .getTapChannelName()); + assertEquals("tap:stream:foo.bar.module.0", channels.iterator().next().getTapChannelName()); } @Test @@ -139,12 +132,10 @@ public class MessageBusAdapterConfigurationTests { this.module.setOutputChannelName("queue:bar"); this.context.registerSingleton("output.topic:foo", new DirectChannel()); refresh(); - Collection channels = this.adapter.getChannelsMetadata() - .getOutputChannels(); + 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()); + assertEquals("topic:foo.bar", channels.iterator().next().getRemoteName()); + assertEquals("tap:stream:foo.bar.module.0", channels.iterator().next().getTapChannelName()); } @Configuration