GH-1062 Added initial support for actuator endpoint for input bindings

Resolves #1062
Resolves #1239

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binding.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2018 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.
@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder;
 
+import org.springframework.context.Lifecycle;
+
 /**
  * Represents a binding between an input or output and an adapter endpoint that connects
  * via a Binder. The binding could be for a consumer or a producer. A consumer binding
@@ -26,9 +28,46 @@ package org.springframework.cloud.stream.binder;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Marius Bogoevici
+ * @author Oleg Zhurakousky
+ * 
  * @see org.springframework.cloud.stream.annotation.EnableBinding
  */
-public interface Binding<T> {
+public interface Binding<T> extends Lifecycle {
+	
+	/**
+	 * Stops the target component represented by this instance.
+	 * NOTE: At the time the instance is created the component is already started. 
+	 * This operation is typically used by actuator to re-bind/re-start.
+	 * 
+	 * @see BindingsEndpoint
+	 */
+	default void start() {}
+	
+	/**
+	 * Starts the target component represented by this instance.
+	 * NOTE: At the time the instance is created the component is already started. 
+	 * This operation is typically used by actuator to re-bind/re-start.
+	 * 
+	 * @see BindingsEndpoint
+	 */
+	default void stop() {}
+	
+	/**
+	 * Returns 'true' if the target component represented by this instance is running.
+	 */
+	default boolean isRunning() {
+		return false;
+	}
+	
+	/**
+	 * Returns the name of this binding  (i.e., channel name)
+	 * 
+	 * @return binding name
+	 */
+	default String getName() {
+		return null;
+	}
+		
 
 	/**
 	 * Unbinds the target component represented by this instance and stops any active

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultBinding.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2016 the original author or authors.
+ * Copyright 2013-2018 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.
@@ -16,10 +16,14 @@
 
 package org.springframework.cloud.stream.binder;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.support.context.NamedComponent;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * Default implementation for a {@link Binding}.
@@ -27,9 +31,13 @@ import org.springframework.util.ObjectUtils;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Marius Bogoevici
+ * @author Oleg Zhurakousky
+ * 
  * @see org.springframework.cloud.stream.annotation.EnableBinding
  */
 public class DefaultBinding<T> implements Binding<T> {
+	
+	private final Log logger = LogFactory.getLog(this.getClass().getName());
 
 	protected final String name;
 
@@ -64,12 +72,33 @@ public class DefaultBinding<T> implements Binding<T> {
 	public String getGroup() {
 		return this.group;
 	}
+	
+	public boolean isRunning() {
+		return this.lifecycle != null && this.lifecycle.isRunning();
+	}
+	
+	@Override
+	public final synchronized void start() {
+		if (!this.isRunning()) {
+			if (this.lifecycle != null && StringUtils.hasText(this.group)) {
+				this.lifecycle.start();
+			}
+			else {
+				logger.warn("Can not re-bind an anonymous binding");
+			}
+		}
+	}
+	
+	@Override
+	public final synchronized void stop() {
+		if (this.isRunning()) {
+			this.lifecycle.stop();
+		}
+	}
 
 	@Override
 	public final void unbind() {
-		if (this.lifecycle != null) {
-			this.lifecycle.stop();
-		}
+		this.stop();
 		afterUnbind();
 	}
 

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/Bindable.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -16,9 +16,12 @@
 
 package org.springframework.cloud.stream.binding;
 
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Set;
 
+import org.springframework.cloud.stream.binder.Binding;
+
 /**
  * Marker interface for instances that can bind/unbind groups of inputs and outputs.
  *
@@ -31,8 +34,23 @@ public interface Bindable {
 
 	/**
 	 * Binds all the inputs associated with this instance.
+	 * @deprecated as of 2.0 in favor of {@link #createAndBindInputs(BindingService)}
 	 */
-	default void bindInputs(BindingService adapter) {}
+	@Deprecated
+	default void bindInputs(BindingService adapter) {
+		this.createAndBindInputs(adapter);
+	}
+	
+	/**
+	 * Binds all the inputs associated with this instance.
+	 * @param adapter instance of {@link BindingService}
+	 * @return collection of {@link Binding}s
+	 * 
+	 * @since 2.0
+	 */
+	default Collection<Binding<Object>> createAndBindInputs(BindingService adapter) {
+		return Collections.<Binding<Object>>emptyList();
+	}
 
 	/**
 	 * Binds all the outputs associated with this instance.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -18,6 +18,7 @@ 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;
@@ -37,6 +38,7 @@ import org.springframework.cloud.stream.aggregate.SharedBindingTargetRegistry;
 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;
@@ -50,6 +52,7 @@ import org.springframework.util.StringUtils;
  * @author Marius Bogoevici
  * @author David Syer
  * @author Ilayaperumal Gopinathan
+ * @author Oleg Zhurakousky
  *
  * @see EnableBinding
  */
@@ -206,8 +209,18 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
 		return true;
 	}
 
+	/**
+	 * @deprecated in favor of {@link #bindInputBindings(BindingService)}
+	 */
 	@Override
+	@Deprecated
 	public void bindInputs(BindingService bindingService) {
+		this.createAndBindInputs(bindingService);
+	}
+	
+	@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));
 		}
