GH-3086 Initial functionality to define bindings dynamically

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<String, Bindable> bindables;
+	final Map<String, Bindable> 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);

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

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<InputBindingLifecycle> inputBindingLifecycles;
 
@@ -50,6 +55,8 @@ public class BindingsLifecycleController {
 
 	private final ObjectMapper objectMapper;
 
+	private ApplicationContext applicationContext;
+
 	@SuppressWarnings("unchecked")
 	public BindingsLifecycleController(List<InputBindingLifecycle> inputBindingLifecycles,
 			List<OutputBindingLifecycle> 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

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() {