From 96f96f7621d6d4ccd24b37036a3707a0fe41d669 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Mon, 30 Nov 2015 16:59:21 -0500 Subject: [PATCH] Initial implementation of multibinder support - 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 --- .../{spring.factories => spring.binders} | 2 +- .../{spring.factories => spring.binders} | 2 +- .../{spring.factories => spring.binders} | 2 +- .../multibinder-differentsystems/pom.xml | 102 ++++++++++ .../java/multibinder/BridgeTransformer.java | 33 ++++ .../multibinder/MultibinderApplication.java | 29 +++ .../src/main/resources/application.yml | 33 ++++ .../TwoKafkaBindersApplicationTest.java | 90 +++++++++ .../multibinder/pom.xml | 79 ++++++++ .../java/multibinder/BridgeTransformer.java | 33 ++++ .../multibinder/MultibinderApplication.java | 29 +++ .../src/main/resources/application.yml | 12 ++ .../RabbitAndRedisBinderApplicationTests.java | 78 ++++++++ spring-cloud-stream-samples/pom.xml | 2 + .../test/junit/kafka/KafkaTestSupport.java | 20 +- .../{spring.factories => spring.binders} | 2 +- .../cloud/stream/test/ExampleTest.java | 14 +- .../stream/annotation/EnableBinding.java | 5 +- .../stream/binder/BinderConfiguration.java | 51 +++++ .../cloud/stream/binder/BinderFactory.java | 32 ++++ .../cloud/stream/binder/BinderType.java | 70 +++++++ .../stream/binder/BinderTypeRegistry.java | 33 ++++ .../stream/binder/DefaultBinderFactory.java | 161 ++++++++++++++++ .../binder/DefaultBinderTypeRegistry.java | 45 +++++ .../binding/BinderAwareChannelResolver.java | 31 +++- .../BinderAwareRouterBeanPostProcessor.java | 2 +- .../stream/binding/ChannelBindingService.java | 31 +++- .../config/BinderFactoryConfiguration.java | 134 ++++++++++++++ .../cloud/stream/config/BinderProperties.java | 47 +++++ .../stream/config/BindingProperties.java | 9 + .../ChannelBindingServiceConfiguration.java | 12 +- .../ChannelBindingServiceProperties.java | 54 ++++-- ...terfaceBindingTestsWithBindingTargets.java | 4 +- ...raryInterfaceBindingTestsWithDefaults.java | 6 +- .../BinderAwareChannelResolverTests.java | 24 ++- .../BinderFactoryConfigurationTests.java | 174 ++++++++++++++++++ .../binder/InputOutputBindingOrderTest.java | 7 +- .../stream/binder/LifecycleBinderTests.java | 5 +- ...ocessorBindingTestsWithBindingTargets.java | 6 +- .../ProcessorBindingTestsWithDefaults.java | 6 +- ...rBindingTestsWithPubSubBindingTargets.java | 4 +- .../stream/binder/SinkBindingPubSubTests.java | 4 +- .../SinkBindingTestsWithBindingTargets.java | 6 +- .../binder/SinkBindingTestsWithDefaults.java | 4 +- .../SourceBindingTestsWithBindingTargets.java | 6 +- .../SourceBindingTestsWithDefaults.java | 4 +- .../stream/binder/stub1/StubBinder1.java | 92 +++++++++ .../stub1/StubBinder1Configuration.java | 33 ++++ .../stream/binder/stub2/StubBinder2.java | 98 ++++++++++ .../stub2/StubBinder2ConfigurationA.java | 34 ++++ .../stub2/StubBinder2ConfigurationB.java | 32 ++++ .../binder/stub2/StubBinder2Dependency.java | 24 +++ .../binding/ChannelBindingServiceTests.java | 25 ++- .../BinderConfigurationParsingTests.java | 114 ++++++++++++ .../BoundChannelsInterceptedTest.java | 5 +- .../partitioning/PartitionedConsumerTest.java | 6 +- .../partitioning/PartitionedProducerTest.java | 4 +- .../stream/utils/MockBinderConfiguration.java | 5 +- .../MockBinderRegistryConfiguration.java | 50 +++++ .../resources/binder1/META-INF/spring.binders | 1 + .../resources/binder2/META-INF/spring.binders | 3 + 61 files changed, 1948 insertions(+), 117 deletions(-) rename spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/{spring.factories => spring.binders} (56%) rename spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/{spring.factories => spring.binders} (56%) rename spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/{spring.factories => spring.binders} (56%) create mode 100644 spring-cloud-stream-samples/multibinder-differentsystems/pom.xml create mode 100644 spring-cloud-stream-samples/multibinder-differentsystems/src/main/java/multibinder/BridgeTransformer.java create mode 100644 spring-cloud-stream-samples/multibinder-differentsystems/src/main/java/multibinder/MultibinderApplication.java create mode 100644 spring-cloud-stream-samples/multibinder-differentsystems/src/main/resources/application.yml create mode 100644 spring-cloud-stream-samples/multibinder-differentsystems/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java create mode 100644 spring-cloud-stream-samples/multibinder/pom.xml create mode 100644 spring-cloud-stream-samples/multibinder/src/main/java/multibinder/BridgeTransformer.java create mode 100644 spring-cloud-stream-samples/multibinder/src/main/java/multibinder/MultibinderApplication.java create mode 100644 spring-cloud-stream-samples/multibinder/src/main/resources/application.yml create mode 100644 spring-cloud-stream-samples/multibinder/src/test/java/multibinder/RabbitAndRedisBinderApplicationTests.java rename spring-cloud-stream-test-support/src/main/resources/META-INF/{spring.factories => spring.binders} (55%) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderConfiguration.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderFactory.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderType.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderTypeRegistry.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderTypeRegistry.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1Configuration.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationA.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationB.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2Dependency.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderConfigurationParsingTests.java create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java create mode 100644 spring-cloud-stream/src/test/resources/binder1/META-INF/spring.binders create mode 100644 spring-cloud-stream/src/test/resources/binder2/META-INF/spring.binders diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders similarity index 56% rename from spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories rename to spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders index 7f00520be..375468587 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/resources/META-INF/spring.binders @@ -1,2 +1,2 @@ -org.springframework.boot.autoconfigure.EnableAutoConfiguration:\ +kafka:\ org.springframework.cloud.stream.binder.kafka.config.KafkaServiceAutoConfiguration diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders similarity index 56% rename from spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.factories rename to spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders index cde4a6309..7a65c4682 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-rabbit/src/main/resources/META-INF/spring.binders @@ -1,2 +1,2 @@ -org.springframework.boot.autoconfigure.EnableAutoConfiguration:\ +rabbit:\ org.springframework.cloud.stream.binder.rabbit.config.RabbitServiceAutoConfiguration diff --git a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.factories b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.binders similarity index 56% rename from spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.factories rename to spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.binders index acd5c3a0c..12ae7ef61 100644 --- a/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.factories +++ b/spring-cloud-stream-binders/spring-cloud-stream-binder-redis/src/main/resources/META-INF/spring.binders @@ -1,2 +1,2 @@ -org.springframework.boot.autoconfigure.EnableAutoConfiguration:\ +redis:\ org.springframework.cloud.stream.binder.redis.config.RedisServiceAutoConfiguration diff --git a/spring-cloud-stream-samples/multibinder-differentsystems/pom.xml b/spring-cloud-stream-samples/multibinder-differentsystems/pom.xml new file mode 100644 index 000000000..35addd9c3 --- /dev/null +++ b/spring-cloud-stream-samples/multibinder-differentsystems/pom.xml @@ -0,0 +1,102 @@ + + + 4.0.0 + + spring-cloud-stream-sample-multibinder-differentsystems + jar + spring-cloud-stream-sample-multibinder-differentsystems + + Demo project for multiple binders of the same type (Kafka), connecting to different systems (broker groups) + + + + org.springframework.cloud + spring-cloud-stream-samples + 1.0.0.BUILD-SNAPSHOT + + + + multibinder.MultibinderApplication + + + + + org.springframework.cloud + spring-cloud-stream + + + org.springframework.cloud + spring-cloud-stream-sample-source + + + org.springframework.cloud + spring-cloud-stream-sample-transform + + + org.springframework.cloud + spring-cloud-stream-sample-sink + + + org.springframework.cloud + spring-cloud-stream-binder-redis + + + org.springframework.cloud + spring-cloud-stream-binder-kafka + + + org.springframework.cloud + spring-cloud-stream-test-support-internal + test + + + org.apache.kafka + kafka_2.10 + test + 0.8.2.1 + test + + + org.slf4j + slf4j-log4j12 + + + + + org.apache.curator + curator-test + 2.6.0 + test + + + org.slf4j + slf4j-log4j12 + + + + + org.springframework.boot + spring-boot-configuration-processor + true + + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + exec + + + + + + diff --git a/spring-cloud-stream-samples/multibinder-differentsystems/src/main/java/multibinder/BridgeTransformer.java b/spring-cloud-stream-samples/multibinder-differentsystems/src/main/java/multibinder/BridgeTransformer.java new file mode 100644 index 000000000..d68720ffe --- /dev/null +++ b/spring-cloud-stream-samples/multibinder-differentsystems/src/main/java/multibinder/BridgeTransformer.java @@ -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; + } +} diff --git a/spring-cloud-stream-samples/multibinder-differentsystems/src/main/java/multibinder/MultibinderApplication.java b/spring-cloud-stream-samples/multibinder-differentsystems/src/main/java/multibinder/MultibinderApplication.java new file mode 100644 index 000000000..7aba5f67b --- /dev/null +++ b/spring-cloud-stream-samples/multibinder-differentsystems/src/main/java/multibinder/MultibinderApplication.java @@ -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); + } + +} diff --git a/spring-cloud-stream-samples/multibinder-differentsystems/src/main/resources/application.yml b/spring-cloud-stream-samples/multibinder-differentsystems/src/main/resources/application.yml new file mode 100644 index 000000000..ca88423b9 --- /dev/null +++ b/spring-cloud-stream-samples/multibinder-differentsystems/src/main/resources/application.yml @@ -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} diff --git a/spring-cloud-stream-samples/multibinder-differentsystems/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java b/spring-cloud-stream-samples/multibinder-differentsystems/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java new file mode 100644 index 000000000..62dd8dc0b --- /dev/null +++ b/spring-cloud-stream-samples/multibinder-differentsystems/src/test/java/multibinder/TwoKafkaBindersApplicationTest.java @@ -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 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)); + } + +} diff --git a/spring-cloud-stream-samples/multibinder/pom.xml b/spring-cloud-stream-samples/multibinder/pom.xml new file mode 100644 index 000000000..36ebe1d7f --- /dev/null +++ b/spring-cloud-stream-samples/multibinder/pom.xml @@ -0,0 +1,79 @@ + + + 4.0.0 + + spring-cloud-stream-sample-multibinder + jar + spring-cloud-stream-sample-multibinder + Demo project for multiple binders of different types (Redis and Rabbit) + + + org.springframework.cloud + spring-cloud-stream-samples + 1.0.0.BUILD-SNAPSHOT + + + + multibinder.MultibinderApplication + + + + + org.springframework.cloud + spring-cloud-stream + + + org.springframework.cloud + spring-cloud-stream-sample-source + + + org.springframework.cloud + spring-cloud-stream-sample-transform + + + org.springframework.cloud + spring-cloud-stream-sample-sink + + + org.springframework.cloud + spring-cloud-stream-binder-redis + + + org.springframework.cloud + spring-cloud-stream-binder-rabbit + + + org.springframework.cloud + spring-cloud-stream-test-support-internal + test + + + org.springframework.boot + spring-boot-starter-redis + + + org.springframework.boot + spring-boot-configuration-processor + true + + + + org.springframework.boot + spring-boot-starter-test + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + exec + + + + + + diff --git a/spring-cloud-stream-samples/multibinder/src/main/java/multibinder/BridgeTransformer.java b/spring-cloud-stream-samples/multibinder/src/main/java/multibinder/BridgeTransformer.java new file mode 100644 index 000000000..d68720ffe --- /dev/null +++ b/spring-cloud-stream-samples/multibinder/src/main/java/multibinder/BridgeTransformer.java @@ -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; + } +} diff --git a/spring-cloud-stream-samples/multibinder/src/main/java/multibinder/MultibinderApplication.java b/spring-cloud-stream-samples/multibinder/src/main/java/multibinder/MultibinderApplication.java new file mode 100644 index 000000000..7aba5f67b --- /dev/null +++ b/spring-cloud-stream-samples/multibinder/src/main/java/multibinder/MultibinderApplication.java @@ -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); + } + +} diff --git a/spring-cloud-stream-samples/multibinder/src/main/resources/application.yml b/spring-cloud-stream-samples/multibinder/src/main/resources/application.yml new file mode 100644 index 000000000..9c3cb2180 --- /dev/null +++ b/spring-cloud-stream-samples/multibinder/src/main/resources/application.yml @@ -0,0 +1,12 @@ +server: + port: 8082 +spring: + cloud: + stream: + bindings: + input: + destination: dataIn + binder: redis + output: + destination: dataOut + binder: rabbit diff --git a/spring-cloud-stream-samples/multibinder/src/test/java/multibinder/RabbitAndRedisBinderApplicationTests.java b/spring-cloud-stream-samples/multibinder/src/test/java/multibinder/RabbitAndRedisBinderApplicationTests.java new file mode 100644 index 000000000..7ebfbe922 --- /dev/null +++ b/spring-cloud-stream-samples/multibinder/src/test/java/multibinder/RabbitAndRedisBinderApplicationTests.java @@ -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 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)); + } + +} diff --git a/spring-cloud-stream-samples/pom.xml b/spring-cloud-stream-samples/pom.xml index d506c9197..53c0f3b66 100644 --- a/spring-cloud-stream-samples/pom.xml +++ b/spring-cloud-stream-samples/pom.xml @@ -25,6 +25,8 @@ tap double extended + multibinder + multibinder-differentsystems rxjava-processor diff --git a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java b/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java index a922dc2a3..b6ef10a14 100644 --- a/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java +++ b/spring-cloud-stream-test-support-internal/src/main/java/org/springframework/cloud/stream/test/junit/kafka/KafkaTestSupport.java @@ -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 binderFactory; @Test @SuppressWarnings("unchecked") public void testWiring() { Message message = new GenericMessage<>("hello"); processor.input().send(message); - Message received = (Message) messageCollector.forChannel(processor.output()).poll(); + Message received = (Message) ((TestSupportBinder) binderFactory.getBinder(null)).messageCollector().forChannel(processor.output()).poll(); assertThat(received.getPayload(), equalTo("hello world")); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java index 2f94e08a0..18e02eda2 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java @@ -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 { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderConfiguration.java new file mode 100644 index 000000000..5cea72874 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderConfiguration.java @@ -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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderFactory.java new file mode 100644 index 000000000..b0bf33804 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderFactory.java @@ -0,0 +1,32 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.stream.binder; + +/** + * @author Marius Bogoevici + */ +public interface BinderFactory { + + /** + * 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 getBinder(String configurationName); +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderType.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderType.java new file mode 100644 index 000000000..2a910239b --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderType.java @@ -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; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderTypeRegistry.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderTypeRegistry.java new file mode 100644 index 000000000..ad0846531 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/BinderTypeRegistry.java @@ -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 getAll(); + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java new file mode 100644 index 000000000..daba7a1c8 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderFactory.java @@ -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 implements BinderFactory, DisposableBean, EnvironmentAware { + + private final Map binderConfigurations; + + private final Map> binderInstanceCache = new HashMap<>(); + + private volatile Environment environment; + + private volatile String defaultBinder; + + public DefaultBinderFactory(Map 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> entry : binderInstanceCache.entrySet()) { + BinderInstanceHolder binderInstanceHolder = entry.getValue(); + binderInstanceHolder.getBinderContext().close(); + } + } + + @Override + public synchronized Binder 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 args = new ArrayList<>(); + for (Map.Entry 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 binder = (Binder) 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 + */ + private static class BinderInstanceHolder { + + private final Binder binderInstance; + + private final ConfigurableApplicationContext binderContext; + + public BinderInstanceHolder(Binder binderInstance, ConfigurableApplicationContext binderContext) { + this.binderInstance = binderInstance; + this.binderContext = binderContext; + } + + public Binder 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 { + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderTypeRegistry.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderTypeRegistry.java new file mode 100644 index 000000000..ad6aac101 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinderTypeRegistry.java @@ -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 binderTypes; + + public DefaultBinderTypeRegistry(Map binderTypes) { + this.binderTypes = Collections.unmodifiableMap(new HashMap<>(binderTypes)); + } + + @Override + public BinderType get(String name) { + return binderTypes.get(name); + } + + @Override + public Map getAll() { + return binderTypes; + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java index 136053612..02c5dfba0 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareChannelResolver.java @@ -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 binder; + private final BinderFactory binderFactory; private final Properties producerProperties; - public BinderAwareChannelResolver(Binder binder, Properties producerProperties) { - this.binder = binder; + public BinderAwareChannelResolver(BinderFactory 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" + + " [:]:"); + } + Binder 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); diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java index b01a70bd9..cf573be64 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BinderAwareRouterBeanPostProcessor.java @@ -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. diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java index ac22d826c..0192df052 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/ChannelBindingService.java @@ -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 binder; + private BinderFactory binderFactory; private final ChannelBindingServiceProperties channelBindingServiceProperties; - public ChannelBindingService(ChannelBindingServiceProperties channelBindingServiceProperties, Binder binder) { + public ChannelBindingService(ChannelBindingServiceProperties channelBindingServiceProperties, + BinderFactory binderFactory) { this.channelBindingServiceProperties = channelBindingServiceProperties; - this.binder = binder; + this.binderFactory = binderFactory; } public void bindConsumer(MessageChannel inputChannel, String inputChannelName) { String channelBindingTarget = this.channelBindingServiceProperties.getBindingDestination(inputChannelName); + Binder 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 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 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 binder = getBinderForChannel(outputChannelName); + binder.unbindProducers(outputChannelName); + } + + private Binder getBinderForChannel(String channelName) { + String transport = this.channelBindingServiceProperties.getBinder(channelName); + return binderFactory.getBinder(transport); } private String consumerGroup(String inputChannelName) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java new file mode 100644 index 000000000..e91b218c5 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryConfiguration.java @@ -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 binderConfigurations = new HashMap<>(); + if (!CollectionUtils.isEmpty(channelBindingServiceProperties.getBinders())) { + for (Map.Entry 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 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 binderTypes = new HashMap<>(); + ClassLoader classLoader = configurableApplicationContext.getClassLoader(); + if (classLoader == null) { + classLoader = ChannelBindingAutoConfiguration.class.getClassLoader(); + } + try { + Enumeration 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 parseBinderConfigurations(ClassLoader classLoader, Resource resource) + throws IOException, ClassNotFoundException { + Properties properties = PropertiesLoaderUtils.loadProperties(resource); + Collection 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; + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java new file mode 100644 index 000000000..0323ca731 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderProperties.java @@ -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; + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java index 23fb216ae..40c3e9e7b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingProperties.java @@ -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; + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java index 3ceeec658..f26d9ed1c 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceConfiguration.java @@ -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 binder) { - return new ChannelBindingService(channelBindingServiceProperties, binder); + BinderFactory binderFactory) { + return new ChannelBindingService(channelBindingServiceProperties, binderFactory); } @Bean @@ -103,8 +103,8 @@ public class ChannelBindingServiceConfiguration { @Bean public BinderAwareChannelResolver binderAwareChannelResolver( - Binder binder) { - return new BinderAwareChannelResolver(binder, new Properties()); + BinderFactory binderFactory) { + return new BinderAwareChannelResolver(binderFactory, new Properties()); } @Bean diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java index a60abd733..2a2a9dfd4 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ChannelBindingServiceProperties.java @@ -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 bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); - private Properties getConsumerProperties() { + private Map binders = new HashMap<>(); + + private String defaultBinder; + + public Properties getConsumerProperties() { return this.consumerProperties; } @@ -72,6 +76,22 @@ public class ChannelBindingServiceProperties { this.bindings = bindings; } + public Map getBinders() { + return binders; + } + + public void setBinders(Map 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(); + } + } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithBindingTargets.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithBindingTargets.java index d86a2c813..906d541cb 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithBindingTargets.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithBindingTargets.java @@ -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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithDefaults.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithDefaults.java index 5b04bece4..120cc3112 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithDefaults.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceBindingTestsWithDefaults.java @@ -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 { } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java index 74c5d3b8b..83963a164 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderAwareChannelResolverTests.java @@ -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() { + @Override + public Binder 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); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java new file mode 100644 index 000000000..73e4fcf57 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java @@ -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 { + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java index 6ed07bc39..69a7135b8 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java @@ -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.any()); @@ -59,7 +60,7 @@ public class InputOutputBindingOrderTest { @EnableBinding(Processor.class) @EnableAutoConfiguration - @Import(MockBinderConfiguration.class) + @Import(MockBinderRegistryConfiguration.class) public static class TestSource { @Bean diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/LifecycleBinderTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/LifecycleBinderTests.java index b61df2f7c..ce0554d35 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/LifecycleBinderTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/LifecycleBinderTests.java @@ -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 diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithBindingTargets.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithBindingTargets.java index 9527a9310..33320c063 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithBindingTargets.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithBindingTargets.java @@ -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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithDefaults.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithDefaults.java index 3a2db5733..c86227daa 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithDefaults.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithDefaults.java @@ -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 { } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithPubSubBindingTargets.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithPubSubBindingTargets.java index a8db8519c..df69433f5 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithPubSubBindingTargets.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingTestsWithPubSubBindingTargets.java @@ -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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingPubSubTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingPubSubTests.java index ac1941e47..a81261cf2 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingPubSubTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingPubSubTests.java @@ -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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithBindingTargets.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithBindingTargets.java index ee660c89b..ef6f0f182 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithBindingTargets.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithBindingTargets.java @@ -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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithDefaults.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithDefaults.java index e3a4c0392..d54c53ba5 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithDefaults.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingTestsWithDefaults.java @@ -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 { } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java index 0fd304ac0..f0fdf048d 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithBindingTargets.java @@ -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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithDefaults.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithDefaults.java index 5a4567d44..b59bbcb5a 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithDefaults.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingTestsWithDefaults.java @@ -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 { } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java new file mode 100644 index 000000000..687ffdebb --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1.java @@ -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; + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1Configuration.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1Configuration.java new file mode 100644 index 000000000..cda49299f --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub1/StubBinder1Configuration.java @@ -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(); + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java new file mode 100644 index 000000000..f20c4e0c8 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2.java @@ -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; + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationA.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationA.java new file mode 100644 index 000000000..5f896ef6f --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationA.java @@ -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); + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationB.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationB.java new file mode 100644 index 000000000..8a8fac197 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2ConfigurationB.java @@ -0,0 +1,32 @@ +/* + * Copyright 2015 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.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(); + } + } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2Dependency.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2Dependency.java new file mode 100644 index 000000000..b9573a36a --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/stub2/StubBinder2Dependency.java @@ -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 { + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java index 73e7756a3..8d8dca487 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/ChannelBindingServiceTests.java @@ -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 bindings = new HashMap<>(); BindingProperties props = new BindingProperties(); @@ -45,17 +50,20 @@ public class ChannelBindingServiceTests { bindings.put(name, props); properties.setBindings(bindings); @SuppressWarnings("unchecked") - Binder binder = mock(Binder.class); - ChannelBindingService service = new ChannelBindingService(properties, binder); + DefaultBinderFactory binderFactory = + new DefaultBinderFactory<>(Collections.singletonMap("mock", new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}), new Properties()))); + Binder 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 bindings = new HashMap<>(); BindingProperties props = new BindingProperties(); @@ -64,13 +72,16 @@ public class ChannelBindingServiceTests { bindings.put(name, props); properties.setBindings(bindings); @SuppressWarnings("unchecked") - Binder binder = mock(Binder.class); - ChannelBindingService service = new ChannelBindingService(properties, binder); + DefaultBinderFactory binderFactory = + new DefaultBinderFactory<>(Collections.singletonMap("mock", new BinderConfiguration(new BinderType("mock", new Class[]{MockBinderConfiguration.class}), new Properties()))); + Binder 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(); } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderConfigurationParsingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderConfigurationParsingTests.java new file mode 100644 index 000000000..91ec8dfb9 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BinderConfigurationParsingTests.java @@ -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 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 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 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))) + )); + + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java index ca3d47971..8342381fc 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java @@ -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) diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java index 9f5b68e80..dd5bda040 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java @@ -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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java index b21d52fa5..09ff396e5 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedProducerTest.java @@ -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 { diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java index ef1eca15e..6abbcb858 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderConfiguration.java @@ -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 diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java new file mode 100644 index 000000000..4b41fed85 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/utils/MockBinderRegistryConfiguration.java @@ -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 binderFactory) { + return binderFactory.getBinder(null); + } +} diff --git a/spring-cloud-stream/src/test/resources/binder1/META-INF/spring.binders b/spring-cloud-stream/src/test/resources/binder1/META-INF/spring.binders new file mode 100644 index 000000000..d3bb4866b --- /dev/null +++ b/spring-cloud-stream/src/test/resources/binder1/META-INF/spring.binders @@ -0,0 +1 @@ +binder1=org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration \ No newline at end of file diff --git a/spring-cloud-stream/src/test/resources/binder2/META-INF/spring.binders b/spring-cloud-stream/src/test/resources/binder2/META-INF/spring.binders new file mode 100644 index 000000000..f5bb5abaa --- /dev/null +++ b/spring-cloud-stream/src/test/resources/binder2/META-INF/spring.binders @@ -0,0 +1,3 @@ +binder2:\ +org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationA,\ +org.springframework.cloud.stream.binder.stub2.StubBinder2ConfigurationB \ No newline at end of file