From 6c772bba0833ee1033eb6fcc0fc2a9896a8b16aa Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 7 Aug 2019 16:00:04 +0200 Subject: [PATCH] GH-1742 Adjusted previous commit to new function infrustructure Resolves #1738 Resolves #1742 --- ...cationJsonMessageMarshallingConverter.java | 26 + .../stream/function/FunctionInvoker.java | 331 ------- .../IntegrationFlowFunctionSupport.java | 289 ------ .../stream/function/FunctionInvokerTests.java | 836 ------------------ 4 files changed, 26 insertions(+), 1456 deletions(-) delete mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java delete mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java delete mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java 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 96e37406b..2ab36f2d5 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 @@ -20,12 +20,15 @@ import java.io.IOException; import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; import java.nio.charset.StandardCharsets; +import java.util.Collection; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.stream.Collectors; import com.fasterxml.jackson.databind.JavaType; import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.cloud.function.context.catalog.FunctionTypeUtils; import org.springframework.core.MethodParameter; import org.springframework.core.ParameterizedTypeReference; import org.springframework.lang.Nullable; @@ -118,6 +121,9 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC try { JavaType type = this.typeCache.get(conversionHint); if (type == null) { + conversionHint = FunctionTypeUtils.isMessage(conversionHint) + ? FunctionTypeUtils.getImmediateGenericType(conversionHint, 0) + : conversionHint; type = objectMapper.getTypeFactory() .constructType(conversionHint); this.typeCache.put(conversionHint, type); @@ -129,6 +135,26 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC return objectMapper.readValue((String) payload, type); } else { + final JavaType typeToUse = type; + if (payload instanceof Collection) { + Collection collection = (Collection) ((Collection) payload).stream() + .map(value -> { + try { + if (value instanceof byte[]) { + return objectMapper.readValue((byte[]) value, typeToUse.getContentType()); + } + else if (value instanceof String) { + return objectMapper.readValue((String) value, typeToUse.getContentType()); + } + } + catch (Exception e) { + logger.error("Failed to convert payload " + value, e); + } + return null; + }).collect(Collectors.toList()); + + return collection; + } return null; } } 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 deleted file mode 100644 index feddfffab..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java +++ /dev/null @@ -1,331 +0,0 @@ -/* - * 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.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.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.BindingServiceProperties; -import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.converter.CompositeMessageConverter; -import org.springframework.messaging.support.ErrorMessage; -import org.springframework.util.Assert; -import org.springframework.util.MimeType; -import org.springframework.util.ReflectionUtils; - -/** - * @param the payload type of the input Message - * @param the payload type of the output Message - * @author Oleg Zhurakousky - * @author David Turanski - * @author Tolga Kavukcu - * @author Gary Russell - * @since 2.1 - */ -class FunctionInvoker implements Function>, Flux>> { - - private static final Log logger = LogFactory.getLog(FunctionInvoker.class); - - private static final Field MESSAGE_HEADERS_FIELD; - - static { - MESSAGE_HEADERS_FIELD = ReflectionUtils.findField(MessageHeaders.class, - "headers"); - MESSAGE_HEADERS_FIELD.setAccessible(true); - } - - private final Class inputClass; - - private final ParameterizedType inputParameterizedType; - - private final Class outputClass; - - private final Function, Flux> userFunction; - - private final CompositeMessageConverter messageConverter; - - private final MessageChannel errorChannel; - - private final boolean isInputArgumentMessage; - - private final Class messagePayloadClass; - - private final Type messagePayloadType; - - private final ConsumerProperties consumerProperties; - - private final ProducerProperties producerProperties; - - private final BindingServiceProperties bindingServiceProperties; - - private final StreamFunctionProperties functionProperties; - - private final boolean batchMode; - - private final Type listContentParameterizedType; - - private final Class listContentClass; - - FunctionInvoker(StreamFunctionProperties functionProperties, - FunctionCatalog functionCatalog, FunctionInspector functionInspector, - CompositeMessageConverterFactory compositeMessageConverterFactory) { - this(functionProperties, functionCatalog, functionInspector, - compositeMessageConverterFactory, null); - } - - @SuppressWarnings("unchecked") - FunctionInvoker(StreamFunctionProperties functionProperties, - FunctionCatalog functionCatalog, FunctionInspector functionInspector, - CompositeMessageConverterFactory compositeMessageConverterFactory, - MessageChannel errorChannel) { - - this.functionProperties = functionProperties; - Object originalUserFunction = functionCatalog - .lookup(functionProperties.getDefinition()); - - this.userFunction = (Function, Flux>) originalUserFunction; - - Assert.isInstanceOf(Function.class, this.userFunction); - this.messageConverter = compositeMessageConverterFactory - .getMessageConverterForAllRegistered(); - FunctionType functionType = functionInspector - .getRegistration(originalUserFunction).getType(); - this.isInputArgumentMessage = functionType.isMessage(); - this.inputClass = functionType.getInputType(); - this.outputClass = functionType.getOutputType(); - this.errorChannel = errorChannel; - this.bindingServiceProperties = functionProperties.getBindingServiceProperties(); - this.consumerProperties = this.bindingServiceProperties - .getConsumerProperties(functionProperties.getInputDestinationName()); - this.producerProperties = this.bindingServiceProperties - .getProducerProperties(functionProperties.getOutputDestinationName()); - this.batchMode = this.consumerProperties.isBatchMode(); - Type type = functionType.getType(); - ParameterizedType functionInputParameterizedType = null; - Type listContainsType = null; - Type payloadType = null; - if (type instanceof ParameterizedType) { - Type functionInputType = ((ParameterizedType) type).getActualTypeArguments()[0]; - if (functionInputType instanceof ParameterizedType) { - functionInputParameterizedType = (ParameterizedType) functionInputType; - Type rawType = ((ParameterizedType) functionInputType).getRawType(); - if (rawType.equals(List.class)) { - listContainsType = ((ParameterizedType) functionInputType).getActualTypeArguments()[0]; - } - else if (rawType.equals(Message.class)) { - payloadType = determinePayloadType(functionInputType); - } - } - } - if (listContainsType instanceof Class) { - this.listContentClass = (Class) 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 - public Flux> apply(Flux> input) { - AtomicReference> originalMessageRef = new AtomicReference<>(); - - return input.concatMap(message -> { - return Flux.just(message).doOnNext(originalMessageRef::set) - .map(this::resolveArgument) - .transform(this.userFunction::apply) - .retryBackoff(this.consumerProperties.getMaxAttempts(), - Duration.ofMillis( - this.consumerProperties.getBackOffInitialInterval()), - Duration.ofMillis( - this.consumerProperties.getBackOffMaxInterval())) - .onErrorResume(e -> { - onError(e, originalMessageRef.get()); - return Mono.empty(); - }); - }).map(resultMessage -> toMessage(resultMessage, originalMessageRef.get())); // create - // output - // message - } - - private void onError(Throwable t, Message originalMessage) { - if (this.errorChannel != null) { - ErrorMessage em = new ErrorMessage(t, originalMessage); - logger.error(em); - this.errorChannel.send(em); - } - else { - logger.error(t); - } - } - - @SuppressWarnings("unchecked") - private Message toMessage(T value, Message originalMessage) { - if (logger.isDebugEnabled()) { - logger.debug("Converting result back to message using the original message: " - + originalMessage); - } - - Message returnMessage; - if (this.producerProperties.isUseNativeEncoding()) { - if (logger.isDebugEnabled()) { - logger.debug( - "Native encoding enabled wrapping result to message using the original message: " - + originalMessage); - } - returnMessage = wrapOutputToMessage(value, originalMessage); - } - else { - returnMessage = (Message) (value instanceof Message ? value - : this.messageConverter.toMessage(value, - originalMessage.getHeaders())); - if (returnMessage == null - && value.getClass().isAssignableFrom(this.outputClass)) { - returnMessage = wrapOutputToMessage(value, originalMessage); - } - else if (this.bindingServiceProperties != null - && this.bindingServiceProperties.getBindingProperties( - this.functionProperties.getOutputDestinationName()) != null - && !returnMessage.getHeaders() - .containsKey(MessageHeaders.CONTENT_TYPE)) { - - ((Map) ReflectionUtils.getField(MESSAGE_HEADERS_FIELD, - returnMessage.getHeaders())).put( - MessageHeaders.CONTENT_TYPE, - MimeType.valueOf(this.bindingServiceProperties - .getBindingProperties("output") - .getContentType())); - - } - Assert.notNull(returnMessage, - "Failed to convert result value '" + value + "' to message."); - } - return returnMessage; - } - - @SuppressWarnings("unchecked") - private Message wrapOutputToMessage(T value, Message originalMessage) { - Message returnMessage = (Message) MessageBuilder.withPayload(value) - .copyHeaders(originalMessage.getHeaders()) - .removeHeader(MessageHeaders.CONTENT_TYPE).build(); - return returnMessage; - } - - @SuppressWarnings("unchecked") - private T resolveArgument(Message message) { - if (logger.isDebugEnabled()) { - logger.debug("Resolving input argument from message: " + message); - } - - T argument = (T) (shouldConvertFromMessage(message) - ? this.messageConverter.fromMessage(message, this.inputClass, this.inputParameterizedType) : message); - Assert.notNull(argument, "Failed to resolve argument type '" + this.inputClass - + "' from message: " + message); - if (this.batchMode - && this.messagePayloadClass != null - && this.isInputArgumentMessage - && argument instanceof Message - && ((Message) 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()) - && !this.inputClass.isAssignableFrom(Object.class); - } - -} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java deleted file mode 100644 index 0d54b035c..000000000 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java +++ /dev/null @@ -1,289 +0,0 @@ -/* - * 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.util.concurrent.atomic.AtomicReference; -import java.util.function.Consumer; -import java.util.function.Function; -import java.util.function.Supplier; - -import org.reactivestreams.Publisher; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.core.publisher.MonoSink; - -import org.springframework.cloud.function.context.FunctionCatalog; -import org.springframework.cloud.function.context.FunctionRegistration; -import org.springframework.cloud.function.context.FunctionType; -import org.springframework.cloud.function.context.catalog.FunctionInspector; -import org.springframework.cloud.function.core.FluxSupplier; -import org.springframework.cloud.stream.binder.BindingCreatedEvent; -import org.springframework.cloud.stream.config.BindingServiceProperties; -import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; -import org.springframework.context.support.GenericApplicationContext; -import org.springframework.integration.context.IntegrationObjectSupport; -import org.springframework.integration.dsl.IntegrationFlowBuilder; -import org.springframework.integration.dsl.IntegrationFlows; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.SubscribableChannel; -import org.springframework.util.Assert; -import org.springframework.util.StringUtils; - -/** - * @author Oleg Zhurakousky - * @author David Turanski - * @author Ilayaperumal Gopinathan - * @since 2.1 - */ -public class IntegrationFlowFunctionSupport { - - private final FunctionCatalog functionCatalog; - - private final FunctionInspector functionInspector; - - private final CompositeMessageConverterFactory messageConverterFactory; - - private final StreamFunctionProperties functionProperties; - - - //private final AtomicReference> triggerRef = new AtomicReference<>(); - - private final Publisher trigger; - - private final GenericApplicationContext context; - - IntegrationFlowFunctionSupport(FunctionCatalog functionCatalog, - FunctionInspector functionInspector, - CompositeMessageConverterFactory messageConverterFactory, - StreamFunctionProperties functionProperties, - BindingServiceProperties bindingServiceProperties, - GenericApplicationContext context) { - - Assert.notNull(functionCatalog, "'functionCatalog' must not be null"); - Assert.notNull(functionInspector, "'functionInspector' must not be null"); - Assert.notNull(messageConverterFactory, - "'messageConverterFactory' must not be null"); - Assert.notNull(functionProperties, "'functionProperties' must not be null"); - this.functionCatalog = functionCatalog; - this.functionInspector = functionInspector; - this.messageConverterFactory = messageConverterFactory; - this.functionProperties = functionProperties; - this.context = context; - this.functionProperties.setBindingServiceProperties(bindingServiceProperties); - AtomicReference> triggerRef = null; - trigger = Mono.create(emmiter -> { - triggerRef.set(emmiter); - }); - context.addApplicationListener(event -> { - if (event instanceof BindingCreatedEvent) { - if (triggerRef.get() != null) { - triggerRef.get().success(); - } - } - }); - } - - - /** - * Determines if function specified via 'spring.cloud.stream.function.definition' - * property can be located in {@link FunctionCatalog}s. - * @param type of function - * @param typeOfFunction must be Supplier, Function or Consumer - * @return {@code true} if function is already stored - */ - public boolean containsFunction(Class typeOfFunction) { - return StringUtils.hasText(this.functionProperties.getDefinition()) - && this.catalogContains(typeOfFunction, - this.functionProperties.getDefinition()); - } - - /** - * Determines if function specified via 'spring.cloud.stream.function.definition' - * property can be located in {@link FunctionCatalog}. - * @param type of function - * @param typeOfFunction must be Supplier, Function or Consumer - * @param functionName the function name to check - * @return {@code true} if function is already stored - */ - public boolean containsFunction(Class typeOfFunction, String functionName) { - return StringUtils.hasText(functionName) - && this.catalogContains(typeOfFunction, functionName); - } - - /** - * @return the function type basing on the function definition. - */ - public FunctionType getCurrentFunctionType() { - FunctionType functionType = this.functionInspector.getRegistration( - this.functionCatalog.lookup(this.functionProperties.getDefinition())) - .getType(); - return functionType; - } - - /** - * Create an instance of the {@link IntegrationFlowBuilder} from a {@link Supplier} - * bean available in the context. The name of the bean must be provided via - * `spring.cloud.stream.function.definition` property. - * @return instance of {@link IntegrationFlowBuilder} - * @throws IllegalStateException if the named Supplier can not be located. - */ - public IntegrationFlowBuilder integrationFlowFromNamedSupplier() { - if (StringUtils.hasText(this.functionProperties.getDefinition())) { - Supplier supplier = this.functionCatalog.lookup(Supplier.class, - this.functionProperties.getDefinition()); - if (supplier instanceof FluxSupplier) { - supplier = ((FluxSupplier) supplier).getTarget(); - } - return integrationFlowFromProvidedSupplier(supplier).split(); - } - - throw new IllegalStateException( - "A Supplier is not specified in the 'spring.cloud.stream.function.definition' property."); - } - - /** - * Create an instance of the {@link IntegrationFlowBuilder} from a provided - * {@link Supplier}. - * @param supplier supplier from which the flow builder will be built - * @return instance of {@link IntegrationFlowBuilder} - */ - @SuppressWarnings({ "rawtypes", "unchecked" }) - public IntegrationFlowBuilder integrationFlowFromProvidedSupplier( - Supplier supplier) { - String supplierName = this.functionProperties.getDefinition().split("\\|")[0]; - FunctionRegistration fr = this.functionInspector.getRegistration(this.functionCatalog.lookup(supplierName)); - if (fr != null && fr.getType().isWrapper()) { - Publisher publisher = (Publisher) supplier.get(); - publisher = publisher instanceof Flux - ? ((Flux) publisher).delaySubscription(trigger) - : ((Mono) publisher).delaySubscription(trigger); - - return IntegrationFlows.from(publisher); - } - return IntegrationFlows.from(supplier); - } - - /** - * @param inputChannel channel for which flow we be built - * @return instance of {@link IntegrationFlowBuilder} - */ - public IntegrationFlowBuilder integrationFlowFromChannel( - SubscribableChannel inputChannel) { - IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(inputChannel); - return flowBuilder; - } - - /** - * @param inputChannel channel for which flow we be built - * @param outputChannel channel for which flow we be built - * @return instance of {@link IntegrationFlowBuilder} - */ - public IntegrationFlowBuilder integrationFlowForFunction( - SubscribableChannel inputChannel, MessageChannel outputChannel) { - - if (inputChannel instanceof IntegrationObjectSupport) { - String inputBindingName = ((IntegrationObjectSupport) inputChannel) - .getComponentName(); - if (StringUtils.hasText(inputBindingName)) { - this.functionProperties.setInputDestinationName(inputBindingName); - } - } - - if (outputChannel instanceof IntegrationObjectSupport) { - String outputBindingName = ((IntegrationObjectSupport) outputChannel) - .getComponentName(); - if (StringUtils.hasText(outputBindingName)) { - this.functionProperties.setOutputDestinationName(outputBindingName); - } - } - - IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(inputChannel); - - if (!this.andThenFunction(flowBuilder, outputChannel, this.functionProperties)) { - flowBuilder = flowBuilder.channel(outputChannel); - } - return flowBuilder; - } - - /** - * Add a {@link Function} bean to the end of an integration flow. The name of the bean - * must be provided via `spring.cloud.stream.function.definition` property. - *

