diff --git a/pom.xml b/pom.xml index 494cffc2b..dc30ae0f2 100644 --- a/pom.xml +++ b/pom.xml @@ -32,8 +32,10 @@ spring-cloud-stream spring-cloud-stream-binder-test + spring-cloud-stream-integration-tests docs diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeOutboundSourceTests.java b/spring-cloud-stream-integration-tests/deleted-tests/ContentTypeOutboundSourceTests.java similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/ContentTypeOutboundSourceTests.java rename to spring-cloud-stream-integration-tests/deleted-tests/ContentTypeOutboundSourceTests.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomHeaderPropagationTests.java b/spring-cloud-stream-integration-tests/deleted-tests/CustomHeaderPropagationTests.java similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomHeaderPropagationTests.java rename to spring-cloud-stream-integration-tests/deleted-tests/CustomHeaderPropagationTests.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomMessageConverterTests.java b/spring-cloud-stream-integration-tests/deleted-tests/CustomMessageConverterTests.java similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/CustomMessageConverterTests.java rename to spring-cloud-stream-integration-tests/deleted-tests/CustomMessageConverterTests.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationTests.java b/spring-cloud-stream-integration-tests/deleted-tests/DefaultHeaderPropagationTests.java similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationTests.java rename to spring-cloud-stream-integration-tests/deleted-tests/DefaultHeaderPropagationTests.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationWithApplicationProvidedHeaderTests.java b/spring-cloud-stream-integration-tests/deleted-tests/DefaultHeaderPropagationWithApplicationProvidedHeaderTests.java similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DefaultHeaderPropagationWithApplicationProvidedHeaderTests.java rename to spring-cloud-stream-integration-tests/deleted-tests/DefaultHeaderPropagationWithApplicationProvidedHeaderTests.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DeserializeJSONToJavaTypeTests.java b/spring-cloud-stream-integration-tests/deleted-tests/DeserializeJSONToJavaTypeTests.java similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/DeserializeJSONToJavaTypeTests.java rename to spring-cloud-stream-integration-tests/deleted-tests/DeserializeJSONToJavaTypeTests.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java b/spring-cloud-stream-integration-tests/deleted-tests/MessageChannelConfigurerTests.java similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelConfigurerTests.java rename to spring-cloud-stream-integration-tests/deleted-tests/MessageChannelConfigurerTests.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeDecodingTests.java b/spring-cloud-stream-integration-tests/deleted-tests/MessageChannelWithNativeDecodingTests.java similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeDecodingTests.java rename to spring-cloud-stream-integration-tests/deleted-tests/MessageChannelWithNativeDecodingTests.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeEncodingTests.java b/spring-cloud-stream-integration-tests/deleted-tests/MessageChannelWithNativeEncodingTests.java similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/MessageChannelWithNativeEncodingTests.java rename to spring-cloud-stream-integration-tests/deleted-tests/MessageChannelWithNativeEncodingTests.java diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainConversionTest.java b/spring-cloud-stream-integration-tests/deleted-tests/TextPlainConversionTest.java similarity index 100% rename from spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/TextPlainConversionTest.java rename to spring-cloud-stream-integration-tests/deleted-tests/TextPlainConversionTest.java diff --git a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/disable/AutoconfigurationDisabledTest.java b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/disable/AutoconfigurationDisabledTest.java deleted file mode 100644 index b524ec749..000000000 --- a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/disable/AutoconfigurationDisabledTest.java +++ /dev/null @@ -1,75 +0,0 @@ -/* - * Copyright 2017-2019 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 - * - * https://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.test.disable; - -import java.util.concurrent.TimeUnit; - -import org.junit.jupiter.api.Test; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.cloud.stream.test.binder.MessageCollector; -import org.springframework.cloud.stream.test.binder.TestSupportBinderAutoConfiguration; -import org.springframework.integration.annotation.Transformer; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.messaging.Message; -import org.springframework.test.annotation.DirtiesContext; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * @author Marius Bogoevici - */ -@SpringBootTest(classes = AutoconfigurationDisabledTest.MyProcessor.class, properties = { - "server.port=-1", "spring.cloud.stream.defaultBinder=test", - "--spring.cloud.stream.bindings.input.contentType=text/plain", - "--spring.cloud.stream.bindings.output.contentType=text/plain" }) -@DirtiesContext -public class AutoconfigurationDisabledTest { - - @Autowired - public MessageCollector messageCollector; - - @Autowired - public Processor processor; - - @SuppressWarnings("unchecked") - @Test - public void testAutoconfigurationDisabled() throws Exception { - this.processor.input().send(MessageBuilder.withPayload("Hello").build()); - // Since the interaction is synchronous, the result should be immediate - Message response = (Message) this.messageCollector - .forChannel(this.processor.output()).poll(1000, TimeUnit.MILLISECONDS); - assertThat(response).isNotNull(); - assertThat(response.getPayload()).isEqualTo("Hello world"); - } - - @SpringBootApplication(exclude = TestSupportBinderAutoConfiguration.class) - @EnableBinding(Processor.class) - public static class MyProcessor { - - @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) - public String transform(String in) { - return in + " world"; - } - - } - -} diff --git a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java b/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java deleted file mode 100644 index a95887e28..000000000 --- a/spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java +++ /dev/null @@ -1,72 +0,0 @@ -/* - * Copyright 2015-2019 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 - * - * https://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.test.example; - -import org.junit.jupiter.api.Test; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.cloud.stream.test.binder.MessageCollector; -import org.springframework.integration.annotation.Transformer; -import org.springframework.messaging.Message; -import org.springframework.messaging.support.GenericMessage; -import org.springframework.test.annotation.DirtiesContext; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * Integration test that validates that - * {@link org.springframework.cloud.stream.test.binder.TestSupportBinder} applies - * correctly. - */ -@SpringBootTest(classes = ExampleTest.MyProcessor.class, webEnvironment = SpringBootTest.WebEnvironment.NONE, properties = { - "--spring.cloud.stream.bindings.input.contentType=text/plain", - "--spring.cloud.stream.bindings.output.contentType=text/plain" }) -@DirtiesContext -public class ExampleTest { - - @Autowired - private MessageCollector messageCollector; - - @Autowired - private Processor processor; - - @Test - @SuppressWarnings("unchecked") - public void testWiring() { - Message message = new GenericMessage<>("hello"); - this.processor.input().send(message); - Message received = (Message) this.messageCollector - .forChannel(this.processor.output()).poll(); - assertThat(received.getPayload()).isEqualTo("hello world"); - } - - @SpringBootApplication - @EnableBinding(Processor.class) - public static class MyProcessor { - - @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) - public String transform(String in) { - return in + " world"; - } - - } - -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceWithBindingTargetsTests.java b/spring-cloud-stream/deleted-tests/ArbitraryInterfaceWithBindingTargetsTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceWithBindingTargetsTests.java rename to spring-cloud-stream/deleted-tests/ArbitraryInterfaceWithBindingTargetsTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceWithDefaultsTests.java b/spring-cloud-stream/deleted-tests/ArbitraryInterfaceWithDefaultsTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ArbitraryInterfaceWithDefaultsTests.java rename to spring-cloud-stream/deleted-tests/ArbitraryInterfaceWithDefaultsTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryAutoConfigurationTests.java b/spring-cloud-stream/deleted-tests/BinderFactoryAutoConfigurationTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryAutoConfigurationTests.java rename to spring-cloud-stream/deleted-tests/BinderFactoryAutoConfigurationTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingHandlerAdviseTests.java b/spring-cloud-stream/deleted-tests/BindingHandlerAdviseTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/BindingHandlerAdviseTests.java rename to spring-cloud-stream/deleted-tests/BindingHandlerAdviseTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/spring-cloud-stream/deleted-tests/BindingServiceTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java rename to spring-cloud-stream/deleted-tests/BindingServiceTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java b/spring-cloud-stream/deleted-tests/CustomPartitionedProducerTest.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/CustomPartitionedProducerTest.java rename to spring-cloud-stream/deleted-tests/CustomPartitionedProducerTest.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/FooChannels.java b/spring-cloud-stream/deleted-tests/FooChannels.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/FooChannels.java rename to spring-cloud-stream/deleted-tests/FooChannels.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/HealthIndicatorsConfigurationTests.java b/spring-cloud-stream/deleted-tests/HealthIndicatorsConfigurationTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/HealthIndicatorsConfigurationTests.java rename to spring-cloud-stream/deleted-tests/HealthIndicatorsConfigurationTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java b/spring-cloud-stream/deleted-tests/InputOutputBindingOrderTest.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java rename to spring-cloud-stream/deleted-tests/InputOutputBindingOrderTest.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/InvalidBindingConfigurationTests.java b/spring-cloud-stream/deleted-tests/InvalidBindingConfigurationTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/InvalidBindingConfigurationTests.java rename to spring-cloud-stream/deleted-tests/InvalidBindingConfigurationTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/LifecycleBinderTests.java b/spring-cloud-stream/deleted-tests/LifecycleBinderTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/LifecycleBinderTests.java rename to spring-cloud-stream/deleted-tests/LifecycleBinderTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java b/spring-cloud-stream/deleted-tests/PartitionedConsumerTest.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/partitioning/PartitionedConsumerTest.java rename to spring-cloud-stream/deleted-tests/PartitionedConsumerTest.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingWithBindingTargetsTests.java b/spring-cloud-stream/deleted-tests/ProcessorBindingWithBindingTargetsTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingWithBindingTargetsTests.java rename to spring-cloud-stream/deleted-tests/ProcessorBindingWithBindingTargetsTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingsWithDefaultsTests.java b/spring-cloud-stream/deleted-tests/ProcessorBindingsWithDefaultsTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/ProcessorBindingsWithDefaultsTests.java rename to spring-cloud-stream/deleted-tests/ProcessorBindingsWithDefaultsTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java b/spring-cloud-stream/deleted-tests/RetryTemplateTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/RetryTemplateTests.java rename to spring-cloud-stream/deleted-tests/RetryTemplateTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingWithDefaultTargetsTests.java b/spring-cloud-stream/deleted-tests/SinkBindingWithDefaultTargetsTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingWithDefaultTargetsTests.java rename to spring-cloud-stream/deleted-tests/SinkBindingWithDefaultTargetsTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingWithDefaultsTests.java b/spring-cloud-stream/deleted-tests/SinkBindingWithDefaultsTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SinkBindingWithDefaultsTests.java rename to spring-cloud-stream/deleted-tests/SinkBindingWithDefaultsTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithBindingTargetsTests.java b/spring-cloud-stream/deleted-tests/SourceBindingWithBindingTargetsTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithBindingTargetsTests.java rename to spring-cloud-stream/deleted-tests/SourceBindingWithBindingTargetsTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithDefaultsTests.java b/spring-cloud-stream/deleted-tests/SourceBindingWithDefaultsTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithDefaultsTests.java rename to spring-cloud-stream/deleted-tests/SourceBindingWithDefaultsTests.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesOnlyTest.java b/spring-cloud-stream/deleted-tests/SourceBindingWithGlobalPropertiesOnlyTest.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesOnlyTest.java rename to spring-cloud-stream/deleted-tests/SourceBindingWithGlobalPropertiesOnlyTest.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java b/spring-cloud-stream/deleted-tests/SourceBindingWithGlobalPropertiesTest.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/SourceBindingWithGlobalPropertiesTest.java rename to spring-cloud-stream/deleted-tests/SourceBindingWithGlobalPropertiesTest.java diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/SpelExpressionConverterConfigurationTests.java b/spring-cloud-stream/deleted-tests/SpelExpressionConverterConfigurationTests.java similarity index 100% rename from spring-cloud-stream/src/test/java/org/springframework/cloud/stream/config/SpelExpressionConverterConfigurationTests.java rename to spring-cloud-stream/deleted-tests/SpelExpressionConverterConfigurationTests.java 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 deleted file mode 100644 index d69eac548..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/EnableBinding.java +++ /dev/null @@ -1,60 +0,0 @@ -/* - * Copyright 2015-2019 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 - * - * https://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.annotation; - -import java.lang.annotation.Documented; -import java.lang.annotation.ElementType; -import java.lang.annotation.Inherited; -import java.lang.annotation.Retention; -import java.lang.annotation.RetentionPolicy; -import java.lang.annotation.Target; - -import org.springframework.cloud.stream.config.BinderFactoryAutoConfiguration; -import org.springframework.cloud.stream.config.BindingBeansRegistrar; -import org.springframework.context.annotation.Configuration; -import org.springframework.context.annotation.Import; -import org.springframework.integration.config.EnableIntegration; - -/** - * Enables the binding of targets annotated with {@link Input} and {@link Output} to a - * broker, according to the list of interfaces passed as value to the annotation. - * - * @author Dave Syer - * @author Marius Bogoevici - * @author David Turanski - * @author Soby Chacko - * - * @deprecated as of 3.1 in favor of functional programming model - */ -@Target({ ElementType.TYPE, ElementType.ANNOTATION_TYPE }) -@Retention(RetentionPolicy.RUNTIME) -@Documented -@Inherited -@Configuration(proxyBeanMethods = false) -@Import({ BindingBeansRegistrar.class, BinderFactoryAutoConfiguration.class }) -@EnableIntegration -@Deprecated -public @interface EnableBinding { - - /** - * A list of interfaces having methods annotated with {@link Input} and/or - * {@link Output} to indicate binding targets. - * @return list of interfaces - */ - Class[] value() default {}; - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java deleted file mode 100644 index aca079b30..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java +++ /dev/null @@ -1,53 +0,0 @@ -/* - * Copyright 2015-2017 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 - * - * https://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.annotation; - -import java.lang.annotation.Documented; -import java.lang.annotation.ElementType; -import java.lang.annotation.Inherited; -import java.lang.annotation.Retention; -import java.lang.annotation.RetentionPolicy; -import java.lang.annotation.Target; - -import org.springframework.beans.factory.annotation.Qualifier; - -/** - * Indicates that an input binding target will be created by the framework. - * - * @author Dave Syer - * @author Marius Bogoevici - * - * @deprecated as of 3.1 in favor of functional programming model - */ - -@Qualifier -@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE, - ElementType.PARAMETER }) -@Retention(RetentionPolicy.RUNTIME) -@Inherited -@Documented -@Deprecated -public @interface Input { - - /** - * Specify the binding target name; used as a bean name for binding target and as a - * destination name by default. - * @return the binding target name - */ - String value() default ""; - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java deleted file mode 100644 index c376ad3ac..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * Copyright 2015-2017 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 - * - * https://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.annotation; - -import java.lang.annotation.Documented; -import java.lang.annotation.ElementType; -import java.lang.annotation.Inherited; -import java.lang.annotation.Retention; -import java.lang.annotation.RetentionPolicy; -import java.lang.annotation.Target; - -import org.springframework.beans.factory.annotation.Qualifier; - -/** - * Indicates that an output binding target will be created by the framework. - * - * @author Dave Syer - * @author Marius Bogoevici - * @author Artem Bilan - * - * @deprecated as of 3.1 in favor of functional programming model - */ - -@Qualifier -@Target({ ElementType.FIELD, ElementType.METHOD, ElementType.ANNOTATION_TYPE, - ElementType.PARAMETER }) -@Retention(RetentionPolicy.RUNTIME) -@Inherited -@Documented -@Deprecated -public @interface Output { - - /** - * Specify the binding target name; used as a bean name for binding target and as a - * destination name by default. - * @return the binding target name - */ - String value() default ""; - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java index 5804bde71..0610d9b77 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java @@ -31,14 +31,9 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.annotation.Output; -import org.springframework.core.annotation.AnnotationUtils; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.util.Assert; -import org.springframework.util.ReflectionUtils; /** * {@link FactoryBean} for instantiating the interfaces specified via @@ -77,24 +72,24 @@ public class BindableProxyFactory extends AbstractBindableProxyFactory return boundTarget; } - Input input = AnnotationUtils.findAnnotation(method, Input.class); - if (input != null) { - String name = BindingBeanDefinitionRegistryUtils.getBindingTargetName(input, - method); - boundTarget = this.inputHolders.get(name).getBoundTarget(); - this.targetCache.put(method, boundTarget); - return boundTarget; - } - else { - Output output = AnnotationUtils.findAnnotation(method, Output.class); - if (output != null) { - String name = BindingBeanDefinitionRegistryUtils - .getBindingTargetName(output, method); - boundTarget = this.outputHolders.get(name).getBoundTarget(); - this.targetCache.put(method, boundTarget); - return boundTarget; - } - } +// Input input = AnnotationUtils.findAnnotation(method, Input.class); +// if (input != null) { +// String name = BindingBeanDefinitionRegistryUtils.getBindingTargetName(input, +// method); +// boundTarget = this.inputHolders.get(name).getBoundTarget(); +// this.targetCache.put(method, boundTarget); +// return boundTarget; +// } +// else { +// Output output = AnnotationUtils.findAnnotation(method, Output.class); +// if (output != null) { +// String name = BindingBeanDefinitionRegistryUtils +// .getBindingTargetName(output, method); +// boundTarget = this.outputHolders.get(name).getBoundTarget(); +// this.targetCache.put(method, boundTarget); +// return boundTarget; +// } +// } return null; } @@ -121,30 +116,30 @@ public class BindableProxyFactory extends AbstractBindableProxyFactory populateBindingTargetFactories(this.beanFactory); Assert.notEmpty(BindableProxyFactory.this.bindingTargetFactories, "'bindingTargetFactories' cannot be empty"); - ReflectionUtils.doWithMethods(this.type, method -> { - Input input = AnnotationUtils.findAnnotation(method, Input.class); - if (input != null) { - String name = BindingBeanDefinitionRegistryUtils - .getBindingTargetName(input, method); - Class returnType = method.getReturnType(); - - BindableProxyFactory.this.inputHolders.put(name, - new BoundTargetHolder(getBindingTargetFactory(returnType) - .createInput(name), true)); - } - }); - ReflectionUtils.doWithMethods(this.type, method -> { - Output output = AnnotationUtils.findAnnotation(method, Output.class); - if (output != null) { - String name = BindingBeanDefinitionRegistryUtils - .getBindingTargetName(output, method); - Class returnType = method.getReturnType(); - - BindableProxyFactory.this.outputHolders.put(name, - new BoundTargetHolder(getBindingTargetFactory(returnType) - .createOutput(name), true)); - } - }); +// ReflectionUtils.doWithMethods(this.type, method -> { +// Input input = AnnotationUtils.findAnnotation(method, Input.class); +// if (input != null) { +// String name = BindingBeanDefinitionRegistryUtils +// .getBindingTargetName(input, method); +// Class returnType = method.getReturnType(); +// +// BindableProxyFactory.this.inputHolders.put(name, +// new BoundTargetHolder(getBindingTargetFactory(returnType) +// .createInput(name), true)); +// } +// }); +// ReflectionUtils.doWithMethods(this.type, method -> { +// Output output = AnnotationUtils.findAnnotation(method, Output.class); +// if (output != null) { +// String name = BindingBeanDefinitionRegistryUtils +// .getBindingTargetName(output, method); +// Class returnType = method.getReturnType(); +// +// BindableProxyFactory.this.outputHolders.put(name, +// new BoundTargetHolder(getBindingTargetFactory(returnType) +// .createOutput(name), true)); +// } +// }); } @Override diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java index 184e09bda..9fb8b92f3 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingBeanDefinitionRegistryUtils.java @@ -24,10 +24,7 @@ import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.beans.factory.support.AutowireCandidateQualifier; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.annotation.Output; import org.springframework.core.annotation.AnnotationUtils; -import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; /** @@ -42,17 +39,19 @@ public abstract class BindingBeanDefinitionRegistryUtils { public static void registerInputBindingTargetBeanDefinition(String qualifierValue, String name, String bindingTargetInterfaceBeanName, String bindingTargetInterfaceMethodName, BeanDefinitionRegistry registry) { - registerBindingTargetBeanDefinition(Input.class, qualifierValue, name, - bindingTargetInterfaceBeanName, bindingTargetInterfaceMethodName, - registry); + System.out.println(); +// registerBindingTargetBeanDefinition(Input.class, qualifierValue, name, +// bindingTargetInterfaceBeanName, bindingTargetInterfaceMethodName, +// registry); } public static void registerOutputBindingTargetBeanDefinition(String qualifierValue, String name, String bindingTargetInterfaceBeanName, String bindingTargetInterfaceMethodName, BeanDefinitionRegistry registry) { - registerBindingTargetBeanDefinition(Output.class, qualifierValue, name, - bindingTargetInterfaceBeanName, bindingTargetInterfaceMethodName, - registry); + System.out.println(); +// registerBindingTargetBeanDefinition(Output.class, qualifierValue, name, +// bindingTargetInterfaceBeanName, bindingTargetInterfaceMethodName, +// registry); } private static void registerBindingTargetBeanDefinition( @@ -77,24 +76,24 @@ public abstract class BindingBeanDefinitionRegistryUtils { public static void registerBindingTargetBeanDefinitions(Class type, final String bindingTargetInterfaceBeanName, final BeanDefinitionRegistry registry) { - ReflectionUtils.doWithMethods(type, method -> { - Input input = AnnotationUtils.findAnnotation(method, Input.class); - if (input != null) { - String name = getBindingTargetName(input, method); - if (!registry.containsBeanDefinition(name)) { - registerInputBindingTargetBeanDefinition(input.value(), name, - bindingTargetInterfaceBeanName, method.getName(), registry); - } - } - Output output = AnnotationUtils.findAnnotation(method, Output.class); - if (output != null) { - String name = getBindingTargetName(output, method); - if (!registry.containsBeanDefinition(name)) { - registerOutputBindingTargetBeanDefinition(output.value(), name, - bindingTargetInterfaceBeanName, method.getName(), registry); - } - } - }); +// ReflectionUtils.doWithMethods(type, method -> { +// Input input = AnnotationUtils.findAnnotation(method, Input.class); +// if (input != null) { +// String name = getBindingTargetName(input, method); +// if (!registry.containsBeanDefinition(name)) { +// registerInputBindingTargetBeanDefinition(input.value(), name, +// bindingTargetInterfaceBeanName, method.getName(), registry); +// } +// } +// Output output = AnnotationUtils.findAnnotation(method, Output.class); +// if (output != null) { +// String name = getBindingTargetName(output, method); +// if (!registry.containsBeanDefinition(name)) { +// registerOutputBindingTargetBeanDefinition(output.value(), name, +// bindingTargetInterfaceBeanName, method.getName(), registry); +// } +// } +// }); } public static void registerBindingTargetsQualifiedBeanDefinitions(Class parent, diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamAnnotationCommonMethodUtils.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamAnnotationCommonMethodUtils.java index ae63e728f..8c10967aa 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamAnnotationCommonMethodUtils.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamAnnotationCommonMethodUtils.java @@ -18,13 +18,11 @@ package org.springframework.cloud.stream.binding; import java.lang.reflect.Method; -import org.springframework.cloud.stream.annotation.Output; import org.springframework.core.MethodParameter; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.messaging.handler.annotation.SendTo; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; -import org.springframework.util.StringUtils; /** * Common methods that can be used across various Stream annotations. @@ -45,12 +43,12 @@ public abstract class StreamAnnotationCommonMethodUtils { StreamAnnotationErrorMessages.SEND_TO_EMPTY_DESTINATION); return sendTo.value()[0]; } - Output output = AnnotationUtils.findAnnotation(method, Output.class); - if (output != null) { - Assert.isTrue(StringUtils.hasText(output.value()), - StreamAnnotationErrorMessages.ATLEAST_ONE_OUTPUT); - return output.value(); - } +// Output output = AnnotationUtils.findAnnotation(method, Output.class); +// if (output != null) { +// Assert.isTrue(StringUtils.hasText(output.value()), +// StreamAnnotationErrorMessages.ATLEAST_ONE_OUTPUT); +// return output.value(); +// } return null; } @@ -60,9 +58,9 @@ public abstract class StreamAnnotationCommonMethodUtils { .getParameterTypes().length; parameterIndex++) { MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); - if (methodParameter.hasParameterAnnotation(Output.class)) { - outputAnnotationCount++; - } +// if (methodParameter.hasParameterAnnotation(Output.class)) { +// outputAnnotationCount++; +// } } return outputAnnotationCount; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java index 8d54ea089..f95910569 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingBeansRegistrar.java @@ -20,14 +20,8 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.support.BeanDefinitionRegistry; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils; import org.springframework.context.annotation.ImportBeanDefinitionRegistrar; -import org.springframework.core.annotation.AnnotatedElementUtils; -import org.springframework.core.annotation.AnnotationAttributes; -import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.type.AnnotationMetadata; -import org.springframework.util.ClassUtils; /** * @author Marius Bogoevici @@ -43,32 +37,32 @@ public class BindingBeansRegistrar implements ImportBeanDefinitionRegistrar { @Override public void registerBeanDefinitions(AnnotationMetadata metadata, BeanDefinitionRegistry registry) { - AnnotationAttributes attrs = AnnotatedElementUtils.getMergedAnnotationAttributes( - ClassUtils.resolveClassName(metadata.getClassName(), null), - EnableBinding.class); - try { - for (Class type : collectClasses(attrs, metadata.getClassName())) { - if (!registry.containsBeanDefinition(type.getName())) { - BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions( - type, type.getName(), registry); - BindingBeanDefinitionRegistryUtils - .registerBindingTargetsQualifiedBeanDefinitions(ClassUtils - .resolveClassName(metadata.getClassName(), null), type, - registry); - } - } - } - catch (Throwable e) { - logger.warn("Failed to proxy EnableBinding annotation. If you are using functional programming style, " - + "ignore this warning, otherwise, annotation-based programming model is not and will " - + "not be suppported in native images."); - // happens in native images, but we do not intend supporting annotation-based model for much longer - } +// AnnotationAttributes attrs = AnnotatedElementUtils.getMergedAnnotationAttributes( +// ClassUtils.resolveClassName(metadata.getClassName(), null), +// EnableBinding.class); +// try { +// for (Class type : collectClasses(attrs, metadata.getClassName())) { +// if (!registry.containsBeanDefinition(type.getName())) { +// BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions( +// type, type.getName(), registry); +// BindingBeanDefinitionRegistryUtils +// .registerBindingTargetsQualifiedBeanDefinitions(ClassUtils +// .resolveClassName(metadata.getClassName(), null), type, +// registry); +// } +// } +// } +// catch (Throwable e) { +// logger.warn("Failed to proxy EnableBinding annotation. If you are using functional programming style, " +// + "ignore this warning, otherwise, annotation-based programming model is not and will " +// + "not be suppported in native images."); +// // happens in native images, but we do not intend supporting annotation-based model for much longer +// } } - private Class[] collectClasses(AnnotationAttributes attrs, String className) { - EnableBinding enableBinding = AnnotationUtils.synthesizeAnnotation(attrs, - EnableBinding.class, ClassUtils.resolveClassName(className, null)); - return enableBinding.value(); - } +// private Class[] collectClasses(AnnotationAttributes attrs, String className) { +// EnableBinding enableBinding = AnnotationUtils.synthesizeAnnotation(attrs, +// EnableBinding.class, ClassUtils.resolveClassName(className, null)); +// return enableBinding.value(); +// } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index ae87ef2f1..0985b6faa 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -64,7 +64,6 @@ import org.springframework.cloud.function.context.config.ContextFunctionCatalogA import org.springframework.cloud.function.context.config.FunctionContextUtils; import org.springframework.cloud.function.context.config.RoutingFunction; import org.springframework.cloud.function.context.message.MessageUtils; -import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.BinderFactory; import org.springframework.cloud.stream.binder.BindingCreatedEvent; import org.springframework.cloud.stream.binder.ConsumerProperties; @@ -152,8 +151,9 @@ public class FunctionConfiguration { BindingServiceProperties serviceProperties, ConfigurableApplicationContext applicationContext, StreamBridge streamBridge) { - boolean shouldCreateInitializer = applicationContext.containsBean("output") - || ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class)); + boolean shouldCreateInitializer = true; +// boolean shouldCreateInitializer = applicationContext.containsBean("output"); +// || ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class)); return shouldCreateInitializer ? new FunctionToDestinationBinder(functionCatalog, functionProperties, @@ -170,7 +170,7 @@ public class FunctionConfiguration { @Nullable List proxyFactories, StreamBridge streamBridge, TaskScheduler taskScheduler) { - if (!ObjectUtils.isEmpty(context.getBeanNamesForAnnotation(EnableBinding.class)) || CollectionUtils.isEmpty(proxyFactories)) { + if (CollectionUtils.isEmpty(proxyFactories)) { return null; } @@ -814,46 +814,46 @@ public class FunctionConfiguration { @Override public void afterPropertiesSet() throws Exception { - if (ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class))) { - this.determineFunctionName(functionCatalog, environment); - BeanDefinitionRegistry registry = (BeanDefinitionRegistry) applicationContext.getBeanFactory(); +// if (ObjectUtils.isEmpty(applicationContext.getBeanNamesForAnnotation(EnableBinding.class))) { + this.determineFunctionName(functionCatalog, environment); + BeanDefinitionRegistry registry = (BeanDefinitionRegistry) applicationContext.getBeanFactory(); - if (StringUtils.hasText(streamFunctionProperties.getDefinition())) { - String[] functionDefinitions = this.filterEligibleFunctionDefinitions(); - for (String functionDefinition : functionDefinitions) { - RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); - FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition); - if (function != null) { - Type functionType = function.getFunctionType(); - if (function.isSupplier()) { - this.inputCount = 0; - this.outputCount = this.getOutputCount(functionType, true); - } - else if (function.isConsumer() || functionDefinition.equals(RoutingFunction.FUNCTION_NAME)) { - this.inputCount = FunctionTypeUtils.getInputCount(functionType); - this.outputCount = 0; - } - else { - this.inputCount = FunctionTypeUtils.getInputCount(functionType); - this.outputCount = this.getOutputCount(functionType, false); - } - - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(functionDefinition); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount); - functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties); - registry.registerBeanDefinition(functionDefinition + "_binding", functionBindableProxyDefinition); + if (StringUtils.hasText(streamFunctionProperties.getDefinition())) { + String[] functionDefinitions = this.filterEligibleFunctionDefinitions(); + for (String functionDefinition : functionDefinitions) { + RootBeanDefinition functionBindableProxyDefinition = new RootBeanDefinition(BindableFunctionProxyFactory.class); + FunctionInvocationWrapper function = functionCatalog.lookup(functionDefinition); + if (function != null) { + Type functionType = function.getFunctionType(); + if (function.isSupplier()) { + this.inputCount = 0; + this.outputCount = this.getOutputCount(functionType, true); + } + else if (function.isConsumer() || functionDefinition.equals(RoutingFunction.FUNCTION_NAME)) { + this.inputCount = FunctionTypeUtils.getInputCount(functionType); + this.outputCount = 0; } else { - logger.warn("The function definition '" + streamFunctionProperties.getDefinition() + - "' is not valid. The referenced function bean or one of its components does not exist"); + this.inputCount = FunctionTypeUtils.getInputCount(functionType); + this.outputCount = this.getOutputCount(functionType, false); } + + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(functionDefinition); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.inputCount); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.outputCount); + functionBindableProxyDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.streamFunctionProperties); + registry.registerBeanDefinition(functionDefinition + "_binding", functionBindableProxyDefinition); + } + else { + logger.warn("The function definition '" + streamFunctionProperties.getDefinition() + + "' is not valid. The referenced function bean or one of its components does not exist"); } } - - this.createStandAloneBindingsIfNecessary(registry, applicationContext.getBean(BindingServiceProperties.class)); - } + + this.createStandAloneBindingsIfNecessary(registry, applicationContext.getBean(BindingServiceProperties.class)); + } +// } else { logger.info("Functional binding is disabled due to the presense of @EnableBinding annotation in your configuration"); } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Processor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Processor.java deleted file mode 100644 index a13266f2a..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Processor.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * Copyright 2015-2019 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 - * - * https://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.messaging; - -/** - * Bindable interface with one input and one output channel. - * - * @author Dave Syer - * @author Marius Bogoevici - * @see org.springframework.cloud.stream.annotation.EnableBinding - */ -public interface Processor extends Source, Sink { - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Sink.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Sink.java deleted file mode 100644 index f871473d4..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Sink.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2015-2019 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 - * - * https://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.messaging; - -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.messaging.SubscribableChannel; - -/** - * Bindable interface with one input channel. - * - * @author Dave Syer - * @author Marius Bogoevici - * @see org.springframework.cloud.stream.annotation.EnableBinding - */ -public interface Sink { - - /** - * Input channel name. - */ - String INPUT = "input"; - - /** - * @return input channel. - */ - @Input(Sink.INPUT) - SubscribableChannel input(); - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Source.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Source.java deleted file mode 100644 index 8e60775e7..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Source.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright 2015-2019 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 - * - * https://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.messaging; - -import org.springframework.cloud.stream.annotation.Output; -import org.springframework.messaging.MessageChannel; - -/** - * Bindable interface with one output channel. - * - * @author Dave Syer - * @author Marius Bogoevici - * @see org.springframework.cloud.stream.annotation.EnableBinding - */ -public interface Source { - - /** - * Name of the output channel. - */ - String OUTPUT = "output"; - - /** - * @return output channel - */ - @Output(Source.OUTPUT) - MessageChannel output(); - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetersPublisherBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetersPublisherBinding.java index 45d968189..ca65b6293 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetersPublisherBinding.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetersPublisherBinding.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.micrometer; -import org.springframework.cloud.stream.annotation.Output; import org.springframework.messaging.MessageChannel; /** @@ -33,7 +32,7 @@ public interface MetersPublisherBinding { /** * @return Channel for application metrics. */ - @Output(APPLICATION_METRICS) +// @Output(APPLICATION_METRICS) MessageChannel applicationMetrics(); } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderConfiguration.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderConfiguration.java index b4feed23f..d13d61a3c 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderConfiguration.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderConfiguration.java @@ -19,11 +19,9 @@ package org.springframework.cloud.stream.binder.test; import java.util.ArrayList; import java.util.Arrays; import java.util.List; -import java.util.Map; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.test.context.runner.ApplicationContextRunner; -import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.ProducerProperties; @@ -32,7 +30,6 @@ import org.springframework.cloud.stream.config.BindingServiceConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; -import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.config.EnableIntegration; /** @@ -67,12 +64,6 @@ public class TestChannelBinderConfiguration { Class... additionalConfigurationClasses) { List> configClasses = new ArrayList<>(); configClasses.add(TestChannelBinderConfiguration.class); - Import annotation = AnnotationUtils.getAnnotation(EnableBinding.class, - Import.class); - Map annotationAttributes = AnnotationUtils - .getAnnotationAttributes(annotation); - configClasses - .addAll(Arrays.asList((Class[]) annotationAttributes.get("value"))); configClasses.add(BindingServiceConfiguration.class); if (additionalConfigurationClasses != null) { configClasses.addAll(Arrays.asList(additionalConfigurationClasses)); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java index 94e6f50dd..581e65814 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java @@ -16,42 +16,25 @@ package org.springframework.cloud.stream.function; -import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.Date; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; -import com.fasterxml.jackson.databind.ObjectMapper; -import org.junit.Ignore; import org.junit.Test; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.boot.test.web.client.TestRestTemplate; -import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.test.InputDestination; import org.springframework.cloud.stream.binder.test.OutputDestination; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; -import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; -import org.springframework.cloud.stream.messaging.Source; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; -import org.springframework.http.HttpMethod; -import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.dsl.IntegrationFlow; -import org.springframework.integration.dsl.IntegrationFlows; -import org.springframework.integration.http.dsl.Http; -import org.springframework.integration.http.dsl.HttpRequestHandlerEndpointSpec; -import org.springframework.integration.http.inbound.HttpRequestHandlingEndpointSupport; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; -import org.springframework.messaging.support.MessageBuilder; import static org.assertj.core.api.Assertions.assertThat; @@ -64,30 +47,6 @@ import static org.assertj.core.api.Assertions.assertThat; */ public class GreenfieldFunctionEnableBindingTests { -// @Test - public void testSourceFromSupplier() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration - .getCompleteConfiguration(SourceFromSupplier.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.function.definition=date", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - Message sourceMessage = target.receive(10000); - Date date = (Date) new CompositeMessageConverterFactory() - .getMessageConverterForAllRegistered() - .fromMessage(sourceMessage, Date.class); - assertThat(date).isEqualTo(new Date(12345L)); - - sourceMessage = target.receive(10000); - date = (Date) new CompositeMessageConverterFactory() - .getMessageConverterForAllRegistered() - .fromMessage(sourceMessage, Date.class); - assertThat(date).isEqualTo(new Date(12345L)); - } - } - @Test public void testProcessorFromFunction() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( @@ -120,49 +79,6 @@ public class GreenfieldFunctionEnableBindingTests { } } - @Test - @Ignore - public void testHttpEndpoint() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration( - HttpInboundEndpoint.class)).web(WebApplicationType.SERVLET).run( - "--spring.cloud.function.definition=upperCase", - "--spring.jmx.enabled=false", "--server.port=0")) { - TestRestTemplate restTemplate = new TestRestTemplate(); - restTemplate.postForLocation( - "http://localhost:" - + context.getEnvironment().getProperty("local.server.port"), - "hello"); - OutputDestination target = context.getBean(OutputDestination.class); - String result = new String(target.receive(10000).getPayload()); - System.out.println(result); - assertThat(result).isEqualTo("HELLO"); - } - } - - @Test - @Ignore - public void testPojoReturn() throws IOException { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration( - FooTransform.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.function.definition=fooFunction", - "--spring.jmx" + ".enabled=false", - "--logging.level.org.springframework.integration=TRACE")) { - MessageChannel input = context.getBean("input", MessageChannel.class); - OutputDestination target = context.getBean(OutputDestination.class); - - ObjectMapper mapper = context.getBean(ObjectMapper.class); - - input.send(MessageBuilder.withPayload("bar").build()); - byte[] payload = target.receive(2000).getPayload(); - - Foo result = mapper.readValue(payload, Foo.class); - - assertThat(result.getBar()).isEqualTo("bar"); - } - } - @EnableAutoConfiguration public static class SourceFromSupplier { @@ -201,53 +117,6 @@ public class GreenfieldFunctionEnableBindingTests { } - @EnableAutoConfiguration - @EnableBinding(Source.class) - public static class HttpInboundEndpoint { - - @Bean - public Function upperCase() { - return String::toUpperCase; - } - - @Bean - public HttpRequestHandlingEndpointSupport doFoo(Source source) { - HttpRequestHandlerEndpointSpec httpRequestHandler = Http - .inboundChannelAdapter("/*") - .requestMapping(requestMapping -> requestMapping - .methods(HttpMethod.POST).consumes("*/*")) - .requestChannel(source.output()); - return httpRequestHandler.get(); - } - - } - - @EnableAutoConfiguration - @EnableBinding(Source.class) - public static class FooTransform { - - @Bean - public MessageChannel input() { - return new DirectChannel(); - } - - @Bean - public IntegrationFlow flow() { - - return IntegrationFlows.from(input()).bridge().channel(Source.OUTPUT).get(); - } - - @Bean - public Function, Message> fooFunction() { - return m -> { - Foo foo = new Foo(); - foo.setBar(m.getPayload().toString()); - return MessageBuilder.withPayload(foo).setHeader("foo", "foo").build(); - }; - } - - } - static class Foo { String bar; diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java deleted file mode 100644 index f3cff5517..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java +++ /dev/null @@ -1,186 +0,0 @@ -/* - * Copyright 2019-2020 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 - * - * https://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.function; - -import java.nio.charset.StandardCharsets; -import java.util.List; -import java.util.function.Consumer; -import java.util.function.Function; - -import org.junit.After; -import org.junit.Test; - -import org.springframework.beans.DirectFieldAccessor; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.WebApplicationType; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.binder.test.InputDestination; -import org.springframework.cloud.stream.binder.test.OutputDestination; -import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; -import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Import; -import org.springframework.integration.dsl.IntegrationFlow; -import org.springframework.integration.dsl.IntegrationFlows; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.support.GenericMessage; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * @author Oleg Zhurakousky - * @author David Turanski - * @since 2.1 - */ -public class ProcessorToFunctionsSupportTests { - - private ConfigurableApplicationContext context; - - @After - public void cleanUp() { - this.context.close(); - } - - @Test - public void testPathThrough() { - this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration - .getCompleteConfiguration(FunctionsConfiguration.class)) - .web(WebApplicationType.NONE).run("--spring.jmx.enabled=false"); - InputDestination source = this.context.getBean(InputDestination.class); - OutputDestination target = this.context.getBean(OutputDestination.class); - source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); - assertThat(target.receive(1000).getPayload()) - .isEqualTo("hello".getBytes(StandardCharsets.UTF_8)); - } - - @Test - public void testSingleFunction() { - this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration - .getCompleteConfiguration(FunctionsConfiguration.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=toUpperCase", - "--spring.jmx.enabled=false"); - - InputDestination source = this.context.getBean(InputDestination.class); - OutputDestination target = this.context.getBean(OutputDestination.class); - source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); - assertThat(target.receive(1000).getPayload()) - .isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8)); - //to ensure there is no possibility of load balancing to the EnableBinding - source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); - assertThat(target.receive(1000).getPayload()) - .isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8)); - } - - @Test - public void testComposedFunction() { - this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration - .getCompleteConfiguration(FunctionsConfiguration.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=toUpperCase|concatWithSelf", - "--spring.jmx" + ".enabled=false", - "--logging.level.org.springframework.integration=DEBUG"); - - InputDestination source = this.context.getBean(InputDestination.class); - OutputDestination target = this.context.getBean(OutputDestination.class); - source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); - String result = new String(target.receive(1000).getPayload()); - assertThat(result).isEqualTo("HELLO:HELLO"); - } - - @Test - public void testConsumer() { - this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration - .getCompleteConfiguration(ConsumerConfiguration.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.stream.function.definition=log", - "--spring.jmx.enabled=false"); - - InputDestination source = this.context.getBean(InputDestination.class); - OutputDestination target = this.context.getBean(OutputDestination.class); - source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); - source.send( - new GenericMessage("hello1".getBytes(StandardCharsets.UTF_8))); - source.send( - new GenericMessage("hello2".getBytes(StandardCharsets.UTF_8))); - assertThat(target.receive(1000).getPayload()) - .isEqualTo("hello".getBytes(StandardCharsets.UTF_8)); - assertThat(target.receive(1000).getPayload()) - .isEqualTo("hello1".getBytes(StandardCharsets.UTF_8)); - assertThat(target.receive(1000).getPayload()) - .isEqualTo("hello2".getBytes(StandardCharsets.UTF_8)); - } - - @EnableAutoConfiguration - @Import(BaseProcessorConfiguration.class) - public static class FunctionsConfiguration { - - @Bean - public Function toUpperCase() { - return v -> { - System.out.println(); - return v.toUpperCase(); - }; - } - - @Bean - public Function concatWithSelf() { - return x -> x + ":" + x; - } - - } - - @EnableAutoConfiguration - @Import(BaseProcessorConfiguration.class) - public static class ConsumerConfiguration { - - @SuppressWarnings("unchecked") - @Bean - public Consumer log(OutputDestination out) { - return x -> { - DirectFieldAccessor dfa = new DirectFieldAccessor(out); - MessageChannel channel = ((List) dfa.getPropertyValue("channels")).get(0); - channel.send(new GenericMessage(x.getBytes())); - }; - } - - } - - /** - * This configuration essentially emulates our existing app-starters for Processor and - * essentially demonstrates how a function(s) could be applied to an existing - * processor app via {@link IntegrationFlowFunctionSupport} class. - */ - @EnableBinding(Processor.class) - public static class BaseProcessorConfiguration { - - @Autowired - private Processor processor; - - @Bean - public IntegrationFlow fromChannel() { - - return IntegrationFlows.from(this.processor.input()) - .channel(this.processor.output()).get(); - } - - } - -} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java deleted file mode 100644 index e507f99b4..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java +++ /dev/null @@ -1,391 +0,0 @@ -/* - * Copyright 2018-2019 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 - * - * https://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.function; - -import java.nio.charset.StandardCharsets; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.function.Function; -import java.util.function.Supplier; - -import org.junit.Ignore; -import org.junit.Rule; -import org.junit.Test; -import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeEach; -import org.junit.rules.ExpectedException; -import reactor.core.publisher.Flux; - -import org.springframework.boot.WebApplicationType; -import org.springframework.boot.autoconfigure.EnableAutoConfiguration; -import org.springframework.boot.builder.SpringApplicationBuilder; -import org.springframework.cloud.function.context.PollableBean; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.binder.test.OutputDestination; -import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; -import org.springframework.cloud.stream.binding.BindingsLifecycleController; -import org.springframework.cloud.stream.messaging.Source; -import org.springframework.context.ConfigurableApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.integration.dsl.IntegrationFlow; -import org.springframework.integration.dsl.IntegrationFlows; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.support.GenericMessage; -import org.springframework.util.MimeTypeUtils; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * @author Oleg Zhurakousky - * @author David Turanski - * @since 2.1 - */ -public class SourceToFunctionsSupportTests { - - @Rule - public ExpectedException expectedException = ExpectedException.none(); - - @BeforeEach - public void before() { - System.clearProperty("spring.cloud.stream.function.definition"); - } - @AfterEach - public void after() { - System.clearProperty("spring.cloud.stream.function.definition"); - } - - @Test - @Ignore - public void testFunctionIsAppliedToExistingMessageSource() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration( - FunctionsConfiguration.class, ExistingMessageSourceConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=|toUpperCase", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - assertThat(target.receive(5000).getPayload()) - .isEqualTo("HELLO FUNCTION".getBytes(StandardCharsets.UTF_8)); - } - } - - //@Test - public void testFunctionsAreAppliedToExistingMessageSource() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration( - FunctionsConfiguration.class, ExistingMessageSourceConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=|toUpperCase|concatWithSelf", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - assertThat(target.receive(1000).getPayload()) - .isEqualTo("HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8)); - } - } - - @Test - @Ignore // fails intermittently - public void testFunctionsAreAppliedToExistingMessageSourceReactive() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration( - FunctionsConfiguration.class, ExistingMessageSourceConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=|toUpperCaseReactive|concatWithSelf", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - assertThat(target.receive(1000).getPayload()) - .isEqualTo("HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8)); - } - } - - @Test - @Ignore - public void testImperativeSupplier() throws Exception { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration( - FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=number", - "--spring.jmx.enabled=false")) { - BindingsLifecycleController lifecycle = context .getBean(BindingsLifecycleController.class); - OutputDestination target = context.getBean(OutputDestination.class); - String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("1"); - result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("2"); - - lifecycle.stop("number-out-0"); - for (int i = 0; i < 2; i++) { //drain - target.receive(1000); - } - - Thread.sleep(2000); - assertThat(target.receive(1000)).isNull(); - } - } - - @Test - public void testImperativeSupplierComposedWithFunctions() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration( - FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=number|toUpperCase|concatWithSelf", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("1:1"); - } - } - - @Test - public void testImperativeSupplierComposedWithMixedFunctions() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration( - FunctionsConfiguration.class, SupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=number|toUpperCaseReactive|concatWithSelf", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("1:1"); - } - } - - @Test - public void testReactiveSupplier() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=numberReactive", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("0"); - result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("1"); - result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("2"); - } - } - - @Test - public void testReactiveSupplierComposedWithImperativeFunctions() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class, - SupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=numberReactive|toUpperCase|concatWithSelf", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("0:0"); - result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("1:1"); - result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("2:2"); - } - } - - @Test - public void testReactiveSupplierComposedWithMixedFunctions() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class, - SupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=numberReactive|concatWithSelf|toUpperCaseReactive", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("0:0"); - result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("1:1"); - result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("2:2"); - } - } - - @Test - public void testReactiveSupplierComposedWithMixedFunctions2() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class, - SupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=numberReactive|toUpperCaseReactive|concatWithSelf", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - String result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("0:0"); - result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("1:1"); - result = new String(target.receive(1000).getPayload(), StandardCharsets.UTF_8); - assertThat(result).isEqualTo("2:2"); - } - } - - @Test - @Ignore - public void testFiniteFluxSupplierMessage() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class, - MessageFluxSupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=messageStreamSupplier", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - - assertThat(new String(target.receive(2000).getPayload())).isEqualTo("1"); - assertThat(new String(target.receive(2000).getPayload())).isEqualTo("2"); - assertThat(new String(target.receive(2000).getPayload())).isEqualTo("3"); - assertThat(new String(target.receive(2000).getPayload())).isEqualTo("4"); - assertThat(new String(target.receive(2000).getPayload())).isEqualTo("5"); - assertThat(new String(target.receive(2000).getPayload())).isEqualTo("6"); - } - } - - @Test - public void testFiniteFluxSupplierSimple() { - try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class, - SimpleFluxSupplierConfiguration.class)).web(WebApplicationType.NONE).run( - "--spring.cloud.stream.function.definition=simpleStreamSupplier", - "--spring.jmx.enabled=false")) { - - OutputDestination target = context.getBean(OutputDestination.class); - - assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("1"); - assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("2"); - assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("3"); - assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("4"); - assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("5"); - assertThat(new String(target.receive(2000, "simpleStreamSupplier-out-0").getPayload())).isEqualTo("6"); - } - } - - - @EnableAutoConfiguration - public static class MessageFluxSupplierConfiguration { - AtomicInteger counter = new AtomicInteger(); - - @PollableBean(splittable = true) - public Supplier>> messageStreamSupplier() { - return () -> { - Message m1 = new GenericMessage<>(String.valueOf(counter.incrementAndGet())); - Message m2 = new GenericMessage<>(String.valueOf(counter.incrementAndGet())); - Message m3 = new GenericMessage<>(String.valueOf(counter.incrementAndGet())); - return Flux.just(m1, m2, m3); - }; - } - } - - @EnableAutoConfiguration - public static class MultipleSupplierConfiguration { - - @Bean - public Supplier supplier1() { - return () -> "supplier1"; - } - - @Bean - public Supplier supplier2() { - return () -> "supplier2"; - } - } - - @EnableAutoConfiguration - public static class SimpleFluxSupplierConfiguration { - AtomicInteger counter = new AtomicInteger(); - - @PollableBean(splittable = true) - public Supplier> simpleStreamSupplier() { - return () -> { - return Flux.just(String.valueOf(counter.incrementAndGet()), - String.valueOf(counter.incrementAndGet()), - String.valueOf(counter.incrementAndGet())); - }; - } - } - - @EnableAutoConfiguration - public static class SupplierConfiguration { - - AtomicInteger counter = new AtomicInteger(); - - @Bean - public Supplier number() { - return () -> { - System.out.println("Supplying"); - return String.valueOf(this.counter.incrementAndGet()); - }; - } - - @Bean - public Supplier> numberReactive() { - return () -> Flux.create(emitter -> { - for (int i = 0; i < 3; i++) { - emitter.next(String.valueOf(i)); - } - }); - } - - - @Bean - public Function, Flux> multiplyByTwo() { - return x -> x.map(i -> String.valueOf(Integer.valueOf(i) * 2)); - } - - } - - @EnableAutoConfiguration - public static class FunctionsConfiguration { - - @Bean - public Function toUpperCase() { - return String::toUpperCase; - } - - @Bean - public Function, Flux> toUpperCaseReactive() { - return flux -> flux.map(String::toUpperCase); - } - - @Bean - public Function concatWithSelf() { - return x -> x + ":" + x; - } - - } - - /** - * This configuration essentially emulates our existing app-starters for Sources and - * essentially demonstrates how a function(s) could be applied to an existing source - * via {@link IntegrationFlowFunctionSupport} class. - */ - @EnableBinding(Source.class) - public static class ExistingMessageSourceConfiguration { - - @Bean - public IntegrationFlow messageSourceFlow(Source source) { - Supplier> messageSource = () -> MessageBuilder - .withPayload("hello function") - .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) - .build(); - return IntegrationFlows.fromSupplier(messageSource).channel(source.output()).get(); - } - - } -} 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 deleted file mode 100644 index 28cd96c20..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/interceptor/BoundChannelsInterceptedTest.java +++ /dev/null @@ -1,89 +0,0 @@ -/* - * Copyright 2015-2017 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 - * - * https://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.interceptor; - -import org.junit.Test; -import org.junit.runner.RunWith; -import org.mockito.Mockito; - -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.context.annotation.Bean; -import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.config.GlobalChannelInterceptor; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.support.ChannelInterceptor; -import org.springframework.messaging.support.MessageBuilder; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import org.springframework.util.MimeTypeUtils; - -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.verifyNoMoreInteractions; - -/** - * Verifies that interceptors used by modules are applied correctly to generated channels. - * - * @author Marius Bogoevici - * @author Oleg Zhurakousky - */ -@RunWith(SpringJUnit4ClassRunner.class) -// @checkstyle:off -@SpringBootTest(classes = BoundChannelsInterceptedTest.Foo.class, properties = "spring.cloud.stream.default-binder=mock") -public class BoundChannelsInterceptedTest { - - // @checkstyle:on - - public static final Message TEST_MESSAGE = MessageBuilder.withPayload("bar") - .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON) - .build(); - - @Autowired - ChannelInterceptor channelInterceptor; - - @Autowired - private Sink sink; - - @Test - public void testBoundChannelsIntercepted() { - this.sink.input().send(TEST_MESSAGE); - verify(this.channelInterceptor).preSend(Mockito.any(), - Mockito.eq(this.sink.input())); - verifyNoMoreInteractions(this.channelInterceptor); - } - - @SpringBootApplication - @EnableBinding(Sink.class) - public static class Foo { - - @ServiceActivator(inputChannel = Sink.INPUT) - public void fooSink(Message message) { - } - - @Bean - @GlobalChannelInterceptor - public ChannelInterceptor globalChannelInterceptor() { - return mock(ChannelInterceptor.class); - } - - } - -} diff --git a/spring-cloud-stream/work.txt b/spring-cloud-stream/work.txt new file mode 100644 index 000000000..8fb75740b --- /dev/null +++ b/spring-cloud-stream/work.txt @@ -0,0 +1 @@ +- MetersPublisherBinding uses Output. Need to rework \ No newline at end of file