@@ -218,9 +231,10 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
 				if (log.isDebugEnabled()) {
 					log.debug(String.format("Binding %s:%s:%s", this.namespace, this.type, inputTargetName));
 				}
-				bindingService.bindConsumer(boundTargetHolder.getBoundTarget(), inputTargetName);
+				bindings.addAll(bindingService.bindConsumer(boundTargetHolder.getBoundTarget(), inputTargetName));
 			}
 		}
+		return bindings;
 	}
 
 	@Override

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/InputBindingLifecycle.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2017 the original author or authors.
+ * Copyright 2015-2018 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.
@@ -16,8 +16,11 @@
 
 package org.springframework.cloud.stream.binding;
 
+import java.util.Collection;
 import java.util.Map;
 
+import org.springframework.cloud.stream.binder.Binding;
+
 /**
  * Coordinates binding/unbinding of input binding targets in accordance to the lifecycle
  * of the host context.
@@ -26,6 +29,10 @@ import java.util.Map;
  * @author Oleg Zhurakousky
  */
 public class InputBindingLifecycle extends AbstractBindingLifecycle {
+	
+	@SuppressWarnings("unused")
+	//It is actually used reflectively since at the moment we do not want to expose it via public method
+	private Collection<Binding<Object>> inputBindings;
 
 	public InputBindingLifecycle(BindingService bindingService, Map<String, Bindable> bindables) {
 		super(bindingService, bindables);
@@ -42,7 +49,7 @@ public class InputBindingLifecycle extends AbstractBindingLifecycle {
 
 	@Override
 	void doStartWithBindable(Bindable bindable) {
-		bindable.bindInputs(bindingService);
+		this.inputBindings = bindable.createAndBindInputs(bindingService);
 	}
 
 	@Override

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingsEndpointAutoConfiguration.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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;
+
+import java.util.List;
+
+import org.springframework.boot.actuate.autoconfigure.endpoint.EndpointAutoConfiguration;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.cloud.stream.binding.BindingService;
+import org.springframework.cloud.stream.binding.InputBindingLifecycle;
+import org.springframework.cloud.stream.endpoint.BindingsEndpoint;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author Oleg Zhurakousky
+ * 
+ * @since 2.0
+ */
+@Configuration
+@ConditionalOnClass(name = "org.springframework.boot.actuate.endpoint.annotation.Endpoint")
+@ConditionalOnBean(BindingService.class)
+@AutoConfigureAfter(EndpointAutoConfiguration.class)
+public class BindingsEndpointAutoConfiguration {
+
+	@Bean
+	public BindingsEndpoint bindingsEndpoint(List<InputBindingLifecycle> inputBindings) {
+		return new BindingsEndpoint(inputBindings);
+	}
+	
+	@Bean
+	public BindingsEndpoint.StopEndpoint stopEndpoint(BindingsEndpoint bindingEndpoint) {
+		return bindingEndpoint.getStopEndpoint();
+	}
+	
+	@Bean
+	public BindingsEndpoint.StartEndpoint startEndpoint(BindingsEndpoint bindingEndpoint) {
+		return bindingEndpoint.getStartEndpoint();
+	}
+	
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/endpoint/BindingsEndpoint.java

@@ -0,0 +1,111 @@
+/*
+ * Copyright 2018 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
+ *
+ *      http://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.endpoint;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.boot.actuate.endpoint.annotation.Endpoint;
+import org.springframework.boot.actuate.endpoint.annotation.ReadOperation;
+import org.springframework.boot.actuate.endpoint.annotation.WriteOperation;
+import org.springframework.cloud.stream.binder.Binding;
+import org.springframework.cloud.stream.binding.InputBindingLifecycle;
+import org.springframework.util.Assert;
+
+/**
+ * 
+ * @author Oleg Zhurakousky
+ * 
+ * @since 2.0
+ *
+ */
+@Endpoint(id = BindingsEndpoint.BASE_ENPOINT_ID)
+public class BindingsEndpoint {
+	
+	static final String BASE_ENPOINT_ID = "bindings";
+	
+	static final String START_ENPOINT_ID = BASE_ENPOINT_ID + "/" + "start/{name}";
+	
+	static final String STOP_ENPOINT_ID = BASE_ENPOINT_ID + "/" + "stop/{name}";
+
+	private final List<InputBindingLifecycle> inputBindingLifecycles;
+	
+	public BindingsEndpoint(List<InputBindingLifecycle> inputBindingLifecycles) {
+		Assert.notEmpty(inputBindingLifecycles, "'inputBindingLifecycles' must not be null or empty");
+		this.inputBindingLifecycles = inputBindingLifecycles;
+	}
+	
+	@ReadOperation
+	public List<Object> bindings() {
+		return new ObjectMapper().convertValue(this.gatherInputBindings(), new TypeReference<List<Object>>() {});	
+	}
+	
+	public StopEndpoint getStopEndpoint() {
+		return new StopEndpoint();
+	}
+	
+	public StartEndpoint getStartEndpoint() {
+		return new StartEndpoint();
+	}
+	
+	@SuppressWarnings("unchecked")
+	private List<Binding<Object>> gatherInputBindings() {
+		List<Binding<Object>> inputBindings = new ArrayList<>();
+		for (InputBindingLifecycle inputBindingLifecycle : inputBindingLifecycles) {
+			Collection<Binding<Object>> lifecycleInputBindings = 
+					(Collection<Binding<Object>>) new DirectFieldAccessor(inputBindingLifecycle).getPropertyValue("inputBindings");
+			inputBindings.addAll(lifecycleInputBindings);
+		}
+		return inputBindings;
+	}
+	
+	@Endpoint(id = BindingsEndpoint.STOP_ENPOINT_ID)
+	public class StopEndpoint {	
+		@WriteOperation
+		public boolean stop(String name) {
+			List<Binding<Object>> inputBindings = BindingsEndpoint.this.gatherInputBindings();	
+			for (Binding<Object> binding : inputBindings) {
+				if (name.equals(binding.getName())) {
+					binding.stop();
+					return true;
+				}
+			}
+			return false;
+		}
+	}
+	
+	@Endpoint(id = BindingsEndpoint.START_ENPOINT_ID)
+	public class StartEndpoint {	
+		@WriteOperation
+		public boolean start(String name) {
+			List<Binding<Object>> inputBindings = BindingsEndpoint.this.gatherInputBindings();
+			for (Binding<Object> binding : inputBindings) {
+				if (name.equals(binding.getName())) {
+					binding.start();
+					return true;
+				}
+			}
+			return false;
+		}
+	}
+
+}

spring-cloud-stream/src/main/resources/META-INF/spring.factories

@@ -1,5 +1,6 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration:\
 org.springframework.cloud.stream.config.ChannelBindingAutoConfiguration,\
 org.springframework.cloud.stream.config.BindersHealthIndicatorAutoConfiguration,\
-org.springframework.cloud.stream.config.ChannelsEndpointAutoConfiguration
+org.springframework.cloud.stream.config.ChannelsEndpointAutoConfiguration,\
+org.springframework.cloud.stream.config.BindingsEndpointAutoConfiguration