GH-1302 Ensure reactive/imperative composition is not subscribed twise

Resolves #1302

Author: olegz

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java

@@ -741,7 +741,7 @@ public String toString() {
 		 * Returns true if this function wrapper represents a composed function.
 		 * @return true if this function wrapper represents a composed function otherwise false
 		 */
-		boolean isComposed() {
+		public boolean isComposed() {
 			return this.composed;
 		}
 

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebRequestProcessingHelper.java

@@ -132,7 +132,7 @@ public static Publisher<?> processRequest(FunctionWrapper wrapper, Object argume
 
 		Object result = function.apply(inputMessage);
 		if (function.isConsumer()) {
-			if (result instanceof Publisher) {
+			if (result instanceof Publisher && !function.isComposed()) {
 				Mono.from((Publisher) result).subscribe();
 			}
 			return "DELETE".equals(wrapper.getMethod()) ?

spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/flux/HttpPostIntegrationTests.java

@@ -182,8 +182,6 @@ public void headers() throws Exception {
 		ResponseEntity<String> result = this.rest.exchange(RequestEntity
 				.post(new URI("/headers")).contentType(MediaType.APPLICATION_JSON)
 				.body("[\"foo\",\"bar\"]"), String.class);
-//		assertThat(result.getHeaders().getFirst("foo")).isEqualTo("bar");
-//		assertThat(result.getHeaders()).doesNotContainKey("id");
 		assertThat(result.getBody()).isEqualTo("[\"(FOO)\",\"(BAR)\"]");
 	}
 
@@ -377,7 +375,6 @@ public void multipart() throws Exception {
 
 	@Test
 	@DirtiesContext
-	@Disabled
 	public void count() throws Exception {
 		List<String> list = Arrays.asList("A", "B", "A");
 		assertThat(this.rest.exchange(
@@ -388,7 +385,6 @@ public void count() throws Exception {
 
 	@Test
 	@DirtiesContext
-	@Disabled
 	public void fluxWithList() throws Exception {
 		List<String> list = Arrays.asList("A", "B", "A");
 		assertThat(this.rest.exchange(
@@ -397,6 +393,14 @@ public void fluxWithList() throws Exception {
 				String.class).getBody()).isEqualTo("[\"A\",\"B\",\"A\"]");
 	}
 
+	@Test
+	@DirtiesContext
+	public void testReactiveFunctionComposdWithImperativeConsumer() throws Exception {
+		RequestEntity entity = RequestEntity.post(new URI("/functionReactive,consumerImperative")).build();
+		this.rest.exchange(entity, String.class);
+		assertThat(ApplicationConfiguration.functionReactiveInvocations).isEqualTo(1);
+	}
+
 	private String sse(String... values) {
 		return "[\"" + StringUtils.arrayToDelimitedString(values, "\",\"") + "\"]";
 	}
@@ -407,11 +411,26 @@ public static class ApplicationConfiguration {
 
 		private List<String> list = new ArrayList<>();
 
+		private static int functionReactiveInvocations;
+
 		public static void main(String[] args) throws Exception {
 			SpringApplication.run(HttpPostIntegrationTests.ApplicationConfiguration.class,
 					args);
 		}
 
+		@Bean
+		public Function<Flux<String>, Flux<String>> functionReactive() {
+			functionReactiveInvocations = 0;
+			return flux -> flux.doOnNext(x -> functionReactiveInvocations++);
+		}
+
+		@Bean
+		public Consumer<String> consumerImperative() {
+			return value -> {
+				System.out.println(value);
+			};
+		}
+
 		@Bean({ "uppercase", "transform", "post/more" })
 		public Function<Flux<String>, Flux<String>> uppercase() {
 			return flux -> flux.log()