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

Bypassing target type binding

Browse Source 
When functional model is used and no EnableBinding is provided, this PR allows
core Spring Cloud Stream to skip any target type binding if the downstream
binder is capable of such binding on the target type. In that case, the target
binders provide their own BindableTargetProxyFactory.

Introduce additional properties in StreamFunction properties to allow overriding the
input/output bindings.

Resolves #1751
This commit is contained in:
Soby Chacko
2019-06-28 15:29:07 -04:00
committed by Oleg Zhurakousky
parent 35dff8762a
commit 8af53028f5
6 changed files with 351 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java Normal file
View File

@@ -0,0 +1,188 @@
/*
* Copyright 2019-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binding;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.internal.InternalPropertyNames;
import org.springframework.util.StringUtils;
/**
* Base class for bindable proxy factories. This class is mainly refactored from the
* {@link BindableProxyFactory} so that other downstream binders who want to bind their own
* targets can make use of it.
*
* Original authors in {@link BindableProxyFactory}
* @author Soby Chacko
* @since 3.0.0
*/
public class AbstractBindableProxyFactory implements Bindable {
private static Log log = LogFactory.getLog(AbstractBindableProxyFactory.class);
@Value("${" + InternalPropertyNames.NAMESPACE_PROPERTY_NAME + ":}")
private String namespace;
@Autowired
protected Map<String, BindingTargetFactory> bindingTargetFactories;
protected Map<String, BoundTargetHolder> inputHolders = new LinkedHashMap<>();
protected Map<String, BoundTargetHolder> outputHolders = new LinkedHashMap<>();
protected Class<?> type;
public AbstractBindableProxyFactory(Class<?> type) {
this.type = type;
}
protected BindingTargetFactory getBindingTargetFactory(Class<?> bindingTargetType) {
List<String> candidateBindingTargetFactories = new ArrayList<>();
for (Map.Entry<String, BindingTargetFactory> bindingTargetFactoryEntry : this.bindingTargetFactories
.entrySet()) {
if (bindingTargetFactoryEntry.getValue().canCreate(bindingTargetType)) {
candidateBindingTargetFactories.add(bindingTargetFactoryEntry.getKey());
}
}
if (candidateBindingTargetFactories.size() == 1) {
return this.bindingTargetFactories
.get(candidateBindingTargetFactories.get(0));
}
else {
if (candidateBindingTargetFactories.size() == 0) {
throw new IllegalStateException(
"No factory found for binding target type: "
+ bindingTargetType.getName()
+ " among registered factories: "
+ StringUtils.collectionToCommaDelimitedString(
this.bindingTargetFactories.keySet()));
}
else {
throw new IllegalStateException(
"Multiple factories found for binding target type: "
+ bindingTargetType.getName() + ": "
+ StringUtils.collectionToCommaDelimitedString(
candidateBindingTargetFactories));
}
}
}
@Override
public Collection<Binding<Object>> createAndBindInputs(
BindingService bindingService) {
List<Binding<Object>> bindings = new ArrayList<>();
if (log.isDebugEnabled()) {
log.debug(
String.format("Binding inputs for %s:%s", this.namespace, this.type));
}
for (Map.Entry<String, BoundTargetHolder> boundTargetHolderEntry : this.inputHolders
.entrySet()) {
String inputTargetName = boundTargetHolderEntry.getKey();
BoundTargetHolder boundTargetHolder = boundTargetHolderEntry.getValue();
if (boundTargetHolder.isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.namespace, this.type,
inputTargetName));
}
bindings.addAll(bindingService.bindConsumer(
boundTargetHolder.getBoundTarget(), inputTargetName));
}
}
return bindings;
}
@Override
public Collection<Binding<Object>> createAndBindOutputs(
BindingService bindingService) {
List<Binding<Object>> bindings = new ArrayList<>();
if (log.isDebugEnabled()) {
log.debug(String.format("Binding outputs for %s:%s", this.namespace,
this.type));
}
for (Map.Entry<String, BoundTargetHolder> boundTargetHolderEntry : this.outputHolders
.entrySet()) {
BoundTargetHolder boundTargetHolder = boundTargetHolderEntry.getValue();
String outputTargetName = boundTargetHolderEntry.getKey();
if (boundTargetHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.namespace, this.type,
outputTargetName));
}
bindings.add(bindingService.bindProducer(
boundTargetHolder.getBoundTarget(), outputTargetName));
}
}
return bindings;
}
@Override
public void unbindInputs(BindingService bindingService) {
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding inputs for %s:%s", this.namespace,
this.type));
}
for (Map.Entry<String, BoundTargetHolder> boundTargetHolderEntry : this.inputHolders
.entrySet()) {
if (boundTargetHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding %s:%s:%s", this.namespace,
this.type, boundTargetHolderEntry.getKey()));
}
bindingService.unbindConsumers(boundTargetHolderEntry.getKey());
}
}
}
@Override
public void unbindOutputs(BindingService bindingService) {
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding outputs for %s:%s", this.namespace,
this.type));
}
for (Map.Entry<String, BoundTargetHolder> boundTargetHolderEntry : this.outputHolders
.entrySet()) {
if (boundTargetHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.namespace, this.type,
boundTargetHolderEntry.getKey()));
}
bindingService.unbindProducers(boundTargetHolderEntry.getKey());
}
}
}
@Override
public Set<String> getInputs() {
return this.inputHolders.keySet();
}
@Override
public Set<String> getOutputs() {
return this.outputHolders.keySet();
}
}