- * NOTE: If this method returns true, the integration flow is now represented as a - * Reactive Streams {@link Publisher} bean. - *

- * @param flowBuilder instance of the {@link IntegrationFlowBuilder} representing the - * current state of the integration flow - * @param outputChannel channel where the output of a function will be sent - * @param functionProperties the function properties - * @return true if {@link Function} was located and added and false if it wasn't. - */ - public boolean andThenFunction(IntegrationFlowBuilder flowBuilder, - MessageChannel outputChannel, StreamFunctionProperties functionProperties) { - return andThenFunction(flowBuilder.toReactivePublisher(), outputChannel, - functionProperties); - } - - /** - * @param publisher publisher to subscribe to - * @param outputChannel output channel to which a message will be sent - * @param functionProperties function properties - * @param input of the function - * @param output of the function - * @return whether the function was properly invoked - */ - public boolean andThenFunction(Publisher publisher, - MessageChannel outputChannel, StreamFunctionProperties functionProperties) { - if (!StringUtils.hasText(functionProperties.getDefinition())) { - return false; - } - FunctionInvoker functionInvoker = new FunctionInvoker<>(functionProperties, - this.functionCatalog, this.functionInspector, - this.messageConverterFactory, this.context.getBeanFactory()); - - if (outputChannel != null) { - subscribeToInput(functionInvoker, publisher, outputChannel::send); - } - else { - subscribeToInput(functionInvoker, publisher, null); - } - return true; - } - - private boolean catalogContains(Class functionType, String name) { - return this.functionCatalog.lookup(functionType, name) != null; - } - - private Mono subscribeToOutput(Consumer> outputProcessor, - Publisher> outputPublisher) { - - Flux> output = outputProcessor == null ? Flux.from(outputPublisher) - : Flux.from(outputPublisher).doOnNext(outputProcessor); - return output.then(); - } - - @SuppressWarnings("unchecked") - private void subscribeToInput(FunctionInvoker functionInvoker, - Publisher publisher, Consumer> outputProcessor) { - - Flux inputPublisher = Flux.from(publisher); - subscribeToOutput(outputProcessor, - functionInvoker.apply((Flux>) inputPublisher)).subscribe(); - } - -} 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 deleted file mode 100644 index efcb62ffd..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java +++ /dev/null @@ -1,836 +0,0 @@ -/* - * 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 { - - } - -}