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

Introduce special handling for unconverted messages

Browse Source 
Prevents odd looking exceptions when null pops into the stream
unexpectedly.
This commit is contained in:
Dave Syer
2017-10-06 07:21:13 +02:00
parent 08f2cb2491
commit 30b0d3a1c6
1 changed files with 11 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamListeningFunctionInvoker.java
View File

@@ -63,6 +63,8 @@ public class StreamListeningFunctionInvoker implements SmartInitializingSingleto
private static final FluxMessageProcessor NOENDPOINT = flux -> Flux.empty();
private static final Object UNCONVERTED = new Object();
public StreamListeningFunctionInvoker(FunctionCatalog functionCatalog,
FunctionInspector functionInspector,
CompositeMessageConverterFactory converterFactory, String defaultEndpoint) {
@@ -105,7 +107,8 @@ public class StreamListeningFunctionInvoker implements SmartInitializingSingleto
private Flux<Message<?>> consumer(String name, Flux<Message<?>> flux) {
Consumer<Object> consumer = functionCatalog.lookupConsumer(name);
consumer.accept(flux.map(message -> convertInput(consumer).apply(message)));
consumer.accept(flux.map(message -> convertInput(consumer).apply(message))
.filter(transformed -> transformed != UNCONVERTED));
return Flux.empty();
}
@@ -200,12 +203,17 @@ public class StreamListeningFunctionInvoker implements SmartInitializingSingleto
}
private Object convertPayload(Class<?> inputType, Message<?> m) {
Object result;
if (inputType.isAssignableFrom(m.getPayload().getClass())) {
return m.getPayload();
result = m.getPayload();
}
else {
return this.converter.fromMessage(m, inputType);
result = this.converter.fromMessage(m, inputType);
}
if (result==null) {
result = UNCONVERTED;
}
return result;
}
interface FluxMessageProcessor {