224
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java
View File

@@ -17,12 +17,8 @@
package org.springframework.cloud.stream.binding;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
@@ -32,17 +28,12 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.cloud.stream.binder.Binding;
import org.springframework.cloud.stream.internal.InternalPropertyNames;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
/**
* {@link FactoryBean} for instantiating the interfaces specified via
@@ -54,33 +45,22 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
* @see EnableBinding
*/
public class BindableProxyFactory
implements MethodInterceptor, FactoryBean<Object>, Bindable, InitializingBean {
public class BindableProxyFactory extends AbstractBindableProxyFactory
implements MethodInterceptor, FactoryBean<Object>, InitializingBean {
private static Log log = LogFactory.getLog(BindableProxyFactory.class);
private final Map<Method, Object> targetCache = new HashMap<>(2);
@Value("${" + InternalPropertyNames.NAMESPACE_PROPERTY_NAME + ":}")
private String namespace;
@Autowired
private Map<String, BindingTargetFactory> bindingTargetFactories;
private Class<?> type;
private Object proxy;
private Map<String, BoundTargetHolder> inputHolders = new HashMap<>();
private Map<String, BoundTargetHolder> outputHolders = new HashMap<>();
public BindableProxyFactory(Class<?> type) {
super(type);
this.type = type;
}
@Override
public synchronized Object invoke(MethodInvocation invocation) throws Throwable {
public synchronized Object invoke(MethodInvocation invocation) {
Method method = invocation.getMethod();
// try to use cached target
@@ -111,74 +91,37 @@ public class BindableProxyFactory
}
@Override
public void afterPropertiesSet() throws Exception {
public void afterPropertiesSet() {
Assert.notEmpty(BindableProxyFactory.this.bindingTargetFactories,
"'bindingTargetFactories' cannot be empty");
ReflectionUtils.doWithMethods(this.type, new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = BindingBeanDefinitionRegistryUtils
.getBindingTargetName(input, method);
Class<?> returnType = method.getReturnType();
ReflectionUtils.doWithMethods(this.type, method -> {
Input input = AnnotationUtils.findAnnotation(method, Input.class);
if (input != null) {
String name = BindingBeanDefinitionRegistryUtils
.getBindingTargetName(input, method);
Class<?> returnType = method.getReturnType();
BindableProxyFactory.this.inputHolders.put(name,
new BoundTargetHolder(getBindingTargetFactory(returnType)
.createInput(name), true));
}
BindableProxyFactory.this.inputHolders.put(name,
new BoundTargetHolder(getBindingTargetFactory(returnType)
.createInput(name), true));
}
});
ReflectionUtils.doWithMethods(this.type, new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException {
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = BindingBeanDefinitionRegistryUtils
.getBindingTargetName(output, method);
Class<?> returnType = method.getReturnType();
ReflectionUtils.doWithMethods(this.type, method -> {
Output output = AnnotationUtils.findAnnotation(method, Output.class);
if (output != null) {
String name = BindingBeanDefinitionRegistryUtils
.getBindingTargetName(output, method);
Class<?> returnType = method.getReturnType();
BindableProxyFactory.this.outputHolders.put(name,
new BoundTargetHolder(getBindingTargetFactory(returnType)
.createOutput(name), true));
}
BindableProxyFactory.this.outputHolders.put(name,
new BoundTargetHolder(getBindingTargetFactory(returnType)
.createOutput(name), true));
}
});
}
private BindingTargetFactory getBindingTargetFactory(Class<?> bindingTargetType) {
List<String> candidateBindingTargetFactories = new ArrayList<>();
for (Map.Entry<String, BindingTargetFactory> bindingTargetFactoryEntry : this.bindingTargetFactories
.entrySet()) {
if (bindingTargetFactoryEntry.getValue().canCreate(bindingTargetType)) {
candidateBindingTargetFactories.add(bindingTargetFactoryEntry.getKey());
}
}
if (candidateBindingTargetFactories.size() == 1) {
return this.bindingTargetFactories
.get(candidateBindingTargetFactories.get(0));
}
else {
if (candidateBindingTargetFactories.size() == 0) {
throw new IllegalStateException(
"No factory found for binding target type: "
+ bindingTargetType.getName()
+ " among registered factories: "
+ StringUtils.collectionToCommaDelimitedString(
this.bindingTargetFactories.keySet()));
}
else {
throw new IllegalStateException(
"Multiple factories found for binding target type: "
+ bindingTargetType.getName() + ": "
+ StringUtils.collectionToCommaDelimitedString(
candidateBindingTargetFactories));
}
}
}
@Override
public synchronized Object getObject() throws Exception {
public synchronized Object getObject() {
if (this.proxy == null) {
ProxyFactory factory = new ProxyFactory(this.type, this);
this.proxy = factory.getProxy();
@@ -196,123 +139,4 @@ public class BindableProxyFactory
return true;
}
@Override
public Collection<Binding<Object>> createAndBindInputs(
BindingService bindingService) {
List<Binding<Object>> bindings = new ArrayList<>();
if (log.isDebugEnabled()) {
log.debug(
String.format("Binding inputs for %s:%s", this.namespace, this.type));
}
for (Map.Entry<String, BoundTargetHolder> boundTargetHolderEntry : this.inputHolders
.entrySet()) {
String inputTargetName = boundTargetHolderEntry.getKey();
BoundTargetHolder boundTargetHolder = boundTargetHolderEntry.getValue();
if (boundTargetHolder.isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.namespace, this.type,
inputTargetName));
}
bindings.addAll(bindingService.bindConsumer(
boundTargetHolder.getBoundTarget(), inputTargetName));
}
}
return bindings;
}
@Override
public Collection<Binding<Object>> createAndBindOutputs(
BindingService bindingService) {
List<Binding<Object>> bindings = new ArrayList<>();
if (log.isDebugEnabled()) {
log.debug(String.format("Binding outputs for %s:%s", this.namespace,
this.type));
}
for (Map.Entry<String, BoundTargetHolder> boundTargetHolderEntry : this.outputHolders
.entrySet()) {
BoundTargetHolder boundTargetHolder = boundTargetHolderEntry.getValue();
String outputTargetName = boundTargetHolderEntry.getKey();
if (boundTargetHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.namespace, this.type,
outputTargetName));
}
bindings.add(bindingService.bindProducer(
boundTargetHolder.getBoundTarget(), outputTargetName));
}
}
return bindings;
}
@Override
public void unbindInputs(BindingService bindingService) {
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding inputs for %s:%s", this.namespace,
this.type));
}
for (Map.Entry<String, BoundTargetHolder> boundTargetHolderEntry : this.inputHolders
.entrySet()) {
if (boundTargetHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding %s:%s:%s", this.namespace,
this.type, boundTargetHolderEntry.getKey()));
}
bindingService.unbindConsumers(boundTargetHolderEntry.getKey());
}
}
}
@Override
public void unbindOutputs(BindingService bindingService) {
if (log.isDebugEnabled()) {
log.debug(String.format("Unbinding outputs for %s:%s", this.namespace,
this.type));
}
for (Map.Entry<String, BoundTargetHolder> boundTargetHolderEntry : this.outputHolders
.entrySet()) {
if (boundTargetHolderEntry.getValue().isBindable()) {
if (log.isDebugEnabled()) {
log.debug(String.format("Binding %s:%s:%s", this.namespace, this.type,
boundTargetHolderEntry.getKey()));
}
bindingService.unbindProducers(boundTargetHolderEntry.getKey());
}
}
}
@Override
public Set<String> getInputs() {
return this.inputHolders.keySet();
}
@Override
public Set<String> getOutputs() {
return this.outputHolders.keySet();
}
/**
* Holds information about the binding targets exposed by the interface proxy, as well
* as their status.
*/
private final class BoundTargetHolder {
private Object boundTarget;
private boolean bindable;
private BoundTargetHolder(Object boundTarget, boolean bindable) {
this.boundTarget = boundTarget;
this.bindable = bindable;
}
public Object getBoundTarget() {
return this.boundTarget;
}
public boolean isBindable() {
return this.bindable;
}
}
}

