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

Polished error handling in FunctionInvoker

Browse Source
This commit is contained in:
Oleg Zhurakousky
2018-08-20 20:19:14 +02:00
parent 0fc9d09baf
commit c73f311a12
4 changed files with 98 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -21,8 +21,10 @@ import org.springframework.util.Assert;
/**
* @author David Turanski
*
* @since 2.1
**/
public class FunctionCatalogWrapper {
class FunctionCatalogWrapper {
private final FunctionCatalog catalog;

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

@@ -28,7 +28,9 @@ 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.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.Assert;
/**
@@ -51,23 +53,41 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
private final CompositeMessageConverter messageConverter;
private final MessageChannel errorChannel;
FunctionInvoker(String functionName, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory compositeMessageConverterFactory) {
this(functionName, functionCatalog, functionInspector, compositeMessageConverterFactory, null);
}
FunctionInvoker(String functionName, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory compositeMessageConverterFactory, MessageChannel errorChannel) {
this.userFunction = functionCatalog.lookup(functionName);
Assert.isInstanceOf(Function.class, this.userFunction);
this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
FunctionType functionType = functionInspector.getRegistration(this.userFunction).getType();
this.inputClass = functionType.getInputType();
this.errorChannel = errorChannel;
}
@Override
public Flux<Message<O>> apply(Flux<Message<I>> input) {
AtomicReference<Message<I>> originalMessage = new AtomicReference<>();
AtomicReference<Message<I>> originalMessageRef = new AtomicReference<>();
return input
.doOnNext(originalMessage::set) // to preserve the original message
.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();
}
})
.transform(this.userFunction::apply) // invoke user function
.map(resultMessage -> toMessage(resultMessage, originalMessage.get())); // create output message
.map(resultMessage -> toMessage(resultMessage, originalMessageRef.get())); // create output message
}
@SuppressWarnings("unchecked")
@@ -87,9 +107,12 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
logger.debug("Resolving input argument from message: " + message);
}
return (T) (shouldConvertFromMessage(message)
T argument = (T) (shouldConvertFromMessage(message)
? this.messageConverter.fromMessage(message, this.inputClass)
: message);
Assert.notNull(argument, "Failed to resolve argument type '" + this.inputClass + "' from message: " + message );
return argument;
}
private boolean shouldConvertFromMessage(Message<?> message) {

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

@@ -25,6 +25,7 @@ import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.cloud.function.core.FluxSupplier;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
@@ -52,6 +53,9 @@ public class IntegrationFlowFunctionSupport {
private final FunctionProperties functionProperties;
@Autowired
private MessageChannel errorChannel;
/**
* @param functionCatalog
* @param functionInspector
@@ -115,7 +119,7 @@ public class IntegrationFlowFunctionSupport {
if (StringUtils.hasText(this.functionProperties.getName())) {
FunctionInvoker<I,O> functionInvoker =
new FunctionInvoker<>(this.functionProperties.getName(), this.functionCatalog,
this.functionInspector, this.messageConverterFactory);
this.functionInspector, this.messageConverterFactory, this.errorChannel);
subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), outputChannel::send);
return true;

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

@@ -39,11 +39,13 @@ import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.util.Assert;
import org.springframework.util.MimeTypeUtils;
@@ -52,6 +54,8 @@ import static org.hamcrest.CoreMatchers.allOf;
import static org.hamcrest.CoreMatchers.endsWith;
import static org.hamcrest.CoreMatchers.isA;
import static org.hamcrest.Matchers.hasProperty;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
/**
* @author Oleg Zhurakousky
@@ -87,6 +91,21 @@ public class SourceToFunctionsSupportTests {
}
}
@Test
public void testFailedInputTypeConversion() {
try (ConfigurableApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoContentType.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 testMessageSourceIsCreatedFromProvidedSupplier() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
@@ -140,7 +159,7 @@ public class SourceToFunctionsSupportTests {
allOf(isA(BeanInstantiationException.class), hasProperty("cause", isA(IllegalArgumentException.class)),
hasProperty("message", endsWith("'doesNotExist' cannot be located."))));
ConfigurableApplicationContext context = new SpringApplicationBuilder(
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(
WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=doesNotExist", "--spring.jmx.enabled=false");
@@ -185,6 +204,27 @@ public class SourceToFunctionsSupportTests {
}
@EnableAutoConfiguration
@Import(ExistingMessageSourceConfigurationNoContentTypeSet.class)
public static class FunctionsConfigurationNoContentType {
@Bean
public PollableChannel errorChannel() {
return new QueueChannel(10);
}
@Bean
public Function<String, String> toUpperCase() {
return String::toUpperCase;
}
@Bean
public Function<String, String> concatWithSelf() {
return x -> x + ":" + x;
}
}
/**
* This configuration essentially emulates our existing app-starters for Sources
* and essentially demonstrates how a function(s) could be applied to an existing
@@ -213,6 +253,28 @@ public class SourceToFunctionsSupportTests {
}
@EnableBinding(Source.class)
public static class ExistingMessageSourceConfigurationNoContentTypeSet {
@Autowired
private Source source;
@Bean
public IntegrationFlow messageSourceFlow(IntegrationFlowFunctionSupport functionSupport) {
Supplier<Message<String>> messageSource = () -> MessageBuilder.withPayload("hello function")
.setHeader(MessageHeaders.CONTENT_TYPE, "application/octet-stream").build();
IntegrationFlowBuilder flowBuilder = functionSupport.integrationFlowFromProvidedSupplier(messageSource);
if (!functionSupport.andThenFunction(flowBuilder, this.source.output())) {
flowBuilder = flowBuilder.channel(this.source.output());
}
return flowBuilder.get();
}
}
@EnableBinding(Source.class)
public static class ProvidedMessageSourceConfiguration {