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

Tentatively fix streaming issues

Browse Source 
* Assume that the Function returned by FunctionCatalog
  is already wrapped as Function<Flux<?>,Flux<?>>
This commit is contained in:
Marius Bogoevici
2017-04-26 15:04:32 -04:00
committed by Dave Syer
parent f02f2eaf95
commit e117cfd5bd
4 changed files with 159 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/StreamConfiguration.java
View File

@@ -89,7 +89,7 @@ public class StreamConfiguration {
public AbstractFunctionInvoker<?, ?> invoker(FunctionCatalog registry, FunctionInspector functionInspector,
@Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) {
String name = properties.getEndpoint();
Function<Object, Object> function = registry.lookupFunction(name);
Function<Flux<?>, Flux<?>> function = registry.lookupFunction(name);
Assert.notNull(function, "no such function: " + name);
return new StreamListeningFunctionInvoker(name, function, functionInspector,
compositeMessageConverterFactory);

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

@@ -23,8 +23,6 @@ import reactor.core.publisher.Flux;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.cloud.function.context.FunctionInspector;
import org.springframework.cloud.function.invoker.AbstractFunctionInvoker;
import org.springframework.cloud.function.support.FluxFunction;
import org.springframework.cloud.function.support.FunctionUtils;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.annotation.StreamListener;
@@ -32,7 +30,6 @@ import org.springframework.cloud.stream.converter.CompositeMessageConverterFacto
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
@@ -51,9 +48,9 @@ public class StreamListeningFunctionInvoker extends AbstractFunctionInvoker<Flux
private Class<?> inputType;
public StreamListeningFunctionInvoker(String name, Function<?, ?> function, FunctionInspector functionInspector,
public StreamListeningFunctionInvoker(String name, Function<Flux<?>, Flux<?>> function, FunctionInspector functionInspector,
CompositeMessageConverterFactory converterFactory) {
super(wrapIfNecessary(function));
super(function);
this.name = name;
this.functionInspector = functionInspector;
this.converterFactory = converterFactory;
@@ -71,15 +68,6 @@ public class StreamListeningFunctionInvoker extends AbstractFunctionInvoker<Flux
return this.doInvoke(input.map(convertInput()));
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private static Function<Flux<?>, Flux<?>> wrapIfNecessary(Function function) {
Assert.notNull(function, "Function must not be null");
if (!FunctionUtils.isFluxFunction(function)) {
function = new FluxFunction(function);
}
return function;
}
private Function<Message<?>, Object> convertInput() {
return m -> {
if (this.inputType.isAssignableFrom(m.getPayload().getClass())) {