48
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BoundTargetHolder.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* Copyright 2015-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.binding;
/**
* Holds information about the binding targets exposed by the interface proxy, as well
* as their status.
*
* Refactored from {@link BindableProxyFactory}.
*
* @author Original authors in {@link BindableProxyFactory}
* @author Soby Chacko
* @since 3.0.0
*/
public final class BoundTargetHolder {
private Object boundTarget;
private boolean bindable;
public BoundTargetHolder(Object boundTarget, boolean bindable) {
this.boundTarget = boundTarget;
this.bindable = bindable;
}
public Object getBoundTarget() {
return this.boundTarget;
}
public boolean isBindable() {
return this.bindable;
}
}

37
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindableProvider.java Normal file
View File

@@ -0,0 +1,37 @@
/*
* Copyright 2019-2019 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.stream.config;
/**
* If downstream binders want to take over the responsibility of binding the target types (such as the Kafka Streams binder),
* then they can implement this functional interface to signal the core framework to bypass any binding.
*
* @author Soby Chacko
* @since 3.0.0
*/
@FunctionalInterface
public interface BindableProvider {
/**
* Based on the type provided, the implementation can determine whether it is capable of
* binding this target type.
*
* @param clazz target type to bind
* @return true if capable of binding
*/
boolean canBind(Class<?> clazz);
}

