GH-1389 Added support for output binding control

Added missing support for visualization and control of output binding
Resolves #1389

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

@@ -176,7 +176,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 						.equals(producerProperties.getHeaderMode()), this.headersToEmbed,
 						producerProperties.isUseNativeEncoding()));
 
-		Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(destination, null, outputChannel,
+		Binding<MessageChannel> binding = new DefaultBinding<MessageChannel>(destination, outputChannel,
 				producerMessageHandler instanceof Lifecycle ? (Lifecycle) producerMessageHandler : null) {
 
 			@Override

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

@@ -55,6 +55,8 @@ public class DefaultBinding<T> implements Binding<T> {
 
 	private boolean paused;
 
+	private boolean restartable;
+
 	/**
 	 * Creates an instance that associates a given name, group and binding target with an
 	 * optional {@link Lifecycle} component, which will be stopped during unbinding.
@@ -70,6 +72,12 @@ public class DefaultBinding<T> implements Binding<T> {
 		this.group = group;
 		this.target = target;
 		this.lifecycle = lifecycle;
+		this.restartable = StringUtils.hasText(group);
+	}
+
+	public DefaultBinding(String name, T target, Lifecycle lifecycle) {
+		this(name, null, target, lifecycle);
+		this.restartable = true;
 	}
 
 	public String getName() {
@@ -104,7 +112,7 @@ public class DefaultBinding<T> implements Binding<T> {
 	@Override
 	public final synchronized void start() {
 		if (!this.isRunning()) {
-			if (this.lifecycle != null && StringUtils.hasText(this.group)) {
+			if (this.lifecycle != null && this.restartable) {
 				this.lifecycle.start();
 			}
 			else {

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

@@ -54,9 +54,22 @@ public interface Bindable {
 
 	/**
 	 * Binds all the outputs associated with this instance.
+	 * @deprecated as of 2.0 in favor of {@link #createAndBindOutputs(BindingService)}
 	 */
+	@Deprecated
 	default void bindOutputs(BindingService adapter) {}
 
+	/**
+	 * Binds all the outputs associated with this instance.
+	 * @param adapter instance of {@link BindingService}
+	 * @return collection of {@link Binding}s
+	 *
+	 * @since 2.0
+	 */
+	default Collection<Binding<Object>> createAndBindOutputs(BindingService adapter) {
+		return Collections.<Binding<Object>>emptyList();
+	}
+
 	/**
 	 * Unbinds all the inputs associated with this instance.
 	 */

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

@@ -237,8 +237,18 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
 		return bindings;
 	}
 
+	/**
+	 * @deprecated in favor of {@link #createAndBindOutputs(BindingService)}
+	 */
 	@Override
+	@Deprecated
 	public void bindOutputs(BindingService bindingService) {
+		this.createAndBindOutputs(bindingService);
+	}
+
+	@Override
+	public Collection<Binding<Object>> createAndBindOutputs(BindingService bindingService) {
+		List<Binding<Object>> bindings = new ArrayList<>();
 		if (log.isDebugEnabled()) {
 			log.debug(String.format("Binding outputs for %s:%s", this.namespace, this.type));
 		}
@@ -249,9 +259,10 @@ public class BindableProxyFactory implements MethodInterceptor, FactoryBean<Obje
 				if (log.isDebugEnabled()) {
 					log.debug(String.format("Binding %s:%s:%s", this.namespace, this.type, outputTargetName));
 				}
-				bindingService.bindProducer(boundTargetHolder.getBoundTarget(), outputTargetName);
+				bindings.add(bindingService.bindProducer(boundTargetHolder.getBoundTarget(), outputTargetName));
 			}
 		}
+		return bindings;
 	}
 
 	@Override

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

@@ -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 output binding targets in accordance to the lifecycle
  * of the host context.
@@ -28,6 +31,10 @@ import java.util.Map;
  */
 public class OutputBindingLifecycle 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>> outputBindings;
+
 	public OutputBindingLifecycle(BindingService bindingService, Map<String, Bindable> bindables) {
 		super(bindingService, bindables);
 	}
