diff --git a/pom.xml b/pom.xml index 6fa73298c..8472ffd77 100644 --- a/pom.xml +++ b/pom.xml @@ -45,6 +45,10 @@ spring-xd-dirt 1.2.0.BUILD-SNAPSHOT + + jackson-core-asl + org.codehaus.jackson + org.springframework.xd spring-xd-spark-streaming @@ -69,6 +73,10 @@ zookeeper org.apache.zookeeper + + spring-boot-starter-security + org.springframework.boot + spring-security-ldap org.springframework.security diff --git a/roadmap.md b/roadmap.md index c17c0f6f5..08ebfa9f9 100644 --- a/roadmap.md +++ b/roadmap.md @@ -99,9 +99,9 @@ The best plan for making progress, where we keep in sight the goal of eventually - [x] There are no defaults for several properties in `xd.messagebus.*` so applications have to have a load of boilerplate configuration in `application.yml`. Fixed by adding `@PropertySources` to the default configuration. - - [ ] The XML is in `spring-xd-dirt` which we don't want to depend on. Maybe it should be in the messagebus implementation jars? + - [ ] The "codec.xml" is in `spring-xd-dirt` which we don't want to depend on. Maybe it should be in the messagebus SPI jar? Or we can make a copy and risk it changing in XD. - - [x] Do we need the analytics configuration? It should at least be optional. Answer "no". + - [x] Do we need the analytics configuration? It should at least be optional. Answer "no" (but support for analytics would be cool). - [ ] The `spring-xd-dirt` library contains some of the primitives we might need, especially when building the bridge to create XD modules as apps. It would be best if they could be extracted into another library. 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 888da6769..1c5503fc7 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 @@ -18,6 +18,8 @@ package org.springframework.bus.runner.adapter; import org.springframework.messaging.MessageChannel; +import com.fasterxml.jackson.annotation.JsonIgnore; + /** * @author Dave Syer * @@ -36,6 +38,7 @@ public class InputChannelSpec { return name; } + @JsonIgnore public MessageChannel getMessageChannel() { return channel; } 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 8b66c5205..f5f5fcfa0 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 @@ -18,7 +18,9 @@ package org.springframework.bus.runner.adapter; import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.Map; import java.util.Properties; import java.util.concurrent.atomic.AtomicBoolean; @@ -59,8 +61,8 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { private MessageBus messageBus; private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory(); - private Map outputChannels = new LinkedHashMap(); - private Map inputChannels = new LinkedHashMap(); + private Collection outputChannels = Collections.emptySet(); + private Collection inputChannels = Collections.emptySet(); private boolean running = false; @@ -95,23 +97,23 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { public void setOutputChannel(MessageChannel outputChannel) { if (outputChannel != null) { String name = module.getOutputChannelName(); - this.outputChannels.put(name, new OutputChannelSpec(name, outputChannel)); + this.outputChannels.add(new OutputChannelSpec(name, outputChannel)); } } public void setInputChannel(MessageChannel inputChannel) { if (inputChannel != null) { String name = module.getInputChannelName(); - this.inputChannels.put(name, new InputChannelSpec(name, inputChannel)); + this.inputChannels.add(new InputChannelSpec(name, inputChannel)); } } - public void setOutputChannels(Map outputChannels) { - this.outputChannels = outputChannels; + public void setOutputChannels(Collection outputChannels) { + this.outputChannels = new LinkedHashSet(outputChannels); } - public void setInputChannels(Map inputChannels) { - this.inputChannels = inputChannels; + public void setInputChannels(Collection inputChannels) { + this.inputChannels = new LinkedHashSet(inputChannels); } @Override @@ -152,13 +154,13 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { } protected final void unbindChannels() { - for (String name : inputChannels.keySet()) { - messageBus.unbindConsumers(name); + for (InputChannelSpec spec : inputChannels) { + messageBus.unbindConsumers(spec.getName()); } - for (String name : outputChannels.keySet()) { - messageBus.unbindProducers(name); - if (outputChannels.get(name).isTapped()) { - String tapChannelName = outputChannels.get(name).getTapChannelName(); + for (OutputChannelSpec spec : outputChannels) { + messageBus.unbindProducers(spec.getName()); + if (spec.isTapped()) { + String tapChannelName = spec.getTapChannelName(); messageBus.unbindProducers(tapChannelName); } } @@ -169,8 +171,8 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { if (trackHistory) { // TODO: addHistoryTag(); } - for (String name : outputChannels.keySet()) { - OutputChannelSpec spec = outputChannels.get(name); + for (OutputChannelSpec spec : outputChannels) { + String name = spec.getName(); MessageChannel outputChannel = spec.getMessageChannel(); bindMessageProducer(outputChannel, name, module.getProducerProperties()); if (spec.isTapped()) { @@ -185,8 +187,9 @@ public class MessageBusAdapter implements Lifecycle, ApplicationContextAware { track(outputChannel, historyProperties); } } - for (String name : inputChannels.keySet()) { - MessageChannel inputChannel = inputChannels.get(name).getMessageChannel(); + for (InputChannelSpec spec : inputChannels) { + String name = spec.getName(); + MessageChannel inputChannel = spec.getMessageChannel(); bindMessageConsumer(inputChannel, name, module.getConsumerProperties()); if (trackHistory && outputChannels.size() != 1) { historyProperties.put("inputChannel", name); 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 256871262..b6b1ea132 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 @@ -15,13 +15,17 @@ */ package org.springframework.bus.runner.config; +import java.util.Collection; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.Map; import java.util.Properties; +import java.util.Set; import org.springframework.beans.factory.BeanFactoryUtils; import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.actuate.endpoint.AbstractEndpoint; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.bus.runner.adapter.InputChannelSpec; import org.springframework.bus.runner.adapter.MessageBusAdapter; @@ -57,17 +61,31 @@ public class MessageBusAdapterConfiguration { return adapter; } - protected Map getOutputChannels() { - Map channels = new LinkedHashMap(); + @Bean + public AbstractEndpoint messageBusEndpoint() { + return new AbstractEndpoint>("messages") { + @Override + public Map invoke() { + LinkedHashMap map = new LinkedHashMap(); + map.put("inputChannels", getInputChannels()); + map.put("outputChannels", getOutputChannels()); + map.put("module", module); + return map; + } + }; + } + + protected Collection getOutputChannels() { + Set channels = new LinkedHashSet(); String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(beanFactory, MessageChannel.class); for (String name : names) { + OutputChannelSpec channel = null; + String prefix = ""; if (name.equals("output")) { String channelName = module.getOutputChannelName(); - channels.put( - channelName, - new OutputChannelSpec(channelName, beanFactory.getBean(name, - MessageChannel.class))); + channel = new OutputChannelSpec(channelName, beanFactory.getBean(name, + MessageChannel.class)); } else if (name.startsWith("output.")) { String channelName = name.substring("output.".length()); @@ -76,18 +94,22 @@ public class MessageBusAdapterConfiguration { String type = tokens[0]; if ("queue".equals(type)) { // omit the type for a queue - channelName = tokens[1] + "." - + getPlainChannelName(module.getOutputChannelName()); + prefix = tokens[1] + "."; } else { - channelName = channelName + "." - + getPlainChannelName(module.getOutputChannelName()); + prefix = channelName + "."; } } - channels.put( - channelName, - new OutputChannelSpec(channelName, beanFactory.getBean(name, - MessageChannel.class))); + channelName = prefix + + getPlainChannelName(module.getOutputChannelName()); + channel = new OutputChannelSpec(channelName, beanFactory.getBean(name, + MessageChannel.class)); + } + if (channel != null) { + String tapChannelName = prefix + module.getTapChannelName(); + channel.setTapChannelName(tapChannelName); + channel.setTapped(true); // TODO: determine when this is the case + channels.add(channel); } } return channels; @@ -95,22 +117,20 @@ public class MessageBusAdapterConfiguration { private String getPlainChannelName(String name) { if (name.contains(":")) { - name = name.substring(name.indexOf(":")+1); + name = name.substring(name.indexOf(":") + 1); } return name; } - protected Map getInputChannels() { - Map channels = new LinkedHashMap(); + protected Collection getInputChannels() { + Set channels = new LinkedHashSet(); String[] names = BeanFactoryUtils.beanNamesForTypeIncludingAncestors(beanFactory, MessageChannel.class); for (String name : names) { if (name.equals("input")) { String channelName = module.getInputChannelName(); - channels.put( - channelName, - new InputChannelSpec(channelName, beanFactory.getBean(name, - MessageChannel.class))); + channels.add(new InputChannelSpec(channelName, beanFactory.getBean(name, + MessageChannel.class))); } else if (name.startsWith("input.")) { String channelName = name.substring("input.".length()); @@ -127,10 +147,8 @@ public class MessageBusAdapterConfiguration { + getPlainChannelName(module.getInputChannelName()); } } - channels.put( - channelName, - new InputChannelSpec(channelName, beanFactory.getBean(name, - MessageChannel.class))); + channels.add(new InputChannelSpec(channelName, beanFactory.getBean(name, + MessageChannel.class))); } } return channels; 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 400a614fa..3cd24267f 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 @@ -22,11 +22,15 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.util.Assert; import org.springframework.xd.dirt.integration.bus.BusUtils; +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.annotation.JsonInclude.Include; + /** * @author Dave Syer * */ @ConfigurationProperties("spring.bus") +@JsonInclude(Include.NON_DEFAULT) public class MessageBusProperties { private String name = "module"; 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 3689af817..7311894c4 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 @@ -18,15 +18,19 @@ package org.springframework.bus.runner.config; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; -import java.util.Map; +import java.util.ArrayList; +import java.util.Collection; +import java.util.List; 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.OutputChannelSpec; import org.springframework.bus.runner.config.MessageBusAdapterConfigurationTests.Empty; +import org.springframework.bus.runner.config.MessageBusProperties.Tap; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; @@ -57,65 +61,78 @@ public class MessageBusAdapterConfigurationTests { @Test public void oneOutput() throws Exception { context.registerSingleton("output", new DirectChannel()); - Map channels = configuration.getOutputChannels(); + Collection channels = configuration.getOutputChannels(); assertEquals(1, channels.size()); - assertTrue(channels.containsKey("group.0")); + assertEquals("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()); - Map channels = configuration.getOutputChannels(); + Collection channels = configuration.getOutputChannels(); + List names = getChannelNames(channels); assertEquals(2, channels.size()); - assertTrue(channels.containsKey("group.0")); - assertTrue(channels.containsKey("topic:foo.group.0")); + assertTrue(names.contains("group.0")); + assertTrue(names.contains("topic:foo.group.0")); } @Test public void twoOutputsWithQueue() throws Exception { context.registerSingleton("output", new DirectChannel()); context.registerSingleton("output.queue:foo", new DirectChannel()); - Map channels = configuration.getOutputChannels(); + Collection channels = configuration.getOutputChannels(); + List names = getChannelNames(channels); assertEquals(2, channels.size()); - assertTrue(channels.containsKey("group.0")); - assertTrue(channels.containsKey("foo.group.0")); + assertTrue(names.contains("group.0")); + assertTrue(names.contains("foo.group.0")); + } + + private List getChannelNames(Collection channels) { + List list = new ArrayList(); + for (InputChannelSpec spec : channels) { + list.add(spec.getName()); + } + return list ; } @Test public void overrideNaturalOutputChannelName() throws Exception { module.setOutputChannelName("bar"); context.registerSingleton("output.queue:foo", new DirectChannel()); - Map channels = configuration.getOutputChannels(); + Collection channels = configuration.getOutputChannels(); assertEquals(1, channels.size()); - assertTrue(channels.containsKey("foo.bar")); + assertEquals("foo.bar", channels.iterator().next().getName()); + // TODO: fix this. What should it be? + // assertEquals("tap:stream:group.module.0", channels.iterator().next().getTapChannelName()); } @Test public void overrideNaturalOutputChannelNamedQueue() throws Exception { module.setOutputChannelName("queue:bar"); context.registerSingleton("output.queue:foo", new DirectChannel()); - Map channels = configuration.getOutputChannels(); + Collection channels = configuration.getOutputChannels(); assertEquals(1, channels.size()); - assertTrue("Wrong key: " + channels.keySet(), channels.containsKey("foo.bar")); + assertEquals("foo.bar", channels.iterator().next().getName()); } @Test public void overrideNaturalOutputChannelNamedQueueWithTopic() throws Exception { module.setOutputChannelName("queue:bar"); context.registerSingleton("output.topic:foo", new DirectChannel()); - Map channels = configuration.getOutputChannels(); + Collection channels = configuration.getOutputChannels(); assertEquals(1, channels.size()); - assertTrue("Wrong key: " + channels.keySet(), channels.containsKey("topic:foo.bar")); + assertEquals("topic:foo.bar", channels.iterator().next().getName()); } @Test public void overrideNaturalOutputChannelNamedTopic() throws Exception { module.setOutputChannelName("topic:bar"); context.registerSingleton("output.queue:foo", new DirectChannel()); - Map channels = configuration.getOutputChannels(); + Collection channels = configuration.getOutputChannels(); assertEquals(1, channels.size()); - assertTrue("Wrong key: " + channels.keySet(), channels.containsKey("foo.bar")); + assertEquals("foo.bar", channels.iterator().next().getName()); } @Configuration