35
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java
View File

@@ -249,7 +249,7 @@ public class BinderFactoryAutoConfiguration {
@Bean
public BeanFactoryPostProcessor implicitFunctionBinder(Environment environment,
@Nullable FunctionRegistry functionCatalog, @Nullable FunctionInspector inspector) {
@Nullable FunctionRegistry functionCatalog, @Nullable FunctionInspector inspector) {
return new BeanFactoryPostProcessor() {
@Override
public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
@@ -260,14 +260,35 @@ public class BinderFactoryAutoConfiguration {
Object definedFunction = functionCatalog.lookup(name);
Class<?> inputType = inspector.getInputType(definedFunction);
Class<?> outputType = inspector.getOutputType(definedFunction);
if (Void.class.isAssignableFrom(outputType)) {
bind(Sink.class, registry);
boolean bindDownstream = false;
try {
Map<String, BindableProvider> bindableProviders = beanFactory.getBeansOfType(BindableProvider.class);
Class<?> inputTypeWrapper = inspector.getInputWrapper(definedFunction);
if (bindableProviders != null) {
Collection<BindableProvider> values = bindableProviders.values();
for (BindableProvider bindableProvider : values) {
if (bindableProvider.canBind(inputTypeWrapper)) {
bindDownstream = true;
break;
}
}
}
}
else if (Void.class.isAssignableFrom(inputType)) {
bind(Source.class, registry);
catch (BeansException be) {
// pass through
}
else {
bind(Processor.class, registry);
if (!bindDownstream) {
if (Void.class.isAssignableFrom(outputType)) {
bind(Sink.class, registry);
}
else if (Void.class.isAssignableFrom(inputType)) {
bind(Source.class, registry);
}
else {
bind(Processor.class, registry);
}
}
}
}

26
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java
View File

@@ -16,6 +16,10 @@
package org.springframework.cloud.stream.function;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.messaging.Processor;
@@ -23,6 +27,7 @@ import org.springframework.cloud.stream.messaging.Processor;
/**
* @author Oleg Zhurakousky
* @author Tolga Kavukcu
* @author Soby Chacko
* @since 2.1
*/
@ConfigurationProperties("spring.cloud.stream.function")
@@ -40,6 +45,10 @@ public class StreamFunctionProperties {
private String outputDestinationName = Processor.OUTPUT;
private Map<String, List<String>> inputBindings = new HashMap<>();
private Map<String, List<String>> outputBindings = new HashMap<>();
public String getDefinition() {
return this.definition;
}
@@ -72,4 +81,21 @@ public class StreamFunctionProperties {
this.outputDestinationName = outputDestinationName;
}
public Map<String, List<String>> getInputBindings() {
return inputBindings;
}
public Map<String, List<String>> getOutputBindings() {
return outputBindings;
}
public void setOutputBindings(Map<String, List<String>> outputBindings) {
this.outputBindings = outputBindings;
}
public void setInputBindings(Map<String, List<String>> inputBindings) {
this.inputBindings = inputBindings;
}
}