Initial removal of EnableBinding

pom.xml

@@ -32,8 +32,10 @@
 	<modules>
 		<module>spring-cloud-stream</module>
 		<module>spring-cloud-stream-binder-test</module>
+		<!--
 		<module>spring-cloud-stream-test-support</module>
 		<module>spring-cloud-stream-test-support-internal</module>
+		-->
 		<module>spring-cloud-stream-integration-tests</module>
 		<module>docs</module>
 	</modules>

spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/disable/AutoconfigurationDisabledTest.java

@@ -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<String> response = (Message<String>) 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";
-		}
-
-	}
-
-}

spring-cloud-stream-test-support/src/test/java/org/springframework/cloud/stream/test/example/ExampleTest.java

@@ -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<String> message = new GenericMessage<>("hello");
-		this.processor.input().send(message);
-		Message<String> received = (Message<String>) 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";
-		}
-
-	}
-
-}

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

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

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Input.java

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

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/annotation/Output.java

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

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

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,

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

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();
+//	}
 }

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<BindableFunctionProxyFactory> 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");
 			}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Processor.java

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

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Sink.java

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

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/Source.java

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

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();
 
 }

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<T> {
 			Class<?>... additionalConfigurationClasses) {
 		List<Class<?>> configClasses = new ArrayList<>();
 		configClasses.add(TestChannelBinderConfiguration.class);
-		Import annotation = AnnotationUtils.getAnnotation(EnableBinding.class,
-				Import.class);
-		Map<String, Object> annotationAttributes = AnnotationUtils
-				.getAnnotationAttributes(annotation);
-		configClasses
-				.addAll(Arrays.asList((Class<?>[]) annotationAttributes.get("value")));
 		configClasses.add(BindingServiceConfiguration.class);
 		if (additionalConfigurationClasses != null) {
 			configClasses.addAll(Arrays.asList(additionalConfigurationClasses));

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<byte[]> 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<String, String> 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<?>, 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;

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java

@@ -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<byte[]>("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<byte[]>("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<byte[]>("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<byte[]>("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<byte[]>("hello".getBytes(StandardCharsets.UTF_8)));
-		source.send(
-				new GenericMessage<byte[]>("hello1".getBytes(StandardCharsets.UTF_8)));
-		source.send(
-				new GenericMessage<byte[]>("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<String, String> toUpperCase() {
-			return v -> {
-				System.out.println();
-				return v.toUpperCase();
-			};
-		}
-
-		@Bean
-		public Function<String, String> concatWithSelf() {
-			return x -> x + ":" + x;
-		}
-
-	}
-
-	@EnableAutoConfiguration
-	@Import(BaseProcessorConfiguration.class)
-	public static class ConsumerConfiguration {
-
-		@SuppressWarnings("unchecked")
-		@Bean
-		public Consumer<String> log(OutputDestination out) {
-			return x -> {
-				DirectFieldAccessor dfa = new DirectFieldAccessor(out);
-				MessageChannel channel = ((List<MessageChannel>) dfa.getPropertyValue("channels")).get(0);
-				channel.send(new GenericMessage<byte[]>(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();
-		}
-
-	}
-
-}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java

@@ -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<Flux<Message<?>>> messageStreamSupplier() {
-			return () -> {
-				Message<String> m1 = new GenericMessage<>(String.valueOf(counter.incrementAndGet()));
-				Message<String> m2 = new GenericMessage<>(String.valueOf(counter.incrementAndGet()));
-				Message<String> m3 = new GenericMessage<>(String.valueOf(counter.incrementAndGet()));
-				return Flux.just(m1, m2, m3);
-			};
-		}
-	}
-
-	@EnableAutoConfiguration
-	public static class MultipleSupplierConfiguration {
-
-		@Bean
-		public Supplier<String> supplier1() {
-			return () -> "supplier1";
-		}
-
-		@Bean
-		public Supplier<String> supplier2() {
-			return () -> "supplier2";
-		}
-	}
-
-	@EnableAutoConfiguration
-	public static class SimpleFluxSupplierConfiguration {
-		AtomicInteger counter = new AtomicInteger();
-
-		@PollableBean(splittable = true)
-		public Supplier<Flux<String>> 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<String> number() {
-			return () -> {
-				System.out.println("Supplying");
-				return String.valueOf(this.counter.incrementAndGet());
-			};
-		}
-
-		@Bean
-		public Supplier<Flux<String>> numberReactive() {
-			return () -> Flux.create(emitter -> {
-				for (int i = 0; i < 3; i++) {
-					emitter.next(String.valueOf(i));
-				}
-			});
-		}
-
-
-		@Bean
-		public Function<Flux<String>, Flux<String>> multiplyByTwo() {
-			return x -> x.map(i -> String.valueOf(Integer.valueOf(i) * 2));
-		}
-
-	}
-
-	@EnableAutoConfiguration
-	public static class FunctionsConfiguration {
-
-		@Bean
-		public Function<String, String> toUpperCase() {
-			return String::toUpperCase;
-		}
-
-		@Bean
-		public Function<Flux<String>, Flux<String>> toUpperCaseReactive() {
-			return flux -> flux.map(String::toUpperCase);
-		}
-
-		@Bean
-		public Function<String, String> 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<Message<String>> messageSource = () -> MessageBuilder
-					.withPayload("hello function")
-					.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
-					.build();
-			return IntegrationFlows.fromSupplier(messageSource).channel(source.output()).get();
-		}
-
-	}
-}

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

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

spring-cloud-stream/work.txt

@@ -0,0 +1 @@
+- MetersPublisherBinding uses Output. Need to rework