GH-783 Fix Consumer processing with webflux

Resolves #783

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

@@ -17,6 +17,7 @@
 package org.springframework.cloud.function.web.function;
 
 import java.net.URI;
+import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
@@ -67,6 +68,18 @@ public class FunctionEndpointInitializerTests {
 		assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND);
 	}
 
+	@Test
+	public void testConsumerMapping() throws Exception {
+		FunctionalSpringApplication.run(ConsumerConfiguration.class);
+		TestRestTemplate testRestTemplate = new TestRestTemplate();
+		String port = System.getProperty("server.port");
+		Thread.sleep(200);
+		ResponseEntity<String> response = testRestTemplate
+				.postForEntity(new URI("http://localhost:" + port + "/uppercase"), "stressed", String.class);
+		assertThat(response.getBody()).isNull();
+		assertThat(response.getStatusCode()).isEqualTo(HttpStatus.ACCEPTED);
+	}
+
 	@Test
 	public void testSingleFunctionMapping() throws Exception {
 		FunctionalSpringApplication.run(ApplicationConfiguration.class);
@@ -114,6 +127,23 @@ public class FunctionEndpointInitializerTests {
 		assertThat(response.getBody()).isEqualTo("Jim Lahey");
 	}
 
+	@SpringBootConfiguration
+	protected static class ConsumerConfiguration
+			implements ApplicationContextInitializer<GenericApplicationContext> {
+
+		public Consumer<String> consume() {
+			return v -> System.out.println(v);
+		}
+
+		@Override
+		public void initialize(GenericApplicationContext applicationContext) {
+			applicationContext.registerBean("consume", FunctionRegistration.class,
+					() -> new FunctionRegistration<>(consume())
+						.type(FunctionType.consumer(String.class)));
+		}
+
+	}
+
 
 	@SpringBootConfiguration
 	protected static class ApplicationConfiguration