@@ -43,7 +50,7 @@ public class OutputBindingLifecycle extends AbstractBindingLifecycle {
 
 	@Override
 	void doStartWithBindable(Bindable bindable) {
-		bindable.bindOutputs(bindingService);
+		this.outputBindings = bindable.createAndBindOutputs(bindingService);
 	}
 
 	@Override

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

@@ -17,10 +17,13 @@
 package org.springframework.cloud.stream.binding;
 
 import java.util.Arrays;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
 
+import org.springframework.cloud.stream.binder.Binding;
+
 /**
  * A {@link Bindable} component that wraps a generic output binding target. Useful for
  * binding targets outside the {@link org.springframework.cloud.stream.annotation.Input}
@@ -42,7 +45,12 @@ public class SingleBindingTargetBindable<T> implements Bindable {
 
 	@Override
 	public void bindOutputs(BindingService bindingService) {
-		bindingService.bindProducer(bindingTarget, name);
+		this.createAndBindOutputs(bindingService);
+	}
+
+	@Override
+	public Collection<Binding<Object>> createAndBindOutputs(BindingService bindingService) {
+		return Collections.singletonList(bindingService.bindProducer(bindingTarget, name));
 	}
 
 	@Override

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

@@ -24,6 +24,7 @@ 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.binding.OutputBindingLifecycle;
 import org.springframework.cloud.stream.endpoint.BindingsEndpoint;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -40,7 +41,7 @@ import org.springframework.context.annotation.Configuration;
 public class BindingsEndpointAutoConfiguration {
 
 	@Bean
-	public BindingsEndpoint bindingsEndpoint(List<InputBindingLifecycle> inputBindings) {
-		return new BindingsEndpoint(inputBindings);
+	public BindingsEndpoint bindingsEndpoint(List<InputBindingLifecycle> inputBindings, List<OutputBindingLifecycle> outputBindings) {
+		return new BindingsEndpoint(inputBindings, outputBindings);
 	}
 }

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

@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.endpoint;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.stream.Stream;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -29,6 +30,7 @@ import org.springframework.boot.actuate.endpoint.annotation.Selector;
 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.cloud.stream.binding.OutputBindingLifecycle;
 import org.springframework.util.Assert;
 
 /**
@@ -45,11 +47,14 @@ public class BindingsEndpoint {
 
 	private final List<InputBindingLifecycle> inputBindingLifecycles;
 
+	private final List<OutputBindingLifecycle> outputBindingsLifecycles;
+
 	private final ObjectMapper objectMapper;
 
-	public BindingsEndpoint(List<InputBindingLifecycle> inputBindingLifecycles) {
+	public BindingsEndpoint(List<InputBindingLifecycle> inputBindingLifecycles, List<OutputBindingLifecycle> outputBindingsLifecycles) {
 		Assert.notEmpty(inputBindingLifecycles, "'inputBindingLifecycles' must not be null or empty");
 		this.inputBindingLifecycles = inputBindingLifecycles;
+		this.outputBindingsLifecycles = outputBindingsLifecycles;
 		this.objectMapper = new ObjectMapper();
 	}
 
@@ -78,7 +83,9 @@ public class BindingsEndpoint {
 
 	@ReadOperation
 	public List<?> queryStates() {
-		return objectMapper.convertValue(gatherInputBindings(), List.class);
+		List<Binding<?>> bindings = new ArrayList<>(gatherInputBindings());
+		bindings.addAll(gatherOutputBindings());
+		return objectMapper.convertValue(bindings, List.class);
 	}
 
 	@ReadOperation
@@ -98,8 +105,20 @@ public class BindingsEndpoint {
 		return inputBindings;
 	}
 
+	@SuppressWarnings("unchecked")
+	private List<Binding<?>> gatherOutputBindings() {
+		List<Binding<?>> outputBindings = new ArrayList<>();
+		for (OutputBindingLifecycle inputBindingLifecycle : this.outputBindingsLifecycles) {
+			Collection<Binding<?>> lifecycleInputBindings = (Collection<Binding<?>>) new DirectFieldAccessor(
+					inputBindingLifecycle).getPropertyValue("outputBindings");
+			outputBindings.addAll(lifecycleInputBindings);
+		}
+		return outputBindings;
+	}
+
 	private Binding<?> locateBinding(String name) {
-		return BindingsEndpoint.this.gatherInputBindings().stream()
+		Stream<Binding<?>> bindings = Stream.concat(this.gatherInputBindings().stream(), this.gatherOutputBindings().stream());
+		return bindings
 			.filter(binding -> name.equals(binding.getName()))
 			.findFirst()
 			.orElse(null);