GH-1054 Added support for display extra binding information

Added support for displaying
 - state
 - pausable
 - name/group
 - destination info
 - extended binding info
 - etc

Resolves #1054

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

@@ -16,6 +16,11 @@
 
 package org.springframework.cloud.stream.binder;
 
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
 import org.apache.commons.logging.Log;
 
 import org.springframework.beans.factory.DisposableBean;
@@ -71,6 +76,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 
 	private final EmbeddedHeadersChannelInterceptor embeddedHeadersChannelInterceptor =
 			new EmbeddedHeadersChannelInterceptor(this.logger);
+	
+	private final ObjectMapper objectMapper = new ObjectMapper();
 
 	/**
 	 * {@link ProvisioningProvider} delegated by the downstream binder implementations.
@@ -155,6 +162,11 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		return new DefaultBinding<MessageChannel>(destination, null, outputChannel,
 				producerMessageHandler instanceof Lifecycle ? (Lifecycle) producerMessageHandler : null) {
 
+			@Override
+			public Map<String, Object> getExtendedInfo() {	
+				return doGetExtendedInfo(destination, producerProperties);
+			}
+			
 			@Override
 			public void afterUnbind() {
 				try {
@@ -255,7 +267,12 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 
 			return new DefaultBinding<MessageChannel>(name, group, inputChannel,
 					consumerEndpoint instanceof Lifecycle ? (Lifecycle) consumerEndpoint : null) {
-
+				
+				@Override
+				public Map<String, Object> getExtendedInfo() {
+					return doGetExtendedInfo(destination, properties);
+				}
+				
 				@Override
 				protected void afterUnbind() {
 					try {
@@ -289,7 +306,6 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 	}
 
-
 	@Override
 	public Binding<PollableSource<MessageHandler>> bindPollableConsumer(String name, String group,
 			final PollableSource<MessageHandler> inboundBindTarget, C properties) {
@@ -322,6 +338,11 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		}
 		return new DefaultBinding<PollableSource<MessageHandler>>(name, group, inboundBindTarget,
 				resources.getSource() instanceof Lifecycle ? (Lifecycle) resources.getSource() : null) {
+			
+			@Override
+			public Map<String, Object> getExtendedInfo() {
+				return doGetExtendedInfo(destination, properties);
+			}
 
 			@Override
 			public void afterUnbind() {
@@ -638,6 +659,13 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	protected String errorsBaseName(ProducerDestination destination) {
 		return destination.getName() + ".errors";
 	}
+	
+	private Map<String, Object> doGetExtendedInfo(Object destination, Object properties) {
+		Map<String, Object> extendedInfo = new LinkedHashMap<>();
+		extendedInfo.put("bindingDestination", destination.toString());
+		extendedInfo.put(properties.getClass().getSimpleName(), objectMapper.convertValue(properties, Map.class));
+		return extendedInfo;
+	}
 
 	private final class SendingHandler extends AbstractMessageHandler implements Lifecycle {
 

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

@@ -16,6 +16,9 @@
 
 package org.springframework.cloud.stream.binder;
 
+import java.util.Collections;
+import java.util.Map;
+
 import org.springframework.integration.endpoint.Pausable;
 
 /**
@@ -34,6 +37,10 @@ import org.springframework.integration.endpoint.Pausable;
  */
 public interface Binding<T> extends Pausable {
 	
+	default Map<String, Object> getExtendedInfo() {
+		return Collections.emptyMap();
+	}
+	
 	/**
 	 * Stops the target component represented by this instance.
 	 * NOTE: At the time the instance is created the component is already started. 
@@ -60,7 +67,9 @@ public interface Binding<T> extends Pausable {
 	 * 
 	 * @see BindingsEndpoint
 	 */
-	default void pause() {}
+	default void pause() {
+		this.stop();
+	}
 	
 	/**
 	 * Resumes the target component represented by this instance if and only if the component 
@@ -70,7 +79,9 @@ public interface Binding<T> extends Pausable {
 	 * 
 	 * @see BindingsEndpoint
 	 */
-	default void resume() {}
+	default void resume() {
+		this.start();
+	}
 	
 	/**
 	 * Returns 'true' if the target component represented by this instance is running.

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

@@ -16,6 +16,9 @@
 
 package org.springframework.cloud.stream.binder;
 
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -36,6 +39,8 @@ import org.springframework.util.StringUtils;
  * 
  * @see org.springframework.cloud.stream.annotation.EnableBinding
  */
+@JsonPropertyOrder({ "name", "group", "pausable", "state"})
+@JsonIgnoreProperties("running")
 public class DefaultBinding<T> implements Binding<T> {
 	
 	private final Log logger = LogFactory.getLog(this.getClass().getName());
@@ -47,7 +52,9 @@ public class DefaultBinding<T> implements Binding<T> {
 	protected final T target;
 
 	protected final Lifecycle lifecycle;
-
+	
+	private boolean paused;
+	
 	/**
 	 * Creates an instance that associates a given name, group and binding target with an
 	 * optional {@link Lifecycle} component, which will be stopped during unbinding.
@@ -55,8 +62,9 @@ public class DefaultBinding<T> implements Binding<T> {
 	 * @param name the name of the binding target
 	 * @param group the group (only for input targets)
 	 * @param target the binding target
-	 * @param lifecycle {@link Lifecycle} that runs while the binding is active and will
-	 * be stopped during unbinding
+	 * @param lifecycle {@link Lifecycle} that runs while the binding is active and will be stopped during unbinding
+	 * @param extBindingInfo additional information related to binding
+	 * 
 	 */
 	public DefaultBinding(String name, String group, T target, Lifecycle lifecycle) {
 		Assert.notNull(target, "target must not be null");
@@ -65,7 +73,7 @@ public class DefaultBinding<T> implements Binding<T> {
 		this.target = target;
 		this.lifecycle = lifecycle;
 	}
-
+	
 	public String getName() {
 		return this.name;
 	}
@@ -74,10 +82,27 @@ public class DefaultBinding<T> implements Binding<T> {
 		return this.group;
 	}
 	
+	public String getState() {
+		String state = "N/A";
+		if (this.lifecycle != null) {
+			if (isPausable()) {
+				state = this.paused ? "paused" : this.getRunningState();
+			}
+			else {
+				state = this.getRunningState();
+			}
+		}
+		return state;
+	}
+	
 	public boolean isRunning() {
 		return this.lifecycle != null && this.lifecycle.isRunning();
 	}
 	
+	public boolean isPausable() {
+		return this.lifecycle instanceof Pausable;
+	}
+	
 	@Override
 	public final synchronized void start() {
 		if (!this.isRunning()) {
@@ -101,6 +126,7 @@ public class DefaultBinding<T> implements Binding<T> {
 	public final synchronized void pause() {
 		if (this.lifecycle instanceof Pausable) {
 			( (Pausable) this.lifecycle).pause();
+			this.paused = true;
 		}
 		else {
 			logger.warn("Attempted to pause a component that does not support Pausable " + this.lifecycle);
@@ -111,6 +137,7 @@ public class DefaultBinding<T> implements Binding<T> {
 	public final synchronized void resume() {
 		if (this.lifecycle instanceof Pausable) {
 			( (Pausable) this.lifecycle).resume();
+			this.paused = false;
 		}
 		else {
 			logger.warn("Attempted to resume a component that does not support Pausable " + this.lifecycle);
@@ -127,13 +154,6 @@ public class DefaultBinding<T> implements Binding<T> {
 		return this.lifecycle;
 	}
 
-	/**
-	 * Listener method that executes after unbinding. Subclasses can implement their own
-	 * behaviour on unbinding by overriding this method.
-	 */
-	protected void afterUnbind() {
-	}
-
 	@Override
 	public String toString() {
 		return " Binding [name=" + this.name + ", target=" + this.target + ", lifecycle="
@@ -142,4 +162,15 @@ public class DefaultBinding<T> implements Binding<T> {
 						: ObjectUtils.nullSafeToString(this.lifecycle))
 				+ "]";
 	}
+	
+	/**
+	 * Listener method that executes after unbinding. Subclasses can implement their own
+	 * behaviour on unbinding by overriding this method.
+	 */
+	protected void afterUnbind() {
+	}
+	
+	private String getRunningState() {
+		return isRunning() ? "running" : "stopped";
+	}
 }