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

Initial implementation of multibinder support

Browse Source 
- Addresses #138,#172,#173 with support for multiple binders in an application
- Binders can be of multiple types (e.g. Rabbit or Kafka), as well as multiple binders of the same type connecting to different system (e.g. multiple Rabbit binders connecting to different Rabbit clusters)
- Binders are now created in a separate context, allowing for multiple configurations
- Adds the `spring.cloud.spring.binders` namespace that allows for creating configuration properties for multiple binders
- Add support for default binder, modify Kafka tests to use multiple brokers when running embedded
This commit is contained in:
Marius Bogoevici
2015-11-30 16:59:21 -05:00
committed by Mark Fisher
parent ac395f84e5
commit 96f96f7621
61 changed files with 1948 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories → spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders
View File

@@ -1,2 +1,2 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration:\
kafka:\
org.springframework.cloud.stream.binder.kafka.config.KafkaServiceAutoConfiguration

2
spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.factories → spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders
View File

@@ -1,2 +1,2 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration:\
rabbit:\
org.springframework.cloud.stream.binder.rabbit.config.RabbitServiceAutoConfiguration

2
spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.factories → spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.binders
View File

@@ -1,2 +1,2 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration:\
redis:\
org.springframework.cloud.stream.binder.redis.config.RedisServiceAutoConfiguration

102
spring-cloud-stream-samples/multibinder-differentsystems/pom.xml Normal file
View File

@@ -0,0 +1,102 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-sample-multibinder-differentsystems</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-sample-multibinder-differentsystems</name>
<description>
Demo project for multiple binders of the same type (Kafka), connecting to different systems (broker groups)
</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-samples</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<properties>
<start-class>multibinder.MultibinderApplication</start-class>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-sample-source</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-sample-transform</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-sample-sink</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.10</artifactId>
<classifier>test</classifier>
<version>0.8.2.1</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-test</artifactId>
<version>2.6.0</version>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<classifier>exec</classifier>
</configuration>
</plugin>
</plugins>
</build>
</project>

33
spring-cloud-stream-samples/multibinder-differentsystems/src/main/java/multibinder/BridgeTransformer.java Normal file
View File

