From 93d4c21757df633ecb8a313f21096daa1201bbb3 Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 1 Apr 2025 16:01:07 +0200 Subject: [PATCH] GH-3086 Initial functionality to define bindings dynamically --- .../binding/AbstractBindingLifecycle.java | 10 +++++- .../cloud/stream/binding/BindingService.java | 4 --- .../binding/BindingsLifecycleController.java | 36 ++++++++++++++++++- .../BindableFunctionProxyFactory.java | 7 +++- 4 files changed, 50 insertions(+), 7 deletions(-) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindingLifecycle.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindingLifecycle.java index a49fc4a5a..605d07e21 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindingLifecycle.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/AbstractBindingLifecycle.java @@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binding; import java.util.Map; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.cloud.stream.function.BindableFunctionProxyFactory; import org.springframework.context.ApplicationContext; import org.springframework.context.SmartLifecycle; @@ -36,7 +37,7 @@ public abstract class AbstractBindingLifecycle implements SmartLifecycle { final BindingService bindingService; - private final Map bindables; + final Map bindables; @Autowired private ApplicationContext context; @@ -90,6 +91,13 @@ public abstract class AbstractBindingLifecycle implements SmartLifecycle { } } + public void startBindable(Bindable bindable) { + if (bindable instanceof BindableFunctionProxyFactory functionProxy) { + this.bindables.put(functionProxy.getFunctionDefinition(), bindable); + } + this.doStartWithBindable(bindable); + } + abstract void doStartWithBindable(Bindable bindable); abstract void doStopWithBindable(Bindable bindable); diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java index 80c84a6c4..5e7b9d41b 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingService.java @@ -72,8 +72,6 @@ import static org.springframework.cloud.stream.utils.CacheKeyCreatorUtils.getBin */ public class BindingService { -// private final CustomValidatorBean validator; - private final Log log = LogFactory.getLog(BindingService.class); private final BindingServiceProperties bindingServiceProperties; @@ -97,8 +95,6 @@ public class BindingService { BinderFactory binderFactory, TaskScheduler taskScheduler, ObjectMapper objectMapper) { this.bindingServiceProperties = bindingServiceProperties; this.binderFactory = binderFactory; -// this.validator = new CustomValidatorBean(); -// this.validator.afterPropertiesSet(); this.taskScheduler = taskScheduler; this.objectMapper = objectMapper; } diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java index 009323328..2a545ff55 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/BindingsLifecycleController.java @@ -27,8 +27,13 @@ import com.fasterxml.jackson.databind.Module; import com.fasterxml.jackson.databind.ObjectMapper; import org.springframework.beans.BeanUtils; +import org.springframework.beans.BeansException; import org.springframework.beans.DirectFieldAccessor; import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.function.BindableFunctionProxyFactory; +import org.springframework.cloud.stream.function.StreamFunctionProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; @@ -42,7 +47,7 @@ import org.springframework.util.CollectionUtils; * @author Soby Chacko * @since 3.x */ -public class BindingsLifecycleController { +public class BindingsLifecycleController implements ApplicationContextAware { private final List inputBindingLifecycles; @@ -50,6 +55,8 @@ public class BindingsLifecycleController { private final ObjectMapper objectMapper; + private ApplicationContext applicationContext; + @SuppressWarnings("unchecked") public BindingsLifecycleController(List inputBindingLifecycles, List outputBindingsLifecycles) { @@ -74,6 +81,28 @@ public class BindingsLifecycleController { } } + public void defineBinding(String bindingName, boolean isInputBinding) { + BindableFunctionProxyFactory bindingProxyFactory = new BindableFunctionProxyFactory(bindingName, + isInputBinding ? 1 : 0, isInputBinding ? 0 : 1, this.applicationContext.getBean(StreamFunctionProperties.class)); + bindingProxyFactory.setApplicationContext(this.applicationContext); + + bindingProxyFactory.afterPropertiesSet(); + BindingService bindingService = this.applicationContext.getBean(BindingService.class); + + + AbstractBindingLifecycle bindingLifecycle; + if (isInputBinding) { + bindingProxyFactory.createAndBindInputs(bindingService); + bindingLifecycle = this.applicationContext.getBean(InputBindingLifecycle.class); + } + else { + bindingProxyFactory.createAndBindOutputs(bindingService); + bindingLifecycle = this.applicationContext.getBean(OutputBindingLifecycle.class); + } + + bindingLifecycle.startBindable(bindingProxyFactory); + } + /** * Will return producer or consumer properties for a specified binding. For example, calling `getExtensionProperties("foo-in-0")` * on Kafka binding will return an instance of KafkaConsumerProperties. @@ -173,6 +202,11 @@ public class BindingsLifecycleController { return this.locateBinding(name); } + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + /** * Queries for all input {@link Binding}s. * @return the list of input {@link Binding}s diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java index 153356ead..6b2a57ba1 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/BindableFunctionProxyFactory.java @@ -96,6 +96,10 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement this.functionExist = functionExist; } + public String toString() { + return "Bindable: " + this.functionDefinition; + } + @Override public void afterPropertiesSet() { populateBindingTargetFactories(beanFactory); @@ -125,7 +129,7 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement return true; } - protected String getFunctionDefinition() { + public String getFunctionDefinition() { return this.isFunctionExist() ? this.functionDefinition : null; } @@ -224,6 +228,7 @@ public class BindableFunctionProxyFactory extends BindableProxyFactory implement @Override public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.context = (GenericApplicationContext) applicationContext; + this.setBeanFactory(this.context.getBeanFactory()); } public boolean isFunctionExist() {