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

Add get*Names() to FunctionCatalog

Browse Source 
We could go wit a separate interface, but it seems sensible to
keep it together really, so that callers don't have to downcast.
This commit is contained in:
Dave Syer
2017-06-29 10:25:31 +01:00
parent 2dd7dfb900
commit da0c954135
7 changed files with 69 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -25,7 +25,6 @@ import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionOutcome;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
@@ -63,10 +62,8 @@ public class StreamConfiguration {
@Bean
public SupplierInvokingMessageProducer<Object> supplierInvoker(
ListableBeanFactory beanFactory, FunctionCatalog registry) {
String[] names = beanFactory.getBeanNamesForType(Supplier.class, false,
false);
return new SupplierInvokingMessageProducer<Object>(registry, names);
FunctionCatalog registry) {
return new SupplierInvokingMessageProducer<Object>(registry);
}
}
@@ -78,14 +75,11 @@ public class StreamConfiguration {
private StreamConfigurationProperties properties;
@Bean
public StreamListeningFunctionInvoker functionInvoker(
ListableBeanFactory beanFactory, FunctionCatalog registry,
public StreamListeningFunctionInvoker functionInvoker(FunctionCatalog registry,
FunctionInspector functionInspector,
@Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) {
String[] names = beanFactory.getBeanNamesForType(Function.class, false,
false);
return new StreamListeningFunctionInvoker(registry, functionInspector,
compositeMessageConverterFactory, properties.getEndpoint(), names);
compositeMessageConverterFactory, properties.getEndpoint());
}
}
@@ -97,14 +91,11 @@ public class StreamConfiguration {
private StreamConfigurationProperties properties;
@Bean
public StreamListeningConsumerInvoker consumerInvoker(
ListableBeanFactory beanFactory, FunctionCatalog registry,
public StreamListeningConsumerInvoker consumerInvoker(FunctionCatalog registry,
FunctionInspector functionInspector,
@Lazy CompositeMessageConverterFactory compositeMessageConverterFactory) {
String[] names = beanFactory.getBeanNamesForType(Consumer.class, false,
false);
return new StreamListeningConsumerInvoker(registry, functionInspector,
compositeMessageConverterFactory, properties.getEndpoint(), names);
compositeMessageConverterFactory, properties.getEndpoint());
}
}

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

@@ -16,6 +16,7 @@
package org.springframework.cloud.function.stream;
import java.util.Set;
import java.util.function.Function;
import org.springframework.beans.factory.SmartInitializingSingleton;
@@ -46,19 +47,15 @@ public class StreamListeningConsumerInvoker implements SmartInitializingSingleto
private final String defaultEndpoint;
private final String[] names;
private static final String NOENDPOINT = "__NOENDPOINT__";
public StreamListeningConsumerInvoker(FunctionCatalog functionCatalog,
FunctionInspector functionInspector,
CompositeMessageConverterFactory converterFactory, String defaultEndpoint,
String... names) {
CompositeMessageConverterFactory converterFactory, String defaultEndpoint) {
this.functionCatalog = functionCatalog;
this.functionInspector = functionInspector;
this.converterFactory = converterFactory;
this.defaultEndpoint = defaultEndpoint;
this.names = names;
}
@Override
@@ -81,8 +78,9 @@ public class StreamListeningConsumerInvoker implements SmartInitializingSingleto
private String select(Message<?> input) {
String name = defaultEndpoint;
if (name == null) {
if (names.length == 1) {
name = names[0];
Set<String> names = functionCatalog.getConsumerNames();
if (names.size() == 1) {
name = names.iterator().next();
}
else {
for (String candidate : names) {

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

@@ -16,6 +16,7 @@
package org.springframework.cloud.function.stream;
import java.util.Set;
import java.util.function.Function;
import org.springframework.beans.factory.SmartInitializingSingleton;
@@ -48,19 +49,15 @@ public class StreamListeningFunctionInvoker implements SmartInitializingSingleto
private final String defaultEndpoint;
private final String[] names;
private static final String NOENDPOINT = "__NOENDPOINT__";
public StreamListeningFunctionInvoker(FunctionCatalog functionCatalog,
FunctionInspector functionInspector,
CompositeMessageConverterFactory converterFactory, String defaultEndpoint,
String... names) {
CompositeMessageConverterFactory converterFactory, String defaultEndpoint) {
this.functionCatalog = functionCatalog;
this.functionInspector = functionInspector;
this.converterFactory = converterFactory;
this.defaultEndpoint = defaultEndpoint;
this.names = names;
}
@Override
@@ -84,8 +81,9 @@ public class StreamListeningFunctionInvoker implements SmartInitializingSingleto
private String select(Message<?> input) {
String name = defaultEndpoint;
if (name == null) {
if (names.length == 1) {
name = names[0];
Set<String> names = functionCatalog.getFunctionNames();
if (names.size() == 1) {
name = names.iterator().next();
}
else {
for (String candidate : names) {

7
spring-cloud-function-stream/src/main/java/org/springframework/cloud/function/stream/SupplierInvokingMessageProducer.java
View File

@@ -33,11 +33,8 @@ public class SupplierInvokingMessageProducer<T> extends MessageProducerSupport {
private final FunctionCatalog functionCatalog;
private final String[] names;
public SupplierInvokingMessageProducer(FunctionCatalog registry, String... names) {
public SupplierInvokingMessageProducer(FunctionCatalog registry) {
this.functionCatalog = registry;
this.names = names;
this.setOutputChannelName(Source.OUTPUT);
}
@@ -50,7 +47,7 @@ public class SupplierInvokingMessageProducer<T> extends MessageProducerSupport {
private Flux<?> supplier() {
Supplier<Flux<?>> supplier = null;
Flux<?> result = Flux.empty();
for (String name : names) {
for (String name : functionCatalog.getSupplierNames()) {
supplier = functionCatalog.lookupSupplier(name);
Assert.notNull(supplier, "Supplier must not be null");
result = Flux.merge(result, supplier.get());