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 044c9e658..426a13aa6 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 @@ -27,8 +27,10 @@ import reactor.core.publisher.Flux; import org.springframework.cloud.function.context.FunctionType; import org.springframework.cloud.function.context.catalog.FunctionInspector; 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; @@ -49,12 +51,16 @@ class FunctionInvoker implements Function>, Flux inputClass; + private final Class outputClass; + private final Function, Flux> userFunction; private final CompositeMessageConverter messageConverter; private final MessageChannel errorChannel; + private final boolean isInputArgumentMessage; + FunctionInvoker(String functionName, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector, CompositeMessageConverterFactory compositeMessageConverterFactory) { this(functionName, functionCatalog, functionInspector, compositeMessageConverterFactory, null); @@ -66,39 +72,51 @@ class FunctionInvoker implements Function>, Flux> apply(Flux> input) { AtomicReference> originalMessageRef = new AtomicReference<>(); return input .doOnNext(originalMessageRef::set) // to preserve the original message .map(this::resolveArgument) // resolves argument type before invocation of user function - .onErrorContinue((exception, originalMessage) -> { - if (this.errorChannel != null) { - ErrorMessage em = new ErrorMessage(exception, (Message) originalMessage); - logger.error(em); - this.errorChannel.send(em); - } - else { - exception.printStackTrace(); - } - }) + .onErrorContinue((x, y) -> onError(x, (Message) y)) .transform(this.userFunction::apply) // invoke user function .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, (Message) 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); } - return (Message) + Message returnMessage = (Message) (value instanceof Message ? value - : this.messageConverter.toMessage(value, originalMessage.getHeaders())); + : this.messageConverter.toMessage(value, originalMessage.getHeaders(), this.outputClass)); + if (returnMessage == null) { + if (value.getClass().isAssignableFrom(this.outputClass)) { + returnMessage = (Message) MessageBuilder.withPayload(value).copyHeaders(originalMessage.getHeaders()).removeHeader(MessageHeaders.CONTENT_TYPE).build(); + } + } + Assert.notNull(returnMessage, "Failed to convert result value '" + value + "' to message."); + return returnMessage; } @SuppressWarnings("unchecked") @@ -110,13 +128,15 @@ class FunctionInvoker implements Function>, Flux)argument).getPayload(); + } return argument; } private boolean shouldConvertFromMessage(Message message) { - return !this.inputClass.isAssignableFrom(byte[].class) && + return !message.getPayload().getClass().isAssignableFrom(this.inputClass) && !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 index 4b43ad51d..f9d6f4b6b 100644 --- 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 @@ -33,6 +33,7 @@ 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; @@ -103,6 +104,15 @@ public class IntegrationFlowFunctionSupport { return IntegrationFlows.from(supplier); } + /** + * @param inputChannel + * @return + */ + public IntegrationFlowBuilder integrationFlowFromChannel(SubscribableChannel inputChannel) { + IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(inputChannel).bridge(); + 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.name` property. 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..5f397aee3 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java @@ -0,0 +1,93 @@ +/* + * Copyright 2018 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 + * + * http://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.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.binder.test.TestChannelBinderConfiguration; +import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * + * @author Oleg Zhurakousky + * + */ +public class FunctionInvokerTests { + + @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()); + + FunctionInvoker messageToMessageSameType = new FunctionInvoker<>("messageToMessageSameType", + new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)), context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class)); + Message outputMessage = messageToMessageSameType.apply(Flux.just(inputMessage)).blockFirst(); + assertThat(inputMessage).isSameAs(outputMessage); + + FunctionInvoker pojoToPojoSameType = new FunctionInvoker<>("pojoToPojoSameType", + new FunctionCatalogWrapper(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()); + } + } + + @EnableAutoConfiguration + public static class MyFunctionsConfiguration { + + @Bean + public Function, Message> messageToMessageDifferentType() { + 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; + } + } + + private static class Foo { + + } + + private static class Bar { + + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java new file mode 100644 index 000000000..0e738d8d2 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java @@ -0,0 +1,178 @@ +/* + * Copyright 2018 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 + * + * http://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.nio.charset.StandardCharsets; +import java.util.function.Consumer; +import java.util.function.Function; + +import org.junit.Test; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.cloud.stream.messaging.Processor; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlowBuilder; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.lang.Nullable; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * + * @author Oleg Zhurakousky + * + * @since 2.1 + * + */ +public class ProcessorToFunctionsSupportTests { + + @Test + public void testPathThrough() { + ApplicationContext context = + new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.jmx.enabled=false"); + InputDestination source = context.getBean(InputDestination.class); + OutputDestination target = context.getBean(OutputDestination.class); + source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); + assertThat(target.receive(1000).getPayload()).isEqualTo("hello".getBytes(StandardCharsets.UTF_8)); + } + + @Test + public void testSingleFunction() { + ApplicationContext context = + new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.name=toUpperCase", "--spring.jmx.enabled=false"); + + InputDestination source = context.getBean(InputDestination.class); + OutputDestination target = context.getBean(OutputDestination.class); + source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); + assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8)); + } + + @Test + public void testComposedFunction() { + ApplicationContext context = + new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf", "--spring.jmx.enabled=false"); + + InputDestination source = context.getBean(InputDestination.class); + OutputDestination target = context.getBean(OutputDestination.class); + source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); + assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO:HELLO".getBytes(StandardCharsets.UTF_8)); + } + + @Test + public void testConsumer() { + ApplicationContext context = + new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(ConsumerConfiguration.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.name=log", "--spring.jmx.enabled=false"); + + InputDestination source = context.getBean(InputDestination.class); + OutputDestination target = context.getBean(OutputDestination.class); + source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); + source.send(new GenericMessage("hello1".getBytes(StandardCharsets.UTF_8))); + source.send(new GenericMessage("hello2".getBytes(StandardCharsets.UTF_8))); + assertThat(target.receive(1000).getPayload()).isEqualTo("hello".getBytes(StandardCharsets.UTF_8)); + assertThat(target.receive(1000).getPayload()).isEqualTo("hello1".getBytes(StandardCharsets.UTF_8)); + assertThat(target.receive(1000).getPayload()).isEqualTo("hello2".getBytes(StandardCharsets.UTF_8)); + } + + @EnableAutoConfiguration + @Import(BaseProcessorConfiguration.class) + public static class FunctionsConfiguration { + + @Bean + public Function toUpperCase() { + return String::toUpperCase; + } + + @Bean + public Function concatWithSelf() { + return x -> x + ":" + x; + } + } + + @EnableAutoConfiguration + @Import(BaseProcessorConfiguration.class) + public static class ConsumerConfiguration { + + @Autowired + OutputDestination out; + + @Bean + public Consumer log() { + return x -> { + DirectFieldAccessor dfa = new DirectFieldAccessor(out); + MessageChannel channel = (MessageChannel) dfa.getPropertyValue("channel"); + channel.send(new GenericMessage(x.getBytes())); + }; + } + } + + /** + * This configuration essentially emulates our existing app-starters for Processor + * and essentially demonstrates how a function(s) could be applied to an existing + * processor app via {@link IntegrationFlowFunctionSupport} class. + */ + @EnableBinding(Processor.class) + public static class BaseProcessorConfiguration { + + @Autowired + private Processor processor; + + @Bean + public IntegrationFlow fromChannel(@Nullable IntegrationFlowFunctionSupport functionSupport) { + + IntegrationFlowBuilder flowBuilder = null; + if (functionSupport == null) { + flowBuilder = IntegrationFlows.from(processor.input()).bridge().channel(processor.output()); + } + else { + flowBuilder = functionSupport.integrationFlowFromChannel(processor.input()); + if (!functionSupport.andThenFunction(flowBuilder, processor.output())) { + flowBuilder = flowBuilder.channel(processor.output()); + } + } + + return flowBuilder.get(); + } + + } +} diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java index fda4d81ca..47b40bbb7 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java @@ -95,7 +95,22 @@ public class SourceToFunctionsSupportTests { public void testFailedInputTypeConversion() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( - TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoContentType.class)) + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoConversionPossible.class)) + .web(WebApplicationType.NONE) + .run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf", + "--spring.jmx.enabled=false")) { + PollableChannel errorChannel = context.getBean("errorChannel", PollableChannel.class); + OutputDestination target = context.getBean(OutputDestination.class); + assertNull(target.receive(1000)); + assertNotNull(errorChannel.receive(1000)); + } + } + + @Test + public void testComposedFunctionIsAppliedToExistingMessageSourceFailedTypeConversion() { + try (ConfigurableApplicationContext context = + new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoConversionPossible.class)) .web(WebApplicationType.NONE) .run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf", "--spring.jmx.enabled=false")) { @@ -206,7 +221,7 @@ public class SourceToFunctionsSupportTests { @EnableAutoConfiguration @Import(ExistingMessageSourceConfigurationNoContentTypeSet.class) - public static class FunctionsConfigurationNoContentType { + public static class FunctionsConfigurationNoConversionPossible { @Bean public PollableChannel errorChannel() { @@ -214,13 +229,13 @@ public class SourceToFunctionsSupportTests { } @Bean - public Function toUpperCase() { - return String::toUpperCase; + public Function toUpperCase() { + return x -> true; } @Bean - public Function concatWithSelf() { - return x -> x + ":" + x; + public Function concatWithSelf() { + return x -> 1; } }