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

GH-1141 Add support for composing reactive Supplier/Function with imperative Consumer

Browse Source 
Resolves #1141
This commit is contained in:
Oleg Zhurakousky
2024-06-06 11:38:19 +02:00
parent faeb0f77ad
commit 6ffae9397a
2 changed files with 101 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java
View File

@@ -394,7 +394,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
@SuppressWarnings("rawtypes")
public class FunctionInvocationWrapper implements Function<Object, Object>, Consumer<Object>, Supplier<Object>, Runnable {
private final Object target;
private Object target;
private Type inputType;
@@ -658,13 +658,21 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
|| FunctionTypeUtils.isMultipleArgumentType(((FunctionInvocationWrapper) after).outputType)) {
throw new UnsupportedOperationException("Composition of functions with multiple arguments is not supported at the moment");
}
FunctionInvocationWrapper afterWrapper = (FunctionInvocationWrapper) after;
//see GH-1141 for this code snippet
if ((this.getTarget() instanceof Supplier || this.getTarget() instanceof Function) && FunctionTypeUtils.isPublisher(this.getOutputType())
&& afterWrapper.getTarget() instanceof Consumer && !FunctionTypeUtils.isPublisher(afterWrapper.getInputType())) {
Consumer wrapper = new ConsumerWrapper((Consumer) afterWrapper.getTarget());
afterWrapper.target = wrapper;
afterWrapper.inputType = this.outputType;
}
//
this.setSkipOutputConversion(true);
((FunctionInvocationWrapper) after).setSkipOutputConversion(true);
Function rawComposedFunction = v -> ((FunctionInvocationWrapper) after).doApply(doApply(v));
FunctionInvocationWrapper afterWrapper = (FunctionInvocationWrapper) after;
Type composedFunctionType;
if (afterWrapper.outputType == null) {
composedFunctionType = (this.inputType == null) ?
@@ -1551,4 +1559,20 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
return t;
}
}
@SuppressWarnings({ "unchecked", "rawtypes" })
private static class ConsumerWrapper implements Consumer<Flux<Object>> {
private final Consumer targetConsumer;
ConsumerWrapper(Consumer targetConsumer) {
this.targetConsumer = targetConsumer;
}
@Override
public void accept(Flux messageFlux) {
messageFlux.doOnNext(this.targetConsumer).subscribe();
}
}
}