@@ -0,0 +1,33 @@
/*
* 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 multibinder;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.integration.annotation.ServiceActivator;
/**
* @author Marius Bogoevici
*/
@EnableBinding(Processor.class)
public class BridgeTransformer {
@ServiceActivator(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
public Object transform(Object payload) {
return payload;
}
}

29
spring-cloud-stream-samples/multibinder-differentsystems/src/main/java/multibinder/MultibinderApplication.java Normal file
View File

@@ -0,0 +1,29 @@
/*
* 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 multibinder;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class MultibinderApplication {
public static void main(String[] args) {
SpringApplication.run(MultibinderApplication.class, args);
}
}

33
spring-cloud-stream-samples/multibinder-differentsystems/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,33 @@
server:
port: 8082
spring:
cloud:
stream:
bindings:
input:
destination: dataIn
binder: kafka1
output:
destination: dataOut
binder: kafka2
binders:
kafka1:
type: kafka
environment:
spring:
cloud:
stream:
binder:
kafka:
brokers: ${kafkaBroker1}
zkNodes: ${zk1}
kafka2:
type: kafka
environment:
spring:
cloud:
stream:
binder:
kafka:
brokers: ${kafkaBroker2}
zkNodes: ${zk2}

90
spring-cloud-stream-samples/multibinder-differentsystems/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java Normal file
View File

@@ -0,0 +1,90 @@
/*
* 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 multibinder;
import java.util.UUID;
import org.hamcrest.CoreMatchers;
import org.hamcrest.Matchers;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.test.junit.kafka.KafkaTestSupport;
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.web.WebAppConfiguration;
@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = MultibinderApplication.class)
@WebAppConfiguration
@DirtiesContext
public class TwoKafkaBindersApplicationTest {
@ClassRule
public static KafkaTestSupport kafkaTestSupport1 = new KafkaTestSupport();
@ClassRule
public static KafkaTestSupport kafkaTestSupport2 = new KafkaTestSupport();
@ClassRule
public static RedisTestSupport redisTestSupport = new RedisTestSupport();
@BeforeClass
public static void setupEnvironment() {
System.setProperty("kafkaBroker1", kafkaTestSupport1.getBrokerAddress());
System.setProperty("zk1", kafkaTestSupport1.getZkConnectString());
System.setProperty("kafkaBroker2", kafkaTestSupport2.getBrokerAddress());
System.setProperty("zk2", kafkaTestSupport2.getZkConnectString());
}
@Autowired
private BinderFactory<MessageChannel> binderFactory;
@Test
public void contextLoads() {
}
@Test
public void messagingWorks() {
DirectChannel dataProducer = new DirectChannel();
binderFactory.getBinder("kafka1").bindProducer("dataIn", dataProducer, null);
QueueChannel dataConsumer = new QueueChannel();
binderFactory.getBinder("kafka2").bindPubSubConsumer("dataOut", dataConsumer,
UUID.randomUUID().toString(), null);
String testPayload = "testFoo" + UUID.randomUUID().toString();
dataProducer.send(MessageBuilder.withPayload(testPayload).build());
Message<?> receive = dataConsumer.receive(2000);
Assert.assertThat(receive, Matchers.notNullValue());
Assert.assertThat(receive.getPayload(), CoreMatchers.equalTo(testPayload));
}
}

79
spring-cloud-stream-samples/multibinder/pom.xml Normal file
View File

@@ -0,0 +1,79 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>spring-cloud-stream-sample-multibinder</artifactId>
<packaging>jar</packaging>
<name>spring-cloud-stream-sample-multibinder</name>
<description>Demo project for multiple binders of different types (Redis and Rabbit)</description>
<parent>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-samples</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
</parent>
<properties>
<start-class>multibinder.MultibinderApplication</start-class>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-sample-source</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-sample-transform</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-sample-sink</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support-internal</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-redis</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<configuration>
<classifier>exec</classifier>
</configuration>
</plugin>
</plugins>
</build>
</project>

33
spring-cloud-stream-samples/multibinder/src/main/java/multibinder/BridgeTransformer.java Normal file
View File

@@ -0,0 +1,33 @@
/*
* 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 multibinder;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.integration.annotation.ServiceActivator;
/**
* @author Marius Bogoevici
*/
@EnableBinding(Processor.class)
public class BridgeTransformer {
@ServiceActivator(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
public Object transform(Object payload) {
return payload;
}
}

29
spring-cloud-stream-samples/multibinder/src/main/java/multibinder/MultibinderApplication.java Normal file
View File

@@ -0,0 +1,29 @@
/*
* 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 multibinder;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class MultibinderApplication {
public static void main(String[] args) {
SpringApplication.run(MultibinderApplication.class, args);
}
}

12
spring-cloud-stream-samples/multibinder/src/main/resources/application.yml Normal file
View File

@@ -0,0 +1,12 @@
server:
port: 8082
spring:
cloud:
stream:
bindings:
input:
destination: dataIn
binder: redis
output:
destination: dataOut
binder: rabbit

78
spring-cloud-stream-samples/multibinder/src/test/java/multibinder/RabbitAndRedisBinderApplicationTests.java Normal file
View File

@@ -0,0 +1,78 @@
/*
* 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 multibinder;
import java.util.UUID;
import org.hamcrest.CoreMatchers;
import org.hamcrest.Matchers;
import org.junit.Assert;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.test.junit.rabbit.RabbitTestSupport;
import org.springframework.cloud.stream.test.junit.redis.RedisTestSupport;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.web.WebAppConfiguration;
@RunWith(SpringJUnit4ClassRunner.class)
@SpringApplicationConfiguration(classes = MultibinderApplication.class)
@WebAppConfiguration
@DirtiesContext
public class RabbitAndRedisBinderApplicationTests {
@ClassRule
public static RabbitTestSupport rabbitTestSupport = new RabbitTestSupport();
@ClassRule
public static RedisTestSupport redisTestSupport = new RedisTestSupport();
@Autowired
private BinderFactory<MessageChannel> binderFactory;
@Test
public void contextLoads() {
}
@Test
public void messagingWorks() {
DirectChannel dataProducer = new DirectChannel();
binderFactory.getBinder("redis").bindProducer("dataIn", dataProducer,null);
QueueChannel dataConsumer = new QueueChannel();
binderFactory.getBinder("rabbit").bindPubSubConsumer("dataOut", dataConsumer,
UUID.randomUUID().toString(),null);
String testPayload = "testFoo" + UUID.randomUUID().toString();
dataProducer.send(MessageBuilder.withPayload(testPayload).build());
Message<?> receive = dataConsumer.receive(2000);
Assert.assertThat(receive, Matchers.notNullValue());
Assert.assertThat(receive.getPayload(), CoreMatchers.equalTo(testPayload));
}
}

2
spring-cloud-stream-samples/pom.xml
View File

@@ -25,6 +25,8 @@
<module>tap</module>
<module>double</module>
<module>extended</module>
<module>multibinder</module>
<module>multibinder-differentsystems</module>
<module>rxjava-processor</module>
</modules>
<dependencyManagement>

20
spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java
View File

@@ -19,6 +19,13 @@ package org.springframework.cloud.stream.test.junit.kafka;
import java.util.Properties;
import kafka.server.KafkaConfig;
import kafka.server.KafkaServer;
import kafka.utils.SystemTime$;
import kafka.utils.TestUtils;
import kafka.utils.Utils;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkInterruptedException;
import org.junit.Rule;
@@ -26,15 +33,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport;
import kafka.server.KafkaConfig;
import kafka.server.KafkaServer;
import kafka.utils.SystemTime$;
import kafka.utils.TestUtils;
import kafka.utils.TestZKUtils;
import kafka.utils.Utils;
import kafka.utils.ZKStringSerializer$;
import kafka.utils.ZkUtils;
import org.springframework.util.SocketUtils;
/**
@@ -105,7 +104,7 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
if (embedded) {
try {
log.debug("Starting Zookeeper");
zookeeper = new EmbeddedZookeeper(TestZKUtils.zookeeperConnect());
zookeeper = new EmbeddedZookeeper("127.0.0.1:" + SocketUtils.findAvailableTcpPort());
log.debug("Started Zookeeper at " + zookeeper.getConnectString());
try {
int zkConnectionTimeout = 10000;
@@ -119,6 +118,7 @@ public class KafkaTestSupport extends AbstractExternalResourceTestSupport<String
try {
log.debug("Creating Kafka server");
Properties brokerConfigProperties = brokerConfig;
brokerConfig.put("zookeeper.connect", zookeeper.getConnectString());
kafkaServer = TestUtils.createServer(new KafkaConfig(brokerConfigProperties), SystemTime$.MODULE$);
log.debug("Created Kafka server at " + kafkaServer.config().hostName() + ":" + kafkaServer.config().port());
}

2
spring-cloud-stream-test-support/src/main/resources/META-INF/spring.factories → spring-cloud-stream-test-support/src/main/resources/META-INF/spring.binders
View File

@@ -1,2 +1,2 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration:\
test:\
org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration

14
spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/ExampleTest.java
View File

@@ -16,8 +16,8 @@
package org.springframework.cloud.stream.test;
import static org.hamcrest.CoreMatchers.*;
import static org.junit.Assert.*;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.junit.Assert.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -26,12 +26,14 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.IntegrationTest;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.test.binder.MessageCollector;
import org.springframework.cloud.stream.test.binder.TestSupportBinder;
import org.springframework.integration.annotation.Transformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -51,14 +53,14 @@ public class ExampleTest {
private Processor processor;
@Autowired
private MessageCollector messageCollector;
private BinderFactory<MessageChannel> binderFactory;
@Test
@SuppressWarnings("unchecked")
public void testWiring() {
Message<String> message = new GenericMessage<>("hello");
processor.input().send(message);
Message<String> received = (Message<String>) messageCollector.forChannel(processor.output()).poll();
Message<String> received = (Message<String>) ((TestSupportBinder) binderFactory.getBinder(null)).messageCollector().forChannel(processor.output()).poll();
assertThat(received.getPayload(), equalTo("hello world"));
}

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java
View File

@@ -24,6 +24,7 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.cloud.stream.config.AggregateBuilderConfiguration;
import org.springframework.cloud.stream.config.BinderFactoryConfiguration;
import org.springframework.cloud.stream.config.BindingBeansRegistrar;
import org.springframework.cloud.stream.config.ChannelBindingServiceConfiguration;
import org.springframework.context.annotation.Configuration;
@@ -33,7 +34,6 @@ import org.springframework.integration.config.EnableIntegration;
/**
* Enables the binding of inputs and outputs to a broker, according to the list
* of interfaces passed as value to the annotation.
*
* @author Dave Syer
* @author Marius Bogoevici
* @author David Turanski
@@ -43,7 +43,8 @@ import org.springframework.integration.config.EnableIntegration;
@Documented
@Inherited
@Configuration
@Import({ChannelBindingServiceConfiguration.class, AggregateBuilderConfiguration.class, BindingBeansRegistrar.class})
@Import({ChannelBindingServiceConfiguration.class, AggregateBuilderConfiguration.class, BindingBeansRegistrar.class,
BinderFactoryConfiguration.class})
@EnableIntegration
public @interface EnableBinding {

51
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderConfiguration.java Normal file
View File

@@ -0,0 +1,51 @@
/*
* 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.cloud.stream.binder;
import java.util.Properties;
/**
*
* Configuration for a binder instance, associating a {@link BinderType} with its configuration {@link Properties}.
* An application may contain multiple {@link BinderConfiguration}s per {@link BinderType}, when connecting to multiple
* systems of the same type.
*
* @author Marius Bogoevici
*/
public class BinderConfiguration {
private final BinderType binderType;
private final Properties properties;
/**
* @param binderType the binder type used by this configuration
* @param properties the properties for setting up the binder
*/
public BinderConfiguration(BinderType binderType, Properties properties) {
this.binderType = binderType;
this.properties = properties;
}
public BinderType getBinderType() {
return binderType;
}
public Properties getProperties() {
return properties;
}
}

32
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderFactory.java Normal file
View File

@@ -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.cloud.stream.binder;
/**
* @author Marius Bogoevici
*/
public interface BinderFactory<T> {
/**
* Returns the binder instance associated with the given configuration name. Instance caching is a requirement,
* and implementations must return the same instance on subsequent invocations with the same argument.
*
* @param configurationName the name of a binder configuration
* @return the binder instance
*/
Binder<T> getBinder(String configurationName);
}

70
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderType.java Normal file
View File

@@ -0,0 +1,70 @@
/*
* 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.cloud.stream.binder;
import java.util.Arrays;
/**
* References one or more {@link org.springframework.context.annotation.Configuration}-annotated classes which
* provide a context definition which contains exactly one {@link Binder}, typically for a given type of system (e.g.
* Rabbit, Kafka, Redis, etc.). An application may contain multiple instances of a given {@link BinderType},
* when connecting to multiple systems of the same type.
*
* @author Marius Bogoevici
*/
public class BinderType {
private final String defaultName;
private final Class<?>[] configurationClasses;
public BinderType(String defaultName, Class<?>[] configurationClasses) {
this.defaultName = defaultName;
this.configurationClasses = configurationClasses;
}
public String getDefaultName() {
return defaultName;
}
public Class<?>[] getConfigurationClasses() {
return configurationClasses;
}
@Override
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null || getClass() != o.getClass()) {
return false;
}
BinderType that = (BinderType) o;
if (!defaultName.equals(that.defaultName)) {
return false;
}
return Arrays.equals(configurationClasses, that.configurationClasses);
}
@Override
public int hashCode() {
int result = defaultName.hashCode();
result = 31 * result + Arrays.hashCode(configurationClasses);
return result;
}
}

33
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderTypeRegistry.java Normal file
View File

