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