SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Switch from Map to Collection for *ChannelSpecs

Browse Source
This commit is contained in:
Dave Syer
2015-05-29 15:32:07 +01:00
parent f671737f71
commit 3b4834a2b0
7 changed files with 115 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
pom.xml
View File

@@ -45,6 +45,10 @@
<artifactId>spring-xd-dirt</artifactId>
<version>1.2.0.BUILD-SNAPSHOT</version>
<exclusions>
<exclusion>
<artifactId>jackson-core-asl</artifactId>
<groupId>org.codehaus.jackson</groupId>
</exclusion>
<exclusion>
<groupId>org.springframework.xd</groupId>
<artifactId>spring-xd-spark-streaming</artifactId>
@@ -69,6 +73,10 @@
<artifactId>zookeeper</artifactId>
<groupId>org.apache.zookeeper</groupId>
</exclusion>
<exclusion>
<artifactId>spring-boot-starter-security</artifactId>
<groupId>org.springframework.boot</groupId>
</exclusion>
<exclusion>
<artifactId>spring-security-ldap</artifactId>
<groupId>org.springframework.security</groupId>

4
roadmap.md
View File

@@ -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.

3
spring-bus-core/src/main/java/org/springframework/bus/runner/adapter/InputChannelSpec.java
View File

@@ -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;
}

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

@@ -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<String, OutputChannelSpec> outputChannels = new LinkedHashMap<String, OutputChannelSpec>();
private Map<String, InputChannelSpec> inputChannels = new LinkedHashMap<String, InputChannelSpec>();
private Collection<OutputChannelSpec> outputChannels = Collections.emptySet();
private Collection<InputChannelSpec> 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<String, OutputChannelSpec> outputChannels) {
this.outputChannels = outputChannels;
public void setOutputChannels(Collection<OutputChannelSpec> outputChannels) {
this.outputChannels = new LinkedHashSet<OutputChannelSpec>(outputChannels);
}
public void setInputChannels(Map<String, InputChannelSpec> inputChannels) {
this.inputChannels = inputChannels;
public void setInputChannels(Collection<InputChannelSpec> inputChannels) {
this.inputChannels = new LinkedHashSet<InputChannelSpec>(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);

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

@@ -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<String, OutputChannelSpec> getOutputChannels() {
Map<String, OutputChannelSpec> channels = new LinkedHashMap<String, OutputChannelSpec>();
@Bean
public AbstractEndpoint<?> messageBusEndpoint() {
return new AbstractEndpoint<Map<String, ?>>("messages") {
@Override
public Map<String, ?> invoke() {
LinkedHashMap<String, Object> map = new LinkedHashMap<String, Object>();
map.put("inputChannels", getInputChannels());
map.put("outputChannels", getOutputChannels());
map.put("module", module);
return map;
}
};
}
protected Collection<OutputChannelSpec> getOutputChannels() {
Set<OutputChannelSpec> channels = new LinkedHashSet<OutputChannelSpec>();
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<String, InputChannelSpec> getInputChannels() {
Map<String, InputChannelSpec> channels = new LinkedHashMap<String, InputChannelSpec>();
protected Collection<InputChannelSpec> getInputChannels() {
Set<InputChannelSpec> channels = new LinkedHashSet<InputChannelSpec>();
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;

4
spring-bus-core/src/main/java/org/springframework/bus/runner/config/MessageBusProperties.java
View File

@@ -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";

51
spring-bus-core/src/test/java/org/springframework/bus/runner/config/MessageBusAdapterConfigurationTests.java
View File

@@ -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<String, OutputChannelSpec> channels = configuration.getOutputChannels();
Collection<OutputChannelSpec> 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<String, OutputChannelSpec> channels = configuration.getOutputChannels();
Collection<OutputChannelSpec> channels = configuration.getOutputChannels();
List<String> 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<String, OutputChannelSpec> channels = configuration.getOutputChannels();
Collection<OutputChannelSpec> channels = configuration.getOutputChannels();
List<String> 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<String> getChannelNames(Collection<? extends InputChannelSpec> channels) {
List<String> list = new ArrayList<String>();
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<String, OutputChannelSpec> channels = configuration.getOutputChannels();
Collection<OutputChannelSpec> 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<String, OutputChannelSpec> channels = configuration.getOutputChannels();
Collection<OutputChannelSpec> 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<String, OutputChannelSpec> channels = configuration.getOutputChannels();
Collection<OutputChannelSpec> 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<String, OutputChannelSpec> channels = configuration.getOutputChannels();
Collection<OutputChannelSpec> channels = configuration.getOutputChannels();
assertEquals(1, channels.size());
assertTrue("Wrong key: " + channels.keySet(), channels.containsKey("foo.bar"));
assertEquals("foo.bar", channels.iterator().next().getName());
}
@Configuration