Add test 2106 in ScanrioTests that is currently failing

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

@@ -40,7 +40,6 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.cloud.function.context.FunctionRegistration;
 import org.springframework.cloud.function.context.FunctionType;
-import org.springframework.cloud.function.context.catalog.FunctionAroundWrapper;
 import org.springframework.cloud.function.context.config.ContextFunctionCatalogAutoConfiguration;
 import org.springframework.cloud.stream.annotation.EnableBinding;
 import org.springframework.cloud.stream.annotation.StreamListener;

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

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.function;
 
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -42,6 +43,44 @@ import static org.assertj.core.api.Assertions.assertThat;
  */
 public class ScenarioTests {
 
+	@Test
+	public void test2106() {
+		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration
+				.getCompleteConfiguration(ConsumerConfiguration.class, ConsumerConfiguration.class))
+						.web(WebApplicationType.NONE).run(
+								"--spring.cloud.function.definition=consume;echo",
+								"--spring.cloud.stream.bindings.consume-in-0.destination=input",
+								"--spring.cloud.stream.bindings.echo-in-0.destination=echoin",
+								"--spring.cloud.stream.bindings.echo-out-0.destination=echoout",
+								"--spring.jmx.enabled=false")) {
+
+			ConsumerConfiguration configuration = context.getBean(ConsumerConfiguration.class);
+
+			OutputDestination output = context.getBean(OutputDestination.class);
+
+			StreamBridge bridge = context.getBean(StreamBridge.class);
+			bridge.send("input", "destination");
+			bridge.send("input", "destination");
+			bridge.send("input", "destination");
+
+			bridge.send("consume-in-0", "hello");
+			bridge.send("consume-in-0", "hello");
+			bridge.send("consume-in-0", "hello");
+
+			bridge.send("echoin", "hello");
+			bridge.send("echoin", "hello");
+			bridge.send("echoin", "hello");
+
+			assertThat(configuration.destinationCounter).isEqualTo(3);
+			assertThat(configuration.bindingCounter).isEqualTo(3);
+
+			assertThat(output.receive(1000, "echoout")).isNotNull();
+			assertThat(output.receive(1000, "echoout")).isNotNull();
+			assertThat(output.receive(1000, "echoout")).isNotNull();
+			assertThat(output.receive(1000, "echoout")).isNull();
+		}
+	}
+
 	@Test
 	public void testComposingSupplierWuthTypelessMessageFunction() {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
@@ -129,4 +168,29 @@ public class ScenarioTests {
 			};
 		}
 	}
+
+	@EnableAutoConfiguration
+	public static class ConsumerConfiguration {
+
+		private int destinationCounter;
+
+		private int bindingCounter;
+
+		@Bean
+		public Consumer<String> consume() {
+			return v -> {
+				if (v.equals("destination")) {
+					destinationCounter++;
+				}
+				else {
+					bindingCounter++;
+				}
+			};
+		}
+
+		@Bean
+		public Function<String, String> echo() {
+			return v -> v;
+		}
+	}
 }