From 8af53028f5d69464e7d0d1694fadee9bfb3424f4 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 28 Jun 2019 15:29:07 -0400 Subject: [PATCH] Bypassing target type binding 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 --- .../binding/AbstractBindableProxyFactory.java | 188 +++++++++++++++ .../stream/binding/BindableProxyFactory.java | 224 ++---------------- .../stream/binding/BoundTargetHolder.java | 48 ++++ .../cloud/stream/config/BindableProvider.java | 37 +++ .../BinderFactoryAutoConfiguration.java | 35 ++- .../function/StreamFunctionProperties.java | 26 ++ 6 files changed, 351 insertions(+), 207 deletions(-) create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BoundTargetHolder.java create mode 100644 spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindableProvider.java diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java new file mode 100644 index 000000000..250ef4be1 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindableProxyFactory.java @@ -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 bindingTargetFactories; + + protected Map inputHolders = new LinkedHashMap<>(); + + protected Map outputHolders = new LinkedHashMap<>(); + + protected Class type; + + public AbstractBindableProxyFactory(Class type) { + this.type = type; + } + + protected BindingTargetFactory getBindingTargetFactory(Class bindingTargetType) { + List candidateBindingTargetFactories = new ArrayList<>(); + for (Map.Entry 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> createAndBindInputs( + BindingService bindingService) { + List> bindings = new ArrayList<>(); + if (log.isDebugEnabled()) { + log.debug( + String.format("Binding inputs for %s:%s", this.namespace, this.type)); + } + for (Map.Entry 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> createAndBindOutputs( + BindingService bindingService) { + List> bindings = new ArrayList<>(); + if (log.isDebugEnabled()) { + log.debug(String.format("Binding outputs for %s:%s", this.namespace, + this.type)); + } + for (Map.Entry 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 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 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 getInputs() { + return this.inputHolders.keySet(); + } + + @Override + public Set getOutputs() { + return this.outputHolders.keySet(); + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java index 4f2748ed9..4a94637b2 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindableProxyFactory.java @@ -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, Bindable, InitializingBean { +public class BindableProxyFactory extends AbstractBindableProxyFactory + implements MethodInterceptor, FactoryBean, InitializingBean { private static Log log = LogFactory.getLog(BindableProxyFactory.class); private final Map targetCache = new HashMap<>(2); - @Value("${" + InternalPropertyNames.NAMESPACE_PROPERTY_NAME + ":}") - private String namespace; - - @Autowired - private Map bindingTargetFactories; - - private Class type; - private Object proxy; - private Map inputHolders = new HashMap<>(); - - private Map 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 candidateBindingTargetFactories = new ArrayList<>(); - for (Map.Entry 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> createAndBindInputs( - BindingService bindingService) { - List> bindings = new ArrayList<>(); - if (log.isDebugEnabled()) { - log.debug( - String.format("Binding inputs for %s:%s", this.namespace, this.type)); - } - for (Map.Entry 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> createAndBindOutputs( - BindingService bindingService) { - List> bindings = new ArrayList<>(); - if (log.isDebugEnabled()) { - log.debug(String.format("Binding outputs for %s:%s", this.namespace, - this.type)); - } - for (Map.Entry 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 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 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 getInputs() { - return this.inputHolders.keySet(); - } - - @Override - public Set 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; - } - - } - } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BoundTargetHolder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BoundTargetHolder.java new file mode 100644 index 000000000..f3a384527 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BoundTargetHolder.java @@ -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; + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindableProvider.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindableProvider.java new file mode 100644 index 000000000..0b7e3d52f --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindableProvider.java @@ -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); +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java index 1587f84a9..ccd2e7880 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BinderFactoryAutoConfiguration.java @@ -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 bindableProviders = beanFactory.getBeansOfType(BindableProvider.class); + Class inputTypeWrapper = inspector.getInputWrapper(definedFunction); + if (bindableProviders != null) { + Collection 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); + } } } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java index dab54b2c1..467c99338 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java @@ -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> inputBindings = new HashMap<>(); + + private Map> outputBindings = new HashMap<>(); + public String getDefinition() { return this.definition; } @@ -72,4 +81,21 @@ public class StreamFunctionProperties { this.outputDestinationName = outputDestinationName; } + public Map> getInputBindings() { + return inputBindings; + } + + public Map> getOutputBindings() { + return outputBindings; + } + + public void setOutputBindings(Map> outputBindings) { + this.outputBindings = outputBindings; + } + + public void setInputBindings(Map> inputBindings) { + this.inputBindings = inputBindings; + + + } }