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

stream consumer invoker target now expects Flux

Browse Source
This commit is contained in:
markfisher
2017-03-31 12:09:04 -04:00
parent 0b49edc2fd
commit d208b5e0f3
2 changed files with 4 additions and 4 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

@@ -101,7 +101,7 @@ public class StreamConfiguration {
@ConditionalOnProperty("spring.cloud.stream.bindings.input.destination")
public StreamListeningConsumerInvoker<Object> invoker(FunctionCatalog registry) {
String name = properties.getEndpoint();
Consumer<Object> consumer = registry.lookupConsumer(name);
Consumer<Flux<Object>> consumer = registry.lookupConsumer(name);
return new StreamListeningConsumerInvoker<Object>(consumer);
}
}

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

@@ -29,14 +29,14 @@ import reactor.core.publisher.Flux;
*/
public class StreamListeningConsumerInvoker<T> {
private final Consumer<T> consumer;
private final Consumer<Flux<T>> consumer;
public StreamListeningConsumerInvoker(Consumer<T> consumer) {
public StreamListeningConsumerInvoker(Consumer<Flux<T>> consumer) {
this.consumer = consumer;
}
@StreamListener
public void handle(@Input(Processor.INPUT) Flux<T> input) {
input.subscribe(this.consumer);
this.consumer.accept(input);
}
}