@@ -0,0 +1,33 @@
/*
* 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.cloud.stream.binder;
import java.util.Map;
/**
* A registry of {@link BinderType}s, indexed by name. A {@link BinderTypeRegistry} bean is created automatically
* based on information found in the {@literal META-INF/spring.binders} files provided by binder implementors.
* This can be overridden by registering a {@link BinderTypeRegistry} bean in the context.
*
* @author Marius Bogoevici
*/
public interface BinderTypeRegistry {
BinderType get(String name);
Map<String, BinderType> getAll();
}

161
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java Normal file
View File

@@ -0,0 +1,161 @@
/*
* 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.cloud.stream.binder;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.Banner.Mode;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.EnvironmentAware;
import org.springframework.core.env.Environment;
import org.springframework.util.StringUtils;
/**
* Default {@link BinderFactory} implementation.
*
* @author Marius Bogoevici
*/
public class DefaultBinderFactory<T> implements BinderFactory<T>, DisposableBean, EnvironmentAware {
private final Map<String, BinderConfiguration> binderConfigurations;
private final Map<String, BinderInstanceHolder<T>> binderInstanceCache = new HashMap<>();
private volatile Environment environment;
private volatile String defaultBinder;
public DefaultBinderFactory(Map<String, BinderConfiguration> binderConfigurations) {
this.binderConfigurations = new HashMap<>(binderConfigurations);
}
@Override
public void setEnvironment(Environment environment) {
this.environment = environment;
}
public void setDefaultBinder(String defaultBinder) {
this.defaultBinder = defaultBinder;
}
@Override
public void destroy() throws Exception {
for (Map.Entry<String, BinderInstanceHolder<T>> entry : binderInstanceCache.entrySet()) {
BinderInstanceHolder<T> binderInstanceHolder = entry.getValue();
binderInstanceHolder.getBinderContext().close();
}
}
@Override
public synchronized Binder<T> getBinder(String name) {
String configurationName;
// Fall back to a default if no argument is provided
if (StringUtils.isEmpty(name)) {
if (binderConfigurations.size() == 0) {
throw new IllegalStateException("A default binder has been requested, but there there is no binder available");
}
else if (binderConfigurations.size() == 1) {
configurationName = binderConfigurations.keySet().iterator().next();
}
else {
if (StringUtils.hasText(defaultBinder)) {
configurationName = defaultBinder;
}
else {
throw new IllegalStateException(
"A default binder has been requested, but there is more than one binder available: "
+ StringUtils.collectionToCommaDelimitedString(binderConfigurations.keySet()) + ", and"
+ " no default binder has been set.");
}
}
} else {
configurationName = name;
}
if (!binderInstanceCache.containsKey(configurationName)) {
BinderConfiguration binderConfiguration = binderConfigurations.get(configurationName);
if (binderConfiguration == null) {
throw new IllegalStateException("Unknown binder configuration: " + configurationName);
}
Properties binderProperties = binderConfiguration.getProperties();
// Convert all properties to arguments, so that they receive maximum precedence
ArrayList<String> args = new ArrayList<>();
for (Map.Entry<Object, Object> property : binderProperties.entrySet()) {
args.add(String.format("--%s=%s",property.getKey(),property.getValue()));
}
// Initialize the domain with a unique name based on the bootstrapping context setting
String defaultDomain = environment != null ? environment.getProperty("spring.jmx.default-domain") : null;
if (defaultDomain == null) {
defaultDomain = "";
}
else {
defaultDomain += ".";
}
args.add("--spring.jmx.default-domain=" + defaultDomain + "binder." + configurationName);
SpringApplicationBuilder springApplicationBuilder =
new SpringApplicationBuilder()
.sources(binderConfiguration.getBinderType().getConfigurationClasses())
.sources(SeedConfiguration.class)
.bannerMode(Mode.OFF)
.web(false);
ConfigurableApplicationContext binderProducingContext =
springApplicationBuilder.run(args.toArray(new String[args.size()]));
@SuppressWarnings("unchecked")
Binder<T> binder = (Binder<T>) binderProducingContext.getBean(Binder.class);
binderInstanceCache.put(configurationName, new BinderInstanceHolder<>(binder, binderProducingContext));
}
return binderInstanceCache.get(configurationName).getBinderInstance();
}
/**
* Utility class for storing {@link Binder} instances, along with their associated contexts.
*
* @param <T>
*/
private static class BinderInstanceHolder<T> {
private final Binder<T> binderInstance;
private final ConfigurableApplicationContext binderContext;
public BinderInstanceHolder(Binder<T> binderInstance, ConfigurableApplicationContext binderContext) {
this.binderInstance = binderInstance;
this.binderContext = binderContext;
}
public Binder<T> getBinderInstance() {
return binderInstance;
}
public ConfigurableApplicationContext getBinderContext() {
return binderContext;
}
}
/**
* Configuration class that enables autoconfiguration for the binders
*/
// TODO: Reconsider the use of autoconfiguration as part of binder configuration refactoring
@EnableAutoConfiguration
public static class SeedConfiguration {
}
}

45
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderTypeRegistry.java Normal file
View File

@@ -0,0 +1,45 @@
/*
* 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.cloud.stream.binder;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
/**
* Defult implementation of a {@link BinderTypeRegistry}.
*
* @author Marius Bogoevici
*/
public class DefaultBinderTypeRegistry implements BinderTypeRegistry {
private final Map<String, BinderType> binderTypes;
public DefaultBinderTypeRegistry(Map<String, BinderType> binderTypes) {
this.binderTypes = Collections.unmodifiableMap(new HashMap<>(binderTypes));
}
@Override
public BinderType get(String name) {
return binderTypes.get(name);
}
@Override
public Map<String, BinderType> getAll() {
return binderTypes;
}
}

