diff --git a/pom.xml b/pom.xml
index 9fc7e8548..540da4311 100644
--- a/pom.xml
+++ b/pom.xml
@@ -111,7 +111,7 @@
-->
spring-cloud-stream-schema
spring-cloud-stream-schema-server
-
+
docs
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 2ac663277..b7122ab57 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
@@ -218,15 +218,8 @@ public class FunctionConfiguration {
}
else {
FunctionInvocationWrapper function = functionCatalog.lookup(functionProperties.getDefinition(), "application/json");
- if (!function.isSupplier()) {
- if (function.isConsumer()) {
- throw new UnsupportedOperationException("Consumers are not currently supported");
- }
- else if (function.isFunction()) {
- if ("input".equals(channelName)) {
- this.postProcessForStandAloneFunction(function, messageChannel);
- }
- }
+ if (!function.isSupplier() && "input".equals(channelName)) {
+ this.postProcessForStandAloneFunction(function, messageChannel);
}
}
}
@@ -243,7 +236,9 @@ public class FunctionConfiguration {
ServiceActivatingHandler handler = new ServiceActivatingHandler(new FunctionWrapper(function));
handler.setBeanFactory(context);
handler.afterPropertiesSet();
- handler.setOutputChannelName("output");
+ if (!FunctionTypeUtils.isConsumer(functionType)) {
+ handler.setOutputChannelName("output");
+ }
SubscribableChannel subscribeChannel = (SubscribableChannel) inputChannel;
subscribeChannel.subscribe(handler);
}
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 4f0b03d7c..12ab236fd 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
@@ -67,7 +67,6 @@ import static org.assertj.core.api.Assertions.assertThat;
public class GreenfieldFunctionEnableBindingTests {
@Test
- @Ignore
public void testSourceFromSupplier() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration
@@ -108,7 +107,6 @@ public class GreenfieldFunctionEnableBindingTests {
}
@Test
- @Ignore
public void testSinkFromConsumer() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
index ad00df93f..e2a77ef12 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ImplicitFunctionBindingTests.java
@@ -26,17 +26,14 @@ 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.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.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
-
-
+import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
@@ -54,7 +51,7 @@ public class ImplicitFunctionBindingTests {
}
@Test
- public void testBindingWithNoEnableBindingConfiguration() {
+ public void testSimpleFunctionWithStreamProperty() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
@@ -78,7 +75,7 @@ public class ImplicitFunctionBindingTests {
}
@Test
- public void testBindingWithNoEnableBindingConfigurationWithFunctionNativeDefinitionProperty() {
+ public void testSimpleFunctionWithNativeProperty() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
@@ -102,31 +99,7 @@ public class ImplicitFunctionBindingTests {
}
@Test
- public void testBindingWithEnableBindingConfiguration() {
-
- try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
- TestChannelBinderConfiguration.getCompleteConfiguration(
- EnableBindingConfiguration.class))
- .web(WebApplicationType.NONE)
- .run("--spring.jmx.enabled=false",
- "--spring.cloud.stream.function.definition=func")) {
-
- InputDestination inputDestination = context.getBean(InputDestination.class);
- OutputDestination outputDestination = context
- .getBean(OutputDestination.class);
-
- Message inputMessage = MessageBuilder
- .withPayload("Hello".getBytes()).build();
- inputDestination.send(inputMessage);
-
- Message outputMessage = outputDestination.receive();
- assertThat(outputMessage.getPayload()).isEqualTo("Hello".getBytes());
-
- }
- }
-
- @Test
- public void testBindingWithNoEnableBindingAndNoDefinitionPropertyConfiguration() {
+ public void testSimpleFunctionWithoutDefinitionProperty() {
System.clearProperty("spring.cloud.stream.function.definition");
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
@@ -148,6 +121,20 @@ public class ImplicitFunctionBindingTests {
}
}
+ @Test
+ public void testConsumer() {
+ try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
+ TestChannelBinderConfiguration
+ .getCompleteConfiguration(SingleFunctionConfiguration.class))
+ .web(WebApplicationType.NONE)
+ .run("--spring.cloud.stream.function.definition=consumer",
+ "--spring.jmx.enabled=false")) {
+
+ InputDestination source = context.getBean(InputDestination.class);
+ source.send(new GenericMessage("John Doe".getBytes()));
+ }
+ }
+
@Test
public void testBindingWithReactiveFunction() {
System.clearProperty("spring.cloud.stream.function.definition");
@@ -195,26 +182,6 @@ public class ImplicitFunctionBindingTests {
}
}
- @EnableAutoConfiguration
- @EnableBinding(Processor.class)
- public static class EnableBindingConfiguration {
-
- @Bean
- public Function func() {
- return x -> {
- System.out.println("Function");
- return x;
- };
- }
-
- @Bean
- public Consumer cons() {
- return x -> {
- System.out.println("Consumer");
- };
- }
- }
-
@EnableAutoConfiguration
public static class SingleFunctionConfiguration {
@@ -225,6 +192,14 @@ public class ImplicitFunctionBindingTests {
return x;
};
}
+
+
+ @Bean
+ public Consumer consumer() {
+ return value -> {
+ System.out.println(value);
+ };
+ }
}
@EnableAutoConfiguration