SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1498 Added initial retry logic to reactive pipeline

Browse Source 
Resolves #1498
This commit is contained in:
Oleg Zhurakousky
2018-11-15 18:28:59 +01:00
parent 1d8b92c85f
commit 8ae3ec0041
8 changed files with 114 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
View File

@@ -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<C extends ConsumerProperties,
ConsumerDestination destination = this.provisioningProvider.provisionConsumerDestination(name, group, properties);
// the function support for the inbound channel is only for Sink
if (this.streamFunctionProperties != null && StringUtils.hasText(this.streamFunctionProperties.getDefinition()) && this.processor == null) {
inputChannel = this.postProcessInboundChannelForFunction(inputChannel);
inputChannel = this.postProcessInboundChannelForFunction(inputChannel, (ConsumerProperties) properties);
}
if (HeaderMode.embeddedHeaders.equals(properties.getHeaderMode())) {
enhanceMessageChannel(inputChannel);
@@ -823,14 +825,14 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
moveChannelInterceptors((AbstractMessageChannel) outputChannel, actualOutputChannel);
}
this.integrationFlowFunctionSupport.andThenFunction(publisher, actualOutputChannel,
this.streamFunctionProperties.getDefinition());
this.streamFunctionProperties);
return actualOutputChannel;
}
}
return (SubscribableChannel) outputChannel;
}
private SubscribableChannel postProcessInboundChannelForFunction(MessageChannel inputChannel) {
private SubscribableChannel postProcessInboundChannelForFunction(MessageChannel inputChannel, ConsumerProperties consumerProperties) {
if (this.integrationFlowFunctionSupport != null &&
(this.integrationFlowFunctionSupport.containsFunction(Consumer.class) ||
this.integrationFlowFunctionSupport.containsFunction(Function.class))) {
@@ -838,13 +840,26 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
if (inputChannel instanceof AbstractMessageChannel) {
moveChannelInterceptors((AbstractMessageChannel) inputChannel, actualInputChannel);
}
this.propagateConsumerPropertiesToFunction(consumerProperties);
this.integrationFlowFunctionSupport.andThenFunction(MessageChannelReactiveUtils.toPublisher(actualInputChannel),
inputChannel, this.streamFunctionProperties.getDefinition());
inputChannel, this.streamFunctionProperties);
return actualInputChannel;
}
return (SubscribableChannel) inputChannel;
}
// we're doing it reflectively so we don't expose this as a property to the user
private void propagateConsumerPropertiesToFunction(ConsumerProperties consumerProperties) {
try {
Field f = ReflectionUtils.findField(StreamFunctionProperties.class, "consumerProperties");
f.setAccessible(true);
f.set(this.streamFunctionProperties, consumerProperties);
}
catch (Exception e) {
throw new IllegalStateException(e);
}
}
private void moveChannelInterceptors(AbstractMessageChannel existingMessageChannel,
AbstractMessageChannel actualMessageChannel) {
for (ChannelInterceptor channelInterceptor : existingMessageChannel.getChannelInterceptors()) {

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -27,7 +27,6 @@ import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -50,9 +49,6 @@ public class FunctionConfiguration {
@Autowired(required = false)
private Processor processor;
@Autowired(required = false)
private Sink sink;
@Bean
public IntegrationFlowFunctionSupport functionSupport(FunctionCatalogWrapper functionCatalog,
FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory,

39
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java
View File

@@ -16,6 +16,7 @@
package org.springframework.cloud.stream.function;
import java.time.Duration;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
@@ -23,9 +24,11 @@ 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.FunctionType;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
@@ -61,14 +64,16 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
private final boolean isInputArgumentMessage;
FunctionInvoker(String functionName, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
private final ConsumerProperties consumerProperties;
FunctionInvoker(StreamFunctionProperties functionProperties, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory compositeMessageConverterFactory) {
this(functionName, functionCatalog, functionInspector, compositeMessageConverterFactory, null);
this(functionProperties, functionCatalog, functionInspector, compositeMessageConverterFactory, null);
}
FunctionInvoker(String functionName, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
FunctionInvoker(StreamFunctionProperties functionProperties, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory compositeMessageConverterFactory, MessageChannel errorChannel) {
this.userFunction = functionCatalog.lookup(functionName);
this.userFunction = functionCatalog.lookup(functionProperties.getDefinition());
Assert.isInstanceOf(Function.class, this.userFunction);
this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
FunctionType functionType = functionInspector.getRegistration(this.userFunction).getType();
@@ -76,19 +81,27 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
this.inputClass = functionType.getInputType();
this.outputClass = functionType.getOutputType();
this.errorChannel = errorChannel;
this.consumerProperties = functionProperties.getConsumerProperties() == null
? new ConsumerProperties() : functionProperties.getConsumerProperties();
}
@SuppressWarnings("unchecked")
@Override
public Flux<Message<O>> apply(Flux<Message<I>> input) {
AtomicReference<Message<I>> 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<I>) y))
.transform(this.userFunction::apply) // invoke user function
.onErrorContinue((x, y) -> onError(x, (Message<I>) 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<I> originalMessage) {
@@ -138,7 +151,7 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
private boolean shouldConvertFromMessage(Message<?> message) {
return !this.inputClass.isAssignableFrom(Message.class) &&
!message.getPayload().getClass().isAssignableFrom(this.inputClass) &&
!this.inputClass.isAssignableFrom(message.getPayload().getClass()) &&
!this.inputClass.isAssignableFrom(Object.class);
}

24
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java
View File

@@ -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 <I,O> boolean andThenFunction(IntegrationFlowBuilder flowBuilder, MessageChannel outputChannel,
String functionName) {
return andThenFunction(flowBuilder.toReactivePublisher(), outputChannel, functionName);
StreamFunctionProperties functionProperties) {
return andThenFunction(flowBuilder.toReactivePublisher(), outputChannel, functionProperties);
}
public <I,O> boolean andThenFunction(Publisher<?> publisher, MessageChannel outputChannel,
String functionName) {
if (!StringUtils.hasText(functionName)) {
StreamFunctionProperties functionProperties) {
if (!StringUtils.hasText(functionProperties.getDefinition())) {
return false;
}
FunctionInvoker<I, O> functionInvoker =
new FunctionInvoker<>(functionName, this.functionCatalog,
new FunctionInvoker<>(functionProperties, this.functionCatalog,
this.functionInspector, this.messageConverterFactory, this.errorChannel);
if (outputChannel != null) {

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java
View File

@@ -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;
}
}

56
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java
View File

@@ -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<Foo> inputMessage = new GenericMessage<>(new Foo());
FunctionInvoker<Foo, Foo> messageToMessageSameType = new FunctionInvoker<>("messageToMessageSameType",
StreamFunctionProperties functionProperties = createStreamFunctionProperties();
functionProperties.setDefinition("messageToMessageSameType");
FunctionInvoker<Foo, Foo> messageToMessageSameType = new FunctionInvoker<>(functionProperties,
new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)), context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class));
Message<Foo> outputMessage = messageToMessageSameType.apply(Flux.just(inputMessage)).blockFirst();
assertThat(inputMessage).isSameAs(outputMessage);
FunctionInvoker<Foo, Foo> pojoToPojoSameType = new FunctionInvoker<>("pojoToPojoSameType",
functionProperties.setDefinition("pojoToPojoSameType");
FunctionInvoker<Foo, Foo> 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<Foo, Foo> messageToMessageNoType = new FunctionInvoker<>("messageToMessageNoType",
functionProperties.setDefinition("messageToMessageNoType");
FunctionInvoker<Foo, Foo> 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<Foo, Foo> withException = new FunctionInvoker<>(functionProperties,
new FunctionCatalogWrapper(context.getBean(FunctionCatalog.class)), context.getBean(FunctionInspector.class), context.getBean(CompositeMessageConverterFactory.class));
Flux<Message<Foo>> fluxOfMessages = Flux.just(new GenericMessage<>(new ErrorFoo()), inputMessage);
Message<Foo> 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<Foo, Foo> 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 {
}

2
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/GreenfieldFunctionEnableBindingTests.java
View File

@@ -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);

4
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
View File

@@ -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));
}
}