31
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java
View File

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binding;
import java.util.Properties;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.BeanFactoryMessageChannelDestinationResolver;
import org.springframework.messaging.core.DestinationResolutionException;
@@ -26,18 +27,17 @@ import org.springframework.messaging.core.DestinationResolutionException;
/**
* A {@link org.springframework.messaging.core.DestinationResolver} implementation that first checks for any channel
* whose name begins with a colon in the {@link Binder}.
*
* @author Mark Fisher
* @author Gary Russell
*/
public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestinationResolver {
private final Binder<MessageChannel> binder;
private final BinderFactory<MessageChannel> binderFactory;
private final Properties producerProperties;
public BinderAwareChannelResolver(Binder<MessageChannel> binder, Properties producerProperties) {
this.binder = binder;
public BinderAwareChannelResolver(BinderFactory<MessageChannel> binderFactory, Properties producerProperties) {
this.binderFactory = binderFactory;
this.producerProperties = producerProperties;
}
@@ -49,15 +49,28 @@ public class BinderAwareChannelResolver extends BeanFactoryMessageChannelDestina
}
catch (DestinationResolutionException e) {
}
if (name.indexOf(":") != -1) {
if (binder != null) {
if (name.contains(":")) {
if (binderFactory != null) {
String[] tokens = name.split(":", 2);
String type = tokens[0];
String transport = null;
String type;
if (tokens.length == 2) {
type = tokens[0];
}
else if (tokens.length == 3) {
transport = tokens[0];
type = tokens[1];
}
else {
throw new IllegalArgumentException("Unrecognized channel naming scheme: " + name + " , should be" +
" [<transport>:]<type>:<name>");
}
Binder<MessageChannel> binder = binderFactory.getBinder(transport);
if ("queue".equals(type)) {
channel = this.binder.bindDynamicProducer(name, this.producerProperties);
channel = binder.bindDynamicProducer(name, this.producerProperties);
}
else if ("topic".equals(type)) {
channel = this.binder.bindDynamicPubSubProducer(name, this.producerProperties);
channel = binder.bindDynamicPubSubProducer(name, this.producerProperties);
}
else {
throw new IllegalArgumentException("unrecognized channel type: " + type);

2
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2014 the original author or authors.
* Copyright 2013-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.

31
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java
View File

@@ -17,6 +17,7 @@
package org.springframework.cloud.stream.binding;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.BinderUtils;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
@@ -36,36 +37,39 @@ import org.springframework.util.StringUtils;
*/
public class ChannelBindingService {
private final Binder<MessageChannel> binder;
private BinderFactory<MessageChannel> binderFactory;
private final ChannelBindingServiceProperties channelBindingServiceProperties;
public ChannelBindingService(ChannelBindingServiceProperties channelBindingServiceProperties, Binder<MessageChannel> binder) {
public ChannelBindingService(ChannelBindingServiceProperties channelBindingServiceProperties,
BinderFactory<MessageChannel> binderFactory) {
this.channelBindingServiceProperties = channelBindingServiceProperties;
this.binder = binder;
this.binderFactory = binderFactory;
}
public void bindConsumer(MessageChannel inputChannel, String inputChannelName) {
String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(inputChannelName);
Binder<MessageChannel> binder = getBinderForChannel(inputChannelName);
if (BinderUtils.isChannelPubSub(channelBindingTarget)) {
this.binder.bindPubSubConsumer(removePrefix(channelBindingTarget),
binder.bindPubSubConsumer(removePrefix(channelBindingTarget),
inputChannel, consumerGroup(inputChannelName),
this.channelBindingServiceProperties.getConsumerProperties(inputChannelName));
}
else {
this.binder.bindConsumer(channelBindingTarget, inputChannel,
binder.bindConsumer(channelBindingTarget, inputChannel,
this.channelBindingServiceProperties.getConsumerProperties(inputChannelName));
}
}
public void bindProducer(MessageChannel outputChannel, String outputChannelName) {
String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(outputChannelName);
Binder<MessageChannel> binder = getBinderForChannel(outputChannelName);
if (BinderUtils.isChannelPubSub(channelBindingTarget)) {
this.binder.bindPubSubProducer(removePrefix(channelBindingTarget),
binder.bindPubSubProducer(removePrefix(channelBindingTarget),
outputChannel, this.channelBindingServiceProperties.getProducerProperties(outputChannelName));
}
else {
this.binder.bindProducer(channelBindingTarget, outputChannel,
binder.bindProducer(channelBindingTarget, outputChannel,
this.channelBindingServiceProperties.getProducerProperties(outputChannelName));
}
}
@@ -76,16 +80,23 @@ public class ChannelBindingService {
}
public void unbindConsumers(String inputChannelName) {
Binder<MessageChannel> binder = getBinderForChannel(inputChannelName);
if (BinderUtils.isChannelPubSub(this.channelBindingServiceProperties.getBindingDestination(inputChannelName))) {
this.binder.unbindPubSubConsumers(inputChannelName, consumerGroup(inputChannelName));
binder.unbindPubSubConsumers(inputChannelName, consumerGroup(inputChannelName));
}
else {
this.binder.unbindConsumers(inputChannelName);
binder.unbindConsumers(inputChannelName);
}
}
public void unbindProducers(String outputChannelName) {
this.binder.unbindProducers(outputChannelName);
Binder<MessageChannel> binder = getBinderForChannel(outputChannelName);
binder.unbindProducers(outputChannelName);
}
private Binder<MessageChannel> getBinderForChannel(String channelName) {
String transport = this.channelBindingServiceProperties.getBinder(channelName);
return binderFactory.getBinder(transport);
}
private String consumerGroup(String inputChannelName) {

134
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java Normal file
View File

@@ -0,0 +1,134 @@
/*
* 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.cloud.stream.config;
import java.io.IOException;
import java.net.URL;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.cloud.stream.binder.BinderConfiguration;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.BinderType;
import org.springframework.cloud.stream.binder.BinderTypeRegistry;
import org.springframework.cloud.stream.binder.DefaultBinderFactory;
import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.Resource;
import org.springframework.core.io.UrlResource;
import org.springframework.core.io.support.PropertiesLoaderUtils;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
/**
* @author Marius Bogoevici
*/
@Configuration
public class BinderFactoryConfiguration {
@Bean
@ConditionalOnMissingBean(BinderFactory.class)
public BinderFactory binderFactory(BinderTypeRegistry binderTypeRegistry,
ChannelBindingServiceProperties channelBindingServiceProperties) {
Map<String, BinderConfiguration> binderConfigurations = new HashMap<>();
if (!CollectionUtils.isEmpty(channelBindingServiceProperties.getBinders())) {
for (Map.Entry<String, BinderProperties> binderEntry :
channelBindingServiceProperties.getBinders().entrySet()) {
BinderProperties binderProperties = binderEntry.getValue();
if (binderTypeRegistry.get(binderEntry.getKey()) != null) {
binderConfigurations.put(binderEntry.getKey(),
new BinderConfiguration(binderTypeRegistry.get(binderEntry.getKey()),
binderProperties.getEnvironment()));
}
else {
Assert.hasText(binderProperties.getType(), "No 'type' property present for custom " +
"binder " + binderEntry.getKey());
binderConfigurations.put(binderEntry.getKey(),
new BinderConfiguration(binderTypeRegistry.get(binderProperties.getType()),
binderProperties.getEnvironment()));
}
}
}
else {
for (Map.Entry<String, BinderType> entry : binderTypeRegistry.getAll().entrySet()) {
binderConfigurations.put(entry.getKey(),
new BinderConfiguration(entry.getValue(), new Properties()));
}
}
DefaultBinderFactory binderFactory = new DefaultBinderFactory<>(binderConfigurations);
binderFactory.setDefaultBinder(channelBindingServiceProperties.getDefaultBinder());
return binderFactory;
}
@Bean
@ConditionalOnMissingBean(BinderTypeRegistry.class)
public BinderTypeRegistry binderTypeRegistry(ConfigurableApplicationContext configurableApplicationContext) {
Map<String, BinderType> binderTypes = new HashMap<>();
ClassLoader classLoader = configurableApplicationContext.getClassLoader();
if (classLoader == null) {
classLoader = ChannelBindingAutoConfiguration.class.getClassLoader();
}
try {
Enumeration<URL> resources = classLoader.getResources("META-INF/spring.binders");
if (resources == null || !resources.hasMoreElements()) {
throw new BeanCreationException("Cannot create binder factory, no `META-INF/spring.binders` " +
"resources found on the classpath");
}
while (resources.hasMoreElements()) {
URL url = resources.nextElement();
UrlResource resource = new UrlResource(url);
for (BinderType binderType : parseBinderConfigurations(classLoader, resource)) {
binderTypes.put(binderType.getDefaultName(), binderType);
}
}
}
catch (IOException | ClassNotFoundException e) {
throw new BeanCreationException("Cannot create binder factory:", e);
}
return new DefaultBinderTypeRegistry(binderTypes);
}
static Collection<BinderType> parseBinderConfigurations(ClassLoader classLoader, Resource resource)
throws IOException, ClassNotFoundException {
Properties properties = PropertiesLoaderUtils.loadProperties(resource);
Collection<BinderType> parsedBinderConfigurations = new ArrayList<>();
for (Map.Entry<?, ?> entry : properties.entrySet()) {
String binderType = (String) entry.getKey();
String[] binderConfigurationClassNames =
StringUtils.commaDelimitedListToStringArray((String)entry.getValue());
Class[] binderConfigurationClasses = new Class[binderConfigurationClassNames.length];
int i = 0;
for (String binderConfigurationClassName : binderConfigurationClassNames) {
binderConfigurationClasses[i++] = ClassUtils.forName(binderConfigurationClassName, classLoader);
}
parsedBinderConfigurations.add(new BinderType(binderType, binderConfigurationClasses));
}
return parsedBinderConfigurations;
}
}

47
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java Normal file
View File

@@ -0,0 +1,47 @@
/*
* 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.cloud.stream.config;
import java.util.Properties;
/**
* Contains the properties of a binder.
*
* @author Marius Bogoevici
*/
public class BinderProperties {
private String type;
private Properties environment = new Properties();
public String getType() {
return type;
}
public void setType(String name) {
this.type = name;
}
public Properties getEnvironment() {
return environment;
}
public void setEnvironment(Properties environment) {
this.environment = environment;
}
}

9
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java
View File

@@ -49,6 +49,8 @@ public class BindingProperties {
private String contentType;
private String binder;
public String getDestination() {
return this.destination;
}
@@ -121,4 +123,11 @@ public class BindingProperties {
this.contentType = contentType;
}
public String getBinder() {
return binder;
}
public void setBinder(String binder) {
this.binder = binder;
}
}

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java
View File

@@ -27,13 +27,13 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.ConfigurationPropertiesBinding;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor;
import org.springframework.cloud.stream.binding.ChannelBindingService;
import org.springframework.cloud.stream.binding.ChannelFactory;
import org.springframework.cloud.stream.binding.DefaultChannelFactory;
import org.springframework.cloud.stream.binding.ContextStartAfterRefreshListener;
import org.springframework.cloud.stream.binding.DefaultChannelFactory;
import org.springframework.cloud.stream.binding.InputBindingLifecycle;
import org.springframework.cloud.stream.binding.MessageConverterConfigurer;
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
@@ -69,8 +69,8 @@ public class ChannelBindingServiceConfiguration {
@ConditionalOnMissingBean(ChannelBindingService.class)
public ChannelBindingService bindingService(
ChannelBindingServiceProperties channelBindingServiceProperties,
Binder<MessageChannel> binder) {
return new ChannelBindingService(channelBindingServiceProperties, binder);
BinderFactory<MessageChannel> binderFactory) {
return new ChannelBindingService(channelBindingServiceProperties, binderFactory);
}
@Bean
@@ -103,8 +103,8 @@ public class ChannelBindingServiceConfiguration {
@Bean
public BinderAwareChannelResolver binderAwareChannelResolver(
Binder<MessageChannel> binder) {
return new BinderAwareChannelResolver(binder, new Properties());
BinderFactory<MessageChannel> binderFactory) {
return new BinderAwareChannelResolver(binderFactory, new Properties());
}
@Bean

54
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java
View File

@@ -16,18 +16,18 @@
package org.springframework.cloud.stream.config;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.TreeMap;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.util.StringUtils;
import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.annotation.JsonInclude.Include;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.util.StringUtils;
/**
* @author Dave Syer
* @author Marius Bogoevici
@@ -48,7 +48,11 @@ public class ChannelBindingServiceProperties {
private Map<String, BindingProperties> bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
private Properties getConsumerProperties() {
private Map<String, BinderProperties> binders = new HashMap<>();
private String defaultBinder;
public Properties getConsumerProperties() {
return this.consumerProperties;
}
@@ -72,6 +76,22 @@ public class ChannelBindingServiceProperties {
this.bindings = bindings;
}
public Map<String, BinderProperties> getBinders() {
return binders;
}
public void setBinders(Map<String, BinderProperties> binders) {
this.binders = binders;
}
public String getDefaultBinder() {
return defaultBinder;
}
public void setDefaultBinder(String defaultBinder) {
this.defaultBinder = defaultBinder;
}
public int getInstanceIndex() {
return instanceIndex;
}
@@ -88,7 +108,6 @@ public class ChannelBindingServiceProperties {
this.instanceCount = instanceCount;
}
public String getBindingDestination(String channelName) {
BindingProperties bindingProperties = bindings.get(channelName);
// we may shortcut directly to the path
@@ -124,9 +143,9 @@ public class ChannelBindingServiceProperties {
if (isPartitionedConsumer(inputChannelName)) {
Properties channelConsumerProperties = new Properties();
channelConsumerProperties.putAll(consumerProperties);
channelConsumerProperties.setProperty(BinderProperties.COUNT,
channelConsumerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.COUNT,
Integer.toString(getInstanceCount()));
channelConsumerProperties.setProperty(BinderProperties.PARTITION_INDEX,
channelConsumerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.PARTITION_INDEX,
Integer.toString(getInstanceIndex()));
return channelConsumerProperties;
}
@@ -146,24 +165,24 @@ public class ChannelBindingServiceProperties {
if (isPartitionedProducer(outputChannelName)) {
Properties channelProducerProperties = new Properties();
channelProducerProperties.putAll(this.producerProperties);
channelProducerProperties.setProperty(BinderProperties.NEXT_MODULE_COUNT,
channelProducerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.NEXT_MODULE_COUNT,
Integer.toString(getPartitionCount(outputChannelName)));
BindingProperties bindingProperties = bindings.get(outputChannelName);
if (bindingProperties != null) {
if (bindingProperties.getPartitionKeyExpression() != null) {
channelProducerProperties.setProperty(BinderProperties.PARTITION_KEY_EXPRESSION,
channelProducerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.PARTITION_KEY_EXPRESSION,
bindingProperties.getPartitionKeyExpression());
}
if (bindingProperties.getPartitionKeyExtractorClass() != null) {
channelProducerProperties.setProperty(BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS,
channelProducerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.PARTITION_KEY_EXTRACTOR_CLASS,
bindingProperties.getPartitionKeyExtractorClass());
}
if (bindingProperties.getPartitionSelectorClass() != null) {
channelProducerProperties.setProperty(BinderProperties.PARTITION_SELECTOR_CLASS,
channelProducerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.PARTITION_SELECTOR_CLASS,
bindingProperties.getPartitionSelectorClass());
}
if (bindingProperties.getPartitionSelectorExpression() != null) {
channelProducerProperties.setProperty(BinderProperties.PARTITION_SELECTOR_EXPRESSION,
channelProducerProperties.setProperty(org.springframework.cloud.stream.binder.BinderProperties.PARTITION_SELECTOR_EXPRESSION,
bindingProperties.getPartitionSelectorExpression());
}
}
@@ -174,4 +193,11 @@ public class ChannelBindingServiceProperties {
}
}
public String getBinder(String channelName) {
if (!bindings.containsKey(channelName)) {
return null;
}
return bindings.get(channelName).getBinder();
}
}

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithBindingTargets.java
View File

@@ -31,7 +31,7 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -63,7 +63,7 @@ public class ArbitraryInterfaceBindingTestsWithBindingTargets {
@EnableBinding(FooChannels.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
@PropertySource("classpath:/org/springframework/cloud/stream/binder/arbitrary-binding-test.properties")
public static class TestFooChannels {

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithDefaults.java
View File

@@ -29,9 +29,9 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -62,7 +62,7 @@ public class ArbitraryInterfaceBindingTestsWithDefaults {
@EnableBinding(FooChannels.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
public static class TestFooChannels {
}

24
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java
View File

@@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
@@ -32,6 +33,7 @@ import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
@@ -68,7 +70,12 @@ public class BinderAwareChannelResolverTests {
this.binder = new LocalMessageChannelBinder();
this.binder.setApplicationContext(context);
this.binder.afterPropertiesSet();
this.resolver = new BinderAwareChannelResolver(this.binder, null);
this.resolver = new BinderAwareChannelResolver(new BinderFactory<MessageChannel>() {
@Override
public Binder<MessageChannel> getBinder(String configurationName) {
return binder;
}
}, null);
this.resolver.setBeanFactory(context);
context.getBeanFactory().registerSingleton("channelResolver",
this.resolver);
@@ -157,17 +164,20 @@ public class BinderAwareChannelResolverTests {
public void propertyPassthrough() {
Properties properties = new Properties();
@SuppressWarnings("rawtypes")
Binder binder = mock(Binder.class);
doReturn(new DirectChannel()).when(binder).bindDynamicProducer("queue:foo", properties);
doReturn(new DirectChannel()).when(binder).bindDynamicPubSubProducer("topic:bar", properties);
Binder binderFactory = mock(Binder.class);
doReturn(new DirectChannel()).when(binderFactory).bindDynamicProducer("queue:foo", properties);
doReturn(new DirectChannel()).when(binderFactory).bindDynamicPubSubProducer("topic:bar", properties);
BinderFactory mockBinderFactory = Mockito.mock(BinderFactory.class);
Mockito.when(mockBinderFactory.getBinder(anyString())).thenReturn(binderFactory);
@SuppressWarnings("unchecked")
BinderAwareChannelResolver resolver = new BinderAwareChannelResolver(binder, properties);
BinderAwareChannelResolver resolver =
new BinderAwareChannelResolver(mockBinderFactory, properties);
BeanFactory beanFactory = new DefaultListableBeanFactory();
resolver.setBeanFactory(beanFactory);
resolver.resolveDestination("queue:foo");
resolver.resolveDestination("topic:bar");
verify(binder).bindDynamicProducer("queue:foo", properties);
verify(binder).bindDynamicPubSubProducer("topic:bar", properties);
verify(binderFactory).bindDynamicProducer("queue:foo", properties);
verify(binderFactory).bindDynamicPubSubProducer("topic:bar", properties);
}
}

174
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java Normal file
View File

@@ -0,0 +1,174 @@
/*
* 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.cloud.stream.binder;
import static org.hamcrest.Matchers.arrayContaining;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasKey;
import static org.hamcrest.Matchers.hasProperty;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import java.io.IOException;
import java.net.URL;
import java.net.URLClassLoader;
import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.binder.stub1.StubBinder1;
import org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration;
import org.springframework.cloud.stream.binder.stub2.StubBinder2;
import org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationA;
import org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationB;
import org.springframework.cloud.stream.config.BinderFactoryConfiguration;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Import;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.DefaultResourceLoader;
import org.springframework.util.ObjectUtils;
/**
* @author Marius Bogoevici
*/
public class BinderFactoryConfigurationTests {
@Test
public void loadBinderTypeRegistry() throws Exception {
try {
ConfigurableApplicationContext context = createBinderTestContext(new String[]{});
fail();
}
catch (BeanCreationException e) {
assertThat(e.getMessage(),containsString("Cannot create binder factory, no `META-INF/spring.binders` " +
"resources found on the classpath"));
}
}
@Test
public void loadBinderTypeRegistryWithOneBinder() throws Exception {
ConfigurableApplicationContext context = createBinderTestContext(
new String[] {"binder1"});
BinderTypeRegistry binderTypeRegistry = context.getBean(BinderTypeRegistry.class);
assertThat(binderTypeRegistry, notNullValue());
assertThat(binderTypeRegistry.getAll().size(), equalTo(1));
assertThat(binderTypeRegistry.getAll(), hasKey("binder1"));
assertThat(binderTypeRegistry.get("binder1"),
hasProperty("configurationClasses", arrayContaining(StubBinder1Configuration.class)));
BinderFactory binderFactory = context.getBean(BinderFactory.class);
Binder binder1 = binderFactory.getBinder("binder1");
assertThat(binder1, instanceOf(StubBinder1.class));
Binder defaultBinder = binderFactory.getBinder(null);
assertThat(defaultBinder, is(binder1));
}
@Test
public void loadBinderTypeRegistryWithTwoBinders() throws Exception {
ConfigurableApplicationContext context = createBinderTestContext(
new String[]{"binder1", "binder2"});
BinderTypeRegistry binderTypeRegistry = context.getBean(BinderTypeRegistry.class);
assertThat(binderTypeRegistry, notNullValue());
assertThat(binderTypeRegistry.getAll().size(), equalTo(2));
assertThat(binderTypeRegistry.getAll().keySet(), containsInAnyOrder("binder1", "binder2"));
assertThat(binderTypeRegistry.get("binder1"),
hasProperty("configurationClasses", arrayContaining(StubBinder1Configuration.class)));
assertThat(binderTypeRegistry.get("binder2"),
hasProperty("configurationClasses", arrayContaining(StubBinder2ConfigurationA.class,
StubBinder2ConfigurationB.class)));
BinderFactory binderFactory = context.getBean(BinderFactory.class);
try {
binderFactory.getBinder(null);
fail();
}
catch (Exception e) {
assertThat(e, instanceOf(IllegalStateException.class));
assertThat(e.getMessage(), containsString("A default binder has been requested, but there is more than " +
"one binder available:"));
}
Binder binder1 = binderFactory.getBinder("binder1");
assertThat(binder1, instanceOf(StubBinder1.class));
Binder binder2 = binderFactory.getBinder("binder2");
assertThat(binder2, instanceOf(StubBinder2.class));
}
@Test
public void loadDefaultBinderWithTwoBinders() throws Exception {
ConfigurableApplicationContext context =
createBinderTestContext(
new String[]{"binder1", "binder2"}, "spring.cloud.stream.defaultBinder:binder2");
BinderTypeRegistry binderTypeRegistry = context.getBean(BinderTypeRegistry.class);
assertThat(binderTypeRegistry, notNullValue());
assertThat(binderTypeRegistry.getAll().size(), equalTo(2));
assertThat(binderTypeRegistry.getAll().keySet(), containsInAnyOrder("binder1", "binder2"));
assertThat(binderTypeRegistry.get("binder1"),
hasProperty("configurationClasses", arrayContaining(StubBinder1Configuration.class)));
assertThat(binderTypeRegistry.get("binder2"),
hasProperty("configurationClasses", arrayContaining(StubBinder2ConfigurationA.class,
StubBinder2ConfigurationB.class)));
BinderFactory binderFactory = context.getBean(BinderFactory.class);
Binder binder1 = binderFactory.getBinder("binder1");
assertThat(binder1, instanceOf(StubBinder1.class));
Binder binder2 = binderFactory.getBinder("binder2");
assertThat(binder2, instanceOf(StubBinder2.class));
Binder defaultBinder = binderFactory.getBinder(null);
assertThat(defaultBinder, is(binder2));
}
private static ConfigurableApplicationContext createBinderTestContext(String[] additionalClasspathDirectories,
String... properties)
throws IOException {
URL[] urls = ObjectUtils.isEmpty(additionalClasspathDirectories) ?
new URL[0] : new URL[additionalClasspathDirectories.length];
if (!ObjectUtils.isEmpty(additionalClasspathDirectories)) {
for (int i = 0; i < additionalClasspathDirectories.length; i++) {
urls[i] = new URL(new ClassPathResource(additionalClasspathDirectories[i]).getURL().toString() + "/");
}
}
ClassLoader classLoader = new URLClassLoader(urls, BinderFactoryConfigurationTests.class.getClassLoader());
return new SpringApplicationBuilder(SimpleApplication.class)
.resourceLoader(new DefaultResourceLoader(classLoader))
.properties(properties)
.web(false)
.run();
}
@Import({BinderFactoryConfiguration.class, PropertyPlaceholderAutoConfiguration.class})
@EnableConfigurationProperties(ChannelBindingServiceProperties.class)
public static class SimpleApplication {
}
}

7
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java
View File

@@ -26,12 +26,13 @@ import java.util.Properties;
import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.SmartLifecycle;
import org.springframework.context.annotation.Bean;
@@ -47,7 +48,7 @@ public class InputOutputBindingOrderTest {
public void testInputOutputBindingOrder() {
ConfigurableApplicationContext applicationContext = SpringApplication.run(TestSource.class, "--server.port=-1");
@SuppressWarnings("rawtypes")
Binder binder = applicationContext.getBean(Binder.class);
Binder binder = applicationContext.getBean(BinderFactory.class).getBinder(null);
Processor processor = applicationContext.getBean(Processor.class);
// input is bound after the context has been started
verify(binder).bindConsumer(eq("input"), eq(processor.input()), Mockito.<Properties>any());
@@ -59,7 +60,7 @@ public class InputOutputBindingOrderTest {
@EnableBinding(Processor.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
public static class TestSource {
@Bean

5
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/LifecycleBinderTests.java
View File

@@ -20,11 +20,12 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.Bean;
@@ -46,7 +47,7 @@ public class LifecycleBinderTests {
@EnableBinding(Source.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
public static class TestSource {
@Bean

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithBindingTargets.java
View File

@@ -28,10 +28,10 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -59,7 +59,7 @@ public class ProcessorBindingTestsWithBindingTargets {
@EnableBinding(Processor.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
@PropertySource("classpath:/org/springframework/cloud/stream/binder/processor-binding-test.properties")
public static class TestProcessor {

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithDefaults.java
View File

@@ -28,10 +28,10 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -59,7 +59,7 @@ public class ProcessorBindingTestsWithDefaults {
@EnableBinding(Processor.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
public static class TestProcessor {
}

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithPubSubBindingTargets.java
View File

@@ -33,7 +33,7 @@ import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -63,7 +63,7 @@ public class ProcessorBindingTestsWithPubSubBindingTargets {
@EnableBinding(Processor.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
@PropertySource("classpath:/org/springframework/cloud/stream/binder/processor-binding-test-pubsub.properties")
public static class TestProcessor {

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingPubSubTests.java
View File

@@ -32,7 +32,7 @@ import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -61,7 +61,7 @@ public class SinkBindingPubSubTests {
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
@PropertySource("classpath:/org/springframework/cloud/stream/binder/sink-binding-pubsub-test.properties")
public static class TestSink {

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithBindingTargets.java
View File

@@ -29,10 +29,10 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -60,7 +60,7 @@ public class SinkBindingTestsWithBindingTargets {
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
@PropertySource("classpath:/org/springframework/cloud/stream/binder/sink-binding-test.properties")
public static class TestSink {

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithDefaults.java
View File

@@ -32,7 +32,7 @@ import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -59,7 +59,7 @@ public class SinkBindingTestsWithDefaults {
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
public static class TestSink {
}

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java
View File

@@ -29,10 +29,10 @@ import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -60,7 +60,7 @@ public class SourceBindingTestsWithBindingTargets {
@EnableBinding(Source.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
@PropertySource("classpath:/org/springframework/cloud/stream/binder/source-binding-test.properties")
public static class TestSource {

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithDefaults.java
View File

@@ -32,7 +32,7 @@ import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -59,7 +59,7 @@ public class SourceBindingTestsWithDefaults {
@EnableBinding(Source.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
public static class TestSource {
}

92
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java Normal file
View File

@@ -0,0 +1,92 @@
/*
* 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.cloud.stream.binder.stub1;
import java.util.Properties;
import org.springframework.cloud.stream.binder.Binder;
/**
* @author Marius Bogoevici
*/
public class StubBinder1 implements Binder {
@Override
public void bindConsumer(String name, Object inboundBindTarget, Properties properties) {
}
@Override
public void bindPubSubConsumer(String name, Object inboundBindTarget, String group, Properties properties) {
}
@Override
public void bindProducer(String name, Object outboundBindTarget, Properties properties) {
}
@Override
public void bindPubSubProducer(String name, Object outboundBindTarget, Properties properties) {
}
@Override
public void unbindConsumers(String name) {
}
@Override
public void unbindPubSubConsumers(String name, String group) {
}
@Override
public void unbindProducers(String name) {
}
@Override
public void unbindConsumer(String name, Object inboundBindTarget) {
}
@Override
public void unbindProducer(String name, Object outboundBindTarget) {
}
@Override
public void bindRequestor(String name, Object requests, Object replies, Properties properties) {
}
@Override
public void bindReplier(String name, Object requests, Object replies, Properties properties) {
}
@Override
public Object bindDynamicProducer(String name, Properties properties) {
return null;
}
@Override
public Object bindDynamicPubSubProducer(String name, Properties properties) {
return null;
}
}

33
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1Configuration.java Normal file
View File

@@ -0,0 +1,33 @@
/*
* 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.cloud.stream.binder.stub1;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author Marius Bogoevici
*/
@Configuration
public class StubBinder1Configuration {
@Bean
public Binder binder() {
return new StubBinder1();
}
}

98
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java Normal file
View File

@@ -0,0 +1,98 @@
/*
* 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.cloud.stream.binder.stub2;
import java.util.Properties;
import org.springframework.cloud.stream.binder.Binder;
/**
* @author Marius Bogoevici
*/
public class StubBinder2 implements Binder {
private StubBinder2Dependency stubBinder2Dependency;
public StubBinder2(StubBinder2Dependency stubBinder2Dependency) {
this.stubBinder2Dependency = stubBinder2Dependency;
}
@Override
public void bindConsumer(String name, Object inboundBindTarget, Properties properties) {
}
@Override
public void bindPubSubConsumer(String name, Object inboundBindTarget, String group, Properties properties) {
}
@Override
public void bindProducer(String name, Object outboundBindTarget, Properties properties) {
}
@Override
public void bindPubSubProducer(String name, Object outboundBindTarget, Properties properties) {
}
@Override
public void unbindConsumers(String name) {
}
@Override
public void unbindPubSubConsumers(String name, String group) {
}
@Override
public void unbindProducers(String name) {
}
@Override
public void unbindConsumer(String name, Object inboundBindTarget) {
}
@Override
public void unbindProducer(String name, Object outboundBindTarget) {
}
@Override
public void bindRequestor(String name, Object requests, Object replies, Properties properties) {
}
@Override
public void bindReplier(String name, Object requests, Object replies, Properties properties) {
}
@Override
public Object bindDynamicProducer(String name, Properties properties) {
return null;
}
@Override
public Object bindDynamicPubSubProducer(String name, Properties properties) {
return null;
}
}

34
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationA.java Normal file
View File

@@ -0,0 +1,34 @@
/*
* 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.cloud.stream.binder.stub2;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.stub2.StubBinder2;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author Marius Bogoevici
*/
@Configuration
public class StubBinder2ConfigurationA {
@Bean
public Binder binder(StubBinder2Dependency dependency) {
return new StubBinder2(dependency);
}
}

32
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationB.java Normal file
View File

@@ -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.cloud.stream.binder.stub2;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author Marius Bogoevici
*/
@Configuration
public class StubBinder2ConfigurationB {
@Bean
public StubBinder2Dependency dependency() {
return new StubBinder2Dependency();
}
}

24
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2Dependency.java Normal file
View File

@@ -0,0 +1,24 @@
/*
* 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.cloud.stream.binder.stub2;
/**
* @author Marius Bogoevici
*/
public class StubBinder2Dependency {
}

25
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java
View File

@@ -15,17 +15,22 @@
*/
package org.springframework.cloud.stream.binding;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import org.junit.Test;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderConfiguration;
import org.springframework.cloud.stream.binder.BinderType;
import org.springframework.cloud.stream.binder.DefaultBinderFactory;
import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.ChannelBindingServiceProperties;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.messaging.MessageChannel;
@@ -36,7 +41,7 @@ import org.springframework.messaging.MessageChannel;
public class ChannelBindingServiceTests {
@Test
public void testSimple() {
public void testSimple() throws Exception {
ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties();
Map<String, BindingProperties> bindings = new HashMap<>();
BindingProperties props = new BindingProperties();
@@ -45,17 +50,20 @@ public class ChannelBindingServiceTests {
bindings.put(name, props);
properties.setBindings(bindings);
@SuppressWarnings("unchecked")
Binder<MessageChannel> binder = mock(Binder.class);
ChannelBindingService service = new ChannelBindingService(properties, binder);
DefaultBinderFactory<MessageChannel> binderFactory =
new DefaultBinderFactory<>(Collections.singletonMap("mock", new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}), new Properties())));
Binder<MessageChannel> binder = binderFactory.getBinder("mock");
ChannelBindingService service = new ChannelBindingService(properties, binderFactory);
MessageChannel inputChannel = new DirectChannel();
service.bindConsumer(inputChannel, name);
service.unbindConsumers(name);
verify(binder).bindConsumer(name, inputChannel, properties.getConsumerProperties(name));
verify(binder).unbindConsumers(name);
binderFactory.destroy();
}
@Test
public void testPubSub() {
public void testPubSub() throws Exception {
ChannelBindingServiceProperties properties = new ChannelBindingServiceProperties();
Map<String, BindingProperties> bindings = new HashMap<>();
BindingProperties props = new BindingProperties();
@@ -64,13 +72,16 @@ public class ChannelBindingServiceTests {
bindings.put(name, props);
properties.setBindings(bindings);
@SuppressWarnings("unchecked")
Binder<MessageChannel> binder = mock(Binder.class);
ChannelBindingService service = new ChannelBindingService(properties, binder);
DefaultBinderFactory<MessageChannel> binderFactory =
new DefaultBinderFactory<>(Collections.singletonMap("mock", new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}), new Properties())));
Binder<MessageChannel> binder = binderFactory.getBinder("mock");
ChannelBindingService service = new ChannelBindingService(properties, binderFactory);
MessageChannel inputChannel = new DirectChannel();
service.bindConsumer(inputChannel, name);
service.unbindConsumers(name);
verify(binder).bindPubSubConsumer(name, inputChannel, props.getGroup(), properties.getConsumerProperties(name));
verify(binder).unbindPubSubConsumers(name, props.getGroup());
binderFactory.destroy();
}
}

114
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderConfigurationParsingTests.java Normal file
View File

@@ -0,0 +1,114 @@
/*
* 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.cloud.stream.config;
import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.beans.HasPropertyWithValue.hasProperty;
import static org.hamcrest.collection.IsArrayContaining.hasItemInArray;
import static org.hamcrest.collection.IsArrayContainingInAnyOrder.arrayContainingInAnyOrder;
import static org.hamcrest.core.CombinableMatcher.both;
import java.io.ByteArrayInputStream;
import java.util.Collection;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.cloud.stream.binder.BinderType;
import org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration;
import org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationA;
import org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationB;
import org.springframework.core.io.InputStreamResource;
import org.springframework.core.io.Resource;
/**
* @author Marius Bogoevici
*/
@SuppressWarnings("Duplicates")
public class BinderConfigurationParsingTests {
private static ClassLoader classLoader = BinderConfigurationParsingTests.class.getClassLoader();
@Test
public void testParseOneBinderConfiguration() throws Exception {
// this is just checking that resources are passed and classes are loaded properly
// class values used here are not binder configurations
String oneBinderConfiguration = "binder1=org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration";
Resource resource = new InputStreamResource(new ByteArrayInputStream(oneBinderConfiguration.getBytes()));
Collection<BinderType> binderConfigurations
= BinderFactoryConfiguration.parseBinderConfigurations(classLoader, resource);
Assert.assertNotNull(binderConfigurations);
Assert.assertThat(binderConfigurations.size(), equalTo(1));
Assert.assertThat(binderConfigurations, contains(
both(hasProperty("defaultName", equalTo("binder1"))).and(
hasProperty("configurationClasses", hasItemInArray(StubBinder1Configuration.class)))
));
}
@Test
public void testParseTwoBindersConfigurations() throws Exception {
// this is just checking that resources are passed and classes are loaded properly
// class values used here are not binder configurations
String binderConfiguration = "binder1=org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration\n" +
"binder2=org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationA";
Resource twoBinderConfigurationResource =
new InputStreamResource(new ByteArrayInputStream(binderConfiguration.getBytes()));
Collection<BinderType> twoBinderConfigurations
= BinderFactoryConfiguration.parseBinderConfigurations(classLoader,
twoBinderConfigurationResource);
Assert.assertThat(twoBinderConfigurations.size(), equalTo(2));
Assert.assertThat(twoBinderConfigurations, containsInAnyOrder(
both(hasProperty("defaultName", equalTo("binder1"))).and(
hasProperty("configurationClasses", hasItemInArray(StubBinder1Configuration.class))),
both(hasProperty("defaultName", equalTo("binder2"))).and(
hasProperty("configurationClasses", hasItemInArray(StubBinder2ConfigurationA.class)))
));
}
@Test
@SuppressWarnings("unchecked")
public void testParseTwoBindersWithMultipleClasses() throws Exception {
// this is just checking that resources are passed and classes are loaded properly
// class values used here are not binder configurations
String binderConfiguration = "binder1=org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration\n" +
"binder2=org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationA," +
"org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationB";
Resource binderConfigurationResource =
new InputStreamResource(new ByteArrayInputStream(binderConfiguration.getBytes()));
Collection<BinderType> binderConfigurations
= BinderFactoryConfiguration.parseBinderConfigurations(classLoader, binderConfigurationResource);
Assert.assertThat(binderConfigurations.size(), equalTo(2));
Assert.assertThat(binderConfigurations, containsInAnyOrder(
both(hasProperty("defaultName", equalTo("binder1"))).and(
hasProperty("configurationClasses", hasItemInArray(StubBinder1Configuration.class))),
both(hasProperty("defaultName", equalTo("binder2"))).and(
hasProperty("configurationClasses", arrayContainingInAnyOrder(StubBinder2ConfigurationA.class,
StubBinder2ConfigurationB.class)))
));
}
}

5
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java
View File

@@ -22,13 +22,14 @@ import static org.mockito.Mockito.verifyNoMoreInteractions;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.SpringApplicationConfiguration;
import org.springframework.cloud.stream.annotation.Bindings;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.integration.annotation.ServiceActivator;
@@ -66,7 +67,7 @@ public class BoundChannelsInterceptedTest {
@SpringBootApplication
@EnableBinding(Sink.class)
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
public static class Foo {
@ServiceActivator(inputChannel = Sink.INPUT)

6
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java
View File

@@ -17,7 +17,6 @@
package org.springframework.cloud.stream.partitioning;
import static org.hamcrest.core.IsEqual.equalTo;
import static org.hamcrest.text.IsEqualIgnoringCase.equalToIgnoringCase;
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -38,8 +37,7 @@ import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -73,7 +71,7 @@ public class PartitionedConsumerTest {
@EnableBinding(Sink.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
@PropertySource("classpath:/org/springframework/cloud/stream/binder/partitioned-consumer-test.properties")
public static class TestSink {

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java
View File

@@ -37,7 +37,7 @@ import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderProperties;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.cloud.stream.utils.MockBinderConfiguration;
import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.PropertySource;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -71,7 +71,7 @@ public class PartitionedProducerTest {
@EnableBinding(Source.class)
@EnableAutoConfiguration
@Import(MockBinderConfiguration.class)
@Import(MockBinderRegistryConfiguration.class)
@PropertySource("classpath:/org/springframework/cloud/stream/binder/partitioned-producer-test.properties")
public static class TestSource {

5
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java
View File

@@ -13,19 +13,18 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.utils;
import org.mockito.Mockito;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* A simple configuration that creates mock {@link org.springframework.cloud.stream.binder.Binder}s.
*
* @author Marius Bogoevici
*/
@Configuration
public class MockBinderConfiguration {
@Bean

50
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* 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.cloud.stream.utils;
import java.util.Collections;
import java.util.Properties;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.BinderConfiguration;
import org.springframework.cloud.stream.binder.BinderFactory;
import org.springframework.cloud.stream.binder.BinderType;
import org.springframework.cloud.stream.binder.BinderTypeRegistry;
import org.springframework.cloud.stream.binder.DefaultBinderFactory;
import org.springframework.cloud.stream.binder.DefaultBinderTypeRegistry;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.MessageChannel;
/**
* A simple configuration that creates mock {@link org.springframework.cloud.stream.binder.Binder}s.
* @author Marius Bogoevici
*/
@Configuration
public class MockBinderRegistryConfiguration {
@Bean
public BinderTypeRegistry binderTypeRegistry() {
return new DefaultBinderTypeRegistry(
Collections.singletonMap("mock", new BinderType("", new Class[]{MockBinderConfiguration.class})));
}
@Bean
public Binder<?> defaultBinder(BinderFactory<MessageChannel> binderFactory) {
return binderFactory.getBinder(null);
}
}

1
spring-cloud-stream/src/test/resources/binder1/META-INF/spring.binders Normal file
View File

@@ -0,0 +1 @@
binder1=org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration

3
spring-cloud-stream/src/test/resources/binder2/META-INF/spring.binders Normal file
View File

@@ -0,0 +1,3 @@
binder2:\
org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationA,\
org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationB