diff --git a/docs/src/main/asciidoc/spring-cloud-stream.adoc b/docs/src/main/asciidoc/spring-cloud-stream.adoc index 70d7958b1..632849720 100644 --- a/docs/src/main/asciidoc/spring-cloud-stream.adoc +++ b/docs/src/main/asciidoc/spring-cloud-stream.adoc @@ -587,6 +587,8 @@ support. [[spring_cloud_function]] ==== Spring Cloud Function support +===== Overview + Since Spring Cloud Stream v2.1, another alternative for defining _stream handlers_ and _sources_ is to use build-in support for https://cloud.spring.io/spring-cloud-function/[Spring Cloud Function] where they can be expressed as beans of type `java.util.function.[Supplier/Function/Consumer]`. @@ -712,7 +714,19 @@ For example, the above composition could be defined as such (if both functions p --spring.cloud.stream.function.definition=reactiveUpperCase|wrapInQuotes ---- +===== Batch Consumers +When using a `MessageChannelBinder` that supports batch listeners, and the feature is enabled for the consumer binding, you can set `spring.cloud.stream.function.definition` to `true` to enable the entire batch of messages to be passed to the function in a `List`. + +==== +[source, java] +---- +@Bean +public Function, Person> findFirstPerson() { + return persons -> persons.get(0); +} +---- +==== [[spring-cloud-streams-overview-using-polled-consumers]] ==== Using Polled Consumers diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java index dfed76167..00d293d39 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/ConsumerProperties.java @@ -17,12 +17,15 @@ package org.springframework.cloud.stream.binder; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; import javax.validation.constraints.Min; import com.fasterxml.jackson.annotation.JsonInclude; +import org.springframework.messaging.Message; + /** * Common consumer properties - spring.cloud.stream.bindings.[destinationName].consumer. * @@ -157,6 +160,15 @@ public class ConsumerProperties { */ private boolean multiplex; + /** + * When set to true, if the binder supports it, the messages emitted will have a {@link List} + * payload; When used in conjunction with functions, the function can receive a list of + * objects (or {@link Message}s) with the payloads converted if necessary. + * + * @since 3.0 + */ + private boolean batchMode; + public String getRetryTemplateName() { return retryTemplateName; } @@ -285,4 +297,12 @@ public class ConsumerProperties { this.autoStartup = autoStartup; } + public boolean isBatchMode() { + return this.batchMode; + } + + public void setBatchMode(boolean batchMode) { + this.batchMode = batchMode; + } + } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java index 848dd8500..96e37406b 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/converter/ApplicationJsonMessageMarshallingConverter.java @@ -18,6 +18,7 @@ package org.springframework.cloud.stream.converter; import java.io.IOException; import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Type; import java.nio.charset.StandardCharsets; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; @@ -44,7 +45,7 @@ import org.springframework.messaging.converter.MessageConversionException; */ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageConverter { - private final Map, JavaType> typeCache = new ConcurrentHashMap<>(); + private final Map typeCache = new ConcurrentHashMap<>(); ApplicationJsonMessageMarshallingConverter(@Nullable ObjectMapper objectMapper) { if (objectMapper != null) { @@ -67,8 +68,8 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC } @Override - protected Object convertFromInternal(Message message, Class targetClass, - @Nullable Object conversionHint) { + protected Object convertFromInternal(Message message, Class targetClass, @Nullable Object hint) { + Object conversionHint = hint; Object result = null; if (conversionHint instanceof MethodParameter) { Class conversionHintType = ((MethodParameter) conversionHint) @@ -87,12 +88,14 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC ParameterizedTypeReference forType = ParameterizedTypeReference .forType(((MethodParameter) conversionHint) .getGenericParameterType()); - result = convertParameterizedType(message, targetClass, forType); + result = convertParameterizedType(message, forType.getType()); } } else if (conversionHint instanceof ParameterizedTypeReference) { - result = convertParameterizedType(message, targetClass, - (ParameterizedTypeReference) conversionHint); + result = convertParameterizedType(message, ((ParameterizedTypeReference) conversionHint).getType()); + } + else if (conversionHint instanceof ParameterizedType) { + result = convertParameterizedType(message, (Type) conversionHint); } if (result == null) { @@ -109,15 +112,14 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC return result; } - private Object convertParameterizedType(Message message, Class targetClass, - ParameterizedTypeReference conversionHint) { + private Object convertParameterizedType(Message message, Type conversionHint) { ObjectMapper objectMapper = this.getObjectMapper(); Object payload = message.getPayload(); try { JavaType type = this.typeCache.get(conversionHint); if (type == null) { type = objectMapper.getTypeFactory() - .constructType((conversionHint).getType()); + .constructType(conversionHint); this.typeCache.put(conversionHint, type); } if (payload instanceof byte[]) { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java index a18fa47a8..feddfffab 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java @@ -17,23 +17,26 @@ package org.springframework.cloud.stream.function; import java.lang.reflect.Field; +import java.lang.reflect.ParameterizedType; +import java.lang.reflect.Type; import java.time.Duration; +import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; +import java.util.stream.Collectors; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import org.springframework.beans.factory.BeanFactory; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.FunctionType; import org.springframework.cloud.function.context.catalog.FunctionInspector; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.ProducerProperties; -import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.integration.support.MessageBuilder; @@ -52,6 +55,7 @@ import org.springframework.util.ReflectionUtils; * @author Oleg Zhurakousky * @author David Turanski * @author Tolga Kavukcu + * @author Gary Russell * @since 2.1 */ class FunctionInvoker implements Function>, Flux>> { @@ -68,16 +72,22 @@ class FunctionInvoker implements Function>, Flux inputClass; + private final ParameterizedType inputParameterizedType; + private final Class outputClass; private final Function, Flux> userFunction; private final CompositeMessageConverter messageConverter; - private final BeanFactory beanFactory; + private final MessageChannel errorChannel; private final boolean isInputArgumentMessage; + private final Class messagePayloadClass; + + private final Type messagePayloadType; + private final ConsumerProperties consumerProperties; private final ProducerProperties producerProperties; @@ -86,6 +96,12 @@ class FunctionInvoker implements Function>, Flux listContentClass; + FunctionInvoker(StreamFunctionProperties functionProperties, FunctionCatalog functionCatalog, FunctionInspector functionInspector, CompositeMessageConverterFactory compositeMessageConverterFactory) { @@ -97,7 +113,7 @@ class FunctionInvoker implements Function>, Flux implements Function>, Flux) listContainsType; + this.listContentParameterizedType = null; + } + else { + this.listContentClass = Object.class; + this.listContentParameterizedType = listContainsType; + } + if ((functionInputParameterizedType != null && functionInputParameterizedType.getRawType().equals(Flux.class)) + || payloadType != null) { + functionInputParameterizedType = null; + } + this.inputParameterizedType = functionInputParameterizedType; + if (payloadType instanceof Class) { + this.messagePayloadClass = (Class) payloadType; + this.messagePayloadType = null; + } + else { + this.messagePayloadClass = Object.class; + this.messagePayloadType = payloadType; + } + } + + private Type determinePayloadType(Type functionInputType) { + Type payloadType; + payloadType = ((ParameterizedType) functionInputType).getActualTypeArguments()[0]; + if (payloadType instanceof ParameterizedType) { + Type payloadRawType = ((ParameterizedType) payloadType).getRawType(); + if (payloadRawType.equals(List.class)) { + payloadType = ((ParameterizedType) payloadType).getActualTypeArguments()[0]; + } + } + return payloadType; } @Override @@ -127,7 +194,8 @@ class FunctionInvoker implements Function>, Flux { return Flux.just(message).doOnNext(originalMessageRef::set) - .map(this::resolveArgument).transform(this.userFunction::apply) + .map(this::resolveArgument) + .transform(this.userFunction::apply) .retryBackoff(this.consumerProperties.getMaxAttempts(), Duration.ofMillis( this.consumerProperties.getBackOffInitialInterval()), @@ -143,19 +211,10 @@ class FunctionInvoker implements Function>, Flux originalMessage) { - String inputDestinationName = functionProperties.getInputDestinationName(); - BindingProperties bindingProperties = functionProperties.getBindingServiceProperties().getBindings().get(inputDestinationName); - String destinationName = bindingProperties.getDestination(); - String groupName = bindingProperties.getGroup(); - String bindingErrorChannelName = destinationName + "." + groupName + ".errors"; - - if (beanFactory != null) { - MessageChannel errorChannel = beanFactory.containsBean(bindingErrorChannelName) - ? beanFactory.getBean(bindingErrorChannelName, MessageChannel.class) - : beanFactory.getBean("errorChannel", MessageChannel.class); - ErrorMessage em = new ErrorMessage(t, originalMessage.getHeaders(), (Message) originalMessage); + if (this.errorChannel != null) { + ErrorMessage em = new ErrorMessage(t, originalMessage); logger.error(em); - errorChannel.send(em); + this.errorChannel.send(em); } else { logger.error(t); @@ -221,19 +280,48 @@ class FunctionInvoker implements Function>, Flux) argument).getPayload() instanceof List + && !this.messagePayloadClass.isAssignableFrom(((Message) argument).getPayload().getClass())) { + argument = (T) MessageBuilder + .withPayload(convertListContents(message.getPayload(), this.messagePayloadClass, + this.messagePayloadType)) + .build(); + } + else if (this.isInputArgumentMessage && !(argument instanceof Message)) { + if (shouldBatchConvert(argument)) { + argument = convertListContents(argument, this.messagePayloadClass, this.messagePayloadType); + } argument = (T) MessageBuilder.withPayload(argument) .copyHeaders(message.getHeaders()).build(); } else if (!this.isInputArgumentMessage && argument instanceof Message) { argument = ((Message) argument).getPayload(); + if (shouldBatchConvert(argument)) { + argument = convertListContents(argument, this.listContentClass, this.listContentParameterizedType); + } } return argument; } + private boolean shouldBatchConvert(T argument) { + return this.batchMode && argument instanceof List && this.listContentClass != null; + } + + @SuppressWarnings("unchecked") + private T convertListContents(T argument, Class targetClass, Type hint) { + return (T) ((List) argument).stream() + .map(payload -> this.messageConverter.fromMessage(MessageBuilder.withPayload(payload).build(), + targetClass, hint)) + .collect(Collectors.toList()); + } + private boolean shouldConvertFromMessage(Message message) { return !this.inputClass.isAssignableFrom(Message.class) && !this.inputClass.isAssignableFrom(message.getPayload().getClass()) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java index 1d5012b2f..939727f53 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java @@ -49,6 +49,8 @@ public class StreamFunctionProperties { private Map> outputBindings = new HashMap<>(); + private boolean batchMode; + private boolean composeTo; private boolean composeFrom; @@ -110,7 +112,13 @@ public class StreamFunctionProperties { public void setInputBindings(Map> inputBindings) { this.inputBindings = inputBindings; + } + public boolean isBatchMode() { + return this.batchMode; + } + public void setBatchMode(boolean batchMode) { + this.batchMode = batchMode; } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java index b5f326771..53086951e 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/InputDestination.java @@ -33,7 +33,7 @@ public class InputDestination extends AbstractDestination { * destination (e.g., Processor.INPUT). * @param message message to send */ - public void send(Message message) { + public void send(Message message) { this.getChannel().send(message); } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java new file mode 100644 index 000000000..d96bbe4ac --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionBatchingTests.java @@ -0,0 +1,257 @@ +/* + * Copyright 2019-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.util.ArrayList; +import java.util.List; +import java.util.function.Function; + +import org.junit.Test; + +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +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.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * + * @author Gary Russel + * @author Oleg Zhurakousky + * + * @since 3.0 + */ +public class FunctionBatchingTests { + + @Test + public void testMessageBatchConfiguration() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + MessageBatchConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + List list = new ArrayList<>(); + list.add("{\"name\":\"bob\"}".getBytes()); + list.add("{\"name\":\"jill\"}".getBytes()); + Message> inputMessage = MessageBuilder + .withPayload(list) + .build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()) + .isEqualTo("{\"name\":\"bob\"}".getBytes()); + + context.stop(); + } + } + + @Test + public void testListPayloadConfiguration() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ListPayloadNotBatchConfiguration.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("[{\"name\":\"bob\"},{\"name\":\"jill\"}]".getBytes()) + .build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()) + .isEqualTo("{\"name\":\"bob\"}".getBytes()); + + context.stop(); + } + } + + @Test + public void testSimpleBatchConfiguration() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + SimpleBatchConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + List list = new ArrayList<>(); + list.add("{\"name\":\"bob\"}".getBytes()); + list.add("{\"name\":\"jill\"}".getBytes()); + Message> inputMessage = MessageBuilder + .withPayload(list) + .build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()) + .isEqualTo("{\"name\":\"bob\"}".getBytes()); + context.stop(); + } + } + + @Test + public void testNestedBatchConfiguration() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + NestedBatchConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + List list = new ArrayList<>(); + list.add("[{\"name\":\"bob\"},{\"name\":\"jill\"}]".getBytes()); + Message> inputMessage = MessageBuilder + .withPayload(list) + .build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()) + .isEqualTo("{\"name\":\"bob\"}".getBytes()); + context.stop(); + } + } + + @EnableAutoConfiguration + public static class SimpleBatchConfiguration { + + @Bean + public Function, Person> func() { + return x -> x.get(0); + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + } + + } + + + @EnableAutoConfiguration + public static class ListPayloadNotBatchConfiguration { + + @Bean + public Function, Person> func() { + return x -> x.get(0); + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + } + + } + + @EnableAutoConfiguration + public static class NestedBatchConfiguration { + + @Bean + public Function>, Person> func() { + return x -> x.get(0).get(0); + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + } + + } + + @EnableAutoConfiguration + public static class MessageBatchConfiguration { + + @Bean + public Function>, Person> func() { + return x -> x.getPayload().get(0); + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + } + + } + +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java new file mode 100644 index 000000000..efcb62ffd --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java @@ -0,0 +1,836 @@ +/* + * 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.lang.reflect.Field; +import java.util.ArrayList; +import java.util.List; +import java.util.function.Consumer; +import java.util.function.Function; + +import org.junit.Test; +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.FunctionCatalog; +import org.springframework.cloud.function.context.catalog.FunctionInspector; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamMessageConverter; +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.config.BindingServiceProperties; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.cloud.stream.function.pojo.Baz; +import org.springframework.cloud.stream.function.pojo.ErrorBaz; +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.MessageHeaders; +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.ReflectionUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Oleg Zhurakousky + * @author Tolga Kavukcu + * @author Gary Russell + * + */ +public class FunctionInvokerTests { + + private static String testWithFluxedConsumerValue; + + @Test + public void testSimpleEchoConfiguration() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + SimpleEchoConfiguration.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("{\"name\":\"bob\"}".getBytes()).build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()) + .isEqualTo("{\"name\":\"bob\"}".getBytes()); + + } + } + + @Test + public void testFluxPojoFunction() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(SimpleFluxFunctionConfiguration.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("{\"name\":\"bob\"}".getBytes()).build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()).isEqualTo("Person: bob".getBytes()); + + } + } + + @Test + public void testFluxMessagePojoFunction() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + SimpleFluxMessageFunctionConfiguration.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("{\"name\":\"bob\"}".getBytes()).build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()).isEqualTo("Person: bob".getBytes()); + } + } + + @Test + public void testFunctionHonorsOutboundBindingContentType() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ConverterDoesNotProduceCTConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.output.contentType=text/plain")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder + .withPayload("{\"name\":\"bob\"}".getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE) + .toString()).isEqualTo("text/plain"); + + } + } + + @Test + public void testFunctionHonorsConverterSetContentType() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ConverterInjectingCTConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.output.contentType=text/plain")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder + .withPayload("{\"name\":\"bob\"}".getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE) + .toString()).isEqualTo("ping/pong"); + + } + } + + @Test + public void testSameMessageTypesAreNotConverted() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(MyFunctionsConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false")) { + + Message inputMessage = new GenericMessage<>(new Foo()); + + StreamFunctionProperties functionProperties = createStreamFunctionProperties(); + + functionProperties.setDefinition("messageToMessageSameType"); + FunctionInvoker messageToMessageSameType = new FunctionInvoker<>( + functionProperties, + context.getBean(FunctionCatalog.class), + context.getBean(FunctionInspector.class), + context.getBean(CompositeMessageConverterFactory.class)); + Message outputMessage = messageToMessageSameType + .apply(Flux.just(inputMessage)).blockFirst(); + assertThat(inputMessage).isSameAs(outputMessage); + + functionProperties.setDefinition("pojoToPojoSameType"); + FunctionInvoker pojoToPojoSameType = new FunctionInvoker<>( + functionProperties, + context.getBean(FunctionCatalog.class), + context.getBean(FunctionInspector.class), + context.getBean(CompositeMessageConverterFactory.class)); + outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage)) + .blockFirst(); + assertThat(inputMessage.getPayload()).isEqualTo(outputMessage.getPayload()); + + functionProperties.setDefinition("messageToMessageNoType"); + FunctionInvoker messageToMessageNoType = new FunctionInvoker<>( + functionProperties, + context.getBean(FunctionCatalog.class), + context.getBean(FunctionInspector.class), + context.getBean(CompositeMessageConverterFactory.class)); + outputMessage = messageToMessageNoType.apply(Flux.just(inputMessage)) + .blockFirst(); + assertThat(outputMessage).isInstanceOf(Message.class); + + functionProperties.setDefinition("withException"); + FunctionInvoker withException = new FunctionInvoker<>( + functionProperties, + context.getBean(FunctionCatalog.class), + context.getBean(FunctionInspector.class), + context.getBean(CompositeMessageConverterFactory.class)); + + Flux> fluxOfMessages = Flux + .just(new GenericMessage<>(new ErrorFoo()), inputMessage); + Message resultMessage = withException.apply(fluxOfMessages).blockFirst(); + assertThat(resultMessage.getPayload()).isNotInstanceOf(ErrorFoo.class); + } + } + + @Test + public void testNativeEncodingEnabled() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(MyFunctionsConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false")) { + + Message inputMessage = new GenericMessage<>(new Baz()); + + StreamFunctionProperties functionProperties = createStreamFunctionPropertiesWithNativeEncoding(); + + functionProperties.setDefinition("pojoToPojoNonEmptyPojo"); + FunctionInvoker pojoToPojoSameType = new FunctionInvoker<>( + functionProperties, + context.getBean(FunctionCatalog.class), + context.getBean(FunctionInspector.class), + context.getBean(CompositeMessageConverterFactory.class)); + Message outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage)) + .blockFirst(); + assertThat(inputMessage.getPayload()).isEqualTo(outputMessage.getPayload()); + + Message inputMessageWithBaz = new GenericMessage<>(new Baz()); + + functionProperties.setDefinition("messageToMessageNoType"); + FunctionInvoker messageToMessageNoType = new FunctionInvoker<>( + functionProperties, + context.getBean(FunctionCatalog.class), + context.getBean(FunctionInspector.class), + context.getBean(CompositeMessageConverterFactory.class)); + outputMessage = messageToMessageNoType.apply(Flux.just(inputMessageWithBaz)) + .blockFirst(); + assertThat(outputMessage).isInstanceOf(Message.class); + + functionProperties.setDefinition("withExceptionNativeEncodingEnabled"); + FunctionInvoker withException = new FunctionInvoker<>( + functionProperties, + context.getBean(FunctionCatalog.class), + context.getBean(FunctionInspector.class), + context.getBean(CompositeMessageConverterFactory.class)); + + Flux> fluxOfMessages = Flux + .just(new GenericMessage<>(new ErrorBaz()), inputMessage); + Message resultMessage = withException.apply(fluxOfMessages).blockFirst(); + assertThat(resultMessage.getPayload()).isNotInstanceOf(ErrorFoo.class); + } + } + + @Test + public void testWithOutNativeEncodingEnabled() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(MyFunctionsConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false")) { + + Message inputMessage = new GenericMessage<>(new Baz()); + + StreamFunctionProperties functionProperties = createStreamFunctionProperties(); + + functionProperties.setDefinition("pojoToPojoNonEmptyPojo"); + FunctionInvoker pojoToPojoSameType = new FunctionInvoker<>( + functionProperties, + context.getBean(FunctionCatalog.class), + context.getBean(FunctionInspector.class), + context.getBean(CompositeMessageConverterFactory.class)); + Message outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage)) + .blockFirst(); + assertThat(outputMessage).isNotNull(); + assertThat(inputMessage.getPayload()) + .isNotEqualTo(outputMessage.getPayload()); + + } + } + + @Test + public void testWithFluxedConsumer() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration + .getCompleteConfiguration(MyFunctionsConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false")) { + + String value = "Hello"; + Message inputMessage = new GenericMessage<>(value); + + StreamFunctionProperties functionProperties = createStreamFunctionProperties(); + + functionProperties.setDefinition("fluxConsumer"); + FunctionInvoker fluxedConsumer = new FunctionInvoker<>( + functionProperties, + context.getBean(FunctionCatalog.class), + context.getBean(FunctionInspector.class), + context.getBean(CompositeMessageConverterFactory.class)); + + fluxedConsumer.apply(Flux.just(inputMessage)).blockFirst(); + + assertThat(testWithFluxedConsumerValue).isEqualTo(value); + } + } + + @Test + public void testListPayloadConfiguration() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + ListPayloadNotBatchConfiguration.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("[{\"name\":\"bob\"},{\"name\":\"jill\"}]".getBytes()) + .build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()) + .isEqualTo("{\"name\":\"bob\"}".getBytes()); + + } + } + + @Test + public void testSimpleBatchConfiguration() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + SimpleBatchConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + List list = new ArrayList<>(); + list.add("{\"name\":\"bob\"}".getBytes()); + list.add("{\"name\":\"jill\"}".getBytes()); + Message> inputMessage = MessageBuilder + .withPayload(list) + .build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()) + .isEqualTo("{\"name\":\"bob\"}".getBytes()); + + } + } + + @Test + public void testNestedBatchConfiguration() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + NestedBatchConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + List list = new ArrayList<>(); + list.add("[{\"name\":\"bob\"},{\"name\":\"jill\"}]".getBytes()); + Message> inputMessage = MessageBuilder + .withPayload(list) + .build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()) + .isEqualTo("{\"name\":\"bob\"}".getBytes()); + + } + } + + @Test + public void testMessageBatchConfiguration() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + MessageBatchConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context + .getBean(OutputDestination.class); + + List list = new ArrayList<>(); + list.add("{\"name\":\"bob\"}".getBytes()); + list.add("{\"name\":\"jill\"}".getBytes()); + Message> inputMessage = MessageBuilder + .withPayload(list) + .build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertThat(outputMessage).isNotNull(); + assertThat(outputMessage.getPayload()) + .isEqualTo("{\"name\":\"bob\"}".getBytes()); + + } + } + + private StreamFunctionProperties createStreamFunctionProperties() { + StreamFunctionProperties functionProperties = new StreamFunctionProperties(); + functionProperties.setInputDestinationName("input"); + functionProperties.setOutputDestinationName("output"); + BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); + bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3); + try { + Field f = ReflectionUtils.findField(StreamFunctionProperties.class, + "bindingServiceProperties"); + f.setAccessible(true); + f.set(functionProperties, bindingServiceProperties); + return functionProperties; + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + + private StreamFunctionProperties createStreamFunctionPropertiesWithNativeEncoding() { + StreamFunctionProperties functionProperties = new StreamFunctionProperties(); + functionProperties.setInputDestinationName("input"); + functionProperties.setOutputDestinationName("output"); + BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); + bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3); + bindingServiceProperties.getProducerProperties("output") + .setUseNativeEncoding(true); + try { + Field bspField = ReflectionUtils.findField(StreamFunctionProperties.class, + "bindingServiceProperties"); + bspField.setAccessible(true); + bspField.set(functionProperties, bindingServiceProperties); + return functionProperties; + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class SimpleEchoConfiguration { + + @Bean + public Function func() { + return x -> x; + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + } + + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class SimpleFluxFunctionConfiguration { + + @Bean + public Function, Flux> func() { + return x -> x.map(person -> person.toString()); + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + @Override + public String toString() { + return "Person: " + name; + } + + } + + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class SimpleFluxMessageFunctionConfiguration { + + @Bean + public Function>, Flux>> func() { + return x -> x.map(personMessage -> { + Person person = personMessage.getPayload(); + Message message = MessageBuilder.withPayload(person.toString()) + .copyHeaders(personMessage.getHeaders()).build(); + return message; + }); + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + @Override + public String toString() { + return "Person: " + name; + } + + } + + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class ConverterDoesNotProduceCTConfiguration { + + @Bean + public Function func() { + return x -> x; + } + + @StreamMessageConverter + public MessageConverter customConverter() { + return new MessageConverter() { + + @Override + public Message toMessage(Object payload, MessageHeaders headers) { + return new GenericMessage(((String) payload).getBytes()); + } + + @Override + public Object fromMessage(Message message, Class targetClass) { + String contentType = message.getHeaders() + .get(MessageHeaders.CONTENT_TYPE).toString(); + if (contentType.equals("foo/bar")) { + return new String((byte[]) message.getPayload()); + } + return null; + } + }; + } + + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class ConverterInjectingCTConfiguration { + + @Bean + public Function func() { + return x -> x; + } + + @StreamMessageConverter + public MessageConverter customConverter() { + return new MessageConverter() { + + @Override + public Message toMessage(Object payload, MessageHeaders headers) { + return MessageBuilder.withPayload(((String) payload).getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, "ping/pong").build(); + } + + @Override + public Object fromMessage(Message message, Class targetClass) { + String contentType = message.getHeaders() + .get(MessageHeaders.CONTENT_TYPE).toString(); + if (contentType.equals("foo/bar")) { + return new String((byte[]) message.getPayload()); + } + return null; + } + }; + } + + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class ListPayloadNotBatchConfiguration { + + @Bean + public Function, Person> func() { + return x -> x.get(0); + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + } + + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class SimpleBatchConfiguration { + + @Bean + public Function, Person> func() { + return x -> x.get(0); + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + } + + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class NestedBatchConfiguration { + + @Bean + public Function>, Person> func() { + return x -> x.get(0).get(0); + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + } + + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class MessageBatchConfiguration { + + @Bean + public Function>, Person> func() { + return x -> x.getPayload().get(0); + } + + public static class Person { + + private String name; + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + } + + } + + @EnableAutoConfiguration + public static class MyFunctionsConfiguration { + + @Bean + public Consumer> fluxConsumer() { + return f -> f.subscribe(v -> { + System.out.println("Consuming flux: " + v); + testWithFluxedConsumerValue = v; + }); + } + + @Bean + public Function, Message> messageToMessageDifferentType() { + return x -> MessageBuilder.withPayload(new Bar()).copyHeaders(x.getHeaders()) + .build(); + } + + @Bean + public Function, Message> messageToMessageAnyType() { + return x -> MessageBuilder.withPayload(new Bar()).copyHeaders(x.getHeaders()) + .build(); + } + + @Bean + public Function, Message> messageToMessageNoType() { + return x -> MessageBuilder.withPayload(new Bar()).copyHeaders(x.getHeaders()) + .build(); + } + + @Bean + public Function, Message> messageToMessageSameType() { + return x -> x; + } + + @Bean + public Function pojoToPojoSameType() { + return x -> x; + } + + @Bean + public Function pojoToPojoNonEmptyPojo() { + return x -> x; + } + + @Bean + public Function withException() { + return x -> { + if (x instanceof ErrorFoo) { + System.out.println("Throwing exception "); + throw new RuntimeException("Boom!"); + } + else { + System.out.println("All is good "); + return x; + } + }; + } + + @Bean + public Function withExceptionNativeEncodingEnabled() { + return x -> { + if (x instanceof ErrorBaz) { + System.out.println("Throwing exception "); + throw new RuntimeException("Boom!"); + } + else { + System.out.println("All is good "); + return x; + } + }; + } + + } + + private static class Foo { + + } + + private static class ErrorFoo extends Foo { + + } + + private static class Bar { + + } + +}