From 8ae3ec0041156646969f1a9ddd0accaacb541e44 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 15 Nov 2018 18:28:59 +0100 Subject: [PATCH] GH-1498 Added initial retry logic to reactive pipeline Resolves #1498 --- .../binder/AbstractMessageChannelBinder.java | 23 ++++++-- .../function/FunctionConfiguration.java | 4 -- .../stream/function/FunctionInvoker.java | 39 ++++++++----- .../IntegrationFlowFunctionSupport.java | 24 ++------ .../function/StreamFunctionProperties.java | 7 +++ .../stream/function/FunctionInvokerTests.java | 56 ++++++++++++++++++- .../GreenfieldFunctionEnableBindingTests.java | 2 +- .../SourceToFunctionsSupportTests.java | 4 +- 8 files changed, 114 insertions(+), 45 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index e6fad5acc..630286cdd 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.binder; +import java.lang.reflect.Field; import java.util.LinkedHashMap; import java.util.Map; import java.util.function.Consumer; @@ -65,6 +66,7 @@ import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.retry.RecoveryCallback; import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; /** @@ -345,7 +347,7 @@ public abstract class AbstractMessageChannelBinder implements Function>, Flux 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((x, y) -> onError(x, (Message) y)) - .transform(this.userFunction::apply) // invoke user function - .onErrorContinue((x, y) -> onError(x, (Message) y)) - .map(resultMessage -> toMessage(resultMessage, originalMessageRef.get())); // create output message + + return input.concatMap(message -> { + return Flux.just(message) + .doOnNext(originalMessageRef::set) + .map(this::resolveArgument) + .transform(this.userFunction::apply) + .retryBackoff(consumerProperties.getMaxAttempts(), + Duration.ofMillis(consumerProperties.getBackOffInitialInterval()), + Duration.ofMillis(consumerProperties.getBackOffMaxInterval())) + .onErrorResume(e -> { + onError(e, originalMessageRef.get()); + return Mono.empty(); + }); + }).log().map(resultMessage -> toMessage(resultMessage, originalMessageRef.get())); // create output message } private void onError(Throwable t, Message originalMessage) { @@ -138,7 +151,7 @@ class FunctionInvoker implements Function>, Flux message) { return !this.inputClass.isAssignableFrom(Message.class) && - !message.getPayload().getClass().isAssignableFrom(this.inputClass) && + !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 index dea9939a0..f987be8dc 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 @@ -30,9 +30,6 @@ 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.converter.CompositeMessageConverterFactory; -import org.springframework.cloud.stream.messaging.Processor; -import org.springframework.cloud.stream.messaging.Sink; -import org.springframework.cloud.stream.messaging.Source; import org.springframework.integration.dsl.IntegrationFlowBuilder; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.messaging.Message; @@ -62,15 +59,6 @@ public class IntegrationFlowFunctionSupport { @Autowired private MessageChannel errorChannel; - @Autowired(required = false) - private Source source; - - @Autowired(required = false) - private Processor processor; - - @Autowired(required = false) - private Sink sink; - /** * @param functionCatalog * @param functionInspector @@ -162,7 +150,7 @@ public class IntegrationFlowFunctionSupport { MessageChannel outputChannel) { IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(inputChannel).bridge(); - if (!this.andThenFunction(flowBuilder, outputChannel, this.functionProperties.getDefinition())) { + if (!this.andThenFunction(flowBuilder, outputChannel, this.functionProperties)) { flowBuilder = flowBuilder.channel(outputChannel); } return flowBuilder; @@ -182,17 +170,17 @@ public class IntegrationFlowFunctionSupport { * @return true if {@link Function} was located and added and false if it wasn't. */ public boolean andThenFunction(IntegrationFlowBuilder flowBuilder, MessageChannel outputChannel, - String functionName) { - return andThenFunction(flowBuilder.toReactivePublisher(), outputChannel, functionName); + StreamFunctionProperties functionProperties) { + return andThenFunction(flowBuilder.toReactivePublisher(), outputChannel, functionProperties); } public boolean andThenFunction(Publisher publisher, MessageChannel outputChannel, - String functionName) { - if (!StringUtils.hasText(functionName)) { + StreamFunctionProperties functionProperties) { + if (!StringUtils.hasText(functionProperties.getDefinition())) { return false; } FunctionInvoker functionInvoker = - new FunctionInvoker<>(functionName, this.functionCatalog, + new FunctionInvoker<>(functionProperties, this.functionCatalog, this.functionInspector, this.messageConverterFactory, this.errorChannel); if (outputChannel != null) { 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 53d1bf195..0fe2be729 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 @@ -17,6 +17,7 @@ package org.springframework.cloud.stream.function; import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.cloud.stream.binder.ConsumerProperties; /** * @@ -32,6 +33,8 @@ public class StreamFunctionProperties { */ private String definition; + private ConsumerProperties consumerProperties; + public String getDefinition() { return this.definition; } @@ -39,4 +42,8 @@ public class StreamFunctionProperties { public void setDefinition(String definition) { this.definition = definition; } + + protected ConsumerProperties getConsumerProperties() { + return consumerProperties; + } } 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 index e53e346ee..d53631fff 100644 --- 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 @@ -16,6 +16,7 @@ package org.springframework.cloud.stream.function; +import java.lang.reflect.Field; import java.util.function.Function; import org.junit.Test; @@ -27,6 +28,7 @@ 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.ConsumerProperties; import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.context.ConfigurableApplicationContext; @@ -34,6 +36,7 @@ import org.springframework.context.annotation.Bean; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.ReflectionUtils; import static org.assertj.core.api.Assertions.assertThat; @@ -53,20 +56,49 @@ public class FunctionInvokerTests { Message inputMessage = new GenericMessage<>(new Foo()); - FunctionInvoker messageToMessageSameType = new FunctionInvoker<>("messageToMessageSameType", + StreamFunctionProperties functionProperties = createStreamFunctionProperties(); + + functionProperties.setDefinition("messageToMessageSameType"); + FunctionInvoker messageToMessageSameType = new FunctionInvoker<>(functionProperties, 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", + functionProperties.setDefinition("pojoToPojoSameType"); + FunctionInvoker pojoToPojoSameType = new FunctionInvoker<>(functionProperties, 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()); - FunctionInvoker messageToMessageNoType = new FunctionInvoker<>("messageToMessageNoType", + + functionProperties.setDefinition("messageToMessageNoType"); + FunctionInvoker messageToMessageNoType = new FunctionInvoker<>(functionProperties, new FunctionCatalogWrapper(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, + new FunctionCatalogWrapper(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); + } + } + + private StreamFunctionProperties createStreamFunctionProperties() { + StreamFunctionProperties functionProperties = new StreamFunctionProperties(); + ConsumerProperties consumerProperties = new ConsumerProperties(); + consumerProperties.setMaxAttempts(3); + try { + Field f = ReflectionUtils.findField(StreamFunctionProperties.class, "consumerProperties"); + f.setAccessible(true); + f.set(functionProperties, consumerProperties); + return functionProperties; + } + catch (Exception e) { + throw new IllegalStateException(e); } } @@ -98,12 +130,30 @@ public class FunctionInvokerTests { 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; + } + }; + } + } private static class Foo { } + private static class ErrorFoo extends Foo { + + } + private static class Bar { } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java index 31e84f617..58fa36cd7 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java @@ -132,7 +132,7 @@ public class GreenfieldFunctionEnableBindingTests { ObjectMapper mapper = context.getBean(ObjectMapper.class); input.send(MessageBuilder.withPayload("bar").build()); - byte[] payload = target.receive(10000).getPayload(); + byte[] payload = target.receive(2000).getPayload(); Foo result = mapper.readValue(payload, Foo.class); 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 772a5a93c..5b47bf548 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 @@ -101,7 +101,7 @@ public class SourceToFunctionsSupportTests { PollableChannel errorChannel = context.getBean("errorChannel", PollableChannel.class); OutputDestination target = context.getBean(OutputDestination.class); assertNull(target.receive(1000)); - assertNotNull(errorChannel.receive(1000)); + assertNotNull(errorChannel.receive(10000)); } } @@ -115,7 +115,7 @@ public class SourceToFunctionsSupportTests { PollableChannel errorChannel = context.getBean("errorChannel", PollableChannel.class); OutputDestination target = context.getBean(OutputDestination.class); assertNull(target.receive(1000)); - assertNotNull(errorChannel.receive(1000)); + assertNotNull(errorChannel.receive(10000)); } }