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; + + + } }